diff --git a/CLAUDE.md b/CLAUDE.md index d853b55c..a47d22b4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -57,7 +57,8 @@ hand-maintained per-file tree drifts out of sync with the code. - **`hive-jobq/`** — persistent job-DAG scheduler, extracted from hive-c0re's in-tree `job_queue` as a domain-agnostic library. One persistent graph for the whole system (not a DAG per job); enqueuing - inserts a self-contained sub-DAG and returns its node ids. Generic over + inserts a self-contained sub-DAG and returns the ids of the nodes the job + asked for, in the order it named them. Generic over the node payload `N` and the resource name `R`; resource deps are named counting semaphores acquired all-or-nothing at node start. `hive-c0re`'s remaining `job_queue/` module is the c0re-specific layer *over* this diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 3ce892d0..c31e7be4 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -181,15 +181,18 @@ fn insert_group( inner: &mut QueueInner, declare: Declare, group_parent: Option, -) -> anyhow::Result> { - let ids = inner +) -> anyhow::Result<()> { + inner .sched - .insert_job(group_parent, declare) + .insert_job(group_parent, |b| { + declare(b); + // c0re names no handles: a DAG is addressed by its container node, + // which `submit` inserts itself, and nothing downstream looks an + // individual step up by id. + Vec::new() + }) .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; - for &id in ids.values() { - inner.node_rt.insert(id, NodeRuntime::default()); - } - Ok(ids.into_values().collect()) + Ok(()) } impl JobQueue { @@ -255,12 +258,11 @@ impl JobQueue { /// gate — the children run once `dep_on` reaches `Finishing`. Because the /// emitting node stays `Finishing` until this appended subtree is terminal and /// the DAG's terminal node deps on the top root, roll-up keeps the DAG from - /// settling early with no explicit wiring. Returns the new node ids; empty if - /// the DAG is gone or `nodes` is empty. - pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) -> Vec { + /// settling early with no explicit wiring. A no-op if the DAG is gone. + pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) { let mut inner = self.lock(); if inner.container(dag_id).is_none() { - return Vec::new(); + return; } // Insert the subgraph as a group rooted under the emitting node: the // subgraph's own root becomes a child of `dep_on`, its steps children of @@ -268,20 +270,16 @@ impl JobQueue { // emitter stays `Finishing` until this appended subtree settles, and the // container node rolls up terminal only once its whole subtree (incl. this // appended work) has settled, so the DAG hook waits for free. - let ids = match insert_group(&mut inner, declare, Some(dep_on)) { - Ok(ids) => ids, - Err(e) => { - tracing::error!( - dag = dag_id, - error = %e, - "job_queue: append_subgraph insert failed" - ); - return Vec::new(); - } - }; + if let Err(e) = insert_group(&mut inner, declare, Some(dep_on)) { + tracing::error!( + dag = dag_id, + error = %e, + "job_queue: append_subgraph insert failed" + ); + return; + } drop(inner); self.notify.notify_one(); - ids } /// Claim every currently-runnable node, acquiring its resources, and mark it diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 718ac1e7..1584c281 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1252,12 +1252,11 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { // BEFORE the emitting node is completed. Completing first would settle the // apply node `Done` with nothing under it, opening the tail's `AfterAny` // gate immediately and letting the deploy "finish" before it had built. - let grown = q.append_subgraph( + q.append_subgraph( id, templates::deploy_rebuild_nodes("agent-a", 11), apply.node_id, ); - assert!(!grown.is_empty(), "subgraph grafted onto the apply node"); q.complete_node(apply.node_id, Ok(())); // The grafted chain runs in rebuild order. `claim_one` asserts exactly one diff --git a/hive-jobq/README.md b/hive-jobq/README.md index 46bcad53..eac5c262 100644 --- a/hive-jobq/README.md +++ b/hive-jobq/README.md @@ -16,8 +16,9 @@ kinds, wires deps, and supplies a runner; the scheduler decides what can start. ## Model One **persistent graph** for the whole system, not a DAG per job. Enqueuing -inserts a self-contained sub-DAG and returns the new node ids; the scheduler -runs a continuous loop, starting every node whose deps are satisfied: +inserts a self-contained sub-DAG and returns the ids of the nodes the job +*asked* for, in the order it named them; the scheduler runs a continuous loop, +starting every node whose deps are satisfied: - **Resource deps** are named counting semaphores over a caller-chosen type `R` — e.g. `build-slot` (capacity N), `agent/` (capacity 1), or any diff --git a/hive-jobq/src/builder.rs b/hive-jobq/src/builder.rs index c95f1673..19d34b4f 100644 --- a/hive-jobq/src/builder.rs +++ b/hive-jobq/src/builder.rs @@ -74,12 +74,40 @@ pub enum BuildError { /// The not-yet-declared parent. parent: NodeGuid, }, + /// A handle named in the closure's return value belongs to a different + /// job. Same cause as a foreign handle on an edge: a [`NodeGuid`] outlives + /// the borrow that tied it to its builder, so one can be carried here. + #[error("handle {node:?} names no node in this job")] + UnknownNode { + /// The handle that resolved to nothing. + node: NodeGuid, + }, /// The graph rejected an otherwise well-formed node — an out-of-group edge, /// an unsatisfiable [`DepWhen`], and so on. #[error(transparent)] Graph(#[from] GraphError), } +/// Look each handle a job asked for up in what the insert actually minted, +/// preserving the order it asked in — the last step of both insertion entry +/// points. +/// +/// # Errors +/// [`BuildError::UnknownNode`] for a handle this job never issued. +pub(crate) fn resolve_wanted( + wanted: &[NodeGuid], + ids: &HashMap, +) -> Result, BuildError> { + wanted + .iter() + .map(|g| { + ids.get(g) + .copied() + .ok_or(BuildError::UnknownNode { node: *g }) + }) + .collect() +} + /// One node as the builder holds it: edges and parent still name *handles*, so /// nothing here depends on ids the graph has not minted yet. #[derive(Debug)] @@ -381,39 +409,41 @@ mod tests { #[test] fn edges_resolve_to_minted_ids() { let mut g = graph(); - let mut named = None; let ids = g .insert_job(None, |b| { let first = b.node("a"); let second = b.node("b").after_ok(first); - named = Some((first.guid(), second.guid())); + vec![first.guid(), second.guid()] }) .expect("insert"); - let (first, second) = named.expect("declared"); + let [first, second] = ids[..] else { + panic!("two ids back, in the order asked for") + }; assert_eq!( - deps_of(&g, ids[&second]), + deps_of(&g, second), vec![Dep::Node { - id: ids[&first], + id: first, when: DepWhen::AFTER_OK }] ); - assert!(deps_of(&g, ids[&first]).is_empty()); + assert!(deps_of(&g, first).is_empty()); } #[test] fn parent_resolves_to_a_minted_id() { let mut g = graph(); - let mut named = None; let ids = g .insert_job(None, |b| { let root = b.node("a"); let child = b.node("b").part_of(root); - named = Some((root.guid(), child.guid())); + vec![root.guid(), child.guid()] }) .expect("insert"); - let (root, child) = named.expect("declared"); - assert_eq!(g.node(ids[&root]).expect("root").parent, None); - assert_eq!(g.node(ids[&child]).expect("child").parent, Some(ids[&root])); + let [root, child] = ids[..] else { + panic!("two ids back") + }; + assert_eq!(g.node(root).expect("root").parent, None); + assert_eq!(g.node(child).expect("child").parent, Some(root)); } /// A handle is `Copy`, so naming the same node as a dependency twice must @@ -421,27 +451,28 @@ mod tests { #[test] fn one_handle_can_be_depended_on_twice() { let mut g = graph(); - let mut named = None; let ids = g .insert_job(None, |b| { let shared = b.node("a"); let ok = b.node("b").after_ok(shared); let any = b.node("c").after_any(shared); - named = Some((shared.guid(), ok.guid(), any.guid())); + vec![shared.guid(), ok.guid(), any.guid()] }) .expect("insert"); - let (shared, ok, any) = named.expect("declared"); + let [shared, ok, any] = ids[..] else { + panic!("three ids back") + }; assert_eq!( - deps_of(&g, ids[&ok]), + deps_of(&g, ok), vec![Dep::Node { - id: ids[&shared], + id: shared, when: DepWhen::AFTER_OK }] ); assert_eq!( - deps_of(&g, ids[&any]), + deps_of(&g, any), vec![Dep::Node { - id: ids[&shared], + id: shared, when: DepWhen::AFTER_ANY }] ); @@ -451,20 +482,21 @@ mod tests { #[test] fn resources_become_resource_deps() { let mut g = graph(); - let mut named = None; let ids = g .insert_job(None, |b| { - named = Some( + vec![ b.node("a") .needs("agent/atlas") .needs_units("build", 2) .guid(), - ); + ] }) .expect("insert"); - let only = named.expect("declared"); + let [only] = ids[..] else { + panic!("one id back") + }; assert_eq!( - deps_of(&g, ids[&only]), + deps_of(&g, only), vec![ Dep::Resource { name: "agent/atlas", @@ -492,6 +524,9 @@ mod tests { let second = b.node("b"); let _ = first.after_any(second); named = Some((first.guid(), second.guid())); + // The insert fails, so nothing comes back to ask for — the + // handles under test travel out through `named` instead. + Vec::new() }) .expect_err("forward edge"); let (first, second) = named.expect("declared"); @@ -514,6 +549,7 @@ mod tests { let parent = b.node("b"); let _ = child.part_of(parent); named = Some((child.guid(), parent.guid())); + Vec::new() }) .expect_err("forward parent"); let (child, parent) = named.expect("declared"); @@ -539,6 +575,7 @@ mod tests { let mut foreign = None; g.insert_job(None, |b| { foreign = Some(b.node("first job").guid()); + Vec::new() }) .expect("first job inserts"); let foreign = foreign.expect("declared"); @@ -546,6 +583,7 @@ mod tests { let err = g .insert_job(None, |b| { let _ = b.node("second job").after_ok(foreign); + Vec::new() }) .expect_err("foreign handle"); assert!( @@ -565,6 +603,7 @@ mod tests { // A child may not depend on its own parent: the parent gate // already orders them, and the edge would deadlock. let _ = b.node("child").part_of(root).after_ok(root); + Vec::new() }) .expect_err("out-of-group dep"); assert!(matches!(err, BuildError::Graph(_)), "{err:?}"); @@ -578,23 +617,47 @@ mod tests { let mut g = graph(); let container = g.insert("container", Vec::new(), None).expect("container"); - let mut named = None; let ids = g .insert_job(Some(container), |b| { let root = b.node("root"); let child = b.node("child").part_of(root); - named = Some((root.guid(), child.guid())); + vec![root.guid(), child.guid()] }) .expect("insert"); - let (root, child) = named.expect("declared"); - assert_eq!(g.node(ids[&root]).expect("root").parent, Some(container)); - assert_eq!(g.node(ids[&child]).expect("child").parent, Some(ids[&root])); + let [root, child] = ids[..] else { + panic!("two ids back") + }; + assert_eq!(g.node(root).expect("root").parent, Some(container)); + assert_eq!(g.node(child).expect("child").parent, Some(root)); + } + + /// A handle that names no node in *this* job is refused rather than + /// silently dropped from the returned ids — the return is positional, so a + /// short vector would misalign every id after it. + #[test] + fn asking_for_a_foreign_handle_is_an_error() { + let mut g = graph(); + let mut foreign = None; + g.insert_job(None, |b| { + foreign = Some(b.node("first job").guid()); + Vec::new() + }) + .expect("first job inserts"); + let foreign = foreign.expect("declared"); + + let err = g + .insert_job(None, |b| { + let _ = b.node("second job"); + vec![foreign] + }) + .expect_err("foreign handle asked for"); + assert_eq!(err, BuildError::UnknownNode { node: foreign }); } #[test] fn an_empty_builder_inserts_nothing() { let mut g = graph(); - let ids = g.insert_job(None, |_| {}).expect("insert"); + let ids = g.insert_job(None, |_| Vec::new()).expect("insert"); assert!(ids.is_empty()); assert_eq!(g.nodes().count(), 0); } diff --git a/hive-jobq/src/lib.rs b/hive-jobq/src/lib.rs index 6f27bfb0..2fcf0d84 100644 --- a/hive-jobq/src/lib.rs +++ b/hive-jobq/src/lib.rs @@ -458,26 +458,33 @@ impl Graph { Ok(id) } - /// Insert a whole job under `root_parent`, returning the id each handle's - /// node was minted as. + /// Insert a whole job under `root_parent`, returning the ids of the nodes + /// `declare` **asked for**, in the order it named them. /// - /// `declare` receives a fresh [`JobBuilder`] and names the job's nodes on - /// it; the builder never leaves this call, so a job cannot be built in one - /// place and inserted in another. The graph-only counterpart of + /// `declare` receives a fresh [`JobBuilder`], names the job's nodes on it, + /// and returns the handles whose ids it wants back. The builder never + /// leaves this call, so a job cannot be built in one place and inserted in + /// another. The graph-only counterpart of /// [`crate::scheduler::Scheduler::insert_job`] — prefer that one when a /// scheduler owns the graph, since it also records what it started. /// + /// Asking is how a caller addresses a node it created: the alternative — a + /// map of everything inserted, or a positional vector — either hands back a + /// lookup nobody performs or reintroduces the counting this API exists to + /// remove. + /// /// # Errors /// Propagates [`BuildError`] — a forward reference in the job's own /// declarations, a handle from a different job, or a graph rejection. pub fn insert_job( &mut self, root_parent: Option, - declare: impl FnOnce(&JobBuilder), - ) -> Result, BuildError> { + declare: impl FnOnce(&JobBuilder) -> Vec, + ) -> Result, BuildError> { let job = JobBuilder::new(); - declare(&job); - job.insert_into(self, root_parent) + let wanted = declare(&job); + let ids = job.insert_into(self, root_parent)?; + crate::builder::resolve_wanted(&wanted, &ids) } /// Borrow a node by id. diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index dd92d07a..d326f382 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -103,11 +103,12 @@ impl Scheduler { /// Insert a whole job under `root_parent`, returning the id each handle's /// node was minted as. /// - /// `declare` receives a fresh [`JobBuilder`] and names the job's nodes on - /// it; the builder never leaves this call. That is the whole insertion - /// API — a caller cannot construct a builder, hold one, or insert one - /// itself, so there is no way to end up with a job-shaped value being - /// passed around as a spec. + /// `declare` receives a fresh [`JobBuilder`], names the job's nodes on it, + /// and returns the handles whose ids it wants back — they come back in + /// that order. The builder never leaves this call. That is the whole + /// insertion API — a caller cannot construct a builder, hold one, or + /// insert one itself, so there is no way to end up with a job-shaped value + /// being passed around as a spec. /// /// The scheduler-side counterpart of [`Graph::insert_job`]: same /// resolution, but each node goes through [`Scheduler::append`], so a @@ -121,13 +122,14 @@ impl Scheduler { pub fn insert_job( &mut self, root_parent: Option, - declare: impl FnOnce(&JobBuilder), - ) -> Result, BuildError> { + declare: impl FnOnce(&JobBuilder) -> Vec, + ) -> Result, BuildError> { let job = JobBuilder::new(); - declare(&job); - job.insert_with(root_parent, |payload, deps, parent| { + let wanted = declare(&job); + let ids = job.insert_with(root_parent, |payload, deps, parent| { self.append(payload, deps, parent) - }) + })?; + crate::builder::resolve_wanted(&wanted, &ids) } /// Claim every currently-runnable pending node and start it: node-deps