diff --git a/Cargo.lock b/Cargo.lock index abf86db0..0999afc6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1662,15 +1662,6 @@ dependencies = [ "serde", ] -[[package]] -name = "hive-jobq" -version = "0.1.0" -dependencies = [ - "serde", - "serde_json", - "thiserror 2.0.18", -] - [[package]] name = "hive-matrix-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 771ae3c3..e671ba8d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,6 @@ members = [ "hive-claude", "hive-forge", "hive-host-sock", - "hive-jobq", "hive-matrix-mcp", "hive-metric", "hive-priv", diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml deleted file mode 100644 index 71e644a4..00000000 --- a/hive-jobq/Cargo.toml +++ /dev/null @@ -1,14 +0,0 @@ -[package] -name = "hive-jobq" -edition.workspace = true -version.workspace = true - -[lints] -workspace = true - -[dependencies] -serde = { workspace = true } -thiserror = { workspace = true } - -[dev-dependencies] -serde_json = { workspace = true } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs deleted file mode 100644 index 6af94cde..00000000 --- a/hive-jobq/src/lib.rs +++ /dev/null @@ -1,508 +0,0 @@ -//! `hive-jobq` — a persistent job-DAG scheduler, extracted from hive-c0re's -//! in-tree `job_queue` as a domain-agnostic library. -//! -//! # Model (v2) -//! -//! One **persistent graph** for the whole system, not a DAG per job. Enqueuing -//! inserts a self-contained **node group** and returns its id; the scheduler -//! runs a continuous loop, starting every node whose [`Dep`]s are satisfied: -//! -//! - **Resource** deps are named counting semaphores ([`ResourceName`]): -//! `build-slot` (capacity N), `agent/` (capacity 1), or any name -//! (capacity 1, created on use). A node acquires *all* its resource deps -//! atomically at start (all-or-nothing) — no hold-and-wait, so no deadlock -//! and no cycle detection needed. -//! - **Node** deps wait on a node/group per [`DepWhen`]: `AfterOk` needs -//! success (a failed dep cancels the dependent), `AfterAny` only terminal. -//! -//! A **node group** is a self-contained sub-graph; things depend on it as a -//! whole (done = every inner node terminal), never on an inner node. Groups -//! nest; a running node may grow its own group but not reach outside it. -//! -//! A [`NodeId`] is opaque, stable, and monotonic — persisted, so it survives -//! restarts. Group membership is a parent edge ([`Node::parent`]), *not* in the -//! id; the `1/1/2` hierarchy is a derived UI label. The node payload is generic -//! (`N`) so the library stays container-agnostic — the caller supplies its own -//! node kind. Resources are held by the acquiring node and released on -//! completion via guard objects, recursive within a group. -//! -//! The resource-acquisition machinery, the guards, and the scheduler loop are -//! follow-ups; this is the data model they build on. - -/// Opaque, stable, monotonic node identifier. -/// -/// Assigned by the [`Graph`] on insert and persisted, so it is stable across -/// restarts. Group membership is a separate parent edge ([`Node::parent`]) — it -/// is deliberately *not* encoded in the id, so the id never changes as the tree -/// grows or collapses. The hierarchical `1/1/2` path used in the UI is derived -/// from the parent tree at render time. -/// -/// The inner field is crate-private: an id can only originate from the graph's -/// monotonic counter (or deserialization of a persisted graph), never be -/// fabricated by a caller — that is what makes it opaque. -#[derive( - Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, -)] -pub struct NodeId(pub(crate) u64); - -/// A named counting semaphore. -/// -/// Examples: `build-slot` (capacity configured to the number of build slots), -/// `agent/` (capacity 1 — the per-agent lifecycle lock), or any other -/// name, which is assumed to have capacity 1 and is created on first use. -#[derive( - Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, -)] -pub struct ResourceName(pub String); - -/// When a [`Dep::Node`] edge is satisfied — the strong/weak distinction the -/// current queue carries as `DepWhen`, load-bearing for failure safety. -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub enum DepWhen { - /// The dependency must reach [`State::Done`]. This is the default chain - /// edge: if the dependency *fails*, the dependent must not run and is - /// cancelled ([`State::Cancelled`]) down the chain — e.g. a failed - /// `Prebuild` must not let `StopForUpdate` stop a healthy container. - AfterOk, - /// The dependency need only be terminal — success or failure both satisfy - /// it. For steps that must converge regardless, e.g. `Reconcile` running - /// even when the preceding `Swap` failed. - AfterAny, -} - -impl DepWhen { - /// Whether a dependency in `dep_state` satisfies this edge. - #[must_use] - pub fn satisfied_by(self, dep_state: State) -> bool { - match self { - DepWhen::AfterOk => dep_state == State::Done, - DepWhen::AfterAny => dep_state.is_terminal(), - } - } -} - -/// One dependency of a node. A node becomes runnable once every [`Dep::Node`] -/// edge it names is satisfied (per its [`DepWhen`]) *and* every [`Dep::Resource`] -/// it names can be acquired (all of them, atomically). -#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub enum Dep { - /// Depend on another node (or a group, by its group node's id). Whether a - /// *failed* dependency satisfies the edge is decided by `when`: `AfterOk` - /// requires success (and cancels this node if the dep fails), `AfterAny` - /// only requires the dep to be terminal. - Node { - /// The node (or group) depended on. - id: NodeId, - /// Strong (`AfterOk`) vs weak (`AfterAny`). - when: DepWhen, - }, - /// Hold `count` units of a named resource for the duration of this node's - /// run. Acquired atomically with the node's other resource deps at start, - /// released when the node completes. - Resource { - /// The resource to acquire. - name: ResourceName, - /// How many units to hold (usually 1). - count: u32, - }, -} - -/// A node's lifecycle state. -#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)] -pub enum State { - /// Waiting on dependencies (node or resource). - Pending, - /// Dependencies satisfied, resources held, currently executing. - Running, - /// Completed successfully. - Done, - /// Completed unsuccessfully. - Failed, - /// Never ran: an `AfterOk` dependency failed, so this node (and the rest of - /// its strong-dependent chain) is cancelled rather than run. - Cancelled, -} - -impl State { - /// A node is *terminal* once it has finished — successfully, unsuccessfully, - /// or cancelled — which is when its resources are released and dependents - /// are re-evaluated. - #[must_use] - pub fn is_terminal(self) -> bool { - matches!(self, State::Done | State::Failed | State::Cancelled) - } -} - -/// A node in the graph, carrying a caller-defined payload `N`. -/// -/// The library schedules over `Node`s and resources without interpreting the -/// payload; the caller supplies `N` (its own node kind) and a runner to execute -/// a claimed node. -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct Node { - /// Stable identity, assigned on insert. - pub id: NodeId, - /// The group this node belongs to, if any. `None` for a top-level group - /// node. Group membership lives here, not in the id. - pub parent: Option, - /// Caller-defined payload (the node's kind / work description). - pub payload: N, - /// What must hold before this node runs (other nodes + resources). - pub deps: Vec, - /// Lifecycle state. - pub state: State, -} - -/// An error from inserting into or loading a [`Graph`] with a dangling id. -/// -/// A [`NodeId`] is only meaningful against the graph that minted it, so both -/// entry points — [`Graph::insert`] and deserialization — reject references to -/// nodes the graph does not contain. That is what lets internal iteration trust -/// every id the graph holds. -#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] -pub enum GraphError { - /// A node's dependency named an id not present in the graph. - #[error("dependency references unknown node {0:?}")] - UnknownDep(NodeId), - /// A node's parent named an id not present in the graph. - #[error("parent references unknown node {0:?}")] - UnknownParent(NodeId), - /// A loaded graph's `next_id` counter is not past the largest existing id, - /// so the next minted id would collide with one already in the graph. - #[error("next_id {next_id} must exceed the largest existing node id {max_id}")] - NextIdTooSmall { - /// The persisted counter value. - next_id: u64, - /// The largest id already present. - max_id: u64, - }, -} - -/// The single persistent graph of all nodes. -/// -/// New jobs are inserted as node groups; the scheduler (added in a follow-up) -/// walks this graph filling open slots. Completed groups are retained (no -/// pruning in v1). -#[derive(Debug, serde::Serialize, serde::Deserialize)] -#[serde(try_from = "GraphData")] -pub struct Graph { - nodes: Vec>, - next_id: u64, -} - -// Deserialization target: the raw fields, turned into a `Graph` by the `TryFrom` -// below — which runs [`Graph::validate`], so a loaded graph can never carry a -// dangling id reference (Serialize does not validate; Deserialize always does). -#[derive(serde::Deserialize)] -struct GraphData { - nodes: Vec>, - next_id: u64, -} - -impl TryFrom> for Graph { - type Error = GraphError; - - fn try_from(data: GraphData) -> Result { - let graph = Graph { - nodes: data.nodes, - next_id: data.next_id, - }; - graph.validate()?; - Ok(graph) - } -} - -// A `derive(Default)` would wrongly require `N: Default` (an empty graph holds -// no payload); an empty `Vec>` needs no such bound, so impl it directly. -impl Default for Graph { - fn default() -> Self { - Self::new() - } -} - -impl Graph { - /// An empty graph. - #[must_use] - pub fn new() -> Self { - Self { - nodes: Vec::new(), - next_id: 0, - } - } - - /// Mint the next stable node id. - fn mint_id(&mut self) -> NodeId { - let id = NodeId(self.next_id); - self.next_id += 1; - id - } - - /// Insert a node with the given payload, deps, and parent group, returning - /// its freshly-minted id. The node starts [`State::Pending`]. - /// - /// Every [`Dep::Node`] id and the `parent` id (if any) must already resolve - /// to a node in the graph — an id is only meaningful against the graph that - /// minted it, so a dangling reference is rejected here rather than surfacing - /// as a broken edge later. - /// - /// # Errors - /// Returns [`GraphError::UnknownParent`] / [`GraphError::UnknownDep`] if the - /// parent or a dependency references a node not in the graph. - pub fn insert( - &mut self, - payload: N, - deps: Vec, - parent: Option, - ) -> Result { - if let Some(parent_id) = parent - && self.node(parent_id).is_none() - { - return Err(GraphError::UnknownParent(parent_id)); - } - for dep in &deps { - if let Dep::Node { id, .. } = dep - && self.node(*id).is_none() - { - return Err(GraphError::UnknownDep(*id)); - } - } - let id = self.mint_id(); - self.nodes.push(Node { - id, - parent, - payload, - deps, - state: State::Pending, - }); - Ok(id) - } - - /// Borrow a node by id. - #[must_use] - pub fn node(&self, id: NodeId) -> Option<&Node> { - self.nodes.iter().find(|n| n.id == id) - } - - /// The direct children of a group node (nodes whose `parent` is `id`). - pub fn children(&self, id: NodeId) -> impl Iterator> { - self.nodes.iter().filter(move |n| n.parent == Some(id)) - } - - /// A group is terminal once the group node itself is terminal *and* every - /// node inside it (recursively) is terminal. The node's own state matters: - /// a group node still `Pending`/`Running` is not terminal even with no - /// children yet, since a running node may still append some. (Deciding when - /// to *settle* a group node to terminal once its children are all done is a - /// separate concern the scheduler owns.) An unknown id is not terminal. - #[must_use] - pub fn group_terminal(&self, id: NodeId) -> bool { - let Some(node) = self.node(id) else { - return false; - }; - node.state.is_terminal() && self.children(id).all(|child| self.group_terminal(child.id)) - } - - /// Check that every id the graph holds resolves: each node's `parent` and - /// every [`Dep::Node`] id names a node present in the graph, and `next_id` - /// is past the largest existing id. Deserialization runs this, so a loaded - /// graph is internally consistent and internal iteration can trust its ids. - /// - /// # Errors - /// Returns [`GraphError`] on a dangling parent / dependency reference, or a - /// `next_id` that would remint an id already in the graph. - pub fn validate(&self) -> Result<(), GraphError> { - for node in &self.nodes { - if let Some(parent_id) = node.parent - && self.node(parent_id).is_none() - { - return Err(GraphError::UnknownParent(parent_id)); - } - for dep in &node.deps { - if let Dep::Node { id, .. } = dep - && self.node(*id).is_none() - { - return Err(GraphError::UnknownDep(*id)); - } - } - } - if let Some(max_id) = self.nodes.iter().map(|n| n.id.0).max() - && self.next_id <= max_id - { - return Err(GraphError::NextIdTooSmall { - next_id: self.next_id, - max_id, - }); - } - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn insert_mints_stable_monotonic_ids() { - let mut g: Graph<&str> = Graph::new(); - let a = g.insert("sweep", vec![], None).unwrap(); - let b = g - .insert( - "update", - vec![Dep::Node { - id: a, - when: DepWhen::AfterOk, - }], - Some(a), - ) - .unwrap(); - assert_eq!(a, NodeId(0)); - assert_eq!(b, NodeId(1)); - // Membership is the parent edge, not the id. - assert_eq!(g.node(b).unwrap().parent, Some(a)); - assert_eq!(g.node(a).unwrap().parent, None); - } - - fn set_state(g: &mut Graph, id: NodeId, state: State) { - let idx = g.nodes.iter().position(|n| n.id == id).unwrap(); - g.nodes[idx].state = state; - } - - #[test] - fn group_terminal_requires_the_group_node_and_all_children_terminal() { - let mut g: Graph<&str> = Graph::new(); - let group = g.insert("group", vec![], None).unwrap(); - let child = g.insert("child", vec![], Some(group)).unwrap(); - // Both pending → not terminal. - assert!(!g.group_terminal(group)); - // Child done, but the group node itself is still pending → NOT terminal: - // the group node's own state is load-bearing, not just its children. - set_state(&mut g, child, State::Done); - assert!(!g.group_terminal(group)); - // Group node terminal too → the whole group is terminal. - set_state(&mut g, group, State::Done); - assert!(g.group_terminal(group)); - } - - #[test] - fn empty_running_group_is_not_terminal() { - // A running node with no children yet may still append some, so it must - // not read as terminal just because its child set is currently empty. - let mut g: Graph<&str> = Graph::new(); - let group = g.insert("group", vec![], None).unwrap(); - set_state(&mut g, group, State::Running); - assert!(!g.group_terminal(group)); - // Once it finishes (having grown no children), it is terminal. - set_state(&mut g, group, State::Done); - assert!(g.group_terminal(group)); - } - - #[test] - fn state_terminality() { - assert!(State::Done.is_terminal()); - assert!(State::Failed.is_terminal()); - assert!(State::Cancelled.is_terminal()); - assert!(!State::Pending.is_terminal()); - assert!(!State::Running.is_terminal()); - } - - #[test] - fn after_ok_needs_success_after_any_needs_terminal() { - // AfterOk: only Done satisfies; a Failed/Cancelled dep does NOT (the - // dependent must be cancelled, not run). - assert!(DepWhen::AfterOk.satisfied_by(State::Done)); - assert!(!DepWhen::AfterOk.satisfied_by(State::Failed)); - assert!(!DepWhen::AfterOk.satisfied_by(State::Cancelled)); - assert!(!DepWhen::AfterOk.satisfied_by(State::Running)); - // AfterAny: any terminal state satisfies. - assert!(DepWhen::AfterAny.satisfied_by(State::Done)); - assert!(DepWhen::AfterAny.satisfied_by(State::Failed)); - assert!(DepWhen::AfterAny.satisfied_by(State::Cancelled)); - assert!(!DepWhen::AfterAny.satisfied_by(State::Pending)); - } - - #[test] - fn insert_rejects_unknown_parent() { - let mut g: Graph<&str> = Graph::new(); - let bogus = NodeId(7); - assert_eq!( - g.insert("x", vec![], Some(bogus)).unwrap_err(), - GraphError::UnknownParent(bogus) - ); - } - - #[test] - fn insert_rejects_unknown_dep() { - let mut g: Graph<&str> = Graph::new(); - let bogus = NodeId(42); - let deps = vec![Dep::Node { - id: bogus, - when: DepWhen::AfterOk, - }]; - assert_eq!( - g.insert("x", deps, None).unwrap_err(), - GraphError::UnknownDep(bogus) - ); - } - - #[test] - fn valid_graph_round_trips_through_serde() { - let mut g: Graph = Graph::new(); - let a = g.insert("a".to_owned(), vec![], None).unwrap(); - g.insert( - "b".to_owned(), - vec![Dep::Node { - id: a, - when: DepWhen::AfterAny, - }], - Some(a), - ) - .unwrap(); - let json = serde_json::to_string(&g).unwrap(); - let back: Graph = serde_json::from_str(&json).unwrap(); - assert!(back.validate().is_ok()); - assert_eq!(back.node(a).unwrap().payload, "a"); - } - - #[test] - fn deserialize_rejects_a_dangling_dependency() { - // Build a graph whose only node depends on a non-existent id, serialize - // it (Serialize does not validate), and confirm deserialize rejects it. - let bad = Graph:: { - nodes: vec![Node { - id: NodeId(0), - parent: None, - payload: "x".to_owned(), - deps: vec![Dep::Node { - id: NodeId(99), - when: DepWhen::AfterOk, - }], - state: State::Pending, - }], - next_id: 1, - }; - let json = serde_json::to_string(&bad).unwrap(); - let err = serde_json::from_str::>(&json).unwrap_err(); - assert!(err.to_string().contains("unknown node")); - } - - #[test] - fn validate_rejects_next_id_that_would_remint() { - let bad = Graph::<&str> { - nodes: vec![Node { - id: NodeId(5), - parent: None, - payload: "x", - deps: vec![], - state: State::Pending, - }], - next_id: 3, - }; - assert_eq!( - bad.validate().unwrap_err(), - GraphError::NextIdTooSmall { - next_id: 3, - max_id: 5, - } - ); - } -}