refactor(job-queue): a job is a recipe, not a value you carry

Follows the jobq change: a builder can no longer be constructed or
inserted outside `hive_jobq`, so `DagSpec` cannot hold one. It carries a
`Declare` — `Box<dyn FnOnce(&Job) + Send>` — and the queue runs it
against a builder jobq owns, at the moment it inserts.

`NodeOutput.append_subgraph` becomes `Vec<Declare>` for the same reason,
and this is where the shape was always heading: that field's doc already
said an executor "cannot reach the queue, so it hands the declaration
back", while its type was a `Vec<Job>` the executor had built itself.
The rejected `build_nodes -> Vec<NodeSpec>` was the first version of that
escape hatch; a recipe is the last one, because there is no job-shaped
value to hand over at all.

Templates and the power-op assemblers move their owned data into the
closure and are otherwise unchanged — `rebuild_nodes`, `node` and the
tail helpers already took `&Job` and returned handles, so only each
template's outermost frame moved.

Two `Debug` impls are hand-written: a closure has nothing to show, and
its nodes do not exist until the queue runs it. `NodeOutput` reports how
many subgraphs were emitted, `DagSpec` its source and reason.

`append_subgraph`'s `is_empty()` early-return is gone — you cannot ask a
recipe whether it will declare anything without running it. It now
inserts and returns an empty id list if nothing was declared, which
takes the queue lock in a case that previously skipped it.

The two in-DAG-growth tests build `Declare`s now, so they exercise the
shape an executor actually produces rather than one only a test could
construct. 45 job-queue tests unchanged and passing.
This commit is contained in:
atlas 2026-08-02 13:40:03 +02:00 committed by mara
commit 9c97365f8f
8 changed files with 336 additions and 284 deletions

View file

@ -10,7 +10,7 @@ use std::sync::Arc;
use anyhow::{Context as _, Result}; use anyhow::{Context as _, Result};
use super::{Claim, Job}; use super::{Claim, Declare};
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
use super::model::NodeKind; use super::model::NodeKind;
@ -27,7 +27,7 @@ pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from
/// Extra signal an executor hands back to the scheduler alongside /// Extra signal an executor hands back to the scheduler alongside
/// success. /// success.
#[derive(Debug, Default)] #[derive(Default)]
pub struct NodeOutput { pub struct NodeOutput {
/// Whole per-agent *subgraphs* to append into *this same* DAG at /// Whole per-agent *subgraphs* to append into *this same* DAG at
/// runtime — the single in-DAG-growth channel. Each [`Job`] is one /// runtime — the single in-DAG-growth channel. Each [`Job`] is one
@ -42,7 +42,17 @@ pub struct NodeOutput {
/// *before* the emitting node's completion so the DAG never rolls terminal /// *before* the emitting node's completion so the DAG never rolls terminal
/// with the appended work still pending — keeping the lease-window /// with the appended work still pending — keeping the lease-window
/// transient held across the sub-step. /// transient held across the sub-step.
pub append_subgraph: Vec<Job>, pub append_subgraph: Vec<Declare>,
}
impl std::fmt::Debug for NodeOutput {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
// The subgraphs are closures — how many were emitted is the only thing
// there is to say about them before the queue runs them.
f.debug_struct("NodeOutput")
.field("append_subgraph", &self.append_subgraph.len())
.finish()
}
} }
/// Build-log sink for one claimed node. /// Build-log sink for one claimed node.
@ -342,17 +352,18 @@ async fn run_meta_lock(
.unwrap_or_default() .unwrap_or_default()
.iter() .iter()
.map(|agent| { .map(|agent| {
let job = Job::new(); let agent = agent.clone();
super::templates::rebuild_nodes( Box::new(move |b: &super::Job| {
&job, super::templates::rebuild_nodes(
agent, b,
super::templates::RebuildOpts { &agent,
relock: true, super::templates::RebuildOpts {
graceful: true, relock: true,
}, graceful: true,
None, },
); None,
job );
}) as Declare
}) })
.collect(); .collect();
return Ok(NodeOutput { append_subgraph }); return Ok(NodeOutput { append_subgraph });
@ -374,17 +385,18 @@ async fn run_meta_lock(
let append_subgraph = cascade let append_subgraph = cascade
.iter() .iter()
.map(|agent| { .map(|agent| {
let job = Job::new(); let agent = agent.clone();
super::templates::rebuild_nodes( Box::new(move |b: &super::Job| {
&job, super::templates::rebuild_nodes(
agent, b,
super::templates::RebuildOpts { &agent,
relock: false, super::templates::RebuildOpts {
graceful: false, relock: false,
}, graceful: false,
None, },
); None,
job );
}) as Declare
}) })
.collect(); .collect();
Ok(NodeOutput { append_subgraph }) Ok(NodeOutput { append_subgraph })
@ -404,10 +416,10 @@ async fn run_reconcile(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOu
// One node targeting this agent, rooted on this reconcile node. `NodeKind` // One node targeting this agent, rooted on this reconcile node. `NodeKind`
// carries the agent it targets, so stamp `claim.agent` into the fanned-out // carries the agent it targets, so stamp `claim.agent` into the fanned-out
// Start/Stop kind (one in-DAG-growth channel). // Start/Stop kind (one in-DAG-growth channel).
let sub = |kind| { let sub = |kind: NodeKind| {
let job = Job::new(); vec![Box::new(move |b: &super::Job| {
let _ = super::templates::node(&job, kind); let _ = super::templates::node(b, kind);
vec![job] }) as Declare]
}; };
let append_subgraph = match reconcile_action(wanted, running) { let append_subgraph = match reconcile_action(wanted, running) {
ReconcileAction::Start => sub(NodeKind::Start { ReconcileAction::Start => sub(NodeKind::Start {

View file

@ -51,10 +51,20 @@ pub use model::{DagSpec, 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
/// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into /// ([`NodeKind`]) and resource ([`Resource`]) types. Templates declare into a
/// one of these; [`JobQueue::submit`] inserts it. /// borrowed one; only `hive_jobq` can make or insert it.
pub type Job = hive_jobq::JobBuilder<NodeKind, Resource>; pub type Job = hive_jobq::JobBuilder<NodeKind, Resource>;
/// A job's shape as a **recipe**: given a builder, declare the nodes.
///
/// What a template returns and what an executor hands back, because neither
/// can build a job itself — `hive_jobq` creates the builder inside its own
/// insertion call and never lets one out. So the transferable thing is the
/// declaring closure, and the queue runs it at the moment it inserts.
///
/// `Send` because an executor's output crosses the scheduler's task boundary.
pub type Declare = Box<dyn FnOnce(&Job) + Send>;
/// A handle to one node a template declared — where its edges, grouping and /// A handle to one node a template declared — where its edges, grouping and
/// resources are declared. `Copy`; naming a node as a dependency does not /// resources are declared. `Copy`; naming a node as a dependency does not
/// consume the ability to name it again. /// consume the ability to name it again.
@ -169,12 +179,12 @@ impl Default for JobQueue {
/// Propagates a crate graph-insert error (malformed dep/parent / dep-scope). /// Propagates a crate graph-insert error (malformed dep/parent / dep-scope).
fn insert_group( fn insert_group(
inner: &mut QueueInner, inner: &mut QueueInner,
job: Job, declare: Declare,
group_parent: Option<NodeId>, group_parent: Option<NodeId>,
) -> anyhow::Result<Vec<NodeId>> { ) -> anyhow::Result<Vec<NodeId>> {
let ids = inner let ids = inner
.sched .sched
.insert_job(job, group_parent) .insert_job(group_parent, declare)
.map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: graph insert failed: {e}"))?;
for &id in ids.values() { for &id in ids.values() {
inner.node_rt.insert(id, NodeRuntime::default()); inner.node_rt.insert(id, NodeRuntime::default());
@ -226,7 +236,7 @@ impl JobQueue {
) )
.map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?; .map_err(|e| anyhow::anyhow!("job_queue: container insert failed: {e}"))?;
inner.node_rt.insert(container, NodeRuntime::default()); inner.node_rt.insert(container, NodeRuntime::default());
insert_group(&mut inner, spec.job, Some(container))?; insert_group(&mut inner, spec.declare, Some(container))?;
// Settle the container's own (no-op) logic immediately so it parks in // Settle the container's own (no-op) logic immediately so it parks in
// `Finishing` and its children become runnable — it never needs claiming // `Finishing` and its children become runnable — it never needs claiming
// or executing, and stays out of `claim_ready`. It rolls up terminal when // or executing, and stays out of `claim_ready`. It rolls up terminal when
@ -247,10 +257,7 @@ impl JobQueue {
/// the DAG's terminal node deps on the top root, roll-up keeps the DAG from /// the DAG's terminal node deps on the top root, roll-up keeps the DAG from
/// settling early with no explicit wiring. Returns the new node ids; empty if /// settling early with no explicit wiring. Returns the new node ids; empty if
/// the DAG is gone or `nodes` is empty. /// the DAG is gone or `nodes` is empty.
pub fn append_subgraph(&self, dag_id: u64, job: Job, dep_on: NodeId) -> Vec<NodeId> { pub fn append_subgraph(&self, dag_id: u64, declare: Declare, dep_on: NodeId) -> Vec<NodeId> {
if job.is_empty() {
return Vec::new();
}
let mut inner = self.lock(); let mut inner = self.lock();
if inner.container(dag_id).is_none() { if inner.container(dag_id).is_none() {
return Vec::new(); return Vec::new();
@ -261,7 +268,7 @@ impl JobQueue {
// emitter stays `Finishing` until this appended subtree settles, and the // emitter stays `Finishing` until this appended subtree settles, and the
// container node rolls up terminal only once its whole subtree (incl. this // container node rolls up terminal only once its whole subtree (incl. this
// appended work) has settled, so the DAG hook waits for free. // appended work) has settled, so the DAG hook waits for free.
let ids = match insert_group(&mut inner, job, Some(dep_on)) { let ids = match insert_group(&mut inner, declare, Some(dep_on)) {
Ok(ids) => ids, Ok(ids) => ids,
Err(e) => { Err(e) => {
tracing::error!( tracing::error!(

View file

@ -470,15 +470,27 @@ impl NodeKind {
/// Type-specific payloads (`PermChange`'s file payload) ride the node that /// Type-specific payloads (`PermChange`'s file payload) ride the node that
/// consumes them ([`NodeKind::WritePermFile`]), not this generic spec. /// consumes them ([`NodeKind::WritePermFile`]), not this generic spec.
/// ///
/// There is no separate per-node spec type: the nodes live in the builder, /// There is no separate per-node spec type, and no built job either: `declare`
/// which inserts them itself. A shape that has been declared is therefore /// is a *recipe* the queue runs against a builder `hive_jobq` owns, at the
/// always insertable — a dangling edge or a cycle cannot be expressed, so /// moment it inserts. A shape that has been declared is therefore always
/// there is nothing left for a submit-time validation pass to reject. /// insertable — a dangling edge or a cycle cannot be expressed, so there is
#[derive(Debug)] /// nothing left for a submit-time validation pass to reject.
pub struct DagSpec { pub struct DagSpec {
pub source: Source, pub source: Source,
/// Free-form "why". /// Free-form "why".
pub reason: String, pub reason: String,
/// The DAG's declared nodes, with their edges, grouping and resources. /// Declares the DAG's nodes — their edges, grouping and resources — onto
pub job: super::Job, /// the builder the queue hands it.
pub declare: super::Declare,
}
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()
}
} }

View file

@ -26,7 +26,7 @@ use std::sync::Arc;
use super::model::{DagSpec, NodeKind}; use super::model::{DagSpec, NodeKind};
use super::templates::{RebuildOpts, node, rebuild_nodes}; use super::templates::{RebuildOpts, node, rebuild_nodes};
use super::{Job, Source, templates}; use super::{Declare, Job, Source, templates};
use crate::coordinator::Coordinator; use crate::coordinator::Coordinator;
use crate::lifecycle; use crate::lifecycle;
@ -169,11 +169,11 @@ fn restart_chain(b: &Job, agent: &str, graceful: bool, running: bool) {
/// and each keeps its own root, so the per-agent subgraphs are independent and /// 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 /// run concurrently, each on its own lease. Rebasing one subgraph's indices
/// onto another's used to be a function. /// onto another's used to be a function.
fn power_dag(source: Source, reason: String, job: Job) -> DagSpec { fn power_dag(source: Source, reason: String, declare: Declare) -> DagSpec {
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare,
} }
} }
@ -189,11 +189,16 @@ pub(crate) fn stop_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let job = Job::new(); let targets = targets.to_vec();
for (agent, running) in targets { power_dag(
stop_chain(&job, agent, graceful, *running); source,
} reason,
power_dag(source, reason, job) Box::new(move |b| {
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.
@ -207,11 +212,16 @@ pub(crate) fn start_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let job = Job::new(); let targets = targets.to_vec();
for (agent, running, stale) in targets { power_dag(
start_chain(&job, agent, *running, *stale); source,
} reason,
power_dag(source, reason, job) Box::new(move |b| {
for (agent, running, stale) in targets {
start_chain(b, &agent, running, stale);
}
}),
)
} }
/// Assemble the restart DAG from explicit `(agent, running)` targets. /// Assemble the restart DAG from explicit `(agent, running)` targets.
@ -221,11 +231,16 @@ pub(crate) fn restart_spec(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let job = Job::new(); let targets = targets.to_vec();
for (agent, running) in targets { power_dag(
restart_chain(&job, agent, graceful, *running); source,
} reason,
power_dag(source, reason, job) Box::new(move |b| {
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`].

View file

@ -24,7 +24,7 @@
use hive_jobq::TerminalState; use hive_jobq::TerminalState;
use super::model::{DagSpec, NodeKind, PermPayload, Source}; use super::model::{DagSpec, NodeKind, PermPayload, Source};
use super::{Handle, Job}; use super::{Declare, Handle, Job};
/// Declare one node carrying `kind`, with the resources that kind needs. /// Declare one node carrying `kind`, with the resources that kind needs.
/// ///
@ -239,27 +239,28 @@ pub(crate) fn rebuild_nodes<'a>(
/// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's /// inside the `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's
/// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor /// `MetaSync` and `FinalizeDeploy` declare is re-entered from the ancestor
/// already holding it rather than deadlocking against it. /// already holding it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Job { pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Declare {
let b = Job::new(); let agent = agent.to_owned();
let roots = rebuild_nodes( Box::new(move |b| {
&b, let roots = rebuild_nodes(
agent, b,
RebuildOpts { &agent,
relock: false, RebuildOpts {
graceful: false, relock: false,
}, graceful: false,
None, },
); None,
let _finalize = node( );
&b, let _finalize = node(
NodeKind::FinalizeDeploy { b,
agent: agent.to_owned(), NodeKind::FinalizeDeploy {
approval_id, agent: agent.clone(),
}, approval_id,
) },
.after_ok(roots.prebuild) )
.after_ok(roots.reconcile); .after_ok(roots.prebuild)
b .after_ok(roots.reconcile);
})
} }
/// One uniform rebuild shape — no `was_running` branch. `StopForUpdate` /// One uniform rebuild shape — no `was_running` branch. `StopForUpdate`
@ -274,21 +275,22 @@ pub(crate) fn deploy_rebuild_nodes(agent: &str, approval_id: i64) -> Job {
/// 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(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec { pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> DagSpec {
let job = Job::new(); let agent = agent.to_owned();
let roots = rebuild_nodes(
&job,
agent,
RebuildOpts {
relock,
graceful: false,
},
None,
);
emit_rebuilt_tails(&job, agent, &roots.all());
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare: Box::new(move |b| {
let roots = rebuild_nodes(
b,
&agent,
RebuildOpts {
relock,
graceful: false,
},
None,
);
emit_rebuilt_tails(b, &agent, &roots.all());
}),
} }
} }
@ -318,48 +320,48 @@ pub fn rebuild(agent: &str, source: Source, reason: String, relock: bool) -> Dag
/// 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(agent: &str, approval_id: i64, reason: String) -> DagSpec { pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec {
let a = || agent.to_owned(); let agent = agent.to_owned();
let job = Job::new();
let window = node(
&job,
NodeKind::DeployWindow {
agent: a(),
approval_id,
},
);
let verify = node(
&job,
NodeKind::MergeVerify {
agent: a(),
approval_id,
},
)
.part_of(window);
let apply = node(
&job,
NodeKind::DeployApply {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_ok(verify);
let _tail = node(
&job,
NodeKind::DeployTail {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_any(apply);
resolve_approval_tails(&job, approval_id, window);
DagSpec { DagSpec {
source: Source::Approval, source: Source::Approval,
reason, reason,
job, declare: Box::new(move |b| {
let a = || agent.clone();
let window = node(
b,
NodeKind::DeployWindow {
agent: a(),
approval_id,
},
);
let verify = node(
b,
NodeKind::MergeVerify {
agent: a(),
approval_id,
},
)
.part_of(window);
let apply = node(
b,
NodeKind::DeployApply {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_ok(verify);
let _tail = node(
b,
NodeKind::DeployTail {
agent: a(),
approval_id,
},
)
.part_of(window)
.after_any(apply);
resolve_approval_tails(b, approval_id, window);
}),
} }
} }
@ -370,17 +372,13 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec
/// in the queue tests); production paths no longer emit a bare reconcile. /// in the queue tests); production paths no longer emit a bare reconcile.
#[cfg(test)] #[cfg(test)]
pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec { pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec {
let job = Job::new(); let agent = agent.to_owned();
let _reconcile = node(
&job,
NodeKind::Reconcile {
agent: agent.to_owned(),
},
);
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare: Box::new(move |b| {
let _reconcile = node(b, NodeKind::Reconcile { agent });
}),
} }
} }
@ -396,21 +394,21 @@ pub fn reconcile_only(agent: &str, source: Source, reason: String) -> DagSpec {
/// `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 { pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec {
let a = || agent.to_owned(); let agent = agent.to_owned();
let job = Job::new();
let provision = node(&job, NodeKind::Provision { agent: a() });
let create = node(&job, NodeKind::Create { agent: a() }).part_of(provision);
let dropin = node(&job, NodeKind::WriteDropin { agent: a() }).part_of(create);
let _reconcile = node(&job, NodeKind::Reconcile { agent: a() })
.part_of(create)
.after_ok(dropin);
resolve_approval_tails(&job, approval_id, provision);
DagSpec { DagSpec {
source: Source::Approval, source: Source::Approval,
reason, reason,
job, declare: Box::new(move |b| {
let a = || agent.clone();
let provision = node(b, NodeKind::Provision { agent: a() });
let create = node(b, NodeKind::Create { agent: a() }).part_of(provision);
let dropin = node(b, NodeKind::WriteDropin { agent: a() }).part_of(create);
let _reconcile = node(b, NodeKind::Reconcile { agent: a() })
.part_of(create)
.after_ok(dropin);
resolve_approval_tails(b, approval_id, provision);
}),
} }
} }
@ -420,32 +418,33 @@ pub fn spawn(agent: &str, approval_id: i64, reason: String) -> DagSpec {
/// 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(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec { pub fn perm_change(agent: &str, source: Source, reason: String, payload: PermPayload) -> DagSpec {
let job = Job::new(); let agent = agent.to_owned();
let write = node(
&job,
NodeKind::WritePermFile {
agent: agent.to_owned(),
payload,
},
);
let roots = rebuild_nodes(
&job,
agent,
RebuildOpts {
relock: true,
graceful: false,
},
Some(write),
);
emit_rebuilt_tails(
&job,
agent,
&[write, roots.meta_sync, roots.prebuild, roots.reconcile],
);
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare: Box::new(move |b| {
let write = node(
b,
NodeKind::WritePermFile {
agent: agent.clone(),
payload,
},
);
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],
);
}),
} }
} }
@ -465,26 +464,27 @@ pub fn meta_update(
reason: String, reason: String,
approval_id: Option<i64>, approval_id: Option<i64>,
) -> DagSpec { ) -> DagSpec {
let job = Job::new();
let lock = node(
&job,
NodeKind::MetaLock {
sweep: false,
fanout: None,
inputs,
},
);
// 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(&job, approval_id, lock);
}
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare: Box::new(move |b| {
let lock = node(
b,
NodeKind::MetaLock {
sweep: false,
fanout: None,
inputs,
},
);
// 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);
}
}),
} }
} }
@ -502,12 +502,12 @@ pub fn reparent(
source: Source, source: Source,
reason: String, reason: String,
) -> DagSpec { ) -> DagSpec {
let job = Job::new();
let _reparent = node(&job, NodeKind::Reparent { moves });
DagSpec { DagSpec {
source, source,
reason, reason,
job, declare: Box::new(move |b| {
let _reparent = node(b, NodeKind::Reparent { moves });
}),
} }
} }

View file

@ -192,22 +192,22 @@ fn rebuild_chain_claims_in_dep_order() {
#[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 job = Job::new();
templates::rebuild_nodes(
&job,
"agent-a",
templates::RebuildOpts {
relock: true,
graceful: true,
},
None,
);
let id = submit( let id = submit(
&q, &q,
DagSpec { DagSpec {
source: Source::AutoUpdate, source: Source::AutoUpdate,
reason: "sweep".to_owned(), reason: "sweep".to_owned(),
job, declare: Box::new(|b| {
templates::rebuild_nodes(
b,
"agent-a",
templates::RebuildOpts {
relock: true,
graceful: true,
},
None,
);
}),
}, },
); );
for expected in [ for expected in [
@ -241,22 +241,22 @@ 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 job = Job::new();
templates::rebuild_nodes(
&job,
"agent-a",
templates::RebuildOpts {
relock: true,
graceful: false,
},
None,
);
let id = submit( let id = submit(
&q, &q,
DagSpec { DagSpec {
source: Source::Manual, source: Source::Manual,
reason: "manual".to_owned(), reason: "manual".to_owned(),
job, declare: Box::new(|b| {
templates::rebuild_nodes(
b,
"agent-a",
templates::RebuildOpts {
relock: true,
graceful: false,
},
None,
);
}),
}, },
); );
let mut kinds = Vec::new(); let mut kinds = Vec::new();
@ -692,19 +692,19 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
// subgraph per stale agent into its OWN DAG. Each subgraph is rooted on // subgraph per stale agent into its OWN DAG. Each subgraph is rooted on
// the emitter and its LOCAL 0-based deps are rebased onto the DAG. // the emitter and its LOCAL 0-based deps are rebased onto the DAG.
let q = JobQueue::new(4); let q = JobQueue::new(4);
let job = Job::new();
let _lock = templates::node(
&job,
NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
},
);
let spec = DagSpec { let spec = DagSpec {
source: Source::AutoUpdate, source: Source::AutoUpdate,
reason: "sweep".to_owned(), reason: "sweep".to_owned(),
job, declare: Box::new(|b| {
let _lock = templates::node(
b,
NodeKind::MetaLock {
sweep: true,
fanout: None,
inputs: Vec::new(),
},
);
}),
}; };
let id = submit(&q, spec); let id = submit(&q, spec);
let emitter = claim_one(&q); let emitter = claim_one(&q);
@ -713,18 +713,19 @@ fn append_subgraph_roots_on_emitter_and_rebases_local_deps() {
// sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain → // sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain →
// StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must // StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must
// match the sweep arm of `run_meta_lock` or this stops tracking production. // match the sweep arm of `run_meta_lock` or this stops tracking production.
let subgraph = |agent: &str| { let subgraph = |agent: &str| -> Declare {
let job = Job::new(); let agent = agent.to_owned();
templates::rebuild_nodes( Box::new(move |b| {
&job, templates::rebuild_nodes(
agent, b,
templates::RebuildOpts { &agent,
relock: true, templates::RebuildOpts {
graceful: true, relock: true,
}, graceful: true,
None, },
); None,
job );
})
}; };
// Must append BEFORE completing the emitter (the documented contract). // Must append BEFORE completing the emitter (the documented contract).
q.append_subgraph(id, subgraph("a"), emitter.node_id); q.append_subgraph(id, subgraph("a"), emitter.node_id);
@ -820,17 +821,18 @@ fn meta_update_grows_cascade_in_dag() {
// Simulate the executor growing the cascade in-DAG (`relock = false` — a // Simulate the executor growing the cascade in-DAG (`relock = false` — a
// cascade child must not re-lock and revert the parent's bump). // cascade child must not re-lock and revert the parent's bump).
for agent in ["alice", "bob"] { for agent in ["alice", "bob"] {
let job = Job::new(); let declare: Declare = Box::new(move |b| {
templates::rebuild_nodes( templates::rebuild_nodes(
&job, b,
agent, agent,
templates::RebuildOpts { templates::RebuildOpts {
relock: false, relock: false,
graceful: false, graceful: false,
}, },
None, None,
); );
q.append_subgraph(id, job, meta_lock.node_id); });
q.append_subgraph(id, declare, meta_lock.node_id);
} }
q.complete_node(meta_lock.node_id, Ok(())); q.complete_node(meta_lock.node_id, Ok(()));
// Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root // Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root

View file

@ -334,7 +334,7 @@ fn submit_boot_tree(
n_deferred: usize, n_deferred: usize,
n_skipped: usize, n_skipped: usize,
) { ) {
use crate::job_queue::{DagSpec, Job, NodeKind, Source, templates}; use crate::job_queue::{DagSpec, NodeKind, Source, templates};
// 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() {
@ -348,28 +348,29 @@ fn submit_boot_tree(
n_skipped, n_skipped,
); );
let job = Job::new(); let declare: crate::job_queue::Declare = Box::new(move |b| {
// Sweep whenever ANY marker is stale — even when every stale agent is // Sweep whenever ANY marker is stale — even when every stale agent is
// wanted-offline: the hyperhive lock bump must land now so their later // wanted-offline: the hyperhive lock bump must land now so their later
// start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock // start-upgrade rebuilds build against it. No stale agents ⇒ no MetaLock
// ⇒ no meta commit on a no-change boot. The `fanout` list rides the // ⇒ no meta commit on a no-change boot. The `fanout` list rides the
// MetaLock into `run_meta_lock`, which appends the rebuild subgraphs. // MetaLock into `run_meta_lock`, which appends the rebuild subgraphs.
if any_stale { if any_stale {
let _ = templates::node( let _ = templates::node(
&job, b,
NodeKind::MetaLock { NodeKind::MetaLock {
sweep: true, sweep: true,
fanout: Some(fanout), fanout: Some(fanout),
// A sweep bumps `hyperhive` alone (`lock_update_hyperhive`), // A sweep bumps `hyperhive` alone (`lock_update_hyperhive`),
// so it names no inputs. // so it names no inputs.
inputs: Vec::new(), inputs: Vec::new(),
}, },
); );
} }
// One boot Reconcile per drifted agent — independent roots. // One boot Reconcile per drifted agent — independent roots.
for name in drifted { for name in drifted {
let _ = templates::node(&job, NodeKind::Reconcile { agent: name }); let _ = templates::node(b, NodeKind::Reconcile { agent: name });
} }
});
let spec = DagSpec { let spec = DagSpec {
// The sweep's own rebuild subgraphs emit their `Rebuilt` events as they // The sweep's own rebuild subgraphs emit their `Rebuilt` events as they
@ -379,7 +380,7 @@ fn submit_boot_tree(
// Rebuilding when the sweep will grow rebuild subgraphs (per-agent // Rebuilding when the sweep will grow rebuild subgraphs (per-agent
// crash-watch suppression during their Swap, applied at claim time); // crash-watch suppression during their Swap, applied at claim time);
// a reconcile-only boot needs no transient. // a reconcile-only boot needs no transient.
job, declare,
}; };
if let Err(e) = coord.job_queue.submit(spec) { 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");

View file

@ -5,11 +5,14 @@
//! than computing where that node landed, and there is no positional index to //! than computing where that node landed, and there is no positional index to
//! get wrong. //! get wrong.
//! //!
//! **An insertion API, not a spec factory.** [`JobBuilder::insert_into`] //! **An insertion API, not a spec factory.** A builder is only ever handed to a
//! consumes the builder and puts the nodes straight into a [`Graph`], returning //! closure by an insertion entry point ([`Graph::insert_job`],
//! the ids the graph minted. Nothing job-shaped comes back out — there is no //! [`crate::scheduler::Scheduler::insert_job`]), which inserts the declared
//! intermediate node-description type to keep in sync with [`Graph::insert`]'s //! nodes and returns the ids the graph minted. It cannot be constructed, held
//! signature. //! or inserted from outside this crate, and there is no intermediate
//! node-description type to keep in sync with [`Graph::insert`]'s signature —
//! so a job has no representation that can be passed around instead of being
//! inserted.
//! //!
//! **Payload-agnostic.** Generic over the same `N` and `R` as [`Graph`]: the //! **Payload-agnostic.** Generic over the same `N` and `R` as [`Graph`]: the
//! builder knows nothing about what a node *does*, only how nodes relate. //! builder knows nothing about what a node *does*, only how nodes relate.