From 10dbdb444df70dcb088fdf657107e89ebbce7c32 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 15:57:51 +0200 Subject: [PATCH 1/2] job_queue: declare a node's resources where the node is constructed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Resources were derived from the node's kind: `templates::node` called `NodeKind::resource_deps()`, which fanned out to `needs_build_slot` / `needs_lease` / `needs_meta_window`. That made the requirement a property of the *kind*, so a kind that happened to run under an ancestor already holding the resource could get away with declaring nothing. Three did. `Start`, `Stop` and `PostSwap` appear in none of the three predicates, and that was only safe because one construction site fans them out from inside a lease-holding `Reconcile` — a fact about today's DAG shape, not about the nodes. Each of the 41 construction sites now says what it holds. `Start` / `Stop` / `PostSwap` declare the agent lease; per the contract that is a re-entrant borrow, which a new test pins rather than argues. `running_transients` reads the node's declared deps instead of re-deriving from the kind. That closes the blank-pill gap: the pill went blank during container start, stop and the post-swap tail because the declaration was missing, not because the filter was wrong. The deleted predicates carried the only written record of three design decisions; each moved to the `Resource` variant it constrains rather than dying with its function. --- docs/coordinator.md | 2 +- hive-c0re/src/job_queue/exec.rs | 20 ++++-- hive-c0re/src/job_queue/mod.rs | 28 ++++---- hive-c0re/src/job_queue/model.rs | 86 ++--------------------- hive-c0re/src/job_queue/resource.rs | 61 ++++++++-------- hive-c0re/src/job_queue/submit.rs | 34 ++++++--- hive-c0re/src/job_queue/templates.rs | 100 +++++++++++++++++++-------- hive-c0re/src/job_queue/tests.rs | 100 +++++++++++++++++++++++++-- hive-c0re/src/meta.rs | 2 +- 9 files changed, 256 insertions(+), 177 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index dc87418d..cda161e3 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -75,7 +75,7 @@ container build: resource declared by every node kind that mutates the meta repo — `MetaSync`, `MetaLock`, `WritePermFile`, `Reparent`, `Provision`'s agent registration, and `DeployWindow` — the deploy subtree's root, which holds it across every - phase below it (`NodeKind::needs_meta_window`). Two meta + phase below it (it declares `Resource::MetaWindow`). Two meta mutations can therefore never interleave, so no commit lands inside another node's staged window. It is a queue resource rather than a runtime mutex because a resource is held by a subtree root across its whole subtree, which diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index d6f44a37..ef962b97 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -14,6 +14,7 @@ use super::{Claim, Declare}; use hive_jobq::TerminalState; use super::model::NodeKind; +use super::resource::Resource; use crate::coordinator::Coordinator; use crate::power::{ReconcileAction, reconcile_action}; @@ -168,8 +169,8 @@ fn run_emit_rebuilt(coord: &Arc, claim: &Claim, ok: bool) -> NodeOu /// Write the agent's durable power intent — the DAG-node form of the old /// pre-submit `set_wanted` side effect. Store-only (no container touch), so -/// build-slot-exempt; but it takes the agent's lifecycle lease (see -/// `NodeKind::needs_lease`) so the whole power-op DAG is atomic per-agent. +/// build-slot-exempt; but it declares the agent's lifecycle lease +/// (`Resource::Agent`) so the whole power-op DAG is atomic per-agent. /// The downstream `Reconcile` reads the intent this writes. Unlike the old /// warn-and-continue write, a failed write fails the node (cancel-downstream /// cancels the `Reconcile`) rather than letting it converge to a stale @@ -189,7 +190,7 @@ fn run_set_wanted(coord: &Arc, claim: &Claim, up: bool) -> Result, claim: &Claim) -> Result, claim: &Claim) -> Result { let name = &claim.agent; @@ -417,8 +418,13 @@ async fn run_reconcile(coord: &Arc, claim: &Claim) -> Result, claim: &Claim) -> Result< let NodeKind::WritePermFile { payload, .. } = &claim.kind else { anyhow::bail!("run_write_perm_file on a non-WritePermFile node"); }; - // Runs under the deploy window (`NodeKind::needs_meta_window`): a + // Runs under the deploy window (it declares `Resource::MetaWindow`): a // perm commit landing inside another node's staged prepare→finalize // window would sweep the staged deploy lock into its commit (the // commits are also path-limited in meta.rs — belt and braces). @@ -592,7 +598,7 @@ async fn run_write_perm_file(coord: &Arc, claim: &Claim) -> Result< /// commit (`Coordinator::reparent_bulk_with_notify`, which already handles /// both the single- and bulk-move case, sends the per-agent move /// notifications, and rescans + diff-emits the container tree). Runs under -/// the deploy window (`NodeKind::needs_meta_window`), same reasoning as +/// the deploy window (it declares `Resource::MetaWindow`), same reasoning as /// `run_write_perm_file`: a topology commit landing inside another node's /// staged deploy window would sweep the staged lock into its commit. async fn run_reparent(coord: &Arc, claim: &Claim) -> Result { diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 9b9db781..3db89559 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -9,7 +9,7 @@ //! - [`model::NodeKind`] **is** the crate payload `N` directly — each variant //! carries the agent it targets ([`NodeKind::agent`]); the two resource //! classes are [`resource::Resource`] (`BuildSlot` node-held, `Agent` lease -//! subtree-held), derived per node by [`NodeKind::resource_deps`]; +//! subtree-held), declared per node at its construction site; //! - a **DAG is a single container node** ([`NodeKind::Dag`], `parent = None`) //! carrying the group's metadata, with the work nodes hung under it as //! its subtree (the **parent axis** groups; `deps` order). So the container's @@ -436,10 +436,13 @@ impl JobQueue { /// `Stop` pill are both pills; only one means a vanished container is /// expected. /// - /// By design, `Start` / `Stop` / `PostSwap` run inside a lease-holding - /// ancestor and re-declare nothing, so they light no pill; closing that is - /// the resources-where-constructed work, not this function. An agent's lease - /// is cap-1, so at most one entry per agent. + /// Read off the node's **declared** resource edges, not off its kind. Those + /// are the same thing now that every construction site states what it holds, + /// and the distinction is the whole point: `Start` / `Stop` / `PostSwap` run + /// inside a lease-holding ancestor, and while the declaration was derived + /// from the kind they re-declared nothing and lit no pill. Asking the node + /// what it holds cannot go stale that way. An agent's lease is cap-1, so at + /// most one entry per agent. #[must_use] pub fn running_transients(&self) -> Vec { let inner = self.lock(); @@ -449,14 +452,13 @@ impl JobQueue { .nodes() .filter(|n| matches!(n.state, State::Running)) .filter_map(|n| { - let agent = - n.payload - .resource_deps() - .into_iter() - .find_map(|(name, _)| match name { - Resource::Agent(a) => Some(a), - _ => None, - })?; + let agent = n.deps.iter().find_map(|dep| match dep { + hive_jobq::Dep::Resource { + name: Resource::Agent(a), + .. + } => Some(a.clone()), + _ => None, + })?; Some(RunningTransient { agent, label: n.payload.as_str().to_owned(), diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 19556271..b6a9fe8d 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -21,7 +21,8 @@ use hive_jobq::TerminalState; /// The primitive operations — each kind maps to one executor fn in /// `exec.rs`, a thin wrapper over existing `lifecycle.rs` / `meta.rs` /// code. Concurrency is gated by two resource classes (see -/// [`NodeKind::needs_build_slot`] / [`NodeKind::needs_lease`]); the +/// [`Resource`](super::resource::Resource), declared per node where the node +/// is constructed rather than derived from its kind); the /// meta *repo* is serialized by `meta::META_LOCK` inside the wrapped /// functions themselves, which is why there is no `GitCommit` node — /// a standalone commit node would open a dirty-working-tree window @@ -62,8 +63,10 @@ pub enum NodeKind { /// swap succeeded; the tail `Reconcile` deps `AfterAny(PostSwap)`, so on /// swap failure this node is cancel-cascaded (a terminal state) and /// recovery still runs. Store/forge/matrix work only — no nix build, so - /// build-slot-exempt; the agent lease taken at `Swap` is held across the - /// whole chain until `Reconcile` settles, so it's not re-declared here. + /// build-slot-exempt. It *does* declare the agent lease: an ancestor in the + /// stop chain already holds it, so this is a re-entrant borrow rather than a + /// second unit — declaring it keeps the requirement true of this node rather + /// than of the one DAG shape it happens to be used in. PostSwap { agent: String }, /// First-spawn pre-create provisioning: proposed/applied repos, /// state subvolume, and meta registration (`sync_agents`). Runs @@ -124,7 +127,7 @@ pub enum NodeKind { /// Topology move(s) — `set-parent` (len 1) or `set-parent-bulk` (len N) — /// as a single queue node. Agentless like [`NodeKind::MetaLock`]: a /// reparent touches the meta repo, not any one container, and a bulk - /// move spans multiple agents anyway. `needs_meta_window() = true`, same + /// move spans multiple agents anyway. Declares the meta window, same /// precedent as [`NodeKind::WritePermFile`] (also a small /// git-commit-under-`META_LOCK` op) — a reparent's commit must not land /// inside another node's staged deploy `prepare_deploy`→`finalize_deploy` @@ -347,42 +350,6 @@ impl NodeKind { } } - /// Nix-heavy kinds hold one of the `buildSlots` semaphore permits - /// for the node's duration. - pub fn needs_build_slot(&self) -> bool { - matches!( - self, - NodeKind::Prebuild { .. } - | NodeKind::Swap { .. } - | NodeKind::Create { .. } - | NodeKind::MetaLock { .. } - | NodeKind::DeployWindow { .. } - ) - } - - /// Container-affecting kinds require the DAG to hold the agent's - /// lifecycle lease (acquired at the first such node, held until the - /// DAG is terminal). Lease-exempt kinds (`MetaSync`, `Prebuild`, - /// `Provision`, `MetaLock`, `WritePermFile`, `Reparent`) touch the store / meta repo, not the - /// running container — which is exactly why a `Prebuild` can overlap - /// another DAG's work on the same agent. `Provision` precedes the - /// container's existence entirely, so the lease is first taken at the - /// `Create` node it feeds. - pub fn needs_lease(&self) -> bool { - matches!( - self, - NodeKind::Swap { .. } - | NodeKind::Create { .. } - | NodeKind::Reconcile { .. } - | NodeKind::StopForUpdate { .. } - | NodeKind::Signal { .. } - | NodeKind::Drain { .. } - | NodeKind::WriteDropin { .. } - | NodeKind::DeployWindow { .. } - | NodeKind::SetWanted { .. } - ) - } - /// Whether running this node is *expected* to take the agent's container /// down. Feeds `TransientState::deliberate_stop`, which the crash watcher /// reads to tell an intentional stop from a crash. @@ -420,45 +387,6 @@ impl NodeKind { // their own answer. // - `DeployWindow` brackets a deploy without itself stopping anything. } - - /// Kinds that **mutate the meta repo** and so must hold the global - /// [`Resource::MetaWindow`](super::resource::Resource::MetaWindow) for - /// their duration: no two meta mutations may interleave, because a commit - /// landing inside another node's staged `prepare_deploy`→`finalize_deploy` - /// window would sweep the staged `flake.lock` into its own commit and - /// neuter `abort_deploy`. - /// - /// This is the queue-primitive replacement for the former runtime - /// `meta::exclusive()` mutex — same global serialisation, but held by the - /// scheduler and therefore able to span a whole subtree, which a - /// `MutexGuard` cannot. - /// - /// Note what is **not** here: [`NodeKind::Prebuild`]. The window must stay - /// off the multi-minute toplevel build, which only *reads* the store — the - /// old mutex was scoped to drop before it, and holding a cap-1 global - /// across it would serialize every agent's rebuild behind every other's. - /// That is why the meta preamble is its own [`NodeKind::MetaSync`] node, - /// and why that node is a sibling rather than `Prebuild`'s parent (a - /// resource held by a parent covers its whole subtree). - /// - /// Two of these kinds run *inside* a [`NodeKind::DeployWindow`]'s subtree - /// (the appended rebuild's `MetaSync`, and [`NodeKind::FinalizeDeploy`]). - /// They still declare the window: a descendant re-enters an ancestor's hold - /// through the crate's recursive lock, exactly as `Start` / `Stop` re-enter - /// a `Reconcile`'s agent lease. Declaring it is what keeps the requirement - /// true of the *node* rather than of one particular DAG shape. - pub fn needs_meta_window(&self) -> bool { - matches!( - self, - NodeKind::MetaSync { .. } - | NodeKind::Provision { .. } - | NodeKind::MetaLock { .. } - | NodeKind::WritePermFile { .. } - | NodeKind::Reparent { .. } - | NodeKind::DeployWindow { .. } - | NodeKind::FinalizeDeploy { .. } - ) - } } /// Submit-time spec for a whole DAG: the group's metadata plus the declared — diff --git a/hive-c0re/src/job_queue/resource.rs b/hive-c0re/src/job_queue/resource.rs index a3118137..720a44aa 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -3,8 +3,16 @@ //! `hive-jobq` is generic over a resource type `R: Clone + Eq + Hash` and a node //! payload `N`; here `R` is [`Resource`] and `N` is [`NodeKind`] directly (each //! variant carries the agent it targets). - -use super::model::NodeKind; +//! +//! **A node's resources are declared where the node is constructed**, not +//! derived from its kind — see `templates::node`. Deriving them made the +//! requirement a property of the *kind*, which let a kind that happened to run +//! under a holding ancestor declare nothing at all. +//! +//! All of a node's resource edges are acquired **atomically** +//! (`try_acquire_all`) — a node never holds one resource while waiting on +//! another, so the multi-resource nodes (a `MetaLock` wants a build slot *and* +//! the meta window) cannot deadlock against each other. /// The two resource classes the queue gates concurrency on, as the crate's /// generic resource type `R`. @@ -19,10 +27,20 @@ pub enum Resource { /// a DAG's first container-affecting node for that agent and re-entered by /// the rest of that agent's subtree via the crate's recursive lock, so two /// DAGs never interleave container ops on one agent. + /// + /// Nodes that touch the *store or meta repo* rather than the running + /// container do not declare it — `MetaSync`, `Prebuild`, `Provision`, + /// `MetaLock`, `WritePermFile`, `Reparent`. That exemption is what lets a + /// `Prebuild` overlap another DAG's work on the same agent. `Provision` + /// precedes the container's existence entirely, so the lease is first taken + /// at the `Create` it feeds. Agent(String), /// The meta-repo mutation window — a global singleton (default capacity 1) /// held by any node that mutates the meta repo, so two meta mutations never - /// interleave. Replaces the former runtime `meta::exclusive()` mutex: a + /// interleave. That exclusion is load-bearing: a commit landing inside + /// another node's staged `prepare_deploy`→`finalize_deploy` window would + /// sweep the staged `flake.lock` into its own commit and neuter + /// `abort_deploy`. Replaces the former runtime `meta::exclusive()` mutex: a /// `MutexGuard` cannot span scheduler nodes, but a resource held by a /// subtree root *can* — which is what lets the two-phase deploy /// (`prepare_deploy` stages `flake.lock` uncommitted across the whole @@ -30,35 +48,12 @@ pub enum Resource { /// decomposed into sub-nodes instead of one opaque node. Descendants of a /// holder re-enter it through the crate's recursive lock, exactly like /// [`Resource::Agent`]. + /// + /// Deliberately **not** declared by `Prebuild`: the window must stay off the + /// multi-minute toplevel build, which only *reads* the store. Holding a + /// hive-global cap-1 across it would serialize every agent's rebuild behind + /// every other's — which is why the meta preamble is its own `MetaSync` + /// node, and a sibling of `Prebuild` rather than its parent (a resource is + /// held across the holder's whole subtree). MetaWindow, } - -impl NodeKind { - /// The resources this node must acquire to run — `(name, units)` — derived from - /// its kind + agent: a build slot for nix-heavy kinds - /// ([`NodeKind::needs_build_slot`]) and the agent lease for - /// container-affecting kinds ([`NodeKind::needs_lease`]). Lease-exempt - /// container ops (`Start` / `Stop`, fanned out by a lease-holding - /// `Reconcile`) hold no lease of their own — they re-enter the ancestor's - /// `Agent` lock through the crate's recursive re-entrancy. Meta-mutating - /// kinds ([`NodeKind::needs_meta_window`]) additionally take the global - /// [`Resource::MetaWindow`]. - /// - /// All of a node's resource edges are acquired **atomically** - /// (`try_acquire_all`) — a node never holds one resource while waiting on - /// another, so the multi-resource kinds (a `MetaLock` wants a build slot - /// *and* the meta window) cannot deadlock against each other. - pub fn resource_deps(&self) -> Vec<(Resource, u32)> { - let mut deps = Vec::new(); - if self.needs_build_slot() { - deps.push((Resource::BuildSlot, 1)); - } - if self.needs_lease() { - deps.push((Resource::Agent(self.agent().to_owned()), 1)); - } - if self.needs_meta_window() { - deps.push((Resource::MetaWindow, 1)); - } - deps - } -} diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 7aaf1878..f7b301af 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -25,6 +25,7 @@ use std::sync::Arc; use super::model::{DagSpec, NodeKind}; +use super::resource::Resource; use super::templates::{RebuildOpts, node, rebuild_nodes}; use super::{Job, Source, templates}; use crate::coordinator::Coordinator; @@ -70,19 +71,26 @@ fn stop_chain(b: &Job, agent: &str, graceful: bool, running: bool) { agent: a(), up: false, }, - ); + ) + .needs(Resource::Agent(a())); // 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 = node(b, NodeKind::Signal { agent: a() }).part_of(wanted); + let signal = node(b, NodeKind::Signal { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(wanted); let drain = node(b, NodeKind::Drain { agent: a() }) + .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(signal); let _ = node(b, NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) .part_of(wanted) .after_ok(drain); } else { - let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(wanted); + let _ = node(b, NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(wanted); } } @@ -97,7 +105,8 @@ fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) { agent: agent.to_owned(), up: true, }, - ); + ) + .needs(Resource::Agent(agent.to_owned())); if !running && stale { // Rebuild subtree chained behind the `SetWanted` head. `MetaSync`, // `Prebuild` + `Reconcile` are their own group roots (top-level, per @@ -118,6 +127,7 @@ fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) { agent: agent.to_owned(), }, ) + .needs(Resource::Agent(agent.to_owned())) .part_of(wanted); } } @@ -135,7 +145,7 @@ fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) { let a = || agent.to_owned(); if !running { // Nothing to bounce — a lone Reconcile converges to intent. - let _ = node(b, NodeKind::Reconcile { agent: a() }); + let _ = node(b, NodeKind::Reconcile { agent: a() }).needs(Resource::Agent(a())); return; } // Running: mechanical stop then Reconcile. The first stop node is the group @@ -147,17 +157,23 @@ fn restart_chain(b: &Job, 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 = node(b, NodeKind::Signal { agent: a() }); - let drain = node(b, NodeKind::Drain { agent: a() }).part_of(signal); + let signal = node(b, NodeKind::Signal { agent: a() }).needs(Resource::Agent(a())); + let drain = node(b, NodeKind::Drain { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(signal); let stop = node(b, NodeKind::StopForUpdate { agent: a() }) + .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain); let _ = node(b, NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) .part_of(signal) .after_ok(stop); } else { - let stop = node(b, NodeKind::StopForUpdate { agent: a() }); - let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(stop); + let stop = node(b, NodeKind::StopForUpdate { agent: a() }).needs(Resource::Agent(a())); + let _ = node(b, NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(stop); } } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 7d29b946..32c90bdd 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -24,28 +24,29 @@ use hive_jobq::TerminalState; use super::model::{DagSpec, NodeKind, PermPayload, Source}; +use super::resource::Resource; use super::{Declare, Handle, Job}; -/// Declare one node carrying `kind`, with the resources that kind needs. +/// Declare one node carrying `kind`. **Resources are not derived here** — the +/// construction site says what the node holds, with `.needs(…)`. /// -/// The resource declaration is [`NodeKind::resource_deps`] applied at the -/// construction site — a build slot for nix-heavy kinds, the agent lease for -/// container-affecting ones, the global meta window for meta-mutating ones. A -/// node that declares a resource an ancestor already holds re-enters that -/// grant rather than taking a fresh unit, so declaring costs nothing. +/// That is the point rather than an omission. Deriving `(name, units)` from the +/// kind made the declaration a property of the *kind*, so a kind that happened +/// to run under an ancestor holding the resource could get away with declaring +/// nothing — which is precisely how `Start` / `Stop` / `PostSwap` ended up +/// lease-exempt: one construction site fans them out from inside a +/// lease-holding `Reconcile`. The requirement is a property of the node, not of +/// the one DAG shape it is used in today. +/// +/// Declaring a resource an ancestor already holds is free: a descendant +/// re-enters that grant through the crate's recursive lock rather than taking a +/// fresh unit. /// /// The returned handle is where edges and grouping are declared, and is `Copy` /// — naming a node as a dependency does not consume the ability to name it /// again. pub(crate) fn node(b: &Job, kind: NodeKind) -> Handle<'_> { - // Read the resources off the kind before handing it over — the payload is - // moved into the node, not cloned for it. - let resources = kind.resource_deps(); - let mut handle = b.node(kind); - for (name, count) in resources { - handle = handle.needs_units(name, count); - } - handle + b.node(kind) } /// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node @@ -181,32 +182,52 @@ 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 }); + let mut meta_sync = + node(b, 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() }).after_ok(meta_sync); + let prebuild = node(b, 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() }).part_of(prebuild); + let signal = node(b, 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() }).part_of(signal); + let drain = node(b, NodeKind::Drain { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(signal); node(b, NodeKind::StopForUpdate { agent: a() }) + .needs(Resource::Agent(a())) .part_of(signal) .after_ok(drain) } else { - node(b, NodeKind::StopForUpdate { agent: a() }).part_of(prebuild) + node(b, NodeKind::StopForUpdate { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(prebuild) }; - let swap = node(b, NodeKind::Swap { agent: a() }).part_of(stop_for_update); + let swap = node(b, NodeKind::Swap { agent: a() }) + .needs(Resource::BuildSlot) + .needs(Resource::Agent(a())) + .part_of(stop_for_update); + // `PostSwap` declares the lease it actually runs under. It is a child of + // `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() }) + .needs(Resource::Agent(a())) .part_of(stop_for_update) .after_ok(swap); - let reconcile = node(b, NodeKind::Reconcile { agent: a() }).after_any(prebuild); + let reconcile = node(b, NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) + .after_any(prebuild); RebuildRoots { meta_sync, @@ -258,6 +279,7 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { approval_id, }, ) + .needs(Resource::MetaWindow) .after_ok(roots.prebuild) .after_ok(roots.reconcile); }) @@ -335,13 +357,22 @@ pub fn approval_deploy( reason, declare: Box::new(move |b: &Job| { let a = || agent.clone(); + // The window is the widest holder in the tree: it brackets a nix + // build (`BuildSlot`), takes the container down across the swap + // (`Agent`), and serialises the meta mutation its subtree performs + // (`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 { agent: a(), approval_id, }, - ); + ) + .needs(Resource::BuildSlot) + .needs(Resource::Agent(a())) + .needs(Resource::MetaWindow); let verify = node( b, NodeKind::MergeVerify { @@ -390,7 +421,9 @@ pub fn reconcile_only( source, reason, declare: Box::new(move |b: &Job| { - let _reconcile = node(b, NodeKind::Reconcile { agent }); + // 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); }), } } @@ -413,10 +446,16 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec Vec { + let inner = q.lock(); + inner + .sched + .graph() + .node(node_id) + .expect("node exists") + .deps + .iter() + .filter_map(|dep| match dep { + hive_jobq::Dep::Resource { name, .. } => Some(name.clone()), + hive_jobq::Dep::Node { .. } => None, + }) + .collect() +} + fn state_of(q: &JobQueue, dag_id: u64) -> State { // A DAG whose nodes have all settled `Done` or `Skipped` drops out of the // snapshot — absence is the completion signal, so map it to `Done`. @@ -708,6 +730,74 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { ); } +#[test] +fn a_fanned_out_start_declares_the_lease_and_re_enters_its_reconciles_grant() { + // `Start` / `Stop` / `PostSwap` were lease-exempt *as kinds*, which was only + // safe because every construction site fans them out from inside a + // lease-holding ancestor. Now they declare the lease themselves. + // + // The contract says that costs nothing — a descendant re-enters the + // ancestor's grant instead of taking a fresh unit. That is exactly the sort + // of claim that is true until a node is used from a second site, so it is + // pinned here rather than argued: the fanned-out `Start` must (a) actually + // carry the declaration, (b) still run under its parent's grant, and + // (c) not have consumed a second unit of a cap-1 lease. + let q = JobQueue::new(4); + let id = submit( + &q, + templates::reconcile_only("agent-a", Source::Manual, "converge".to_owned()), + ); + // A competing DAG on the same agent, to prove the lease is genuinely held + // (and held *once*) across the fan-out. + let rival = submit( + &q, + templates::reconcile_only("agent-a", Source::Manual, "rival".to_owned()), + ); + + let reconcile = claim_one(&q); + assert_eq!(reconcile.dag_id, id); + assert_eq!(reconcile.kind.as_str(), "reconcile"); + + // What `run_reconcile` does on observing a down container with wanted=Up. + q.append_subgraph( + id, + Box::new(|b: &Job| { + let kind = NodeKind::Start { + agent: "agent-a".to_owned(), + }; + let lease = Resource::Agent(kind.agent().to_owned()); + let _ = templates::node(b, kind).needs(lease); + }), + reconcile.node_id, + ); + q.complete_node(reconcile.node_id, Ok(())); + + // (a) + (b): the child runs, under the parent that parked in `Finishing`. + let start = claim_one(&q); + assert_eq!(start.kind.as_str(), "start"); + assert_eq!( + declared_resources(&q, start.node_id), + vec![Resource::Agent("agent-a".to_owned())], + "a fanned-out Start declares the lease it runs under" + ); + + // (c): one unit, not two. `claim_one` above already asserted the rival did + // not come back in the same pass; make the reason explicit. + assert!( + q.claim_ready().is_empty(), + "the rival DAG's Reconcile must still be blocked — the appended Start \ + borrowed the grant rather than acquiring a second unit" + ); + + q.complete_node(start.node_id, Ok(())); + // Subtree terminal → the grant releases and the rival finally runs. + let rival_reconcile = claim_one(&q); + assert_eq!(rival_reconcile.dag_id, rival); + q.complete_node(rival_reconcile.node_id, Ok(())); + assert_eq!(state_of(&q, id), State::Done); + assert_eq!(state_of(&q, rival), State::Done); +} + #[test] fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild @@ -1594,12 +1684,12 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() { let c = claim_one(&q); assert_eq!(c.kind.as_str(), "reparent"); assert_eq!(c.agent, "", "Reparent is agentless — no per-agent lease"); - assert!( - c.kind.needs_meta_window(), - "a topology commit must hold the same MetaWindow as WritePermFile" + assert_eq!( + declared_resources(&q, c.node_id), + vec![Resource::MetaWindow], + "a topology commit must declare the same MetaWindow as WritePermFile, \ + and nothing else — no lease (agentless), no build slot (no nix work)" ); - assert!(!c.kind.needs_lease()); - assert!(!c.kind.needs_build_slot()); q.complete_node(c.node_id, Ok(())); assert_eq!(state_of(&q, id), State::Done); } diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 3b8e0b43..8c4100ba 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -34,7 +34,7 @@ static META_LOCK: Mutex<()> = Mutex::const_new(()); // this module. `META_LOCK` above serializes individual git ops but cannot // keep another op out of that staged window; that window is owned by the // job queue instead, as `Resource::MetaWindow`, declared by every -// meta-mutating node kind (`NodeKind::needs_meta_window`). A resource can +// meta-mutating node kind (it declares `Resource::MetaWindow`). A resource can // be held by a subtree root across its children, which a `MutexGuard` // (bounded by one executor fn) cannot — that's what lets the deploy be // modelled as sub-nodes rather than one opaque node. From 58a9f218f28006562e8a6808b95b0fae55bb2233 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 16:22:56 +0200 Subject: [PATCH 2/2] 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