From 58a9f218f28006562e8a6808b95b0fae55bb2233 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 16:22:56 +0200 Subject: [PATCH] job_queue: fix the boot sweep's lost declarations, drop the node wrapper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review findings on the resources-at-construction change. argus: `workers::auto_update`'s boot sweep constructs nodes through `templates::node` too, and it was not converted. With the kind-derived declaration gone, its sweep `MetaLock` and its per-agent `Reconcile` silently declared no resources at all — so a boot reconcile no longer held the agent lease and could race another DAG's container ops, and the sweep's meta commit could land inside another node's staged deploy window. Nothing failed to compile: removing an implicit behaviour from a helper is invisible at every call site that relied on it. The declarations now live in a pure `boot_nodes`, split out of `submit_boot_tree` so they can be exercised without a `Coordinator`. That path is the only place job nodes are built outside `job_queue/`, which is exactly why it had no coverage; `boot_sweep_nodes_declare_ their_own_resources` closes that, asserting against declared graph edges rather than against the kind. mara: `templates::node` is a redundant redirect now that it no longer derives resources — deleted, and its 43 call sites use `Job::node` directly. The reasoning it documented moved to the module docs of `templates.rs` and `resource.rs`, which is where it stays true. --- hive-c0re/src/job_queue/exec.rs | 2 +- hive-c0re/src/job_queue/resource.rs | 8 +- hive-c0re/src/job_queue/submit.rs | 72 +++++----- hive-c0re/src/job_queue/templates.rs | 193 ++++++++++++--------------- hive-c0re/src/job_queue/tests.rs | 66 +++++++-- hive-c0re/src/workers/auto_update.rs | 70 ++++++---- 6 files changed, 231 insertions(+), 180 deletions(-) diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index ef962b97..95486951 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -424,7 +424,7 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result Handle<'_> { - b.node(kind) -} - /// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node /// gated on every group-root in `roots`, and the failure node gated on *its* /// elimination. @@ -57,13 +38,10 @@ pub(crate) fn node(b: &Job, kind: NodeKind) -> Handle<'_> { /// dropped — see [`hive_jobq::NodeRef::on_elimination_of`]. fn emit_rebuilt_tails(b: &Job, agent: &str, roots: &[Handle<'_>]) { let ok = roots.iter().fold( - node( - b, - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: true, - }, - ), + b.node(NodeKind::EmitRebuilt { + agent: agent.to_owned(), + ok: true, + }), hive_jobq::NodeRef::after_ok, ); // The failure branch needs *both*: the ok branch being ruled out (that is the @@ -73,13 +51,10 @@ fn emit_rebuilt_tails(b: &Job, 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( - node( - b, - NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: false, - }, - ) + b.node(NodeKind::EmitRebuilt { + agent: agent.to_owned(), + ok: false, + }) .on_elimination_of(ok), hive_jobq::NodeRef::after_any, ); @@ -96,14 +71,12 @@ fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) { TerminalState::Failed, TerminalState::Cancelled, ] { - let _ = node( - b, - NodeKind::ResolveApproval { + let _ = b + .node(NodeKind::ResolveApproval { approval_id, outcome, - }, - ) - .on_outcome(root, &[outcome]); + }) + .on_outcome(root, &[outcome]); } } @@ -182,37 +155,42 @@ pub(crate) fn rebuild_nodes<'a>( let a = || agent.to_owned(); let RebuildOpts { relock, graceful } = opts; - let mut meta_sync = - node(b, NodeKind::MetaSync { agent: a(), relock }).needs(Resource::MetaWindow); + let mut meta_sync = b + .node(NodeKind::MetaSync { agent: a(), relock }) + .needs(Resource::MetaWindow); if let Some(after) = after { meta_sync = meta_sync.after_ok(after); } - let prebuild = node(b, NodeKind::Prebuild { agent: a() }) + let prebuild = b + .node(NodeKind::Prebuild { agent: a() }) .needs(Resource::BuildSlot) .after_ok(meta_sync); // 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 = node(b, NodeKind::Signal { agent: a() }) + let signal = b + .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 = node(b, NodeKind::Drain { agent: a() }) + let drain = b + .node(NodeKind::Drain { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal); - node(b, NodeKind::StopForUpdate { agent: a() }) + b.node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain) } else { - node(b, NodeKind::StopForUpdate { agent: a() }) + b.node(NodeKind::StopForUpdate { agent: a() }) .needs(Resource::Agent(a())) .part_of(prebuild) }; - let swap = node(b, NodeKind::Swap { agent: a() }) + let swap = b + .node(NodeKind::Swap { agent: a() }) .needs(Resource::BuildSlot) .needs(Resource::Agent(a())) .part_of(stop_for_update); @@ -220,12 +198,14 @@ pub(crate) fn rebuild_nodes<'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 = node(b, NodeKind::PostSwap { agent: a() }) + let _post_swap = b + .node(NodeKind::PostSwap { agent: a() }) .needs(Resource::Agent(a())) .part_of(stop_for_update) .after_ok(swap); - let reconcile = node(b, NodeKind::Reconcile { agent: a() }) + let reconcile = b + .node(NodeKind::Reconcile { agent: a() }) .needs(Resource::Agent(a())) .after_any(prebuild); @@ -272,16 +252,14 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { }, None, ); - let _finalize = node( - b, - NodeKind::FinalizeDeploy { + let _finalize = b + .node(NodeKind::FinalizeDeploy { agent: agent.clone(), approval_id, - }, - ) - .needs(Resource::MetaWindow) - .after_ok(roots.prebuild) - .after_ok(roots.reconcile); + }) + .needs(Resource::MetaWindow) + .after_ok(roots.prebuild) + .after_ok(roots.reconcile); }) } @@ -363,42 +341,34 @@ pub fn approval_deploy( // (`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 = node( - b, - NodeKind::DeployWindow { + let window = b + .node(NodeKind::DeployWindow { agent: a(), approval_id, - }, - ) - .needs(Resource::BuildSlot) - .needs(Resource::Agent(a())) - .needs(Resource::MetaWindow); - let verify = node( - b, - NodeKind::MergeVerify { + }) + .needs(Resource::BuildSlot) + .needs(Resource::Agent(a())) + .needs(Resource::MetaWindow); + let verify = b + .node(NodeKind::MergeVerify { agent: a(), approval_id, - }, - ) - .part_of(window); - let apply = node( - b, - NodeKind::DeployApply { + }) + .part_of(window); + let apply = b + .node(NodeKind::DeployApply { agent: a(), approval_id, - }, - ) - .part_of(window) - .after_ok(verify); - let _tail = node( - b, - NodeKind::DeployTail { + }) + .part_of(window) + .after_ok(verify); + let _tail = b + .node(NodeKind::DeployTail { agent: a(), approval_id, - }, - ) - .part_of(window) - .after_any(apply); + }) + .part_of(window) + .after_any(apply); resolve_approval_tails(b, approval_id, window); }), @@ -423,7 +393,7 @@ pub fn reconcile_only( declare: Box::new(move |b: &Job| { // Name the lease before the agent string is moved into the kind. let lease = Resource::Agent(agent.clone()); - let _reconcile = node(b, NodeKind::Reconcile { agent }).needs(lease); + let _reconcile = b.node(NodeKind::Reconcile { agent }).needs(lease); }), } } @@ -446,15 +416,20 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec) -> Result<()> { Ok(()) } +/// The boot DAG's node declarations, split out of [`submit_boot_tree`] so they +/// can be exercised without a live [`Coordinator`]. +/// +/// That split is not cosmetic: this path constructs nodes outside +/// `job_queue/`, so it is the one place a resource declaration can be forgotten +/// without any in-module test noticing. It has happened once already — the +/// sweep `MetaLock` and the boot `Reconcile`s silently declared nothing when +/// 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::Job, + any_stale: bool, + fanout: Vec, + drifted: Vec, +) { + use crate::job_queue::NodeKind; + use crate::job_queue::resource::Resource; + + // 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 _ = b + .node(NodeKind::MetaLock { + sweep: true, + fanout: Some(fanout), + // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), + // so it names no inputs. + inputs: Vec::new(), + }) + .needs(Resource::BuildSlot) + .needs(Resource::MetaWindow); + } + // One boot Reconcile per drifted agent — independent roots. + 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); + } +} + /// Submit this boot's work as **one DAG** (no anchor node, no per-agent /// child DAGs). Node 0 is the sweep `MetaLock` (only when /// something is stale) — its executor bumps the hyperhive lock, then grows @@ -334,7 +377,7 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, NodeKind, Source, templates}; + use crate::job_queue::{DagSpec, Source}; // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. if !any_stale && drifted.is_empty() { @@ -348,29 +391,8 @@ fn submit_boot_tree( n_skipped, ); - 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 declare: crate::job_queue::Declare = + Box::new(move |b| boot_nodes(b, any_stale, fanout, drifted)); let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they