vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

jobq: templates swallow the DagSpec layer

Browse source 
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<F> 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.
This commit is contained in:
atlas 2026-08-03 01:23:40 +02:00 • committed by mara
commit 6899f574f6
7 changed files with 347 additions and 550 deletions
Download patch file Download diff file Expand all files Collapse all files

29
hive-c0re/src/actions.rs
View file

@ -55,14 +55,11 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
// nothing). // nothing).
let inputs: Vec<String> = let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default(); serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let submitted = coord let submitted = coord.job_queue.submit(
.job_queue crate::job_queue::Source::Approval,
.submit(crate::job_queue::templates::meta_update( format!("approval #{id} meta input update"),
inputs, |b| crate::job_queue::templates::meta_update(b, inputs, Some(id)),
crate::job_queue::Source::Approval, );
format!("approval #{id} meta input update"),
Some(id),
));
if let Err(e) = submitted { if let Err(e) = submitted {
return Err(e.context("submit meta-update dag")); return Err(e.context("submit meta-update dag"));
} }
@ -78,11 +75,11 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
{ {
tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed"); tracing::warn!(agent = %approval.agent, error = ?e, "agent_power: seed on spawn failed");
} }
let submitted = coord.job_queue.submit(crate::job_queue::templates::spawn( let submitted = coord.job_queue.submit(
approval.agent.as_str(), crate::job_queue::Source::Approval,
id,
format!("approval #{id} spawn"), format!("approval #{id} spawn"),
)); |b| crate::job_queue::templates::spawn(b, approval.agent.as_str(), id),
);
if let Err(e) = submitted { if let Err(e) = submitted {
return Err(e.context("submit spawn dag")); return Err(e.context("submit spawn dag"));
} }
@ -132,11 +129,9 @@ fn enqueue_approval_rebuild(
) { ) {
if let Err(e) = coord if let Err(e) = coord
.job_queue .job_queue
.submit(crate::job_queue::templates::approval_deploy( .submit(crate::job_queue::Source::Approval, reason, |b| {
agent, crate::job_queue::templates::approval_deploy(b, agent, approval_id);
approval_id, })
reason,
))
{ {
tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed"); tracing::error!(%agent, approval_id, error = ?e, "submit approval deploy dag failed");
} }

23
hive-c0re/src/job_queue/mod.rs
View file

@ -47,7 +47,7 @@ use hive_jobq_wire::{GraphNode, GraphWire};
use tokio::sync::Notify; use tokio::sync::Notify;
pub use hive_jobq::TerminalState; 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; use resource::Resource;
/// A job under construction: `hive_jobq`'s builder over this queue's payload /// 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 /// here completes it by hand — a node with no work of its own still goes
/// the way every other node goes. /// the way every other node goes.
/// ///
/// Takes the spec's recipe by generic, not as a boxed closure: a spec /// `source` and `reason` are the container node's own payload — they are
/// travels from the template that built it directly into this call, so /// arguments here rather than fields of a spec struct because that is all
/// there is nothing to allocate for. /// 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 /// # Errors
/// Propagates a graph-insert error (dependencies that aren't /// Propagates a graph-insert error (dependencies that aren't
/// dependency-topological). /// dependency-topological).
pub fn submit<F: FnOnce(&Job)>(&self, spec: DagSpec<F>) -> anyhow::Result<u64> { pub fn submit(
&self,
source: Source,
reason: String,
declare: impl FnOnce(&Job),
) -> anyhow::Result<u64> {
let mut inner = self.lock(); let mut inner = self.lock();
let container = inner let container = inner
.append( .append(
NodeKind::Dag { NodeKind::Dag {
source: spec.source, source,
reason: spec.reason, reason,
created_at: Utc::now(), created_at: Utc::now(),
}, },
Vec::new(), Vec::new(),
None, None,
) )
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; .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); drop(inner);
self.notify.notify_one(); self.notify.notify_one();
Ok(container.get()) Ok(container.get())

41
hive-c0re/src/job_queue/model.rs
View file

@ -426,44 +426,3 @@ impl NodeKind {
// - `DeployWindow` brackets a deploy without itself stopping anything. // - `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<F> {
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<F> std::fmt::Debug for DagSpec<F> {
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()
}
}

135
hive-c0re/src/job_queue/submit.rs
View file

@ -24,18 +24,23 @@
use std::sync::Arc; use std::sync::Arc;
use super::model::{DagSpec, NodeKind}; use super::model::NodeKind;
use super::resource::Resource; use super::resource::Resource;
use super::templates::{RebuildOpts, rebuild_nodes}; use super::templates::{RebuildOpts, rebuild_nodes};
use super::{Job, Source, templates}; use super::{Job, Source, templates};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::lifecycle; use crate::lifecycle;
fn submit_and_emit<F: FnOnce(&Job)>(coord: &Arc<Coordinator>, spec: super::DagSpec<F>) -> u64 { fn submit_and_emit(
coord: &Arc<Coordinator>,
source: Source,
reason: String,
declare: impl FnOnce(&Job),
) -> u64 {
let id = coord let id = coord
.job_queue .job_queue
.submit(spec) .submit(source, reason, declare)
.expect("template-built dag specs are acyclic"); .expect("template-declared shapes are acyclic");
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();
id id
} }
@ -44,7 +49,9 @@ fn submit_and_emit<F: FnOnce(&Job)>(coord: &Arc<Coordinator>, spec: super::DagSp
/// meta input — the meta-update cascade grows its own rebuild subgraphs /// meta input — the meta-update cascade grows its own rebuild subgraphs
/// in-DAG instead of going through this surface). /// in-DAG instead of going through this surface).
pub fn rebuild(coord: &Arc<Coordinator>, agent: &str, source: Source, reason: String) -> u64 { pub fn rebuild(coord: &Arc<Coordinator>, 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 ---------------------------------------- // ---- 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 // The `*_nodes` declarers below are the PURE core the async `*_many` fns call
/// 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<F: FnOnce(&Job)>(source: Source, reason: String, declare: F) -> DagSpec<F> {
DagSpec {
source,
reason,
declare,
}
}
// The `*_spec` builders below are the PURE core the async `*_many` fns call
// after reading live state — they take the per-agent running (and stale) // after reading live state — they take the per-agent running (and stale)
// flags explicitly, so unit tests exercise the online/offline shapes without // 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. /// Declare the stop DAG from explicit `(agent, running)` targets.
pub(crate) fn stop_spec( pub(crate) fn stop_nodes(b: &Job, targets: &[(String, bool)], graceful: bool) {
targets: &[(String, bool)], for (agent, running) in targets {
graceful: bool, stop_chain(b, agent, graceful, *running);
source: Source, }
reason: String,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
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);
}
}),
)
} }
/// Assemble the start DAG from explicit `(agent, running, stale)` targets. /// 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 /// running under its lease, so a down+stale agent that grew a rebuild subgraph
/// reports `rebuilding` during its swap and `starting` at its reconcile, /// reports `rebuilding` during its swap and `starting` at its reconcile,
/// without the DAG having to guess one label covering every target. /// without the DAG having to guess one label covering every target.
pub(crate) fn start_spec( pub(crate) fn start_nodes(b: &Job, targets: &[(String, bool, bool)]) {
targets: &[(String, bool, bool)], for (agent, running, stale) in targets {
source: Source, start_chain(b, agent, *running, *stale);
reason: String, }
) -> DagSpec<impl FnOnce(&Job) + use<>> {
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);
}
}),
)
} }
/// Assemble the restart DAG from explicit `(agent, running)` targets. /// Declare the restart DAG from explicit `(agent, running)` targets.
pub(crate) fn restart_spec( pub(crate) fn restart_nodes(b: &Job, targets: &[(String, bool)], graceful: bool) {
targets: &[(String, bool)], for (agent, running) in targets {
graceful: bool, restart_chain(b, agent, graceful, *running);
source: Source, }
reason: String,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
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);
}
}),
)
} }
/// Restart a single agent. Thin wrapper over [`restart_many`]. /// Restart a single agent. Thin wrapper over [`restart_many`].
@ -302,7 +266,9 @@ pub async fn restart_many(
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); 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`]. /// Start a single agent. Thin wrapper over [`start_many`].
@ -335,7 +301,9 @@ pub async fn start_many(
} }
targets.push((agent.clone(), running, stale)); 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`]. /// Hard stop a single agent. Thin wrapper over [`stop_many`].
@ -371,7 +339,9 @@ pub async fn stop_many(
for agent in agents { for agent in agents {
targets.push((agent.clone(), lifecycle::is_running(agent).await)); 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. /// Perm change: commit the JSON file(s) then rebuild.
@ -382,10 +352,9 @@ pub fn perm_change(
reason: String, reason: String,
payload: super::PermPayload, payload: super::PermPayload,
) -> u64 { ) -> u64 {
submit_and_emit( submit_and_emit(coord, source, reason, |b| {
coord, templates::perm_change(b, agent, payload);
templates::perm_change(agent, source, reason, payload), })
)
} }
/// Meta-input lock bump; cascade rebuilds fan out on completion. /// Meta-input lock bump; cascade rebuilds fan out on completion.
@ -395,7 +364,9 @@ pub fn meta_update(
source: Source, source: Source,
reason: String, reason: String,
) -> u64 { ) -> 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 — /// Topology move(s) as a queue DAG. `moves` is `(child, new_parent)` pairs —
@ -411,5 +382,7 @@ pub fn reparent(
source: Source, source: Source,
reason: String, reason: String,
) -> u64 { ) -> u64 {
submit_and_emit(coord, templates::reparent(moves, source, reason)) submit_and_emit(coord, source, reason, |b| {
templates::reparent(b, moves);
})
} }

283
hive-c0re/src/job_queue/templates.rs
View file

@ -26,7 +26,7 @@
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
use super::model::{DagSpec, NodeKind, PermPayload, Source}; use super::model::{NodeKind, PermPayload};
use super::resource::Resource; use super::resource::Resource;
use super::{Handle, Job}; 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 /// whole `StopForUpdate`→`Swap`→`PostSwap` subtree, so those three cover every
/// node. Edging `Reconcile` alone would not do: it is `AfterAny` `Prebuild`, so /// 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. /// it reaches `Done` even after a failed swap and the tail would report success.
pub fn rebuild( pub fn rebuild(b: &Job, agent: &str, relock: bool) {
agent: &str, let roots = rebuild_nodes(
source: Source, b,
reason: String, agent,
relock: bool, RebuildOpts {
) -> DagSpec<impl FnOnce(&Job) + use<>> { relock,
let agent = agent.to_owned(); graceful: false,
DagSpec { },
source, None,
reason, );
declare: Box::new(move |b: &Job| { emit_rebuilt_tails(b, agent, &roots.all());
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 /// 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` /// The window still spans the container build, as it must: `prepare_deploy`
/// leaves `flake.lock` staged-uncommitted for the build's whole duration. /// leaves `flake.lock` staged-uncommitted for the build's whole duration.
pub fn approval_deploy( pub fn approval_deploy(b: &Job, agent: &str, approval_id: i64) {
agent: &str, let a = || agent.to_owned();
approval_id: i64, // The window is the widest holder in the tree: it brackets a nix
reason: String, // build (`BuildSlot`), takes the container down across the swap
) -> DagSpec<impl FnOnce(&Job) + use<>> { // (`Agent`), and serialises the meta mutation its subtree performs
let agent = agent.to_owned(); // (`MetaWindow`). All three are held for its whole subtree, which
DagSpec { // is what lets the appended rebuild's `MetaSync` and the
source: Source::Approval, // `FinalizeDeploy` re-enter rather than contend.
reason, let window = b
declare: Box::new(move |b: &Job| { .node(NodeKind::DeployWindow {
let a = || agent.clone(); agent: a(),
// The window is the widest holder in the tree: it brackets a nix approval_id,
// build (`BuildSlot`), takes the container down across the swap })
// (`Agent`), and serialises the meta mutation its subtree performs .needs(Resource::BuildSlot)
// (`MetaWindow`). All three are held for its whole subtree, which .needs(Resource::Agent(a()))
// is what lets the appended rebuild's `MetaSync` and the .needs(Resource::MetaWindow);
// `FinalizeDeploy` re-enter rather than contend. let verify = b
let window = b .node(NodeKind::MergeVerify {
.node(NodeKind::DeployWindow { agent: a(),
agent: a(), approval_id,
approval_id, })
}) .part_of(window);
.needs(Resource::BuildSlot) let apply = b
.needs(Resource::Agent(a())) .node(NodeKind::DeployApply {
.needs(Resource::MetaWindow); agent: a(),
let verify = b approval_id,
.node(NodeKind::MergeVerify { })
agent: a(), .part_of(window)
approval_id, .after_ok(verify);
}) let _tail = b
.part_of(window); .node(NodeKind::DeployTail {
let apply = b agent: a(),
.node(NodeKind::DeployApply { approval_id,
agent: a(), })
approval_id, .part_of(window)
}) .after_any(apply);
.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 /// 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 /// 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 /// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up
/// already carries the whole cascade. /// already carries the whole cascade.
pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec<impl FnOnce(&Job) + use<>> { pub fn spawn(b: &Job, agent: &str, approval_id: i64) {
let agent = agent.to_owned(); let a = || agent.to_owned();
DagSpec { let provision = b
source: Source::Approval, .node(NodeKind::Provision { agent: a() })
reason, .needs(Resource::MetaWindow);
declare: Box::new(move |b: &Job| { let create = b
let a = || agent.clone(); .node(NodeKind::Create { agent: a() })
let provision = b .needs(Resource::BuildSlot)
.node(NodeKind::Provision { agent: a() }) .needs(Resource::Agent(a()))
.needs(Resource::MetaWindow); .part_of(provision);
let create = b let dropin = b
.node(NodeKind::Create { agent: a() }) .node(NodeKind::WriteDropin { agent: a() })
.needs(Resource::BuildSlot) .needs(Resource::Agent(a()))
.needs(Resource::Agent(a())) .part_of(create);
.part_of(provision); let _reconcile = b
let dropin = b .node(NodeKind::Reconcile { agent: a() })
.node(NodeKind::WriteDropin { agent: a() }) .needs(Resource::Agent(a()))
.needs(Resource::Agent(a())) .part_of(create)
.part_of(create); .after_ok(dropin);
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 /// 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<impl FnOn
/// effect in the container. Group-roots are `WritePermFile` plus the rebuild /// effect in the container. Group-roots are `WritePermFile` plus the rebuild
/// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail /// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail
/// edges all four. /// edges all four.
pub fn perm_change( pub fn perm_change(b: &Job, agent: &str, payload: PermPayload) {
agent: &str, let write = b
source: Source, .node(NodeKind::WritePermFile {
reason: String, agent: agent.to_owned(),
payload: PermPayload, payload,
) -> DagSpec<impl FnOnce(&Job) + use<>> { })
let agent = agent.to_owned(); .needs(Resource::MetaWindow);
DagSpec { let roots = rebuild_nodes(
source, b,
reason, agent,
declare: Box::new(move |b: &Job| { RebuildOpts {
let write = b relock: true,
.node(NodeKind::WritePermFile { graceful: false,
agent: agent.clone(), },
payload, Some(write),
}) );
.needs(Resource::MetaWindow); emit_rebuilt_tails(
let roots = rebuild_nodes( b,
b, agent,
&agent, &[write, roots.meta_sync, roots.prebuild, roots.reconcile],
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 /// 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 /// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent
/// crash-watch suppression during its `Swap` — the property the old child /// crash-watch suppression during its `Swap` — the property the old child
/// `Rebuild` DAGs carried via their own transient. /// `Rebuild` DAGs carried via their own transient.
pub fn meta_update( pub fn meta_update(b: &Job, inputs: Vec<String>, approval_id: Option<i64>) {
inputs: Vec<String>, let lock = b
source: Source, .node(NodeKind::MetaLock {
reason: String, sweep: false,
approval_id: Option<i64>, fanout: None,
) -> DagSpec<impl FnOnce(&Job) + use<>> { inputs,
DagSpec { })
source, .needs(Resource::BuildSlot)
reason, .needs(Resource::MetaWindow);
declare: Box::new(move |b: &Job| { // The bump itself has no side effect, so an operator-driven one ends
let lock = b // at the `MetaLock`; an approval-driven one still has its row to
.node(NodeKind::MetaLock { // resolve and gets the per-outcome tails edged onto that single
sweep: false, // group-root — whose roll-up covers the rebuild subgraphs `MetaLock`
fanout: None, // grows into itself.
inputs, if let Some(approval_id) = approval_id {
}) resolve_approval_tails(b, approval_id, lock);
.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. /// 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 /// 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. /// off of) and near-instant. No tail node: the write is the whole effect.
pub fn reparent( pub fn reparent(b: &Job, moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>) {
moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>, let _reparent = b
source: Source, .node(NodeKind::Reparent { moves })
reason: String, .needs(Resource::MetaWindow);
) -> DagSpec<impl FnOnce(&Job) + use<>> {
DagSpec {
source,
reason,
declare: Box::new(move |b: &Job| {
let _reparent = b
.node(NodeKind::Reparent { moves })
.needs(Resource::MetaWindow);
}),
}
} }
// The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree` // The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`

356
hive-c0re/src/job_queue/tests.rs
View file

@ -17,40 +17,36 @@
use super::model::NodeKind; use super::model::NodeKind;
use super::*; use super::*;
fn submit<F: FnOnce(&Job)>(q: &JobQueue, spec: DagSpec<F>) -> u64 { /// Submit a declared shape with the metadata every mechanics test uses.
q.submit(spec).expect("valid spec") /// `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 { fn ident(s: &str) -> hive_types::Ident {
hive_types::Ident::parse(s).expect("valid test ident") hive_types::Ident::parse(s).expect("valid test ident")
} }
fn rebuild(agent: &str, reason: &str) -> DagSpec<impl FnOnce(&Job) + use<>> { fn rebuild(b: &Job, agent: &str) {
templates::rebuild(agent, Source::Manual, reason.to_owned(), true) 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) /// shape (`[Signal→Drain→] StopForUpdate → Reconcile`, no `SetWanted` head)
/// most queue-mechanics tests assume. Mirrors the pre-dynamic /// most queue-mechanics tests assume. Mirrors the pre-dynamic
/// `templates::restart` (which is now the state-aware `submit::restart_spec`). /// `templates::restart` (which is now the state-aware `submit::restart_nodes`).
fn restart_online( fn restart_online(b: &Job, agents: &[&str], graceful: bool) {
agents: &[&str],
graceful: bool,
reason: &str,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); 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`). /// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
fn stop_online( fn stop_online(b: &Job, agents: &[&str], graceful: bool) {
agents: &[&str],
graceful: bool,
reason: &str,
) -> DagSpec<impl FnOnce(&Job) + use<>> {
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); 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 // `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
@ -303,9 +299,9 @@ fn state_of(q: &JobQueue, dag_id: u64) -> State {
#[test] #[test]
fn submit_assigns_distinct_ids() { fn submit_assigns_distinct_ids() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "first")); let first = submit(&q, "first", |b| rebuild(b, "agent-a"));
let b = submit(&q, rebuild("agent-b", "second")); let second = submit(&q, "second", |b| rebuild(b, "agent-b"));
assert_ne!(a, b); assert_ne!(first, second);
assert_eq!(q.snapshot().len(), 2); assert_eq!(q.snapshot().len(), 2);
} }
@ -317,20 +313,20 @@ fn submit_assigns_distinct_ids() {
#[test] #[test]
fn identical_resubmit_is_a_distinct_dag() { fn identical_resubmit_is_a_distinct_dag() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "first")); let first = submit(&q, "first", |b| rebuild(b, "agent-a"));
let b = submit(&q, rebuild("agent-a", "again")); let resubmit = submit(&q, "again", |b| rebuild(b, "agent-a"));
assert_ne!(a, b, "no dedup: identical resubmit is a new DAG"); assert_ne!(first, resubmit, "no dedup: identical resubmit is a new DAG");
assert_eq!(q.snapshot().len(), 2); assert_eq!(q.snapshot().len(), 2);
} }
#[test] #[test]
fn distinct_submits_never_collapse() { fn distinct_submits_never_collapse() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "r")); let rebuild_a = submit(&q, "r", |b| rebuild(b, "agent-a"));
let b = submit(&q, rebuild("agent-b", "r")); let rebuild_b = submit(&q, "r", |b| rebuild(b, "agent-b"));
let c = submit(&q, restart_online(&["agent-a"], false, "r")); let restart_a = submit(&q, "r", |b| restart_online(b, &["agent-a"], false));
assert_ne!(a, b); assert_ne!(rebuild_a, rebuild_b);
assert_ne!(a, c); assert_ne!(rebuild_a, restart_a);
assert_eq!(q.snapshot().len(), 3); 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 // swallowed" is the scenario people worry about, and a reader looking for
// it should find it. // it should find it.
let q = JobQueue::new(1); let q = JobQueue::new(1);
let a = submit(&q, rebuild("agent-a", "first")); let a = submit(&q, "first", |b| rebuild(b, "agent-a"));
let again = submit(&q, rebuild("agent-a", "config bumped during build")); let again = submit(&q, "config bumped during build", |b| rebuild(b, "agent-a"));
assert_ne!(a, again); assert_ne!(a, again);
assert_eq!(q.snapshot().len(), 2); 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 // logic"). Both axes are asserted below because a template can break either
// one independently. // one independently.
let q = JobQueue::new(1); 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!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
vec![ vec![
@ -429,24 +425,19 @@ fn rebuild_chain_is_declared_serial() {
#[test] #[test]
fn graceful_rebuild_chain_drains_before_stopping() { fn graceful_rebuild_chain_drains_before_stopping() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = q
&q, .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| {
DagSpec { templates::rebuild_nodes(
source: Source::AutoUpdate, b,
reason: "sweep".to_owned(), "agent-a",
declare: Box::new(|b: &Job| { templates::RebuildOpts {
templates::rebuild_nodes( relock: true,
b, graceful: true,
"agent-a", },
templates::RebuildOpts { None,
relock: true, );
graceful: true, })
}, .expect("valid shape");
None,
);
}),
},
);
assert_eq!( assert_eq!(
declared_shape(&q, id) declared_shape(&q, id)
.iter() .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 // job keeps its nodes to itself and inserts them, so what it built is
// observable where it matters — in what the scheduler runs. // observable where it matters — in what the scheduler runs.
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "manual", |b| {
&q, templates::rebuild_nodes(
DagSpec { b,
source: Source::Manual, "agent-a",
reason: "manual".to_owned(), templates::RebuildOpts {
declare: Box::new(|b: &Job| { relock: true,
templates::rebuild_nodes( graceful: false,
b, },
"agent-a", None,
templates::RebuildOpts { );
relock: true, });
graceful: false,
},
None,
);
}),
},
);
assert_eq!( assert_eq!(
declared_shape(&q, id) declared_shape(&q, id)
.iter() .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 // a resource unit is held for the acquirer's whole subtree, so the slot
// `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it. // `Prebuild` takes covers `StopForUpdate` → `Swap` → `PostSwap` beneath it.
let q = JobQueue::new(1); 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 res = |kind: &str| declared_resources(&q, node_of(&q, id, kind));
let agent = || Resource::Agent("agent-a".to_owned()); let agent = || Resource::Agent("agent-a".to_owned());
@ -610,10 +594,9 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
#[test] #[test]
fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
let id = submit( let id = submit(&q, "hive-wide", |b| {
&q, restart_online(b, &["agent-a", "agent-b"], false);
restart_online(&["agent-a", "agent-b"], false, "hive-wide"), });
);
// A hive-wide restart is ONE DAG, not one-per-agent. // A hive-wide restart is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1); assert_eq!(q.snapshot().len(), 1);
// Each agent's subgraph head (StopForUpdate, since both are running) is a // 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] #[test]
fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
let id = submit( let id = submit(&q, "hive-wide stop", |b| {
&q, stop_online(b, &["agent-a", "agent-b"], false);
stop_online(&["agent-a", "agent-b"], false, "hive-wide stop"), });
);
// A hive-wide stop is ONE DAG, not one-per-agent. // A hive-wide stop is ONE DAG, not one-per-agent.
assert_eq!(q.snapshot().len(), 1); assert_eq!(q.snapshot().len(), 1);
// Same declared story as the restart case above: each agent's subgraph head // 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] #[test]
fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
let id = submit( // fresh: offline + not stale → SetWanted → Reconcile.
&q, // stale: offline + stale → SetWanted → «rebuild subgraph».
// fresh: offline + not stale → SetWanted → Reconcile. let id = submit(&q, "hive-wide start", |b| {
// stale: offline + stale → SetWanted → «rebuild subgraph». submit::start_nodes(
submit::start_spec( b,
&[ &[
("fresh".to_owned(), false, false), ("fresh".to_owned(), false, false),
("stale".to_owned(), false, true), ("stale".to_owned(), false, true),
], ],
Source::Manual, );
"hive-wide start".to_owned(), });
),
);
// One DAG spanning both agents. // One DAG spanning both agents.
assert_eq!(q.snapshot().len(), 1); assert_eq!(q.snapshot().len(), 1);
// The fold is a *declared* difference, readable the moment submit returns: // 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. // read and node exec.
let q = JobQueue::new(4); let q = JobQueue::new(4);
// Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain).
let stop = submit( let stop = submit(&q, "stop down", |b| {
&q, submit::stop_nodes(b, &[("down".to_owned(), false)], true);
submit::stop_spec( });
&[("down".to_owned(), false)],
true,
Source::Manual,
"stop down".to_owned(),
),
);
// Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate): // Offline restart → a lone Reconcile (no SetWanted, no StopForUpdate):
// nothing to bounce, and restart never rewrites intent, so the tail // nothing to bounce, and restart never rewrites intent, so the tail
// Reconcile converges the down agent to its existing `wanted`. // Reconcile converges the down agent to its existing `wanted`.
let restart = submit( let restart = submit(&q, "restart down", |b| {
&q, submit::restart_nodes(b, &[("down2".to_owned(), false)], true);
submit::restart_spec( });
&[("down2".to_owned(), false)],
true,
Source::Manual,
"restart down".to_owned(),
),
);
let shape = |id: u64| -> Vec<String> { let shape = |id: u64| -> Vec<String> {
q.snapshot() q.snapshot()
.iter() .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 // meta commit inside another node's staged deploy window. Nothing failed to
// compile; only an exhaustive caller list would have caught it. // compile; only an exhaustive caller list would have caught it.
let q = JobQueue::new(4); let q = JobQueue::new(4);
let id = submit( let id = q
&q, .submit(Source::AutoUpdate, "boot".to_owned(), |b: &Job| {
DagSpec { crate::workers::auto_update::boot_nodes(
source: Source::AutoUpdate, b,
reason: "boot".to_owned(), true,
declare: Box::new(|b: &Job| { vec!["stale-agent".to_owned()],
crate::workers::auto_update::boot_nodes( vec!["drifted-agent".to_owned()],
b, );
true, })
vec!["stale-agent".to_owned()], .expect("valid shape");
vec!["drifted-agent".to_owned()],
);
}),
},
);
let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock")); let mut lock = declared_resources(&q, node_of(&q, id, "meta_lock"));
lock.sort_by_key(|r| format!("{r:?}")); 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 /// 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 /// 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 /// 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 // easy thing to break — someone flattening the chain would keep every edge
// and still lose the guarantee. // and still lose the guarantee.
let q = JobQueue::new(1); 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 shape = declared_shape(&q, id);
let parent_of = |kind: &str| { let parent_of = |kind: &str| {
shape shape
@ -969,21 +934,14 @@ fn rebuild_reconcile_waits_for_the_whole_build_subtree() {
#[test] #[test]
fn a_fanned_out_mechanical_node_declares_its_agent_lease() { fn a_fanned_out_mechanical_node_declares_its_agent_lease() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
let id = submit( let id = submit(&q, "fan-out", |b| {
&q, templates::fanned_out_mechanical(
DagSpec { b,
source: Source::Manual, NodeKind::Start {
reason: "fan-out".to_owned(), agent: "agent-a".to_owned(),
declare: Box::new(|b: &Job| { },
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_shape(&q, id), vec![row("start", None, &[])]);
assert_eq!( assert_eq!(
declared_resources(&q, node_of(&q, id, "start")), 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() { fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
let q = JobQueue::new(4); let q = JobQueue::new(4);
let agents = vec!["alice".to_owned(), "bob".to_owned()]; let agents = vec!["alice".to_owned(), "bob".to_owned()];
let id = submit( let id = q
&q, .submit(Source::AutoUpdate, "sweep".to_owned(), |b: &Job| {
DagSpec { templates::grown_rebuilds(
source: Source::AutoUpdate, b,
reason: "sweep".to_owned(), &agents,
declare: Box::new(move |b: &Job| { templates::RebuildOpts {
templates::grown_rebuilds( relock: true,
b, graceful: true,
&agents, },
templates::RebuildOpts { );
relock: true, })
graceful: true, .expect("valid shape");
},
);
}),
},
);
// One chain per agent, each an independent group root — so the two rebuild // One chain per agent, each an independent group root — so the two rebuild
// concurrently, each on its own lease. // concurrently, each on its own lease.
@ -1056,7 +1009,7 @@ fn a_meta_lock_grows_one_rebuild_subgraph_per_agent() {
#[test] #[test]
fn cancel_clears_queued_dag() { fn cancel_clears_queued_dag() {
let q = JobQueue::new(1); 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"); assert!(q.cancel(id), "fully-queued dag cancels");
// The operator sees `Cancelled` the moment the cancel returns — the spared // 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 // tail is still `Pending`, and a DAG must not read `Queued` back to the
@ -1085,7 +1038,9 @@ fn cancel_clears_queued_dag() {
#[test] #[test]
fn cancel_drops_one_agents_branch_leaving_the_rest() { fn cancel_drops_one_agents_branch_leaving_the_rest() {
let q = JobQueue::new(2); 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. // Per-agent subgraphs are independent roots; find agent-a's.
let snap = q.snapshot(); let snap = q.snapshot();
let dag = snap.iter().find(|d| d.id == id).expect("dag in 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 case = format!("graceful={graceful} running={running}");
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "bounce", |b| {
&q, submit::restart_nodes(b, &targets, graceful);
submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()), });
);
assert_cancels_clean(&q, id, false, &format!("restart {case}")); assert_cancels_clean(&q, id, false, &format!("restart {case}"));
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "stop", |b| {
&q, submit::stop_nodes(b, &targets, graceful);
submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()), });
);
assert_cancels_clean(&q, id, true, &format!("stop {case}")); assert_cancels_clean(&q, id, true, &format!("stop {case}"));
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "start", |b| {
&q, submit::start_nodes(b, &[("agent-a".to_owned(), running, false)]);
submit::start_spec( });
&[("agent-a".to_owned(), running, false)],
Source::Manual,
"start".to_owned(),
),
);
assert_cancels_clean(&q, id, true, &format!("start {case}")); assert_cancels_clean(&q, id, true, &format!("start {case}"));
} }
} }
@ -1192,10 +1140,9 @@ fn cancelled_power_op_runs_no_compensating_node() {
#[test] #[test]
fn cancelled_dag_still_runs_its_approval_tail() { fn cancelled_dag_still_runs_its_approval_tail() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "approval #7", |b| {
&q, templates::approval_deploy(b, "agent-a", 7);
templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), });
);
assert!(q.cancel(id), "fully-queued dag cancels"); assert!(q.cancel(id), "fully-queued dag cancels");
// The `Cancelled` tail is the only node whose edge accepts a dropped // The `Cancelled` tail is the only node whose edge accepts a dropped
// dependency, so it is the only one `cancel` spares — and *which* tail // 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); assert_eq!(state_of(&q, id), State::Cancelled);
// An unrelated DAG landing in the same graph doesn't disturb this one's // 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. // 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); assert_eq!(state_of(&q, id), State::Cancelled);
} }
@ -1230,10 +1177,9 @@ fn cancelled_dag_still_runs_its_approval_tail() {
#[test] #[test]
fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() { fn deploy_dag_runs_phases_in_order_and_tails_a_failed_apply() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "approval #7", |b| {
&q, templates::approval_deploy(b, "agent-a", 7);
templates::approval_deploy("agent-a", 7, "approval #7".to_owned()), });
);
assert_eq!( assert_eq!(
declared_shape(&q, id), 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`, // children run (`a_completing_node_grows_the_work_it_declared`,
// `parent_parks_in_finishing_until_children_roll_up`). // `parent_parks_in_finishing_until_children_roll_up`).
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "deploy graft", |b| {
&q, templates::deploy_rebuild_nodes(b, "agent-a", 11);
DagSpec { });
source: Source::Manual,
reason: "deploy graft".to_owned(),
declare: Box::new(|b: &Job| templates::deploy_rebuild_nodes(b, "agent-a", 11)),
},
);
assert_eq!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
@ -1461,7 +1402,7 @@ fn error_truncation_cuts_on_a_char_boundary() {
#[test] #[test]
fn graceful_stop_shape_signal_drain_reconcile() { fn graceful_stop_shape_signal_drain_reconcile() {
let q = JobQueue::new(1); 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!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
vec![ vec![
@ -1491,10 +1432,9 @@ fn graceful_stop_shape_signal_drain_reconcile() {
#[test] #[test]
fn spawn_shape_provision_create_dropin_reconcile() { fn spawn_shape_provision_create_dropin_reconcile() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "approval #7 spawn", |b| {
&q, templates::spawn(b, "newbie", 7);
templates::spawn("newbie", 7, "approval #7 spawn".to_owned()), });
);
assert_eq!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
vec![ vec![
@ -1517,18 +1457,16 @@ fn spawn_shape_provision_create_dropin_reconcile() {
#[test] #[test]
fn perm_change_shape_prefixes_rebuild_chain() { fn perm_change_shape_prefixes_rebuild_chain() {
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "perm", |b| {
&q,
templates::perm_change( templates::perm_change(
b,
"agent-a", "agent-a",
Source::Manual,
"perm".to_owned(),
PermPayload::Combined { PermPayload::Combined {
groups: Some(vec![]), groups: Some(vec![]),
caps: None, caps: None,
}, },
), );
); });
assert_eq!( assert_eq!(
declared_shape(&q, id) declared_shape(&q, id)
.iter() .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 // `MetaLock`, and it must declare the meta window — a topology commit
// must not land inside another node's staged deploy window. // must not land inside another node's staged deploy window.
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "set-parent", |b| {
&q, templates::reparent(b, vec![(ident("alice"), Some(ident("bob")))]);
templates::reparent( });
vec![(ident("alice"), Some(ident("bob")))],
Source::Manual,
"set-parent".to_owned(),
),
);
assert_eq!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
vec![row("reparent", None, &[])], 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. // 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 moves = vec![(ident("alice"), Some(ident("bob"))), (ident("carol"), None)];
let q = JobQueue::new(1); let q = JobQueue::new(1);
let id = submit( let id = submit(&q, "set-parent-bulk", |b| {
&q, templates::reparent(b, moves.clone());
templates::reparent(moves.clone(), Source::Manual, "set-parent-bulk".to_owned()), });
);
assert_eq!( assert_eq!(
declared_shape(&q, id), declared_shape(&q, id),
vec![row("reparent", None, &[])], vec![row("reparent", None, &[])],

30
hive-c0re/src/workers/auto_update.rs
View file

@ -109,12 +109,11 @@ pub async fn ensure_root_agent(coord: &Arc<Coordinator>) -> Result<()> {
tracing::warn!( tracing::warn!(
"manager container exists but no applied flake — forcing rebuild to migrate" "manager container exists but no applied flake — forcing rebuild to migrate"
); );
if let Err(e) = coord.job_queue.submit(crate::job_queue::templates::rebuild( if let Err(e) = coord.job_queue.submit(
MANAGER_NAME,
crate::job_queue::Source::AutoUpdate, crate::job_queue::Source::AutoUpdate,
"manager migration: no applied flake".to_owned(), "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"); tracing::warn!(error = ?e, "manager migration rebuild submit failed");
} }
} else { } else {
@ -377,7 +376,7 @@ fn submit_boot_tree(
n_deferred: usize, n_deferred: usize,
n_skipped: usize, n_skipped: usize,
) { ) {
use crate::job_queue::{DagSpec, Source}; use crate::job_queue::Source;
// Fully-quiet boot (nothing stale, nothing drifted) submits nothing. // Fully-quiet boot (nothing stale, nothing drifted) submits nothing.
if !any_stale && drifted.is_empty() { if !any_stale && drifted.is_empty() {
@ -391,19 +390,14 @@ fn submit_boot_tree(
n_skipped, n_skipped,
); );
let declare = move |b: &crate::job_queue::Job| boot_nodes(b, any_stale, fanout, drifted); // 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.
let spec = DagSpec { // The subgraphs also carry their own per-agent crash-watch suppression
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they // during their `Swap` (applied at claim time); a reconcile-only boot needs
// land; the boot DAG as a whole has no terminal side effect, so no tail. // no transient.
source: Source::AutoUpdate, if let Err(e) = coord.job_queue.submit(Source::AutoUpdate, reason, |b| {
reason, boot_nodes(b, any_stale, fanout, drifted);
// 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) {
tracing::warn!(error = ?e, "boot: sweep DAG submit failed"); tracing::warn!(error = ?e, "boot: sweep DAG submit failed");
} }
coord.emit_rebuild_queue_snapshot(); coord.emit_rebuild_queue_snapshot();