diff --git a/docs/coordinator.md b/docs/coordinator.md index cda161e3..dc87418d 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 (it declares `Resource::MetaWindow`). Two meta + phase below it (`NodeKind::needs_meta_window`). 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 95486951..d6f44a37 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -14,7 +14,6 @@ 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}; @@ -169,8 +168,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 declares the agent's lifecycle lease -/// (`Resource::Agent`) so the whole power-op DAG is atomic per-agent. +/// 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. /// 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 @@ -190,7 +189,7 @@ fn run_set_wanted(coord: &Arc, claim: &Claim, up: bool) -> Result, claim: &Claim) -> Result, claim: &Claim) -> Result { let name = &claim.agent; @@ -418,13 +417,8 @@ 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 (it declares `Resource::MetaWindow`): a + // Runs under the deploy window (`NodeKind::needs_meta_window`): 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). @@ -598,7 +592,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 (it declares `Resource::MetaWindow`), same reasoning as +/// the deploy window (`NodeKind::needs_meta_window`), 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 3db89559..9b9db781 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), declared per node at its construction site; +//! subtree-held), derived per node by [`NodeKind::resource_deps`]; //! - 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,13 +436,10 @@ impl JobQueue { /// `Stop` pill are both pills; only one means a vanished container is /// expected. /// - /// 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. + /// 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. #[must_use] pub fn running_transients(&self) -> Vec { let inner = self.lock(); @@ -452,13 +449,14 @@ impl JobQueue { .nodes() .filter(|n| matches!(n.state, State::Running)) .filter_map(|n| { - let agent = n.deps.iter().find_map(|dep| match dep { - hive_jobq::Dep::Resource { - name: Resource::Agent(a), - .. - } => Some(a.clone()), - _ => None, - })?; + let agent = + n.payload + .resource_deps() + .into_iter() + .find_map(|(name, _)| match name { + Resource::Agent(a) => Some(a), + _ => 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 b6a9fe8d..19556271 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -21,8 +21,7 @@ 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 -/// [`Resource`](super::resource::Resource), declared per node where the node -/// is constructed rather than derived from its kind); the +/// [`NodeKind::needs_build_slot`] / [`NodeKind::needs_lease`]); 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 @@ -63,10 +62,8 @@ 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. 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. + /// 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. PostSwap { agent: String }, /// First-spawn pre-create provisioning: proposed/applied repos, /// state subvolume, and meta registration (`sync_agents`). Runs @@ -127,7 +124,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. Declares the meta window, same + /// move spans multiple agents anyway. `needs_meta_window() = true`, 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` @@ -350,6 +347,42 @@ 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. @@ -387,6 +420,45 @@ 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 e86a69c3..a3118137 100644 --- a/hive-c0re/src/job_queue/resource.rs +++ b/hive-c0re/src/job_queue/resource.rs @@ -3,16 +3,8 @@ //! `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). -//! -//! **A node's resources are declared where the node is constructed** with -//! `.needs(…)`, not derived from its kind. 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. + +use super::model::NodeKind; /// The two resource classes the queue gates concurrency on, as the crate's /// generic resource type `R`. @@ -27,20 +19,10 @@ 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. 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 + /// interleave. 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 @@ -48,12 +30,35 @@ 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 6e19c3f6..7aaf1878 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -6,7 +6,7 @@ //! state, which needs an async `lifecycle::is_running` read that a pure/sync //! template can't do. So these fns are async — they read each agent's state, //! assemble a per-agent subgraph out of the shared pure primitives -//! (`Job::node` + `templates::rebuild_nodes`), all declaring into ONE job +//! (`templates::{node, rebuild_nodes}`), all declaring into ONE job //! (independent per-agent roots, concurrent on their own leases). //! //! Dynamic shape rule: `stop`/`start` carry a head `SetWanted(w)` (durable @@ -25,8 +25,7 @@ use std::sync::Arc; use super::model::{DagSpec, NodeKind}; -use super::resource::Resource; -use super::templates::{RebuildOpts, rebuild_nodes}; +use super::templates::{RebuildOpts, node, rebuild_nodes}; use super::{Job, Source, templates}; use crate::coordinator::Coordinator; use crate::lifecycle; @@ -65,34 +64,25 @@ fn stop_chain(b: &Job, agent: &str, graceful: bool, running: bool) { // steps are its children (borrow the lease, run once it reaches `Finishing`, // dep-ordered among themselves). let a = || agent.to_owned(); - let wanted = b - .node(NodeKind::SetWanted { + let wanted = node( + b, + NodeKind::SetWanted { 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 = b - .node(NodeKind::Signal { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(wanted); - let drain = b - .node(NodeKind::Drain { agent: a() }) - .needs(Resource::Agent(a())) + let signal = node(b, NodeKind::Signal { agent: a() }).part_of(wanted); + let drain = node(b, NodeKind::Drain { agent: a() }) .part_of(wanted) .after_ok(signal); - let _ = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) + let _ = node(b, NodeKind::Reconcile { agent: a() }) .part_of(wanted) .after_ok(drain); } else { - let _ = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(wanted); + let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(wanted); } } @@ -101,12 +91,13 @@ fn stop_chain(b: &Job, agent: &str, graceful: bool, running: bool) { /// current derivations), otherwise a plain `Reconcile` (which starts a down /// agent and noops an already-running one). fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) { - let wanted = b - .node(NodeKind::SetWanted { + let wanted = node( + b, + NodeKind::SetWanted { 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 @@ -121,12 +112,13 @@ fn start_chain(b: &Job, agent: &str, running: bool, stale: bool) { Some(wanted), ); } else { - let _ = b - .node(NodeKind::Reconcile { + let _ = node( + b, + NodeKind::Reconcile { agent: agent.to_owned(), - }) - .needs(Resource::Agent(agent.to_owned())) - .part_of(wanted); + }, + ) + .part_of(wanted); } } @@ -143,9 +135,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 _ = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())); + let _ = node(b, NodeKind::Reconcile { agent: a() }); return; } // Running: mechanical stop then Reconcile. The first stop node is the group @@ -157,31 +147,17 @@ 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 = b - .node(NodeKind::Signal { agent: a() }) - .needs(Resource::Agent(a())); - let drain = b - .node(NodeKind::Drain { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(signal); - let stop = b - .node(NodeKind::StopForUpdate { agent: a() }) - .needs(Resource::Agent(a())) + let signal = node(b, NodeKind::Signal { agent: a() }); + let drain = node(b, NodeKind::Drain { agent: a() }).part_of(signal); + let stop = node(b, NodeKind::StopForUpdate { agent: a() }) .part_of(signal) .after_ok(drain); - let _ = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) + let _ = node(b, NodeKind::Reconcile { agent: a() }) .part_of(signal) .after_ok(stop); } else { - let stop = b - .node(NodeKind::StopForUpdate { agent: a() }) - .needs(Resource::Agent(a())); - let _ = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(stop); + let stop = node(b, NodeKind::StopForUpdate { agent: a() }); + let _ = node(b, NodeKind::Reconcile { agent: a() }).part_of(stop); } } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 8607958c..7d29b946 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -1,10 +1,7 @@ //! DAG shape builders — every operation as a template over the shared //! node primitives. Pure (no I/O); each node carries its own `agent` (there is -//! no DAG-level agent) and **declares its own resources** with `.needs(…)` -//! right where it is constructed, rather than having them derived from its -//! kind. Deriving 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. +//! no DAG-level agent), stamped by the [`node`] helper along with the +//! resources that node's kind needs. //! //! ```text //! rebuild(a): MetaSync(a) → Prebuild(a) → StopForUpdate(a) → Swap(a) →(ok) PostSwap(a) →(any) Reconcile(a) @@ -22,14 +19,35 @@ //! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here: //! their per-agent shape depends on live running state (an async //! `lifecycle::is_running` read), so `submit.rs` assembles them out of the -//! primitives this module exports ([`rebuild_nodes`]) over `Job::node`. +//! primitives this module exports ([`node`], [`rebuild_nodes`]). 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. +/// +/// 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. +/// +/// 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 +} + /// 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. @@ -38,10 +56,13 @@ use super::{Declare, Handle, Job}; /// dropped — see [`hive_jobq::NodeRef::on_elimination_of`]. fn emit_rebuilt_tails(b: &Job, agent: &str, roots: &[Handle<'_>]) { let ok = roots.iter().fold( - b.node(NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: true, - }), + node( + b, + 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 @@ -51,10 +72,13 @@ 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( - b.node(NodeKind::EmitRebuilt { - agent: agent.to_owned(), - ok: false, - }) + node( + b, + NodeKind::EmitRebuilt { + agent: agent.to_owned(), + ok: false, + }, + ) .on_elimination_of(ok), hive_jobq::NodeRef::after_any, ); @@ -71,12 +95,14 @@ fn resolve_approval_tails(b: &Job, approval_id: i64, root: Handle<'_>) { TerminalState::Failed, TerminalState::Cancelled, ] { - let _ = b - .node(NodeKind::ResolveApproval { + let _ = node( + b, + NodeKind::ResolveApproval { approval_id, outcome, - }) - .on_outcome(root, &[outcome]); + }, + ) + .on_outcome(root, &[outcome]); } } @@ -155,59 +181,32 @@ pub(crate) fn rebuild_nodes<'a>( let a = || agent.to_owned(); let RebuildOpts { relock, graceful } = opts; - let mut meta_sync = b - .node(NodeKind::MetaSync { agent: a(), relock }) - .needs(Resource::MetaWindow); + let mut meta_sync = node(b, NodeKind::MetaSync { agent: a(), relock }); if let Some(after) = after { meta_sync = meta_sync.after_ok(after); } - let prebuild = b - .node(NodeKind::Prebuild { agent: a() }) - .needs(Resource::BuildSlot) - .after_ok(meta_sync); + let prebuild = node(b, NodeKind::Prebuild { agent: a() }).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 = b - .node(NodeKind::Signal { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(prebuild); + let signal = node(b, NodeKind::Signal { 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 - .node(NodeKind::Drain { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(signal); - b.node(NodeKind::StopForUpdate { agent: a() }) - .needs(Resource::Agent(a())) + let drain = node(b, NodeKind::Drain { agent: a() }).part_of(signal); + node(b, NodeKind::StopForUpdate { agent: a() }) .part_of(signal) .after_ok(drain) } else { - b.node(NodeKind::StopForUpdate { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(prebuild) + node(b, NodeKind::StopForUpdate { agent: a() }).part_of(prebuild) }; - let swap = b - .node(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 = b - .node(NodeKind::PostSwap { agent: a() }) - .needs(Resource::Agent(a())) + let swap = node(b, NodeKind::Swap { agent: a() }).part_of(stop_for_update); + let _post_swap = node(b, NodeKind::PostSwap { agent: a() }) .part_of(stop_for_update) .after_ok(swap); - let reconcile = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) - .after_any(prebuild); + let reconcile = node(b, NodeKind::Reconcile { agent: a() }).after_any(prebuild); RebuildRoots { meta_sync, @@ -252,14 +251,15 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare { }, None, ); - let _finalize = b - .node(NodeKind::FinalizeDeploy { + let _finalize = node( + b, + NodeKind::FinalizeDeploy { agent: agent.clone(), approval_id, - }) - .needs(Resource::MetaWindow) - .after_ok(roots.prebuild) - .after_ok(roots.reconcile); + }, + ) + .after_ok(roots.prebuild) + .after_ok(roots.reconcile); }) } @@ -335,40 +335,39 @@ 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 = b - .node(NodeKind::DeployWindow { + let window = node( + b, + NodeKind::DeployWindow { agent: a(), approval_id, - }) - .needs(Resource::BuildSlot) - .needs(Resource::Agent(a())) - .needs(Resource::MetaWindow); - let verify = b - .node(NodeKind::MergeVerify { + }, + ); + let verify = node( + b, + NodeKind::MergeVerify { agent: a(), approval_id, - }) - .part_of(window); - let apply = b - .node(NodeKind::DeployApply { + }, + ) + .part_of(window); + let apply = node( + b, + NodeKind::DeployApply { agent: a(), approval_id, - }) - .part_of(window) - .after_ok(verify); - let _tail = b - .node(NodeKind::DeployTail { + }, + ) + .part_of(window) + .after_ok(verify); + let _tail = node( + b, + 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); }), @@ -391,9 +390,7 @@ pub fn reconcile_only( source, reason, 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 = b.node(NodeKind::Reconcile { agent }).needs(lease); + let _reconcile = node(b, NodeKind::Reconcile { agent }); }), } } @@ -416,21 +413,10 @@ 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`. @@ -730,125 +708,6 @@ 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 _ = b.node(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 boot_sweep_nodes_declare_their_own_resources() { - // Regression, and the reason it needs its own test: `workers::auto_update` - // is the only place job nodes are constructed *outside* `job_queue/`, so - // nothing in this module covered it. When resource derivation moved to the - // construction sites, this path was missed and both kinds silently declared - // nothing — dropping the agent lease a boot `Reconcile` needs to not race - // another DAG's container ops, and letting the sweep `MetaLock` land its - // meta commit inside another node's staged deploy window. Nothing failed to - // compile; only an exhaustive caller list would have caught it. - let q = JobQueue::new(4); - let _id = submit( - &q, - DagSpec { - source: Source::AutoUpdate, - reason: "boot".to_owned(), - declare: Box::new(|b: &Job| { - crate::workers::auto_update::boot_nodes( - b, - true, - vec!["stale-agent".to_owned()], - vec!["drifted-agent".to_owned()], - ); - }), - }, - ); - - // Both are independent roots on disjoint resources, so both start at once. - let claims = q.claim_ready(); - let by_kind = |kind: &str| { - claims - .iter() - .find(|c| c.kind.as_str() == kind) - .unwrap_or_else(|| panic!("no {kind} claim in {claims:?}")) - }; - - let mut lock = declared_resources(&q, by_kind("meta_lock").node_id); - lock.sort_by_key(|r| format!("{r:?}")); - assert_eq!( - lock, - vec![Resource::BuildSlot, Resource::MetaWindow], - "the sweep MetaLock runs a nix lock bump and commits to meta" - ); - - assert_eq!( - declared_resources(&q, by_kind("reconcile").node_id), - vec![Resource::Agent("drifted-agent".to_owned())], - "a boot Reconcile touches the container, so it holds that agent's lease" - ); -} - #[test] fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { // The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild @@ -859,11 +718,14 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() { source: Source::AutoUpdate, reason: "sweep".to_owned(), declare: Box::new(|b: &Job| { - let _lock = b.node(NodeKind::MetaLock { - sweep: true, - fanout: None, - inputs: Vec::new(), - }); + let _lock = templates::node( + b, + NodeKind::MetaLock { + sweep: true, + fanout: None, + inputs: Vec::new(), + }, + ); }), }; let id = submit(&q, spec); @@ -1732,12 +1594,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_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_meta_window(), + "a topology commit must hold the same MetaWindow as WritePermFile" ); + 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 8c4100ba..3b8e0b43 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 (it declares `Resource::MetaWindow`). A resource can +// meta-mutating node kind (`NodeKind::needs_meta_window`). 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. diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 572b455b..7753c7f5 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -318,49 +318,6 @@ pub async fn run(coord: Arc) -> 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 @@ -377,7 +334,7 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, Source}; + use crate::job_queue::{DagSpec, NodeKind, Source, templates}; // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. if !any_stale && drifted.is_empty() { @@ -391,8 +348,29 @@ fn submit_boot_tree( n_skipped, ); - let declare: crate::job_queue::Declare = - Box::new(move |b| boot_nodes(b, any_stale, fanout, drifted)); + let declare: crate::job_queue::Declare = Box::new(move |b| { + // Sweep whenever ANY marker is stale — even when every stale agent is + // wanted-offline: the hyperhive lock bump must land now so their later + // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock + // ⇒ no meta commit on a no-change boot. The `fanout` list rides the + // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. + if any_stale { + let _ = templates::node( + b, + NodeKind::MetaLock { + sweep: true, + fanout: Some(fanout), + // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), + // so it names no inputs. + inputs: Vec::new(), + }, + ); + } + // One boot Reconcile per drifted agent — independent roots. + for name in drifted { + let _ = templates::node(b, NodeKind::Reconcile { agent: name }); + } + }); let spec = DagSpec { // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they