From 9c97365f8f9d84911c007a6bf94e0534a596de7c Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 13:40:03 +0200 Subject: [PATCH] refactor(job-queue): a job is a recipe, not a value you carry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follows the jobq change: a builder can no longer be constructed or inserted outside `hive_jobq`, so `DagSpec` cannot hold one. It carries a `Declare` — `Box` — and the queue runs it against a builder jobq owns, at the moment it inserts. `NodeOutput.append_subgraph` becomes `Vec` for the same reason, and this is where the shape was always heading: that field's doc already said an executor "cannot reach the queue, so it hands the declaration back", while its type was a `Vec` the executor had built itself. The rejected `build_nodes -> Vec` was the first version of that escape hatch; a recipe is the last one, because there is no job-shaped value to hand over at all. Templates and the power-op assemblers move their owned data into the closure and are otherwise unchanged — `rebuild_nodes`, `node` and the tail helpers already took `&Job` and returned handles, so only each template's outermost frame moved. Two `Debug` impls are hand-written: a closure has nothing to show, and its nodes do not exist until the queue runs it. `NodeOutput` reports how many subgraphs were emitted, `DagSpec` its source and reason. `append_subgraph`'s `is_empty()` early-return is gone — you cannot ask a recipe whether it will declare anything without running it. It now inserts and returns an empty id list if nothing was declared, which takes the queue lock in a case that previously skipped it. The two in-DAG-growth tests build `Declare`s now, so they exercise the shape an executor actually produces rather than one only a test could construct. 45 job-queue tests unchanged and passing. --- hive-c0re/src/job_queue/exec.rs | 70 ++++--- hive-c0re/src/job_queue/mod.rs | 27 ++- hive-c0re/src/job_queue/model.rs | 26 ++- hive-c0re/src/job_queue/submit.rs | 51 +++-- hive-c0re/src/job_queue/templates.rs | 272 +++++++++++++-------------- hive-c0re/src/job_queue/tests.rs | 112 +++++------ hive-c0re/src/workers/auto_update.rs | 49 ++--- hive-jobq/src/builder.rs | 13 +- 8 files changed, 336 insertions(+), 284 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 78451968..d6f44a37 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -10,7 +10,7 @@ use std::sync::Arc; use anyhow::{Context as _, Result}; -use super::{Claim, Job}; +use super::{Claim, Declare}; use hive_jobq::TerminalState; use super::model::NodeKind; @@ -27,7 +27,7 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// Extra signal an executor hands back to the scheduler alongside /// success. -#[derive(Debug, Default)] +#[derive(Default)] pub struct NodeOutput { /// Whole per-agent *subgraphs* to append into *this same* DAG at /// runtime — the single in-DAG-growth channel. Each [`Job`] is one @@ -42,7 +42,17 @@ pub struct NodeOutput { /// *before* the emitting node's completion so the DAG never rolls terminal /// with the appended work still pending — keeping the lease-window /// transient held across the sub-step. - pub append_subgraph: Vec, + pub append_subgraph: Vec, +} + +impl std::fmt::Debug for NodeOutput { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // The subgraphs are closures — how many were emitted is the only thing + // there is to say about them before the queue runs them. + f.debug_struct("NodeOutput") + .field("append_subgraph", &self.append_subgraph.len()) + .finish() + } } /// Build-log sink for one claimed node. @@ -342,17 +352,18 @@ async fn run_meta_lock( .unwrap_or_default() .iter() .map(|agent| { - let job = Job::new(); - super::templates::rebuild_nodes( - &job, - agent, - super::templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); - job + let agent = agent.clone(); + Box::new(move |b: &super::Job| { + super::templates::rebuild_nodes( + b, + &agent, + super::templates::RebuildOpts { + relock: true, + graceful: true, + }, + None, + ); + }) as Declare }) .collect(); return Ok(NodeOutput { append_subgraph }); @@ -374,17 +385,18 @@ async fn run_meta_lock( let append_subgraph = cascade .iter() .map(|agent| { - let job = Job::new(); - super::templates::rebuild_nodes( - &job, - agent, - super::templates::RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - job + let agent = agent.clone(); + Box::new(move |b: &super::Job| { + super::templates::rebuild_nodes( + b, + &agent, + super::templates::RebuildOpts { + relock: false, + graceful: false, + }, + None, + ); + }) as Declare }) .collect(); Ok(NodeOutput { append_subgraph }) @@ -404,10 +416,10 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result sub(NodeKind::Start { diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index ba06fa86..3ce892d0 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -51,10 +51,20 @@ pub use model::{DagSpec, DagView, NodeKind, PermPayload, Source, State}; use resource::Resource; /// A job under construction: `hive_jobq`'s builder over this queue's payload -/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into -/// one of these; [`JobQueue::submit`] inserts it. +/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into a +/// borrowed one; only `hive_jobq` can make or insert it. pub type Job = hive_jobq::JobBuilder; +/// A job's shape as a **recipe**: given a builder, declare the nodes. +/// +/// What a template returns and what an executor hands back, because neither +/// can build a job itself — `hive_jobq` creates the builder inside its own +/// insertion call and never lets one out. So the transferable thing is the +/// declaring closure, and the queue runs it at the moment it inserts. +/// +/// `Send` because an executor's output crosses the scheduler's task boundary. +pub type Declare = Box; + /// A handle to one node a template declared — where its edges, grouping and /// resources are declared. `Copy`; naming a node as a dependency does not /// consume the ability to name it again. @@ -169,12 +179,12 @@ impl Default for JobQueue { /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). fn insert_group( inner: &mut QueueInner, - job: Job, + declare: Declare, group_parent: Option, ) -> anyhow::Result> { let ids = inner .sched - .insert_job(job, group_parent) + .insert_job(group_parent, declare) .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; for &id in ids.values() { inner.node_rt.insert(id, NodeRuntime::default()); @@ -226,7 +236,7 @@ impl JobQueue { ) .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; inner.node_rt.insert(container, NodeRuntime::default()); - insert_group(&mut inner, spec.job, Some(container))?; + insert_group(&mut inner, spec.declare, Some(container))?; // Settle the container's own (no-op) logic immediately so it parks in // `Finishing` and its children become runnable — it never needs claiming // or executing, and stays out of `claim_ready`. It rolls up terminal when @@ -247,10 +257,7 @@ impl JobQueue { /// 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, job: Job, dep_on: NodeId) -> Vec { - if job.is_empty() { - return Vec::new(); - } + pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) -> Vec { let mut inner = self.lock(); if inner.container(dag_id).is_none() { return Vec::new(); @@ -261,7 +268,7 @@ 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, job, Some(dep_on)) { + let ids = match insert_group(&mut inner, declare, Some(dep_on)) { Ok(ids) => ids, Err(e) => { tracing::error!( diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 79a40d14..2463097a 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -470,15 +470,27 @@ impl NodeKind { /// Type-specific payloads (`PermChange`'s file payload) ride the node that /// consumes them ([`NodeKind::WritePermFile`]), not this generic spec. /// -/// There is no separate per-node spec type: the nodes live in the builder, -/// which inserts them itself. A shape that has been declared is therefore -/// always insertable — a dangling edge or a cycle cannot be expressed, so -/// there is nothing left for a submit-time validation pass to reject. -#[derive(Debug)] +/// There is no separate per-node spec type, and no built job either: `declare` +/// is a *recipe* the queue runs against a builder `hive_jobq` owns, at the +/// moment it inserts. A shape that has been declared is therefore always +/// insertable — a dangling edge or a cycle cannot be expressed, so there is +/// nothing left for a submit-time validation pass to reject. pub struct DagSpec { pub source: Source, /// Free-form "why". pub reason: String, - /// The DAG's declared nodes, with their edges, grouping and resources. - pub job: super::Job, + /// Declares the DAG's nodes — their edges, grouping and resources — onto + /// the builder the queue hands it. + pub declare: super::Declare, +} + +impl std::fmt::Debug for DagSpec { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // The recipe is a closure; there is nothing to show of it, and its + // nodes do not exist until the queue runs it. + f.debug_struct("DagSpec") + .field("source", &self.source) + .field("reason", &self.reason) + .finish_non_exhaustive() + } } diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index cddf2db9..edc3444f 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -26,7 +26,7 @@ use std::sync::Arc; use super::model::{DagSpec, NodeKind}; use super::templates::{RebuildOpts, node, rebuild_nodes}; -use super::{Job, Source, templates}; +use super::{Declare, Job, Source, templates}; use crate::coordinator::Coordinator; use crate::lifecycle; @@ -169,11 +169,11 @@ fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) { /// and each keeps its own root, so the per-agent subgraphs are independent and /// run concurrently, each on its own lease. Rebasing one subgraph's indices /// onto another's used to be a function. -fn power_dag(source: Source, reason: String, job: Job) -> DagSpec { +fn power_dag(source: Source, reason: String, declare: Declare) -> DagSpec { DagSpec { source, reason, - job, + declare, } } @@ -189,11 +189,16 @@ pub(crate) fn stop_spec( source: Source, reason: String, ) -> DagSpec { - let job = Job::new(); - for (agent, running) in targets { - stop_chain(&job, agent, graceful, *running); - } - power_dag(source, reason, job) + let targets = targets.to_vec(); + power_dag( + source, + reason, + Box::new(move |b| { + for (agent, running) in targets { + stop_chain(b, &agent, graceful, running); + } + }), + ) } /// Assemble the start DAG from explicit `(agent, running, stale)` targets. @@ -207,11 +212,16 @@ pub(crate) fn start_spec( source: Source, reason: String, ) -> DagSpec { - let job = Job::new(); - for (agent, running, stale) in targets { - start_chain(&job, agent, *running, *stale); - } - power_dag(source, reason, job) + let targets = targets.to_vec(); + power_dag( + source, + reason, + Box::new(move |b| { + for (agent, running, stale) in targets { + start_chain(b, &agent, running, stale); + } + }), + ) } /// Assemble the restart DAG from explicit `(agent, running)` targets. @@ -221,11 +231,16 @@ pub(crate) fn restart_spec( source: Source, reason: String, ) -> DagSpec { - let job = Job::new(); - for (agent, running) in targets { - restart_chain(&job, agent, graceful, *running); - } - power_dag(source, reason, job) + let targets = targets.to_vec(); + power_dag( + source, + reason, + Box::new(move |b| { + for (agent, running) in targets { + restart_chain(b, &agent, graceful, running); + } + }), + ) } /// Restart a single agent. Thin wrapper over [`restart_many`]. diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 5ea6c886..71aedc05 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -24,7 +24,7 @@ use hive_jobq::TerminalState; use super::model::{DagSpec, NodeKind, PermPayload, Source}; -use super::{Handle, Job}; +use super::{Declare, Handle, Job}; /// Declare one node carrying `kind`, with the resources that kind needs. /// @@ -239,27 +239,28 @@ pub(crate) fn rebuild_nodes<'a>( /// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's /// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor /// already holding it rather than deadlocking against it. -pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Job { - let b = Job::new(); - let roots = rebuild_nodes( - &b, - agent, - RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - let _finalize = node( - &b, - NodeKind::FinalizeDeploy { - agent: agent.to_owned(), - approval_id, - }, - ) - .after_ok(roots.prebuild) - .after_ok(roots.reconcile); - b +pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { + let agent = agent.to_owned(); + Box::new(move |b| { + let roots = rebuild_nodes( + b, + &agent, + RebuildOpts { + relock: false, + graceful: false, + }, + None, + ); + let _finalize = node( + b, + NodeKind::FinalizeDeploy { + agent: agent.clone(), + approval_id, + }, + ) + .after_ok(roots.prebuild) + .after_ok(roots.reconcile); + }) } /// One uniform rebuild shape — no `was_running` branch. `StopForUpdate` @@ -274,21 +275,22 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Job { /// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// it reaches `Done` even after a failed swap and the tail would report success. pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { - let job = Job::new(); - let roots = rebuild_nodes( - &job, - agent, - RebuildOpts { - relock, - graceful: false, - }, - None, - ); - emit_rebuilt_tails(&job, agent, &roots.all()); + let agent = agent.to_owned(); DagSpec { source, reason, - job, + declare: Box::new(move |b| { + let roots = rebuild_nodes( + b, + &agent, + RebuildOpts { + relock, + graceful: false, + }, + None, + ); + emit_rebuilt_tails(b, &agent, &roots.all()); + }), } } @@ -318,48 +320,48 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag /// The window still spans the container build, as it must: `prepare_deploy` /// leaves `flake.lock` staged-uncommitted for the build's whole duration. pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec { - let a = || agent.to_owned(); - let job = Job::new(); - - let window = node( - &job, - NodeKind::DeployWindow { - agent: a(), - approval_id, - }, - ); - let verify = node( - &job, - NodeKind::MergeVerify { - agent: a(), - approval_id, - }, - ) - .part_of(window); - let apply = node( - &job, - NodeKind::DeployApply { - agent: a(), - approval_id, - }, - ) - .part_of(window) - .after_ok(verify); - let _tail = node( - &job, - NodeKind::DeployTail { - agent: a(), - approval_id, - }, - ) - .part_of(window) - .after_any(apply); - - resolve_approval_tails(&job, approval_id, window); + let agent = agent.to_owned(); DagSpec { source: Source::Approval, reason, - job, + declare: Box::new(move |b| { + let a = || agent.clone(); + let window = node( + b, + NodeKind::DeployWindow { + agent: a(), + approval_id, + }, + ); + let verify = node( + b, + NodeKind::MergeVerify { + agent: a(), + approval_id, + }, + ) + .part_of(window); + let apply = node( + b, + NodeKind::DeployApply { + agent: a(), + approval_id, + }, + ) + .part_of(window) + .after_ok(verify); + let _tail = node( + b, + NodeKind::DeployTail { + agent: a(), + approval_id, + }, + ) + .part_of(window) + .after_any(apply); + + resolve_approval_tails(b, approval_id, window); + }), } } @@ -370,17 +372,13 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec /// in the queue tests); production paths no longer emit a bare reconcile. #[cfg(test)] pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { - let job = Job::new(); - let _reconcile = node( - &job, - NodeKind::Reconcile { - agent: agent.to_owned(), - }, - ); + let agent = agent.to_owned(); DagSpec { source, reason, - job, + declare: Box::new(move |b| { + let _reconcile = node(b, NodeKind::Reconcile { agent }); + }), } } @@ -396,21 +394,21 @@ pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up /// already carries the whole cascade. pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { - let a = || agent.to_owned(); - let job = Job::new(); - - let provision = node(&job, NodeKind::Provision { agent: a() }); - let create = node(&job, NodeKind::Create { agent: a() }).part_of(provision); - let dropin = node(&job, NodeKind::WriteDropin { agent: a() }).part_of(create); - let _reconcile = node(&job, NodeKind::Reconcile { agent: a() }) - .part_of(create) - .after_ok(dropin); - - resolve_approval_tails(&job, approval_id, provision); + let agent = agent.to_owned(); DagSpec { source: Source::Approval, reason, - job, + declare: Box::new(move |b| { + let a = || agent.clone(); + let provision = node(b, NodeKind::Provision { agent: a() }); + let create = node(b, NodeKind::Create { agent: a() }).part_of(provision); + let dropin = node(b, NodeKind::WriteDropin { agent: a() }).part_of(create); + let _reconcile = node(b, NodeKind::Reconcile { agent: a() }) + .part_of(create) + .after_ok(dropin); + + resolve_approval_tails(b, approval_id, provision); + }), } } @@ -420,32 +418,33 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec { /// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail /// edges all four. pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { - let job = Job::new(); - let write = node( - &job, - NodeKind::WritePermFile { - agent: agent.to_owned(), - payload, - }, - ); - let roots = rebuild_nodes( - &job, - agent, - RebuildOpts { - relock: true, - graceful: false, - }, - Some(write), - ); - emit_rebuilt_tails( - &job, - agent, - &[write, roots.meta_sync, roots.prebuild, roots.reconcile], - ); + let agent = agent.to_owned(); DagSpec { source, reason, - job, + declare: Box::new(move |b| { + let write = node( + b, + NodeKind::WritePermFile { + agent: agent.clone(), + payload, + }, + ); + let roots = rebuild_nodes( + b, + &agent, + RebuildOpts { + relock: true, + graceful: false, + }, + Some(write), + ); + emit_rebuilt_tails( + b, + &agent, + &[write, roots.meta_sync, roots.prebuild, roots.reconcile], + ); + }), } } @@ -465,26 +464,27 @@ pub fn meta_update( reason: String, approval_id: Option, ) -> DagSpec { - let job = Job::new(); - let lock = node( - &job, - NodeKind::MetaLock { - sweep: false, - fanout: None, - inputs, - }, - ); - // The bump itself has no side effect, so an operator-driven one ends at the - // `MetaLock`; an approval-driven one still has its row to resolve and gets the - // per-outcome tails edged onto that single group-root — whose roll-up covers - // the rebuild subgraphs `MetaLock` grows into itself. - if let Some(approval_id) = approval_id { - resolve_approval_tails(&job, approval_id, lock); - } DagSpec { source, reason, - job, + declare: Box::new(move |b| { + let lock = node( + b, + NodeKind::MetaLock { + sweep: false, + fanout: None, + inputs, + }, + ); + // The bump itself has no side effect, so an operator-driven one ends + // at the `MetaLock`; an approval-driven one still has its row to + // resolve and gets the per-outcome tails edged onto that single + // group-root — whose roll-up covers the rebuild subgraphs `MetaLock` + // grows into itself. + if let Some(approval_id) = approval_id { + resolve_approval_tails(b, approval_id, lock); + } + }), } } @@ -502,12 +502,12 @@ pub fn reparent( source: Source, reason: String, ) -> DagSpec { - let job = Job::new(); - let _reparent = node(&job, NodeKind::Reparent { moves }); DagSpec { source, reason, - job, + declare: Box::new(move |b| { + let _reparent = node(b, NodeKind::Reparent { moves }); + }), } } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index a4e3a66b..718ac1e7 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -192,22 +192,22 @@ fn rebuild_chain_claims_in_dep_order() { #[test] fn graceful_rebuild_chain_drains_before_stopping() { let q = JobQueue::new(1); - let job = Job::new(); - templates::rebuild_nodes( - &job, - "agent-a", - templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); let id = submit( &q, DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - job, + declare: Box::new(|b| { + templates::rebuild_nodes( + b, + "agent-a", + templates::RebuildOpts { + relock: true, + graceful: true, + }, + None, + ); + }), }, ); for expected in [ @@ -241,22 +241,22 @@ fn non_graceful_rebuild_has_no_signal_or_drain() { // job keeps its nodes to itself and inserts them, so what it built is // observable where it matters — in what the scheduler runs. let q = JobQueue::new(1); - let job = Job::new(); - templates::rebuild_nodes( - &job, - "agent-a", - templates::RebuildOpts { - relock: true, - graceful: false, - }, - None, - ); let id = submit( &q, DagSpec { source: Source::Manual, reason: "manual".to_owned(), - job, + declare: Box::new(|b| { + templates::rebuild_nodes( + b, + "agent-a", + templates::RebuildOpts { + relock: true, + graceful: false, + }, + None, + ); + }), }, ); let mut kinds = Vec::new(); @@ -692,19 +692,19 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // subgraph per stale agent into its OWN DAG. Each subgraph is rooted on // the emitter and its LOCAL 0-based deps are rebased onto the DAG. let q = JobQueue::new(4); - let job = Job::new(); - let _lock = templates::node( - &job, - NodeKind::MetaLock { - sweep: true, - fanout: None, - inputs: Vec::new(), - }, - ); let spec = DagSpec { source: Source::AutoUpdate, reason: "sweep".to_owned(), - job, + declare: Box::new(|b| { + let _lock = templates::node( + b, + NodeKind::MetaLock { + sweep: true, + fanout: None, + inputs: Vec::new(), + }, + ); + }), }; let id = submit(&q, spec); let emitter = claim_one(&q); @@ -713,18 +713,19 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain → // StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must // match the sweep arm of `run_meta_lock` or this stops tracking production. - let subgraph = |agent: &str| { - let job = Job::new(); - templates::rebuild_nodes( - &job, - agent, - templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); - job + let subgraph = |agent: &str| -> Declare { + let agent = agent.to_owned(); + Box::new(move |b| { + templates::rebuild_nodes( + b, + &agent, + templates::RebuildOpts { + relock: true, + graceful: true, + }, + None, + ); + }) }; // Must append BEFORE completing the emitter (the documented contract). q.append_subgraph(id, subgraph("a"), emitter.node_id); @@ -820,17 +821,18 @@ fn meta_update_grows_cascade_in_dag() { // Simulate the executor growing the cascade in-DAG (`relock = false` — a // cascade child must not re-lock and revert the parent's bump). for agent in ["alice", "bob"] { - let job = Job::new(); - templates::rebuild_nodes( - &job, - agent, - templates::RebuildOpts { - relock: false, - graceful: false, - }, - None, - ); - q.append_subgraph(id, job, meta_lock.node_id); + let declare: Declare = Box::new(move |b| { + templates::rebuild_nodes( + b, + agent, + templates::RebuildOpts { + relock: false, + graceful: false, + }, + None, + ); + }); + q.append_subgraph(id, declare, meta_lock.node_id); } q.complete_node(meta_lock.node_id, Ok(())); // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 0cb146d8..7753c7f5 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -334,7 +334,7 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, Job, NodeKind, Source, templates}; + use crate::job_queue::{DagSpec, NodeKind, Source, templates}; // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. if !any_stale && drifted.is_empty() { @@ -348,28 +348,29 @@ fn submit_boot_tree( n_skipped, ); - let job = Job::new(); - // Sweep whenever ANY marker is stale — even when every stale agent is - // wanted-offline: the hyperhive lock bump must land now so their later - // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock - // ⇒ no meta commit on a no-change boot. The `fanout` list rides the - // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. - if any_stale { - let _ = templates::node( - &job, - NodeKind::MetaLock { - sweep: true, - fanout: Some(fanout), - // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), - // so it names no inputs. - inputs: Vec::new(), - }, - ); - } - // One boot Reconcile per drifted agent — independent roots. - for name in drifted { - let _ = templates::node(&job, NodeKind::Reconcile { agent: name }); - } + let declare: crate::job_queue::Declare = Box::new(move |b| { + // Sweep whenever ANY marker is stale — even when every stale agent is + // wanted-offline: the hyperhive lock bump must land now so their later + // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock + // ⇒ no meta commit on a no-change boot. The `fanout` list rides the + // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. + if any_stale { + let _ = templates::node( + b, + NodeKind::MetaLock { + sweep: true, + fanout: Some(fanout), + // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), + // so it names no inputs. + inputs: Vec::new(), + }, + ); + } + // One boot Reconcile per drifted agent — independent roots. + for name in drifted { + let _ = templates::node(b, NodeKind::Reconcile { agent: name }); + } + }); let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they @@ -379,7 +380,7 @@ fn submit_boot_tree( // Rebuilding when the sweep will grow rebuild subgraphs (per-agent // crash-watch suppression during their Swap, applied at claim time); // a reconcile-only boot needs no transient. - job, + declare, }; if let Err(e) = coord.job_queue.submit(spec) { tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); diff --git a/hive-jobq/src/builder.rs b/hive-jobq/src/builder.rs index 191beef2..8ae84cb0 100644 --- a/hive-jobq/src/builder.rs +++ b/hive-jobq/src/builder.rs @@ -5,11 +5,14 @@ //! than computing where that node landed, and there is no positional index to //! get wrong. //! -//! **An insertion API, not a spec factory.** [`JobBuilder::insert_into`] -//! consumes the builder and puts the nodes straight into a [`Graph`], returning -//! the ids the graph minted. Nothing job-shaped comes back out — there is no -//! intermediate node-description type to keep in sync with [`Graph::insert`]'s -//! signature. +//! **An insertion API, not a spec factory.** A builder is only ever handed to a +//! closure by an insertion entry point ([`Graph::insert_job`], +//! [`crate::scheduler::Scheduler::insert_job`]), which inserts the declared +//! nodes and returns the ids the graph minted. It cannot be constructed, held +//! or inserted from outside this crate, and there is no intermediate +//! node-description type to keep in sync with [`Graph::insert`]'s signature — +//! so a job has no representation that can be passed around instead of being +//! inserted. //! //! **Payload-agnostic.** Generic over the same `N` and `R` as [`Graph`]: the //! builder knows nothing about what a node *does*, only how nodes relate.