diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index a0348d4e..1fa2b382 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -28,7 +28,7 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// scheduler spawns per claim; the `Result` (stringified) becomes the /// node's terminal state. /// -/// `job` is the node's own growth channel: an executor that decides more work +/// `builder` is the node's own growth channel: an executor that decides more work /// is needed declares it here, and the scheduler inserts it under this node /// when the node completes. Most executors never touch it. Nothing is inserted /// while the node runs — the builder is local state, so this stays outside the @@ -48,15 +48,15 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from /// themselves. Nothing here needs a claim to exist as a type. pub(super) async fn run_node( coord: &Arc, - job: super::JobBuilder, + builder: super::JobBuilder, id: NodeId, kind: &NodeKind, ) -> (super::JobBuilder, Result<()>) { // The agent this node targets rides the payload — empty for the agentless // container kinds (`MetaLock`, `Dag`), which never read it. let agent = kind.agent(); - // Every arm is `Result<()>`; the three that grow work declare into `job` - // *synchronously*, after their own awaits have finished. Borrowing `&job` + // Every arm is `Result<()>`; the three that grow work declare into `builder` + // *synchronously*, after their own awaits have finished. Borrowing `&builder` // inside an `.await` would make this future non-`Send` (see above), so the // growth executors return what to grow rather than taking the builder. let result = match kind { @@ -79,14 +79,14 @@ pub(super) async fn run_node( // than returned, since `run_meta_lock` would only be deriving // it from the `sweep` this call site already holds. if *sweep { - super::templates::grown_graceful_rebuilds(&job, &agents, true); + super::templates::grown_graceful_rebuilds(&builder, &agents, true); } else { - super::templates::grown_rebuilds(&job, &agents, false); + super::templates::grown_rebuilds(&builder, &agents, false); } }), NodeKind::Reconcile { .. } => run_reconcile(coord, agent).await.map(|sub| { if let Some(kind) = sub { - super::templates::fanned_out_mechanical(&job, kind); + super::templates::fanned_out_mechanical(&builder, kind); } }), NodeKind::Start { .. } => run_start(coord, agent).await, @@ -106,7 +106,7 @@ pub(super) async fn run_node( NodeKind::MergeVerify { approval_id, .. } => run_merge_verify(coord, *approval_id).await, NodeKind::DeployApply { approval_id, .. } => { run_deploy_apply(coord, *approval_id).await.map(|()| { - super::templates::deploy_rebuild_nodes(&job, agent, *approval_id); + super::templates::deploy_rebuild_nodes(&builder, agent, *approval_id); }) } NodeKind::FinalizeDeploy { approval_id, .. } => { @@ -132,7 +132,7 @@ pub(super) async fn run_node( // and build slot it declares stay held until its subtree settles. NodeKind::Dag { .. } | NodeKind::DeployWindow { .. } => Ok(()), }; - (job, result) + (builder, result) } /// Resolve the DAG's approval row the way this node's own `outcome` says. @@ -368,7 +368,7 @@ async fn run_meta_lock( /// Idempotent power-converge *planner*: compare `wanted` (durable /// intent) against observed state and, when they diverge, fan the /// mechanical `Start` / `Stop` out as a first-class node appended to -/// *this* DAG (a single node declared into `job`, rooted on this node). +/// *this* DAG (a single node declared into `builder`, rooted on this node). /// Does no container work itself — the sub-step becomes visible in the /// DAG and the lease-window transient (or the sub-step's own node-local /// guard) rides across it. diff --git a/hive-c0re/src/job_queue/scheduler.rs b/hive-c0re/src/job_queue/scheduler.rs index 3dc25de1..93341414 100644 --- a/hive-c0re/src/job_queue/scheduler.rs +++ b/hive-c0re/src/job_queue/scheduler.rs @@ -89,7 +89,7 @@ pub async fn run_worker(coord: Arc) { // it. let sched = Arc::clone(coord.job_queue.sched()); let node_coord = Arc::clone(&coord); - hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, job| { + hive_jobq::scheduler::Scheduler::claim_next(&sched, move |id, kind, builder| { let coord = node_coord; async move { tracing::info!( @@ -99,7 +99,7 @@ pub async fn run_worker(coord: Arc) { agent = %kind.agent(), "job_queue: node running" ); - let (grown, result) = exec::run_node(&coord, job, id, &kind).await; + let (grown, result) = exec::run_node(&coord, builder, id, &kind).await; match &result { Ok(()) => tracing::info!(node = id.get(), "job_queue: node done"), Err(e) => tracing::warn!( diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 08241e35..95a36260 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -49,8 +49,8 @@ fn submit_and_emit( /// meta input — the meta-update cascade grows its own rebuild subgraphs /// in-DAG instead of going through this surface). pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - submit_and_emit(coord, source, reason, |b| { - templates::rebuild(b, agent, true); + submit_and_emit(coord, source, reason, |builder| { + templates::rebuild(builder, agent, true); }) } @@ -67,12 +67,12 @@ pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: St /// actually running (nothing to drain on a down container). The `Reconcile` /// stays even for a down agent so a race-up between the state read and exec /// is still stopped in-DAG. -fn stop_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { +fn stop_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { // `SetWanted` is the group root and owns the agent lease; the mechanical // steps are its children (borrow the lease, run once it reaches `Finishing`, // dep-ordered among themselves). let a = || agent.to_owned(); - let wanted = b + let wanted = builder .node(NodeKind::SetWanted { agent: a(), up: false, @@ -81,22 +81,22 @@ fn stop_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { // Declaration order is dependency order: the quiesce steps come first so // the `Reconcile` that waits on them can name them. if graceful && running { - let signal = b + let signal = builder .node(NodeKind::Signal { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted); - let drain = b + let drain = builder .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(signal); - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(drain); } else { - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(wanted); @@ -107,8 +107,8 @@ fn stop_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { /// agent gets the rebuild subgraph (its tail `Reconcile` starts it on /// current derivations), otherwise a plain `Reconcile` (which starts a down /// agent and noops an already-running one). -fn start_chain(b: &JobBuilder, agent: &str, running: bool, stale: bool) { - let wanted = b +fn start_chain(builder: &JobBuilder, agent: &str, running: bool, stale: bool) { + let wanted = builder .node(NodeKind::SetWanted { agent: agent.to_owned(), up: true, @@ -118,9 +118,9 @@ fn start_chain(b: &JobBuilder, agent: &str, running: bool, stale: bool) { // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, // `Prebuild` + `Reconcile` are their own group roots (top-level, per // `rebuild_nodes`). - rebuild_nodes(b, agent, true, Some(wanted)); + rebuild_nodes(builder, agent, true, Some(wanted)); } else { - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: agent.to_owned(), }) @@ -138,11 +138,11 @@ fn start_chain(b: &JobBuilder, agent: &str, running: bool, stale: bool) { /// before `Reconcile`; a down agent gets just `Reconcile`, which /// converges to intent — a stopped (`wanted = Off`) agent stays stopped, /// a crashed (`wanted = Up`) agent comes back up. -fn restart_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { +fn restart_chain(builder: &JobBuilder, agent: &str, graceful: bool, running: bool) { let a = || agent.to_owned(); if !running { // Nothing to bounce — a lone Reconcile converges to intent. - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())); return; @@ -156,28 +156,28 @@ fn restart_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { // that step *is* the root, and the parent gate already orders it — a child // must NOT dep on its own parent (dep-scope), so it takes no sibling edge. if graceful { - let signal = b + let signal = builder .node(NodeKind::Signal { agent: a() }) .needs(Resource::Agent(a())); - let drain = b + let drain = builder .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal); - let stop = b + let stop = builder .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain); - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(stop); } else { - let stop = b + let stop = builder .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())); - let _ = b + let _ = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(stop); @@ -198,9 +198,9 @@ fn restart_chain(b: &JobBuilder, agent: &str, graceful: bool, running: bool) { // subgraph's indices onto another's used to be a function. /// Declare the stop DAG from explicit `(agent, running)` targets. -pub(crate) fn stop_nodes(b: &JobBuilder, targets: &[(String, bool)], graceful: bool) { +pub(crate) fn stop_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { for (agent, running) in targets { - stop_chain(b, agent, graceful, *running); + stop_chain(builder, agent, graceful, *running); } } @@ -210,16 +210,16 @@ pub(crate) fn stop_nodes(b: &JobBuilder, targets: &[(String, bool)], graceful: b /// running under its lease, so a down+stale agent that grew a rebuild subgraph /// reports `rebuilding` during its swap and `starting` at its reconcile, /// without the DAG having to guess one label covering every target. -pub(crate) fn start_nodes(b: &JobBuilder, targets: &[(String, bool, bool)]) { +pub(crate) fn start_nodes(builder: &JobBuilder, targets: &[(String, bool, bool)]) { for (agent, running, stale) in targets { - start_chain(b, agent, *running, *stale); + start_chain(builder, agent, *running, *stale); } } /// Declare the restart DAG from explicit `(agent, running)` targets. -pub(crate) fn restart_nodes(b: &JobBuilder, targets: &[(String, bool)], graceful: bool) { +pub(crate) fn restart_nodes(builder: &JobBuilder, targets: &[(String, bool)], graceful: bool) { for (agent, running) in targets { - restart_chain(b, agent, graceful, *running); + restart_chain(builder, agent, graceful, *running); } } @@ -258,8 +258,8 @@ pub async fn restart_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, source, reason, |b| { - restart_nodes(b, &targets, graceful); + submit_and_emit(coord, source, reason, |builder| { + restart_nodes(builder, &targets, graceful); }) } @@ -293,8 +293,8 @@ pub async fn start_many( } targets.push((agent.clone(), running, stale)); } - submit_and_emit(coord, source, reason, |b| { - start_nodes(b, &targets); + submit_and_emit(coord, source, reason, |builder| { + start_nodes(builder, &targets); }) } @@ -331,8 +331,8 @@ pub async fn stop_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, source, reason, |b| { - stop_nodes(b, &targets, graceful); + submit_and_emit(coord, source, reason, |builder| { + stop_nodes(builder, &targets, graceful); }) } @@ -344,8 +344,8 @@ pub fn perm_change( reason: String, payload: super::PermPayload, ) -> u64 { - submit_and_emit(coord, source, reason, |b| { - templates::perm_change(b, agent, payload); + submit_and_emit(coord, source, reason, |builder| { + templates::perm_change(builder, agent, payload); }) } @@ -356,8 +356,8 @@ pub fn meta_update( source: Source, reason: String, ) -> u64 { - submit_and_emit(coord, source, reason, |b| { - templates::meta_update(b, inputs, None); + submit_and_emit(coord, source, reason, |builder| { + templates::meta_update(builder, inputs, None); }) } @@ -374,7 +374,7 @@ pub fn reparent( source: Source, reason: String, ) -> u64 { - submit_and_emit(coord, source, reason, |b| { - templates::reparent(b, moves); + submit_and_emit(coord, source, reason, |builder| { + templates::reparent(builder, moves); }) } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index a7198835..89a9b206 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -36,9 +36,9 @@ use super::{Handle, JobBuilder}; /// /// Exactly one runs on a DAG that executed, and neither runs on one the operator /// dropped — see [`hive_jobq::NodeRef::on_elimination_of`]. -fn emit_rebuilt_tails(b: &JobBuilder, agent: &str, roots: &[Handle<'_>]) { +fn emit_rebuilt_tails(builder: &JobBuilder, agent: &str, roots: &[Handle<'_>]) { let ok = roots.iter().fold( - b.node(NodeKind::EmitRebuilt { + builder.node(NodeKind::EmitRebuilt { agent: agent.to_owned(), ok: true, }), @@ -51,11 +51,12 @@ fn emit_rebuilt_tails(b: &JobBuilder, agent: &str, roots: &[Handle<'_>]) { // `Reconcile` is still bringing the container back up, so reporting straight // off the elimination would announce the failure mid-recovery. let _failed = roots.iter().fold( - b.node(NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: false, - }) - .on_elimination_of(ok), + builder + .node(NodeKind::EmitRebuilt { + agent: agent.to_owned(), + ok: false, + }) + .on_elimination_of(ok), hive_jobq::NodeRef::after_any, ); } @@ -65,13 +66,13 @@ fn emit_rebuilt_tails(b: &JobBuilder, agent: &str, roots: &[Handle<'_>]) { /// /// The `Cancelled` node is what keeps a dropped approval DAG from dangling its /// row forever — its edge is the only one [`super::JobQueue::cancel`] spares. -fn resolve_approval_tails(b: &JobBuilder, approval_id: i64, root: Handle<'_>) { +fn resolve_approval_tails(builder: &JobBuilder, approval_id: i64, root: Handle<'_>) { for outcome in [ TerminalState::Done, TerminalState::Failed, TerminalState::Cancelled, ] { - let _ = b + let _ = builder .node(NodeKind::ResolveApproval { approval_id, outcome, @@ -90,18 +91,18 @@ fn resolve_approval_tails(b: &JobBuilder, approval_id: i64, root: Handle<'_>) { /// /// Same reason as [`fanned_out_mechanical`] for living here: this was the /// second construction site declaring nodes inline in an executor. -pub(crate) fn grown_rebuilds(b: &JobBuilder, agents: &[String], relock: bool) { +pub(crate) fn grown_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) { for agent in agents { - rebuild_nodes(b, agent, relock, None); + rebuild_nodes(builder, agent, relock, None); } } /// As [`grown_rebuilds`], but each agent gets its `Signal` → `Drain` window /// before being stopped. The boot sweep's flavour: it stops agents that were /// mid-turn when the host came up, so they drain rather than being cut off. -pub(crate) fn grown_graceful_rebuilds(b: &JobBuilder, agents: &[String], relock: bool) { +pub(crate) fn grown_graceful_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) { for agent in agents { - graceful_rebuild_nodes(b, agent, relock, None); + graceful_rebuild_nodes(builder, agent, relock, None); } } @@ -117,9 +118,9 @@ pub(crate) fn grown_graceful_rebuilds(b: &JobBuilder, agents: &[String], relock: /// declaration does: this is the one construction site that was hiding in an /// executor, which meant the only test of it had to re-declare the same two /// calls itself and would have kept passing if the executor changed. -pub(crate) fn fanned_out_mechanical(b: &JobBuilder, kind: NodeKind) { +pub(crate) fn fanned_out_mechanical(builder: &JobBuilder, kind: NodeKind) { let lease = Resource::Agent(kind.agent().to_owned()); - let _ = b.node(kind).needs(lease); + let _ = builder.node(kind).needs(lease); } /// The group-roots a [`rebuild_nodes`] subgraph exposes to its caller: what a @@ -176,7 +177,7 @@ impl<'a> RebuildRoots<'a> { /// takes a fresh lease; the tiny gap is harmless — `Reconcile` converges to /// the persisted `wanted` idempotently. fn rebuild_subtree<'a>( - b: &'a JobBuilder, + builder: &'a JobBuilder, agent: &str, relock: bool, graceful: bool, @@ -184,13 +185,13 @@ fn rebuild_subtree<'a>( ) -> RebuildRoots<'a> { let a = || agent.to_owned(); - let mut meta_sync = b + let mut meta_sync = builder .node(NodeKind::MetaSync { agent: a(), relock }) .needs(Resource::MetaWindow); if let Some(after) = after { meta_sync = meta_sync.after_ok(after); } - let prebuild = b + let prebuild = builder .node(NodeKind::Prebuild { agent: a() }) .needs(Resource::BuildSlot) .after_ok(meta_sync); @@ -198,27 +199,29 @@ fn rebuild_subtree<'a>( // The stop root hangs off `Prebuild` and owns the agent lease for // everything below it. `StopForUpdate` parents the swap pair either way. let stop_for_update = if graceful { - let signal = b + let signal = builder .node(NodeKind::Signal { agent: a() }) .needs(Resource::Agent(a())) .part_of(prebuild); // `Drain` is a *child* of `Signal`, so the parent gate already orders // it — a child must not dep on its own parent (dep-scope). - let drain = b + let drain = builder .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal); - b.node(NodeKind::StopForUpdate { agent: a() }) + builder + .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain) } else { - b.node(NodeKind::StopForUpdate { agent: a() }) + builder + .node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(prebuild) }; - let swap = b + let swap = builder .node(NodeKind::Swap { agent: a() }) .needs(Resource::BuildSlot) .needs(Resource::Agent(a())) @@ -227,13 +230,13 @@ fn rebuild_subtree<'a>( // `StopForUpdate`, which holds it, so this is a re-entrant borrow — no // second unit, no deadlock. Declaring it is what stops the requirement // being true only of this one DAG shape. - let _post_swap = b + let _post_swap = builder .node(NodeKind::PostSwap { agent: a() }) .needs(Resource::Agent(a())) .part_of(stop_for_update) .after_ok(swap); - let reconcile = b + let reconcile = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .after_any(prebuild); @@ -250,12 +253,12 @@ fn rebuild_subtree<'a>( /// when given, is the node this subgraph chains behind. See /// [`rebuild_subtree`] for the structure. pub(crate) fn rebuild_nodes<'a>( - b: &'a JobBuilder, + builder: &'a JobBuilder, agent: &str, relock: bool, after: Option>, ) -> RebuildRoots<'a> { - rebuild_subtree(b, agent, relock, false, after) + rebuild_subtree(builder, agent, relock, false, after) } /// As [`rebuild_nodes`], but the agent gets a `Signal` → `Drain` window to @@ -266,12 +269,12 @@ pub(crate) fn rebuild_nodes<'a>( /// *prepend* nodes — it **re-parents** the stop root, so a caller cannot /// declare it without being handed the internals. Only the boot sweep wants it. pub(crate) fn graceful_rebuild_nodes<'a>( - b: &'a JobBuilder, + builder: &'a JobBuilder, agent: &str, relock: bool, after: Option>, ) -> RebuildRoots<'a> { - rebuild_subtree(b, agent, relock, true, after) + rebuild_subtree(builder, agent, relock, true, after) } /// The rebuild subgraph a [`NodeKind::DeployApply`] grows into its own DAG once @@ -298,9 +301,9 @@ pub(crate) fn graceful_rebuild_nodes<'a>( /// `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(b: &JobBuilder, agent: &str, approval_id: i64) { - let roots = rebuild_nodes(b, agent, false, None); - let _finalize = b +pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_id: i64) { + let roots = rebuild_nodes(builder, agent, false, None); + let _finalize = builder .node(NodeKind::FinalizeDeploy { agent: agent.to_owned(), approval_id, @@ -321,9 +324,9 @@ pub(crate) fn deploy_rebuild_nodes(b: &JobBuilder, agent: &str, approval_id: i64 /// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every /// 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(b: &JobBuilder, agent: &str, relock: bool) { - let roots = rebuild_nodes(b, agent, relock, None); - emit_rebuilt_tails(b, agent, &roots.all()); +pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) { + let roots = rebuild_nodes(builder, agent, relock, None); + emit_rebuilt_tails(builder, agent, &roots.all()); } /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the @@ -351,7 +354,7 @@ pub fn rebuild(b: &JobBuilder, agent: &str, relock: bool) { /// /// 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(b: &JobBuilder, agent: &str, approval_id: i64) { +pub fn approval_deploy(builder: &JobBuilder, agent: &str, approval_id: i64) { let a = || agent.to_owned(); // The window is the widest holder in the tree: it brackets a nix // build (`BuildSlot`), takes the container down across the swap @@ -359,7 +362,7 @@ pub fn approval_deploy(b: &JobBuilder, agent: &str, approval_id: i64) { // (`MetaWindow`). All three are held for its whole subtree, which // is what lets the appended rebuild's `MetaSync` and the // `FinalizeDeploy` re-enter rather than contend. - let window = b + let window = builder .node(NodeKind::DeployWindow { agent: a(), approval_id, @@ -367,20 +370,20 @@ pub fn approval_deploy(b: &JobBuilder, agent: &str, approval_id: i64) { .needs(Resource::BuildSlot) .needs(Resource::Agent(a())) .needs(Resource::MetaWindow); - let verify = b + let verify = builder .node(NodeKind::MergeVerify { agent: a(), approval_id, }) .part_of(window); - let apply = b + let apply = builder .node(NodeKind::DeployApply { agent: a(), approval_id, }) .part_of(window) .after_ok(verify); - let _tail = b + let _tail = builder .node(NodeKind::DeployTail { agent: a(), approval_id, @@ -388,7 +391,7 @@ pub fn approval_deploy(b: &JobBuilder, agent: &str, approval_id: i64) { .part_of(window) .after_any(apply); - resolve_approval_tails(b, approval_id, window); + resolve_approval_tails(builder, approval_id, window); } /// First-deploy spawn (approval-driven): `Provision` (proposed/applied @@ -402,27 +405,27 @@ pub fn approval_deploy(b: &JobBuilder, agent: &str, approval_id: i64) { /// container was never created). Closed by a `ResolveApproval` tail root edged /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up /// already carries the whole cascade. -pub fn spawn(b: &JobBuilder, agent: &str, approval_id: i64) { +pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) { let a = || agent.to_owned(); - let provision = b + let provision = builder .node(NodeKind::Provision { agent: a() }) .needs(Resource::MetaWindow); - let create = b + let create = builder .node(NodeKind::Create { agent: a() }) .needs(Resource::BuildSlot) .needs(Resource::Agent(a())) .part_of(provision); - let dropin = b + let dropin = builder .node(NodeKind::WriteDropin { agent: a() }) .needs(Resource::Agent(a())) .part_of(create); - let _reconcile = b + let _reconcile = builder .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .part_of(create) .after_ok(dropin); - resolve_approval_tails(b, approval_id, provision); + resolve_approval_tails(builder, approval_id, provision); } /// Perm change: commit the JSON file(s), then the rebuild subgraph so @@ -430,16 +433,16 @@ pub fn spawn(b: &JobBuilder, agent: &str, approval_id: i64) { /// effect in the container. Group-roots are `WritePermFile` plus the rebuild /// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail /// edges all four. -pub fn perm_change(b: &JobBuilder, agent: &str, payload: PermPayload) { - let write = b +pub fn perm_change(builder: &JobBuilder, agent: &str, payload: PermPayload) { + let write = builder .node(NodeKind::WritePermFile { agent: agent.to_owned(), payload, }) .needs(Resource::MetaWindow); - let roots = rebuild_nodes(b, agent, true, Some(write)); + let roots = rebuild_nodes(builder, agent, true, Some(write)); emit_rebuilt_tails( - b, + builder, agent, &[write, roots.meta_sync, roots.prebuild, roots.reconcile], ); @@ -455,8 +458,8 @@ pub fn perm_change(b: &JobBuilder, agent: &str, payload: PermPayload) { /// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent /// crash-watch suppression during its `Swap` — the property the old child /// `Rebuild` DAGs carried via their own transient. -pub fn meta_update(b: &JobBuilder, inputs: Vec, approval_id: Option) { - let lock = b +pub fn meta_update(builder: &JobBuilder, inputs: Vec, approval_id: Option) { + let lock = builder .node(NodeKind::MetaLock { sweep: false, fanout: None, @@ -470,7 +473,7 @@ pub fn meta_update(b: &JobBuilder, inputs: Vec, approval_id: Option // 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); + resolve_approval_tails(builder, approval_id, lock); } } @@ -483,8 +486,8 @@ pub fn meta_update(b: &JobBuilder, inputs: Vec, approval_id: Option /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one /// off of) and near-instant. No tail node: the write is the whole effect. -pub fn reparent(b: &JobBuilder, moves: Vec<(hive_types::Ident, Option)>) { - let _reparent = b +pub fn reparent(builder: &JobBuilder, moves: Vec<(hive_types::Ident, Option)>) { + let _reparent = builder .node(NodeKind::Reparent { moves }) .needs(Resource::MetaWindow); } diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index f9a062d4..e297726e 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -29,24 +29,24 @@ fn ident(s: &str) -> hive_types::Ident { hive_types::Ident::parse(s).expect("valid test ident") } -fn rebuild(b: &JobBuilder, agent: &str) { - templates::rebuild(b, agent, true); +fn rebuild(builder: &JobBuilder, agent: &str) { + templates::rebuild(builder, agent, true); } /// Restart shape with every agent treated as **running** — the online /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) /// most queue-mechanics tests assume. Mirrors the pre-dynamic /// `templates::restart` (which is now the state-aware `submit::restart_nodes`). -fn restart_online(b: &JobBuilder, agents: &[&str], graceful: bool) { +fn restart_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::restart_nodes(b, &targets, graceful); + submit::restart_nodes(builder, &targets, graceful); } /// Stop shape with every agent treated as **running** — the online shape /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). -fn stop_online(b: &JobBuilder, agents: &[&str], graceful: bool) { +fn stop_online(builder: &JobBuilder, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::stop_nodes(b, &targets, graceful); + submit::stop_nodes(builder, &targets, graceful); } // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type @@ -299,8 +299,8 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State { #[test] fn submit_assigns_distinct_ids() { let q = JobQueue::new(1); - let first = submit(&q, "first", |b| rebuild(b, "agent-a")); - let second = submit(&q, "second", |b| rebuild(b, "agent-b")); + let first = submit(&q, "first", |builder| rebuild(builder, "agent-a")); + let second = submit(&q, "second", |builder| rebuild(builder, "agent-b")); assert_ne!(first, second); assert_eq!(q.snapshot().len(), 2); } @@ -313,8 +313,8 @@ fn submit_assigns_distinct_ids() { #[test] fn identical_resubmit_is_a_distinct_dag() { let q = JobQueue::new(1); - let first = submit(&q, "first", |b| rebuild(b, "agent-a")); - let resubmit = submit(&q, "again", |b| rebuild(b, "agent-a")); + let first = submit(&q, "first", |builder| rebuild(builder, "agent-a")); + let resubmit = submit(&q, "again", |builder| rebuild(builder, "agent-a")); assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG"); assert_eq!(q.snapshot().len(), 2); } @@ -322,9 +322,11 @@ fn identical_resubmit_is_a_distinct_dag() { #[test] fn distinct_submits_never_collapse() { let q = JobQueue::new(1); - let rebuild_a = submit(&q, "r", |b| rebuild(b, "agent-a")); - let rebuild_b = submit(&q, "r", |b| rebuild(b, "agent-b")); - let restart_a = submit(&q, "r", |b| restart_online(b, &["agent-a"], false)); + let rebuild_a = submit(&q, "r", |builder| rebuild(builder, "agent-a")); + let rebuild_b = submit(&q, "r", |builder| rebuild(builder, "agent-b")); + let restart_a = submit(&q, "r", |builder| { + restart_online(builder, &["agent-a"], false); + }); assert_ne!(rebuild_a, rebuild_b); assert_ne!(rebuild_a, restart_a); assert_eq!(q.snapshot().len(), 3); @@ -342,8 +344,10 @@ fn resubmit_while_running_is_new_dag() { // swallowed" is the scenario people worry about, and a reader looking for // it should find it. let q = JobQueue::new(1); - let a = submit(&q, "first", |b| rebuild(b, "agent-a")); - let again = submit(&q, "config bumped during build", |b| rebuild(b, "agent-a")); + let a = submit(&q, "first", |builder| rebuild(builder, "agent-a")); + let again = submit(&q, "config bumped during build", |builder| { + rebuild(builder, "agent-a"); + }); assert_ne!(a, again); assert_eq!(q.snapshot().len(), 2); } @@ -379,7 +383,7 @@ fn rebuild_chain_is_declared_serial() { // logic"). Both axes are asserted below because a template can break either // one independently. let q = JobQueue::new(1); - let id = submit(&q, "r", |b| rebuild(b, "agent-a")); + let id = submit(&q, "r", |builder| rebuild(builder, "agent-a")); assert_eq!( declared_shape(&q, id), vec![ @@ -426,9 +430,13 @@ fn rebuild_chain_is_declared_serial() { fn graceful_rebuild_chain_drains_before_stopping() { let q = JobQueue::new(1); let id = q - .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &JobBuilder| { - templates::graceful_rebuild_nodes(b, "agent-a", true, None); - }) + .submit( + Source::AutoUpdate, + "sweep".to_owned(), + |builder: &JobBuilder| { + templates::graceful_rebuild_nodes(builder, "agent-a", true, None); + }, + ) .expect("valid shape"); assert_eq!( declared_shape(&q, id) @@ -460,8 +468,8 @@ 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 id = submit(&q, "manual", |b| { - templates::rebuild_nodes(b, "agent-a", true, None); + let id = submit(&q, "manual", |builder| { + templates::rebuild_nodes(builder, "agent-a", true, None); }); assert_eq!( declared_shape(&q, id) @@ -541,7 +549,7 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() { // a resource unit is held for the acquirer's whole subtree, so the slot // `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it. let q = JobQueue::new(1); - let id = submit(&q, "r", |b| rebuild(b, "agent-a")); + let id = submit(&q, "r", |builder| rebuild(builder, "agent-a")); let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind)); let agent = || Resource::Agent("agent-a".to_owned()); @@ -578,8 +586,8 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() { #[test] fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4); - let id = submit(&q, "hive-wide", |b| { - restart_online(b, &["agent-a", "agent-b"], false); + let id = submit(&q, "hive-wide", |builder| { + restart_online(builder, &["agent-a", "agent-b"], false); }); // A hive-wide restart is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); @@ -614,8 +622,8 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { #[test] fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4); - let id = submit(&q, "hive-wide stop", |b| { - stop_online(b, &["agent-a", "agent-b"], false); + let id = submit(&q, "hive-wide stop", |builder| { + stop_online(builder, &["agent-a", "agent-b"], false); }); // A hive-wide stop is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); @@ -647,9 +655,9 @@ fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { let q = JobQueue::new(4); // fresh: offline + not stale → SetWanted → Reconcile. // stale: offline + stale → SetWanted → «rebuild subgraph». - let id = submit(&q, "hive-wide start", |b| { + let id = submit(&q, "hive-wide start", |builder| { submit::start_nodes( - b, + builder, &[ ("fresh".to_owned(), false, false), ("stale".to_owned(), false, true), @@ -694,14 +702,14 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { // read and node exec. let q = JobQueue::new(4); // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). - let stop = submit(&q, "stop down", |b| { - submit::stop_nodes(b, &[("down".to_owned(), false)], true); + let stop = submit(&q, "stop down", |builder| { + submit::stop_nodes(builder, &[("down".to_owned(), false)], true); }); // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): // nothing to bounce, and restart never rewrites intent, so the tail // Reconcile converges the down agent to its existing `wanted`. - let restart = submit(&q, "restart down", |b| { - submit::restart_nodes(b, &[("down2".to_owned(), false)], true); + let restart = submit(&q, "restart down", |builder| { + submit::restart_nodes(builder, &[("down2".to_owned(), false)], true); }); let shape = |id: u64| -> Vec { q.snapshot() @@ -737,14 +745,18 @@ fn boot_sweep_nodes_declare_their_own_resources() { // compile; only an exhaustive caller list would have caught it. let q = JobQueue::new(4); let id = q - .submit(Source::AutoUpdate, "boot".to_owned(), |b: &JobBuilder| { - crate::workers::auto_update::boot_nodes( - b, - true, - vec!["stale-agent".to_owned()], - vec!["drifted-agent".to_owned()], - ); - }) + .submit( + Source::AutoUpdate, + "boot".to_owned(), + |builder: &JobBuilder| { + crate::workers::auto_update::boot_nodes( + builder, + true, + vec!["stale-agent".to_owned()], + vec!["drifted-agent".to_owned()], + ); + }, + ) .expect("valid shape"); let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock")); @@ -846,7 +858,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() { // easy thing to break — someone flattening the chain would keep every edge // and still lose the guarantee. let q = JobQueue::new(1); - let id = submit(&q, "r", |b| rebuild(b, "agent-a")); + let id = submit(&q, "r", |builder| rebuild(builder, "agent-a")); let shape = declared_shape(&q, id); let parent_of = |kind: &str| { shape @@ -918,9 +930,9 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() { #[test] fn a_fanned_out_mechanical_node_declares_its_agent_lease() { let q = JobQueue::new(4); - let id = submit(&q, "fan-out", |b| { + let id = submit(&q, "fan-out", |builder| { templates::fanned_out_mechanical( - b, + builder, NodeKind::Start { agent: "agent-a".to_owned(), }, @@ -954,9 +966,13 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() { let q = JobQueue::new(4); let agents = vec!["alice".to_owned(), "bob".to_owned()]; let id = q - .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &JobBuilder| { - templates::grown_graceful_rebuilds(b, &agents, true); - }) + .submit( + Source::AutoUpdate, + "sweep".to_owned(), + |builder: &JobBuilder| { + templates::grown_graceful_rebuilds(builder, &agents, true); + }, + ) .expect("valid shape"); // One chain per agent, each an independent group root — so the two rebuild @@ -987,7 +1003,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() { #[test] fn cancel_clears_queued_dag() { let q = JobQueue::new(1); - let id = submit(&q, "r", |b| rebuild(b, "agent-a")); + let id = submit(&q, "r", |builder| rebuild(builder, "agent-a")); assert!(q.cancel(id), "fully-queued dag cancels"); // The operator sees `Cancelled` the moment the cancel returns — the spared // tail is still `Pending`, and a DAG must not read `Queued` back to the @@ -1016,8 +1032,8 @@ fn cancel_clears_queued_dag() { #[test] fn cancel_drops_one_agents_branch_leaving_the_rest() { let q = JobQueue::new(2); - let id = submit(&q, "r", |b| { - restart_online(b, &["agent-a", "agent-b"], false); + let id = submit(&q, "r", |builder| { + restart_online(builder, &["agent-a", "agent-b"], false); }); // Per-agent subgraphs are independent roots; find agent-a's. let snap = q.snapshot(); @@ -1085,20 +1101,20 @@ fn cancelled_power_op_runs_no_compensating_node() { let case = format!("graceful={graceful} running={running}"); let q = JobQueue::new(1); - let id = submit(&q, "bounce", |b| { - submit::restart_nodes(b, &targets, graceful); + let id = submit(&q, "bounce", |builder| { + submit::restart_nodes(builder, &targets, graceful); }); assert_cancels_clean(&q, id, false, &format!("restart {case}")); let q = JobQueue::new(1); - let id = submit(&q, "stop", |b| { - submit::stop_nodes(b, &targets, graceful); + let id = submit(&q, "stop", |builder| { + submit::stop_nodes(builder, &targets, graceful); }); assert_cancels_clean(&q, id, true, &format!("stop {case}")); let q = JobQueue::new(1); - let id = submit(&q, "start", |b| { - submit::start_nodes(b, &[("agent-a".to_owned(), running, false)]); + let id = submit(&q, "start", |builder| { + submit::start_nodes(builder, &[("agent-a".to_owned(), running, false)]); }); assert_cancels_clean(&q, id, true, &format!("start {case}")); } @@ -1118,8 +1134,8 @@ fn cancelled_power_op_runs_no_compensating_node() { #[test] fn cancelled_dag_still_runs_its_approval_tail() { let q = JobQueue::new(1); - let id = submit(&q, "approval #7", |b| { - templates::approval_deploy(b, "agent-a", 7); + let id = submit(&q, "approval #7", |builder| { + templates::approval_deploy(builder, "agent-a", 7); }); assert!(q.cancel(id), "fully-queued dag cancels"); // The `Cancelled` tail is the only node whose edge accepts a dropped @@ -1141,7 +1157,7 @@ fn cancelled_dag_still_runs_its_approval_tail() { assert_eq!(state_of(&q, id), State::Cancelled); // An unrelated DAG landing in the same graph doesn't disturb this one's // roll-up — the snapshot is per-DAG, not a global state machine. - let _other = submit(&q, "r", |b| rebuild(b, "agent-b")); + let _other = submit(&q, "r", |builder| rebuild(builder, "agent-b")); assert_eq!(state_of(&q, id), State::Cancelled); } @@ -1155,8 +1171,8 @@ fn cancelled_dag_still_runs_its_approval_tail() { #[test] fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { let q = JobQueue::new(1); - let id = submit(&q, "approval #7", |b| { - templates::approval_deploy(b, "agent-a", 7); + let id = submit(&q, "approval #7", |builder| { + templates::approval_deploy(builder, "agent-a", 7); }); assert_eq!( @@ -1213,8 +1229,8 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { // children run (`a_completing_node_grows_the_work_it_declared`, // `parent_parks_in_finishing_until_children_roll_up`). let q = JobQueue::new(1); - let id = submit(&q, "deploy graft", |b| { - templates::deploy_rebuild_nodes(b, "agent-a", 11); + let id = submit(&q, "deploy graft", |builder| { + templates::deploy_rebuild_nodes(builder, "agent-a", 11); }); assert_eq!( @@ -1380,7 +1396,9 @@ fn error_truncation_cuts_on_a_char_boundary() { #[test] fn graceful_stop_shape_signal_drain_reconcile() { let q = JobQueue::new(1); - let id = submit(&q, "graceful", |b| stop_online(b, &["agent-a"], true)); + let id = submit(&q, "graceful", |builder| { + stop_online(builder, &["agent-a"], true); + }); assert_eq!( declared_shape(&q, id), vec![ @@ -1410,8 +1428,8 @@ fn graceful_stop_shape_signal_drain_reconcile() { #[test] fn spawn_shape_provision_create_dropin_reconcile() { let q = JobQueue::new(1); - let id = submit(&q, "approval #7 spawn", |b| { - templates::spawn(b, "newbie", 7); + let id = submit(&q, "approval #7 spawn", |builder| { + templates::spawn(builder, "newbie", 7); }); assert_eq!( declared_shape(&q, id), @@ -1435,9 +1453,9 @@ fn spawn_shape_provision_create_dropin_reconcile() { #[test] fn perm_change_shape_prefixes_rebuild_chain() { let q = JobQueue::new(1); - let id = submit(&q, "perm", |b| { + let id = submit(&q, "perm", |builder| { templates::perm_change( - b, + builder, "agent-a", PermPayload::Combined { groups: Some(vec![]), @@ -1473,8 +1491,8 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() { // `MetaLock`, and it must declare the meta window — a topology commit // must not land inside another node's staged deploy window. let q = JobQueue::new(1); - let id = submit(&q, "set-parent", |b| { - templates::reparent(b, vec![(ident("alice"), Some(ident("bob")))]); + let id = submit(&q, "set-parent", |builder| { + templates::reparent(builder, vec![(ident("alice"), Some(ident("bob")))]); }); assert_eq!( declared_shape(&q, id), @@ -1497,8 +1515,8 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() { // request is the reason a single node was chosen in the first place. let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)]; let q = JobQueue::new(1); - let id = submit(&q, "set-parent-bulk", |b| { - templates::reparent(b, moves.clone()); + let id = submit(&q, "set-parent-bulk", |builder| { + templates::reparent(builder, moves.clone()); }); assert_eq!( declared_shape(&q, id), diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 9c52e25c..8132f0b7 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -327,7 +327,7 @@ pub async fn run(coord: Arc) -> Result<()> { /// kind-derived resources were removed, which drops the agent lease a boot /// reconcile needs to not race another DAG's container ops. pub(crate) fn boot_nodes( - b: &crate::job_queue::JobBuilder, + builder: &crate::job_queue::JobBuilder, any_stale: bool, fanout: Vec, drifted: Vec, @@ -341,7 +341,7 @@ pub(crate) fn boot_nodes( // ⇒ 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 _ = b + let _ = builder .node(NodeKind::MetaLock { sweep: true, fanout: Some(fanout), @@ -356,7 +356,9 @@ pub(crate) fn boot_nodes( for name in drifted { // Name the lease before the agent string moves into the kind. let lease = Resource::Agent(name.clone()); - let _ = b.node(NodeKind::Reconcile { agent: name }).needs(lease); + let _ = builder + .node(NodeKind::Reconcile { agent: name }) + .needs(lease); } }