From 8f5ccb2882dd924288219c84140c71f0eff57c3d Mon Sep 17 00:00:00 2001 From: atlas Date: Thu, 16 Jul 2026 21:49:24 +0200 Subject: [PATCH 1/4] feat(#2500): scaffold hive-jobq crate with the core graph data model First step of extracting the job-DAG queue into a domain-agnostic `hive-jobq` library, per the operator's v2 design: one persistent graph, named-counter resources, recursive node groups, opaque stable node ids, guard-object locks, a slot-filling scheduler. This commit lands only the data model, so the shape can be reviewed before the machinery is built on it: - NodeId: opaque, stable, monotonic; group membership is a parent edge, not encoded in the id (the 1/1/2 hierarchy is a derived UI label). - ResourceName, Dep (Node | Resource{name,count}), State. - Node: caller-defined payload N so the library stays container-agnostic. - Graph: insert (mints stable ids), node lookup, children, recursive group-terminal check. Retains completed groups (no pruning in v1). The resource-acquisition machinery (atomic all-or-nothing acquire), the recursive-lock guards, and the scheduler loop are follow-ups. Tests cover id minting, group terminality, and state terminality; clippy + rustdoc clean. --- Cargo.lock | 8 ++ Cargo.toml | 1 + hive-jobq/Cargo.toml | 11 ++ hive-jobq/src/lib.rs | 281 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 301 insertions(+) create mode 100644 hive-jobq/Cargo.toml create mode 100644 hive-jobq/src/lib.rs diff --git a/Cargo.lock b/Cargo.lock index 0999afc6..470daab4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1662,6 +1662,14 @@ dependencies = [ "serde", ] +[[package]] +name = "hive-jobq" +version = "0.1.0" +dependencies = [ + "serde", + "thiserror 2.0.18", +] + [[package]] name = "hive-matrix-mcp" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index e671ba8d..771ae3c3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ 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 new file mode 100644 index 00000000..a20512f9 --- /dev/null +++ b/hive-jobq/Cargo.toml @@ -0,0 +1,11 @@ +[package] +name = "hive-jobq" +edition.workspace = true +version.workspace = true + +[lints] +workspace = true + +[dependencies] +serde = { workspace = true } +thiserror = { workspace = true } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs new file mode 100644 index 00000000..73c31a75 --- /dev/null +++ b/hive-jobq/src/lib.rs @@ -0,0 +1,281 @@ +//! `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. +#[derive( + Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize, +)] +pub struct NodeId(pub 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, +} + +/// 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)] +pub struct Graph { + nodes: Vec>, + next_id: u64, +} + +// 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`]. + pub fn insert(&mut self, payload: N, deps: Vec, parent: Option) -> NodeId { + let id = self.mint_id(); + self.nodes.push(Node { + id, + parent, + payload, + deps, + state: State::Pending, + }); + 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 every node inside it (recursively) is terminal. + /// An empty group is terminal. + #[must_use] + pub fn group_terminal(&self, id: NodeId) -> bool { + self.children(id) + .all(|child| child.state.is_terminal() && self.group_terminal(child.id)) + } +} + +#[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); + let b = g.insert( + "update", + vec![Dep::Node { + id: a, + when: DepWhen::AfterOk, + }], + Some(a), + ); + 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); + } + + #[test] + fn group_is_terminal_only_when_all_children_terminal() { + let mut g: Graph<&str> = Graph::new(); + let group = g.insert("group", vec![], None); + let child = g.insert("child", vec![], Some(group)); + // Empty-below or pending child → not terminal. + assert!(!g.group_terminal(group)); + // Mark the child done. + let idx = g.nodes.iter().position(|n| n.id == child).unwrap(); + g.nodes[idx].state = 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)); + } +} From 581737583e977f32824e47d1a50f3203b2906097 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 17 Jul 2026 01:19:35 +0200 Subject: [PATCH 2/4] fix(#2500): group_terminal must require the group node itself terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit group_terminal checked only that every child was terminal, never the group node's own state — so an empty group whose node is still Running returned true (empty .all()), making a running node that has yet to append its subgraph look already-finished. Now it requires the group node itself terminal AND every child recursively terminal. Deciding when to settle a group node to terminal once its children are done stays a scheduler concern; this answers the dependents' question — is the whole group, node included, finished. Adds group-node-pending-with-child-done + empty-running-group tests. --- hive-jobq/src/lib.rs | 45 +++++++++++++++++++++++++++++++++++--------- 1 file changed, 36 insertions(+), 9 deletions(-) diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 73c31a75..b63e125c 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -210,12 +210,18 @@ impl Graph { self.nodes.iter().filter(move |n| n.parent == Some(id)) } - /// A group is terminal once every node inside it (recursively) is terminal. - /// An empty group is terminal. + /// 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 { - self.children(id) - .all(|child| child.state.is_terminal() && self.group_terminal(child.id)) + let Some(node) = self.node(id) else { + return false; + }; + node.state.is_terminal() && self.children(id).all(|child| self.group_terminal(child.id)) } } @@ -242,16 +248,37 @@ mod tests { 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_is_terminal_only_when_all_children_terminal() { + 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); let child = g.insert("child", vec![], Some(group)); - // Empty-below or pending child → not terminal. + // Both pending → not terminal. assert!(!g.group_terminal(group)); - // Mark the child done. - let idx = g.nodes.iter().position(|n| n.id == child).unwrap(); - g.nodes[idx].state = State::Done; + // 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); + 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)); } From 570188fa1a40987ae97486b1e07122ab66e040ef Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 17 Jul 2026 01:59:17 +0200 Subject: [PATCH 3/4] refactor(#2500): make NodeId genuinely opaque via a crate-private field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit argus flagged that `NodeId(pub u64)` contradicted the "opaque" doc — a pub inner field lets callers fabricate `NodeId(42)`. Make the field `pub(crate)` so an id can only originate from the graph's monotonic counter or serde deserialization, never a caller. Tests construct ids in-crate (unaffected); the derived Serialize/Deserialize round-trips fine. Doc keeps "opaque" — now accurate — with a line explaining the enforcement. --- hive-jobq/src/lib.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index b63e125c..4420362b 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -36,10 +36,14 @@ /// 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 u64); +pub struct NodeId(pub(crate) u64); /// A named counting semaphore. /// From 11df4a1bf558858b3b503b117a3f1b62b1ed08d8 Mon Sep 17 00:00:00 2001 From: atlas Date: Fri, 17 Jul 2026 12:09:51 +0200 Subject: [PATCH 4/4] feat(#2500): validate NodeId references on insert and deserialize MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per mara's direction — validate ids as they enter the graph so internal iteration can trust every id the graph holds; the generational route for removal comes later. Adds GraphError; insert() now rejects a dangling Dep::Node / parent id (it is fallible); validate() checks all internal id references resolve and that next_id is past the largest existing id; deserialization runs validate() via #[serde(try_from = "GraphData")], so a loaded graph can never carry a dangling reference. 5 new tests; serde_json added as a dev-dependency for the round-trip cases. --- Cargo.lock | 1 + hive-jobq/Cargo.toml | 3 + hive-jobq/src/lib.rs | 224 ++++++++++++++++++++++++++++++++++++++++--- 3 files changed, 214 insertions(+), 14 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 470daab4..abf86db0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1667,6 +1667,7 @@ name = "hive-jobq" version = "0.1.0" dependencies = [ "serde", + "serde_json", "thiserror 2.0.18", ] diff --git a/hive-jobq/Cargo.toml b/hive-jobq/Cargo.toml index a20512f9..71e644a4 100644 --- a/hive-jobq/Cargo.toml +++ b/hive-jobq/Cargo.toml @@ -9,3 +9,6 @@ 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 index 4420362b..6af94cde 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -153,17 +153,65 @@ pub struct Node { 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 { @@ -191,7 +239,33 @@ impl Graph { /// Insert a node with the given payload, deps, and parent group, returning /// its freshly-minted id. The node starts [`State::Pending`]. - pub fn insert(&mut self, payload: N, deps: Vec, parent: Option) -> NodeId { + /// + /// 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, @@ -200,7 +274,7 @@ impl Graph { deps, state: State::Pending, }); - id + Ok(id) } /// Borrow a node by id. @@ -227,6 +301,40 @@ impl Graph { }; 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)] @@ -236,15 +344,17 @@ mod tests { #[test] fn insert_mints_stable_monotonic_ids() { let mut g: Graph<&str> = Graph::new(); - let a = g.insert("sweep", vec![], None); - let b = g.insert( - "update", - vec![Dep::Node { - id: a, - when: DepWhen::AfterOk, - }], - Some(a), - ); + 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. @@ -260,8 +370,8 @@ mod tests { #[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); - let child = g.insert("child", vec![], Some(group)); + 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: @@ -278,7 +388,7 @@ mod tests { // 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); + 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. @@ -309,4 +419,90 @@ mod tests { 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, + } + ); + } }