From 6899f574f6faf093573f3515700762274df93a26 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 3 Aug 2026 01:23:40 +0200 Subject: [PATCH] jobq: templates swallow the DagSpec layer DagSpec described the graph the templates were about to build, one layer below the templates themselves. Per #2972 the templates should be that unit, so the spec type is gone and every declarer writes onto the job builder directly. - delete DagSpec and its hand-written Debug impl - submit(source, reason, declare: impl FnOnce(&Job)) replaces the pre-built-spec signature; submit_and_emit follows - all six templates take &Job; the Source is now the caller's to pass, which spawn and approval_deploy previously hardcoded while the other four did not - power_dag dissolves into stop_nodes/start_nodes/restart_nodes, which borrow their targets instead of owning them 315 tests pass unchanged. --- hive-c0re/src/actions.rs | 29 +-- hive-c0re/src/job_queue/mod.rs | 23 +- hive-c0re/src/job_queue/model.rs | 41 --- hive-c0re/src/job_queue/submit.rs | 135 ++++------ hive-c0re/src/job_queue/templates.rs | 283 +++++++++------------ hive-c0re/src/job_queue/tests.rs | 356 +++++++++++---------------- hive-c0re/src/workers/auto_update.rs | 30 +-- 7 files changed, 347 insertions(+), 550 deletions(-) diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 62c04aaa..61ee3618 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -55,14 +55,11 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { // nothing). let inputs: Vec = serde_json::from_str(&approval.commit_ref).unwrap_or_default(); - let submitted = coord - .job_queue - .submit(crate::job_queue::templates::meta_update( - inputs, - crate::job_queue::Source::Approval, - format!("approval #{id} meta input update"), - Some(id), - )); + let submitted = coord.job_queue.submit( + crate::job_queue::Source::Approval, + format!("approval #{id} meta input update"), + |b| crate::job_queue::templates::meta_update(b, inputs, Some(id)), + ); if let Err(e) = submitted { return Err(e.context("submit meta-update dag")); } @@ -78,11 +75,11 @@ pub async fn approve(coord: Arc, id: i64) -> Result<()> { { tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed"); } - let submitted = coord.job_queue.submit(crate::job_queue::templates::spawn( - approval.agent.as_str(), - id, + let submitted = coord.job_queue.submit( + crate::job_queue::Source::Approval, format!("approval #{id} spawn"), - )); + |b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id), + ); if let Err(e) = submitted { return Err(e.context("submit spawn dag")); } @@ -132,11 +129,9 @@ fn enqueue_approval_rebuild( ) { if let Err(e) = coord .job_queue - .submit(crate::job_queue::templates::approval_deploy( - agent, - approval_id, - reason, - )) + .submit(crate::job_queue::Source::Approval, reason, |b| { + crate::job_queue::templates::approval_deploy(b, agent, approval_id); + }) { tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed"); } diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 561d8c67..4e2a7930 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -47,7 +47,7 @@ use hive_jobq_wire::{GraphNode, GraphWire}; use tokio::sync::Notify; pub use hive_jobq::TerminalState; -pub use model::{DagSpec, DagView, NodeKind, PermPayload, Source, State}; +pub use model::{DagView, NodeKind, PermPayload, Source, State}; use resource::Resource; /// A job under construction: `hive_jobq`'s builder over this queue's payload @@ -208,27 +208,34 @@ impl JobQueue { /// here completes it by hand — a node with no work of its own still goes /// the way every other node goes. /// - /// Takes the spec's recipe by generic, not as a boxed closure: a spec - /// travels from the template that built it directly into this call, so - /// there is nothing to allocate for. + /// `source` and `reason` are the container node's own payload — they are + /// arguments here rather than fields of a spec struct because that is all + /// they ever were. `declare` is the recipe, taken by generic and run + /// against a builder `hive_jobq` owns: it goes from the template straight + /// into this call, so there is nothing to allocate for. /// /// # Errors /// Propagates a graph-insert error (dependencies that aren't /// dependency-topological). - pub fn submit(&self, spec: DagSpec) -> anyhow::Result { + pub fn submit( + &self, + source: Source, + reason: String, + declare: impl FnOnce(&Job), + ) -> anyhow::Result { let mut inner = self.lock(); let container = inner .append( NodeKind::Dag { - source: spec.source, - reason: spec.reason, + source, + reason, created_at: Utc::now(), }, Vec::new(), None, ) .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; - insert_group(&mut inner, spec.declare, Some(container))?; + insert_group(&mut inner, declare, Some(container))?; drop(inner); self.notify.notify_one(); Ok(container.get()) diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 263ccf2f..e783c05c 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -426,44 +426,3 @@ impl NodeKind { // - `DeployWindow` brackets a deploy without itself stopping anything. } } - -/// Submit-time spec for a whole DAG: the group's metadata plus the declared — -/// not yet inserted — nodes. Built by `templates.rs`, inserted by -/// `JobQueue::submit`. -/// -/// No DAG-level `agent` — every node carries its own (a DAG can span agents), -/// and the queue derives per-agent leasing from [`NodeKind::agent`]. -/// Type-specific payloads (`PermChange`'s file payload) ride the node that -/// consumes them ([`NodeKind::WritePermFile`]), not this generic spec. -/// -/// There is no separate per-node spec type, and no built job either: `declare` -/// is a *recipe* the queue runs against a builder `hive_jobq` owns, at the -/// moment it inserts. A shape that has been declared is therefore always -/// insertable — a dangling edge or a cycle cannot be expressed, so there is -/// nothing left for a submit-time validation pass to reject. -/// -/// Generic over the recipe rather than boxing it: a spec goes from the template -/// that returns it straight to the `submit` that consumes it, so the closure's -/// concrete type is known the whole way and needs neither an allocation nor a -/// `Send` bound. Nothing boxes a recipe any more — a running node grows its DAG -/// by declaring straight onto the builder it was handed, so there is no recipe -/// to store and replay across a task boundary. -pub struct DagSpec { - pub source: Source, - /// Free-form "why". - pub reason: String, - /// Declares the DAG's nodes — their edges, grouping and resources — onto - /// the builder the queue hands it. - pub declare: F, -} - -impl std::fmt::Debug for DagSpec { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - // The recipe is a closure; there is nothing to show of it, and its - // nodes do not exist until the queue runs it. - f.debug_struct("DagSpec") - .field("source", &self.source) - .field("reason", &self.reason) - .finish_non_exhaustive() - } -} diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 6e19c3f6..85fff967 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -24,18 +24,23 @@ use std::sync::Arc; -use super::model::{DagSpec, NodeKind}; +use super::model::NodeKind; use super::resource::Resource; use super::templates::{RebuildOpts, rebuild_nodes}; use super::{Job, Source, templates}; use crate::coordinator::Coordinator; use crate::lifecycle; -fn submit_and_emit(coord: &Arc, spec: super::DagSpec) -> u64 { +fn submit_and_emit( + coord: &Arc, + source: Source, + reason: String, + declare: impl FnOnce(&Job), +) -> u64 { let id = coord .job_queue - .submit(spec) - .expect("template-built dag specs are acyclic"); + .submit(source, reason, declare) + .expect("template-declared shapes are acyclic"); coord.emit_rebuild_queue_snapshot(); id } @@ -44,7 +49,9 @@ fn submit_and_emit(coord: &Arc, spec: super::DagSp /// meta input — the meta-update cascade grows its own rebuild subgraphs /// in-DAG instead of going through this surface). pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - submit_and_emit(coord, templates::rebuild(agent, source, reason, true)) + submit_and_emit(coord, source, reason, |b| { + templates::rebuild(b, agent, true); + }) } // ---- dynamic power-op DAG assembly ---------------------------------------- @@ -185,44 +192,24 @@ fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) { } } -/// Wrap the per-agent subgraphs in a `DagSpec`. No tail node: a power op's -/// effect is its nodes (`SetWanted` + `Reconcile`), with nothing left to do once -/// they settle. -/// -/// There is no concatenation step: every chain declares into the same builder -/// and each keeps its own root, so the per-agent subgraphs are independent and -/// run concurrently, each on its own lease. Rebasing one subgraph's indices -/// onto another's used to be a function. -fn power_dag(source: Source, reason: String, declare: F) -> DagSpec { - DagSpec { - source, - reason, - declare, - } -} - -// The `*_spec` builders below are the PURE core the async `*_many` fns call +// The `*_nodes` declarers below are the PURE core the async `*_many` fns call // after reading live state — they take the per-agent running (and stale) // flags explicitly, so unit tests exercise the online/offline shapes without -// a live container. `*_many` = gather state + call `*_spec` + submit. +// a live container. `*_many` = gather state + declare + submit. +// +// A power op has no tail node: its effect is its nodes (`SetWanted` + +// `Reconcile`), with nothing left to do once they settle. +// +// There is no concatenation step either: every chain declares into the same +// builder and each keeps its own root, so the per-agent subgraphs are +// independent and run concurrently, each on its own lease. Rebasing one +// subgraph's indices onto another's used to be a function. -/// Assemble the stop DAG from explicit `(agent, running)` targets. -pub(crate) fn stop_spec( - targets: &[(String, bool)], - graceful: bool, - source: Source, - reason: String, -) -> DagSpec> { - let targets = targets.to_vec(); - power_dag( - source, - reason, - Box::new(move |b: &Job| { - for (agent, running) in targets { - stop_chain(b, &agent, graceful, running); - } - }), - ) +/// Declare the stop DAG from explicit `(agent, running)` targets. +pub(crate) fn stop_nodes(b: &Job, targets: &[(String, bool)], graceful: bool) { + for (agent, running) in targets { + stop_chain(b, agent, graceful, *running); + } } /// Assemble the start DAG from explicit `(agent, running, stale)` targets. @@ -231,40 +218,17 @@ pub(crate) fn stop_spec( /// running under its lease, so a down+stale agent that grew a rebuild subgraph /// reports `rebuilding` during its swap and `starting` at its reconcile, /// without the DAG having to guess one label covering every target. -pub(crate) fn start_spec( - targets: &[(String, bool, bool)], - source: Source, - reason: String, -) -> DagSpec> { - let targets = targets.to_vec(); - power_dag( - source, - reason, - Box::new(move |b: &Job| { - for (agent, running, stale) in targets { - start_chain(b, &agent, running, stale); - } - }), - ) +pub(crate) fn start_nodes(b: &Job, targets: &[(String, bool, bool)]) { + for (agent, running, stale) in targets { + start_chain(b, agent, *running, *stale); + } } -/// Assemble the restart DAG from explicit `(agent, running)` targets. -pub(crate) fn restart_spec( - targets: &[(String, bool)], - graceful: bool, - source: Source, - reason: String, -) -> DagSpec> { - let targets = targets.to_vec(); - power_dag( - source, - reason, - Box::new(move |b: &Job| { - for (agent, running) in targets { - restart_chain(b, &agent, graceful, running); - } - }), - ) +/// Declare the restart DAG from explicit `(agent, running)` targets. +pub(crate) fn restart_nodes(b: &Job, targets: &[(String, bool)], graceful: bool) { + for (agent, running) in targets { + restart_chain(b, agent, graceful, *running); + } } /// Restart a single agent. Thin wrapper over [`restart_many`]. @@ -302,7 +266,9 @@ pub async fn restart_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, restart_spec(&targets, graceful, source, reason)) + submit_and_emit(coord, source, reason, |b| { + restart_nodes(b, &targets, graceful); + }) } /// Start a single agent. Thin wrapper over [`start_many`]. @@ -335,7 +301,9 @@ pub async fn start_many( } targets.push((agent.clone(), running, stale)); } - submit_and_emit(coord, start_spec(&targets, source, reason)) + submit_and_emit(coord, source, reason, |b| { + start_nodes(b, &targets); + }) } /// Hard stop a single agent. Thin wrapper over [`stop_many`]. @@ -371,7 +339,9 @@ pub async fn stop_many( for agent in agents { targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, stop_spec(&targets, graceful, source, reason)) + submit_and_emit(coord, source, reason, |b| { + stop_nodes(b, &targets, graceful); + }) } /// Perm change: commit the JSON file(s) then rebuild. @@ -382,10 +352,9 @@ pub fn perm_change( reason: String, payload: super::PermPayload, ) -> u64 { - submit_and_emit( - coord, - templates::perm_change(agent, source, reason, payload), - ) + submit_and_emit(coord, source, reason, |b| { + templates::perm_change(b, agent, payload); + }) } /// Meta-input lock bump; cascade rebuilds fan out on completion. @@ -395,7 +364,9 @@ pub fn meta_update( source: Source, reason: String, ) -> u64 { - submit_and_emit(coord, templates::meta_update(inputs, source, reason, None)) + submit_and_emit(coord, source, reason, |b| { + templates::meta_update(b, inputs, None); + }) } /// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs — @@ -411,5 +382,7 @@ pub fn reparent( source: Source, reason: String, ) -> u64 { - submit_and_emit(coord, templates::reparent(moves, source, reason)) + submit_and_emit(coord, source, reason, |b| { + templates::reparent(b, moves); + }) } diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 198b06bf..dc7d4027 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -26,7 +26,7 @@ use hive_jobq::TerminalState; -use super::model::{DagSpec, NodeKind, PermPayload, Source}; +use super::model::{NodeKind, PermPayload}; use super::resource::Resource; use super::{Handle, Job}; @@ -304,29 +304,17 @@ pub(crate) fn deploy_rebuild_nodes(b: &Job, agent: &str, approval_id: i64) { /// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every /// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// it reaches `Done` even after a failed swap and the tail would report success. -pub fn rebuild( - agent: &str, - source: Source, - reason: String, - relock: bool, -) -> DagSpec> { - let agent = agent.to_owned(); - DagSpec { - source, - reason, - declare: Box::new(move |b: &Job| { - let roots = rebuild_nodes( - b, - &agent, - RebuildOpts { - relock, - graceful: false, - }, - None, - ); - emit_rebuilt_tails(b, &agent, &roots.all()); - }), - } +pub fn rebuild(b: &Job, agent: &str, relock: bool) { + let roots = rebuild_nodes( + b, + agent, + RebuildOpts { + relock, + graceful: false, + }, + None, + ); + emit_rebuilt_tails(b, agent, &roots.all()); } /// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the @@ -354,55 +342,44 @@ pub fn rebuild( /// /// The window still spans the container build, as it must: `prepare_deploy` /// leaves `flake.lock` staged-uncommitted for the build's whole duration. -pub fn approval_deploy( - agent: &str, - approval_id: i64, - reason: String, -) -> DagSpec> { - let agent = agent.to_owned(); - DagSpec { - source: Source::Approval, - 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 { - agent: a(), - approval_id, - }) - .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 = b - .node(NodeKind::DeployApply { - agent: a(), - approval_id, - }) - .part_of(window) - .after_ok(verify); - let _tail = b - .node(NodeKind::DeployTail { - agent: a(), - approval_id, - }) - .part_of(window) - .after_any(apply); +pub fn approval_deploy(b: &Job, agent: &str, approval_id: i64) { + let a = || agent.to_owned(); + // The window is the widest holder in the tree: it brackets a nix + // build (`BuildSlot`), takes the container down across the swap + // (`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 { + agent: a(), + approval_id, + }) + .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 = b + .node(NodeKind::DeployApply { + agent: a(), + approval_id, + }) + .part_of(window) + .after_ok(verify); + let _tail = b + .node(NodeKind::DeployTail { + agent: a(), + approval_id, + }) + .part_of(window) + .after_any(apply); - resolve_approval_tails(b, approval_id, window); - }), - } + resolve_approval_tails(b, approval_id, window); } /// First-deploy spawn (approval-driven): `Provision` (proposed/applied @@ -416,34 +393,27 @@ pub fn approval_deploy( /// container was never created). Closed by a `ResolveApproval` tail root edged /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up /// already carries the whole cascade. -pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec> { - let agent = agent.to_owned(); - DagSpec { - source: Source::Approval, - reason, - declare: Box::new(move |b: &Job| { - let a = || agent.clone(); - let provision = b - .node(NodeKind::Provision { agent: a() }) - .needs(Resource::MetaWindow); - let create = b - .node(NodeKind::Create { agent: a() }) - .needs(Resource::BuildSlot) - .needs(Resource::Agent(a())) - .part_of(provision); - let dropin = b - .node(NodeKind::WriteDropin { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(create); - let _reconcile = b - .node(NodeKind::Reconcile { agent: a() }) - .needs(Resource::Agent(a())) - .part_of(create) - .after_ok(dropin); +pub fn spawn(b: &Job, agent: &str, approval_id: i64) { + let a = || agent.to_owned(); + let provision = b + .node(NodeKind::Provision { agent: a() }) + .needs(Resource::MetaWindow); + let create = b + .node(NodeKind::Create { agent: a() }) + .needs(Resource::BuildSlot) + .needs(Resource::Agent(a())) + .part_of(provision); + let dropin = b + .node(NodeKind::WriteDropin { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(create); + let _reconcile = b + .node(NodeKind::Reconcile { agent: a() }) + .needs(Resource::Agent(a())) + .part_of(create) + .after_ok(dropin); - resolve_approval_tails(b, approval_id, provision); - }), - } + resolve_approval_tails(b, approval_id, provision); } /// Perm change: commit the JSON file(s), then the rebuild subgraph so @@ -451,39 +421,27 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec DagSpec> { - let agent = agent.to_owned(); - DagSpec { - source, - reason, - declare: Box::new(move |b: &Job| { - let write = b - .node(NodeKind::WritePermFile { - agent: agent.clone(), - payload, - }) - .needs(Resource::MetaWindow); - let roots = rebuild_nodes( - b, - &agent, - RebuildOpts { - relock: true, - graceful: false, - }, - Some(write), - ); - emit_rebuilt_tails( - b, - &agent, - &[write, roots.meta_sync, roots.prebuild, roots.reconcile], - ); - }), - } +pub fn perm_change(b: &Job, agent: &str, payload: PermPayload) { + let write = b + .node(NodeKind::WritePermFile { + agent: agent.to_owned(), + payload, + }) + .needs(Resource::MetaWindow); + let roots = rebuild_nodes( + b, + agent, + RebuildOpts { + relock: true, + graceful: false, + }, + Some(write), + ); + emit_rebuilt_tails( + b, + agent, + &[write, roots.meta_sync, roots.prebuild, roots.reconcile], + ); } /// Meta-input lock bump. The `MetaLock` executor grows one rebuild subgraph @@ -496,33 +454,22 @@ pub fn perm_change( /// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent /// crash-watch suppression during its `Swap` — the property the old child /// `Rebuild` DAGs carried via their own transient. -pub fn meta_update( - inputs: Vec, - source: Source, - reason: String, - approval_id: Option, -) -> DagSpec> { - DagSpec { - source, - reason, - declare: Box::new(move |b: &Job| { - let lock = b - .node(NodeKind::MetaLock { - sweep: false, - fanout: None, - inputs, - }) - .needs(Resource::BuildSlot) - .needs(Resource::MetaWindow); - // The bump itself has no side effect, so an operator-driven one ends - // at the `MetaLock`; an approval-driven one still has its row to - // resolve and gets the per-outcome tails edged onto that single - // group-root — whose roll-up covers the rebuild subgraphs `MetaLock` - // grows into itself. - if let Some(approval_id) = approval_id { - resolve_approval_tails(b, approval_id, lock); - } - }), +pub fn meta_update(b: &Job, inputs: Vec, approval_id: Option) { + let lock = b + .node(NodeKind::MetaLock { + sweep: false, + fanout: None, + inputs, + }) + .needs(Resource::BuildSlot) + .needs(Resource::MetaWindow); + // The bump itself has no side effect, so an operator-driven one ends + // at the `MetaLock`; an approval-driven one still has its row to + // resolve and gets the per-outcome tails edged onto that single + // group-root — whose roll-up covers the rebuild subgraphs `MetaLock` + // grows into itself. + if let Some(approval_id) = approval_id { + resolve_approval_tails(b, approval_id, lock); } } @@ -535,20 +482,10 @@ pub fn meta_update( /// checks), so a parent move needs no container rebuild to take effect. /// No transient pill either — the node is agentless (no lease to hang one /// off of) and near-instant. No tail node: the write is the whole effect. -pub fn reparent( - moves: Vec<(hive_types::Ident, Option)>, - source: Source, - reason: String, -) -> DagSpec> { - DagSpec { - source, - reason, - declare: Box::new(move |b: &Job| { - let _reparent = b - .node(NodeKind::Reparent { moves }) - .needs(Resource::MetaWindow); - }), - } +pub fn reparent(b: &Job, moves: Vec<(hive_types::Ident, Option)>) { + let _reparent = b + .node(NodeKind::Reparent { moves }) + .needs(Resource::MetaWindow); } // The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree` diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 5ef2a0be..6abc30b5 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -17,40 +17,36 @@ use super::model::NodeKind; use super::*; -fn submit(q: &JobQueue, spec: DagSpec) -> u64 { - q.submit(spec).expect("valid spec") +/// Submit a declared shape with the metadata every mechanics test uses. +/// `Source::Manual` because none of these exercise provenance — the tests that +/// do name their own source at the call site. +fn submit(q: &JobQueue, reason: &str, declare: impl FnOnce(&Job)) -> u64 { + q.submit(Source::Manual, reason.to_owned(), declare) + .expect("valid shape") } fn ident(s: &str) -> hive_types::Ident { hive_types::Ident::parse(s).expect("valid test ident") } -fn rebuild(agent: &str, reason: &str) -> DagSpec> { - templates::rebuild(agent, Source::Manual, reason.to_owned(), true) +fn rebuild(b: &Job, agent: &str) { + templates::rebuild(b, agent, true); } -/// Restart DAG spec with every agent treated as **running** — the online +/// Restart shape with every agent treated as **running** — the online /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head) /// most queue-mechanics tests assume. Mirrors the pre-dynamic -/// `templates::restart` (which is now the state-aware `submit::restart_spec`). -fn restart_online( - agents: &[&str], - graceful: bool, - reason: &str, -) -> DagSpec> { +/// `templates::restart` (which is now the state-aware `submit::restart_nodes`). +fn restart_online(b: &Job, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned()) + submit::restart_nodes(b, &targets, graceful); } -/// Stop DAG spec with every agent treated as **running** — the online shape +/// Stop shape with every agent treated as **running** — the online shape /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). -fn stop_online( - agents: &[&str], - graceful: bool, - reason: &str, -) -> DagSpec> { +fn stop_online(b: &Job, agents: &[&str], graceful: bool) { let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); - submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned()) + submit::stop_nodes(b, &targets, graceful); } // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type @@ -303,9 +299,9 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State { #[test] fn submit_assigns_distinct_ids() { let q = JobQueue::new(1); - let a = submit(&q, rebuild("agent-a", "first")); - let b = submit(&q, rebuild("agent-b", "second")); - assert_ne!(a, b); + let first = submit(&q, "first", |b| rebuild(b, "agent-a")); + let second = submit(&q, "second", |b| rebuild(b, "agent-b")); + assert_ne!(first, second); assert_eq!(q.snapshot().len(), 2); } @@ -317,20 +313,20 @@ fn submit_assigns_distinct_ids() { #[test] fn identical_resubmit_is_a_distinct_dag() { let q = JobQueue::new(1); - let a = submit(&q, rebuild("agent-a", "first")); - let b = submit(&q, rebuild("agent-a", "again")); - assert_ne!(a, b, "no dedup: identical resubmit is a new DAG"); + let first = submit(&q, "first", |b| rebuild(b, "agent-a")); + let resubmit = submit(&q, "again", |b| rebuild(b, "agent-a")); + assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG"); assert_eq!(q.snapshot().len(), 2); } #[test] fn distinct_submits_never_collapse() { let q = JobQueue::new(1); - let a = submit(&q, rebuild("agent-a", "r")); - let b = submit(&q, rebuild("agent-b", "r")); - let c = submit(&q, restart_online(&["agent-a"], false, "r")); - assert_ne!(a, b); - assert_ne!(a, c); + let rebuild_a = submit(&q, "r", |b| rebuild(b, "agent-a")); + let rebuild_b = submit(&q, "r", |b| rebuild(b, "agent-b")); + let restart_a = submit(&q, "r", |b| restart_online(b, &["agent-a"], false)); + assert_ne!(rebuild_a, rebuild_b); + assert_ne!(rebuild_a, restart_a); assert_eq!(q.snapshot().len(), 3); } @@ -346,8 +342,8 @@ fn resubmit_while_running_is_new_dag() { // swallowed" is the scenario people worry about, and a reader looking for // it should find it. let q = JobQueue::new(1); - let a = submit(&q, rebuild("agent-a", "first")); - let again = submit(&q, rebuild("agent-a", "config bumped during build")); + let a = submit(&q, "first", |b| rebuild(b, "agent-a")); + let again = submit(&q, "config bumped during build", |b| rebuild(b, "agent-a")); assert_ne!(a, again); assert_eq!(q.snapshot().len(), 2); } @@ -383,7 +379,7 @@ fn rebuild_chain_is_declared_serial() { // logic"). Both axes are asserted below because a template can break either // one independently. let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); + let id = submit(&q, "r", |b| rebuild(b, "agent-a")); assert_eq!( declared_shape(&q, id), vec![ @@ -429,24 +425,19 @@ fn rebuild_chain_is_declared_serial() { #[test] fn graceful_rebuild_chain_drains_before_stopping() { let q = JobQueue::new(1); - let id = submit( - &q, - DagSpec { - source: Source::AutoUpdate, - reason: "sweep".to_owned(), - declare: Box::new(|b: &Job| { - templates::rebuild_nodes( - b, - "agent-a", - templates::RebuildOpts { - relock: true, - graceful: true, - }, - None, - ); - }), - }, - ); + let id = q + .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| { + templates::rebuild_nodes( + b, + "agent-a", + templates::RebuildOpts { + relock: true, + graceful: true, + }, + None, + ); + }) + .expect("valid shape"); assert_eq!( declared_shape(&q, id) .iter() @@ -477,24 +468,17 @@ fn non_graceful_rebuild_has_no_signal_or_drain() { // job keeps its nodes to itself and inserts them, so what it built is // observable where it matters — in what the scheduler runs. let q = JobQueue::new(1); - let id = submit( - &q, - DagSpec { - source: Source::Manual, - reason: "manual".to_owned(), - declare: Box::new(|b: &Job| { - templates::rebuild_nodes( - b, - "agent-a", - templates::RebuildOpts { - relock: true, - graceful: false, - }, - None, - ); - }), - }, - ); + let id = submit(&q, "manual", |b| { + templates::rebuild_nodes( + b, + "agent-a", + templates::RebuildOpts { + relock: true, + graceful: false, + }, + None, + ); + }); assert_eq!( declared_shape(&q, id) .iter() @@ -573,7 +557,7 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() { // a resource unit is held for the acquirer's whole subtree, so the slot // `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it. let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); + let id = submit(&q, "r", |b| rebuild(b, "agent-a")); let res = |kind: &str| declared_resources(&q, node_of(&q, id, kind)); let agent = || Resource::Agent("agent-a".to_owned()); @@ -610,10 +594,9 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() { #[test] fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4); - let id = submit( - &q, - restart_online(&["agent-a", "agent-b"], false, "hive-wide"), - ); + let id = submit(&q, "hive-wide", |b| { + restart_online(b, &["agent-a", "agent-b"], false); + }); // A hive-wide restart is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); // Each agent's subgraph head (StopForUpdate, since both are running) is a @@ -647,10 +630,9 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { #[test] fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4); - let id = submit( - &q, - stop_online(&["agent-a", "agent-b"], false, "hive-wide stop"), - ); + let id = submit(&q, "hive-wide stop", |b| { + stop_online(b, &["agent-a", "agent-b"], false); + }); // A hive-wide stop is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); // Same declared story as the restart case above: each agent's subgraph head @@ -679,19 +661,17 @@ fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { #[test] fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { let q = JobQueue::new(4); - let id = submit( - &q, - // fresh: offline + not stale → SetWanted → Reconcile. - // stale: offline + stale → SetWanted → «rebuild subgraph». - submit::start_spec( + // fresh: offline + not stale → SetWanted → Reconcile. + // stale: offline + stale → SetWanted → «rebuild subgraph». + let id = submit(&q, "hive-wide start", |b| { + submit::start_nodes( + b, &[ ("fresh".to_owned(), false, false), ("stale".to_owned(), false, true), ], - Source::Manual, - "hive-wide start".to_owned(), - ), - ); + ); + }); // One DAG spanning both agents. assert_eq!(q.snapshot().len(), 1); // The fold is a *declared* difference, readable the moment submit returns: @@ -730,27 +710,15 @@ fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { // read and node exec. let q = JobQueue::new(4); // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). - let stop = submit( - &q, - submit::stop_spec( - &[("down".to_owned(), false)], - true, - Source::Manual, - "stop down".to_owned(), - ), - ); + let stop = submit(&q, "stop down", |b| { + submit::stop_nodes(b, &[("down".to_owned(), false)], true); + }); // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): // nothing to bounce, and restart never rewrites intent, so the tail // Reconcile converges the down agent to its existing `wanted`. - let restart = submit( - &q, - submit::restart_spec( - &[("down2".to_owned(), false)], - true, - Source::Manual, - "restart down".to_owned(), - ), - ); + let restart = submit(&q, "restart down", |b| { + submit::restart_nodes(b, &[("down2".to_owned(), false)], true); + }); let shape = |id: u64| -> Vec { q.snapshot() .iter() @@ -784,21 +752,16 @@ fn boot_sweep_nodes_declare_their_own_resources() { // 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()], - ); - }), - }, - ); + let id = q + .submit(Source::AutoUpdate, "boot".to_owned(), |b: &Job| { + crate::workers::auto_update::boot_nodes( + b, + true, + vec!["stale-agent".to_owned()], + vec!["drifted-agent".to_owned()], + ); + }) + .expect("valid shape"); let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock")); lock.sort_by_key(|r| format!("{r:?}")); @@ -816,7 +779,9 @@ fn boot_sweep_nodes_declare_their_own_resources() { } /// Crash-watch suppression for a cascade rebuild, which the deleted half of -/// `meta_update_grows_cascade_in_dag` used to assert via `DagSpec::transient`. +/// `meta_update_grows_cascade_in_dag` used to assert via a DAG-level +/// `transient` field on the submit-time spec (both the field and the spec type +/// are gone). /// /// The property is unchanged — a container going down under a rebuild must not /// read as a crash — but it is no longer a DAG-level declaration: each node @@ -897,7 +862,7 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() { // easy thing to break — someone flattening the chain would keep every edge // and still lose the guarantee. let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); + let id = submit(&q, "r", |b| rebuild(b, "agent-a")); let shape = declared_shape(&q, id); let parent_of = |kind: &str| { shape @@ -969,21 +934,14 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() { #[test] fn a_fanned_out_mechanical_node_declares_its_agent_lease() { let q = JobQueue::new(4); - let id = submit( - &q, - DagSpec { - source: Source::Manual, - reason: "fan-out".to_owned(), - declare: Box::new(|b: &Job| { - templates::fanned_out_mechanical( - b, - NodeKind::Start { - agent: "agent-a".to_owned(), - }, - ); - }), - }, - ); + let id = submit(&q, "fan-out", |b| { + templates::fanned_out_mechanical( + b, + NodeKind::Start { + agent: "agent-a".to_owned(), + }, + ); + }); assert_eq!(declared_shape(&q, id), vec![row("start", None, &[])]); assert_eq!( declared_resources(&q, node_of(&q, id, "start")), @@ -1010,23 +968,18 @@ fn a_fanned_out_mechanical_node_declares_its_agent_lease() { fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() { let q = JobQueue::new(4); let agents = vec!["alice".to_owned(), "bob".to_owned()]; - let id = submit( - &q, - DagSpec { - source: Source::AutoUpdate, - reason: "sweep".to_owned(), - declare: Box::new(move |b: &Job| { - templates::grown_rebuilds( - b, - &agents, - templates::RebuildOpts { - relock: true, - graceful: true, - }, - ); - }), - }, - ); + let id = q + .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| { + templates::grown_rebuilds( + b, + &agents, + templates::RebuildOpts { + relock: true, + graceful: true, + }, + ); + }) + .expect("valid shape"); // One chain per agent, each an independent group root — so the two rebuild // concurrently, each on its own lease. @@ -1056,7 +1009,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() { #[test] fn cancel_clears_queued_dag() { let q = JobQueue::new(1); - let id = submit(&q, rebuild("agent-a", "r")); + let id = submit(&q, "r", |b| rebuild(b, "agent-a")); assert!(q.cancel(id), "fully-queued dag cancels"); // The operator sees `Cancelled` the moment the cancel returns — the spared // tail is still `Pending`, and a DAG must not read `Queued` back to the @@ -1085,7 +1038,9 @@ fn cancel_clears_queued_dag() { #[test] fn cancel_drops_one_agents_branch_leaving_the_rest() { let q = JobQueue::new(2); - let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r")); + let id = submit(&q, "r", |b| { + restart_online(b, &["agent-a", "agent-b"], false); + }); // Per-agent subgraphs are independent roots; find agent-a's. let snap = q.snapshot(); let dag = snap.iter().find(|d| d.id == id).expect("dag in snapshot"); @@ -1152,28 +1107,21 @@ fn cancelled_power_op_runs_no_compensating_node() { let case = format!("graceful={graceful} running={running}"); let q = JobQueue::new(1); - let id = submit( - &q, - submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()), - ); + let id = submit(&q, "bounce", |b| { + submit::restart_nodes(b, &targets, graceful); + }); assert_cancels_clean(&q, id, false, &format!("restart {case}")); let q = JobQueue::new(1); - let id = submit( - &q, - submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()), - ); + let id = submit(&q, "stop", |b| { + submit::stop_nodes(b, &targets, graceful); + }); assert_cancels_clean(&q, id, true, &format!("stop {case}")); let q = JobQueue::new(1); - let id = submit( - &q, - submit::start_spec( - &[("agent-a".to_owned(), running, false)], - Source::Manual, - "start".to_owned(), - ), - ); + let id = submit(&q, "start", |b| { + submit::start_nodes(b, &[("agent-a".to_owned(), running, false)]); + }); assert_cancels_clean(&q, id, true, &format!("start {case}")); } } @@ -1192,10 +1140,9 @@ fn cancelled_power_op_runs_no_compensating_node() { #[test] fn cancelled_dag_still_runs_its_approval_tail() { let q = JobQueue::new(1); - let id = submit( - &q, - templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), - ); + let id = submit(&q, "approval #7", |b| { + templates::approval_deploy(b, "agent-a", 7); + }); assert!(q.cancel(id), "fully-queued dag cancels"); // The `Cancelled` tail is the only node whose edge accepts a dropped // dependency, so it is the only one `cancel` spares — and *which* tail @@ -1216,7 +1163,7 @@ fn cancelled_dag_still_runs_its_approval_tail() { assert_eq!(state_of(&q, id), State::Cancelled); // An unrelated DAG landing in the same graph doesn't disturb this one's // roll-up — the snapshot is per-DAG, not a global state machine. - let _other = submit(&q, rebuild("agent-b", "r")); + let _other = submit(&q, "r", |b| rebuild(b, "agent-b")); assert_eq!(state_of(&q, id), State::Cancelled); } @@ -1230,10 +1177,9 @@ fn cancelled_dag_still_runs_its_approval_tail() { #[test] fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { let q = JobQueue::new(1); - let id = submit( - &q, - templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), - ); + let id = submit(&q, "approval #7", |b| { + templates::approval_deploy(b, "agent-a", 7); + }); assert_eq!( declared_shape(&q, id), @@ -1289,14 +1235,9 @@ fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() { // children run (`a_completing_node_grows_the_work_it_declared`, // `parent_parks_in_finishing_until_children_roll_up`). let q = JobQueue::new(1); - let id = submit( - &q, - DagSpec { - source: Source::Manual, - reason: "deploy graft".to_owned(), - declare: Box::new(|b: &Job| templates::deploy_rebuild_nodes(b, "agent-a", 11)), - }, - ); + let id = submit(&q, "deploy graft", |b| { + templates::deploy_rebuild_nodes(b, "agent-a", 11); + }); assert_eq!( declared_shape(&q, id), @@ -1461,7 +1402,7 @@ fn error_truncation_cuts_on_a_char_boundary() { #[test] fn graceful_stop_shape_signal_drain_reconcile() { let q = JobQueue::new(1); - let id = submit(&q, stop_online(&["agent-a"], true, "graceful")); + let id = submit(&q, "graceful", |b| stop_online(b, &["agent-a"], true)); assert_eq!( declared_shape(&q, id), vec![ @@ -1491,10 +1432,9 @@ fn graceful_stop_shape_signal_drain_reconcile() { #[test] fn spawn_shape_provision_create_dropin_reconcile() { let q = JobQueue::new(1); - let id = submit( - &q, - templates::spawn("newbie", 7, "approval #7 spawn".to_owned()), - ); + let id = submit(&q, "approval #7 spawn", |b| { + templates::spawn(b, "newbie", 7); + }); assert_eq!( declared_shape(&q, id), vec![ @@ -1517,18 +1457,16 @@ fn spawn_shape_provision_create_dropin_reconcile() { #[test] fn perm_change_shape_prefixes_rebuild_chain() { let q = JobQueue::new(1); - let id = submit( - &q, + let id = submit(&q, "perm", |b| { templates::perm_change( + b, "agent-a", - Source::Manual, - "perm".to_owned(), PermPayload::Combined { groups: Some(vec![]), caps: None, }, - ), - ); + ); + }); assert_eq!( declared_shape(&q, id) .iter() @@ -1557,14 +1495,9 @@ fn reparent_shape_is_a_lone_agentless_meta_window_node() { // `MetaLock`, and it must declare the meta window — a topology commit // must not land inside another node's staged deploy window. let q = JobQueue::new(1); - let id = submit( - &q, - templates::reparent( - vec![(ident("alice"), Some(ident("bob")))], - Source::Manual, - "set-parent".to_owned(), - ), - ); + let id = submit(&q, "set-parent", |b| { + templates::reparent(b, vec![(ident("alice"), Some(ident("bob")))]); + }); assert_eq!( declared_shape(&q, id), vec![row("reparent", None, &[])], @@ -1586,10 +1519,9 @@ fn reparent_bulk_shape_carries_every_move_on_one_node() { // request is the reason a single node was chosen in the first place. let moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)]; let q = JobQueue::new(1); - let id = submit( - &q, - templates::reparent(moves.clone(), Source::Manual, "set-parent-bulk".to_owned()), - ); + let id = submit(&q, "set-parent-bulk", |b| { + templates::reparent(b, moves.clone()); + }); assert_eq!( declared_shape(&q, id), vec![row("reparent", None, &[])], diff --git a/hive-c0re/src/workers/auto_update.rs b/hive-c0re/src/workers/auto_update.rs index 1e0a38e6..c9b705d1 100644 --- a/hive-c0re/src/workers/auto_update.rs +++ b/hive-c0re/src/workers/auto_update.rs @@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc) -> Result<()> { tracing::warn!( "manager container exists but no applied flake — forcing rebuild to migrate" ); - if let Err(e) = coord.job_queue.submit(crate::job_queue::templates::rebuild( - MANAGER_NAME, + if let Err(e) = coord.job_queue.submit( crate::job_queue::Source::AutoUpdate, "manager migration: no applied flake".to_owned(), - true, - )) { + |b| crate::job_queue::templates::rebuild(b, MANAGER_NAME, true), + ) { tracing::warn!(error = ?e, "manager migration rebuild submit failed"); } } else { @@ -377,7 +376,7 @@ fn submit_boot_tree( n_deferred: usize, n_skipped: usize, ) { - use crate::job_queue::{DagSpec, Source}; + use crate::job_queue::Source; // Fully-quiet boot (nothing stale, nothing drifted) submits nothing. if !any_stale && drifted.is_empty() { @@ -391,19 +390,14 @@ fn submit_boot_tree( n_skipped, ); - let declare = move |b: &crate::job_queue::Job| boot_nodes(b, any_stale, fanout, drifted); - - let spec = DagSpec { - // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they - // land; the boot DAG as a whole has no terminal side effect, so no tail. - source: Source::AutoUpdate, - reason, - // Rebuilding when the sweep will grow rebuild subgraphs (per-agent - // crash-watch suppression during their Swap, applied at claim time); - // a reconcile-only boot needs no transient. - declare, - }; - if let Err(e) = coord.job_queue.submit(spec) { + // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they + // land; the boot DAG as a whole has no terminal side effect, so no tail. + // The subgraphs also carry their own per-agent crash-watch suppression + // during their `Swap` (applied at claim time); a reconcile-only boot needs + // no transient. + if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| { + boot_nodes(b, any_stale, fanout, drifted); + }) { tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); } coord.emit_rebuild_queue_snapshot();