d879d3e67a
Three template tests ran a whole DAG -- claim, complete, repeat -- to observe an order that is fully determined the moment submit returns. They now read the graph directly: kinds, parent nesting, and dep edges with the outcome set each accepts. rebuild_chain_claims_in_dep_order is renamed, because the old name was wrong about the mechanism and reading it rather than the graph is how you stay wrong: only half that chain is dep edges. stop_for_update and swap declare no deps at all and are ordered by parent nesting -- a node's sub-nodes run after its own logic. Both axes are asserted now, since a template can break either independently. declared_shape spells out each edge's accepted outcomes rather than bucketing them into ok/any. Bucketing made spawn's three ResolveApproval tails -- which differ only in accepted outcome -- render as identical rows, which would have made the assertion a tautology. Checked by mutation against the code under test rather than the assertion: dropping .after_ok(signal) from the graceful-stop Drain fails graceful_stop_shape with a diff naming the one missing edge.
1980 lines
75 KiB
Rust
1980 lines
75 KiB
Rust
//! Queue-core unit tests: submit / no-dedup, cycle rejection, resource
|
|
//! serialization (build slots / per-agent leases), lease-exempt
|
|
//! overlap, FIFO fairness, cancel semantics, `AfterAny` failure
|
|
//! routing, in-DAG subgraph growth, and history retention. All
|
|
//! synchronous — the
|
|
//! scheduler's async loop is a thin claim/complete pump over the same
|
|
//! methods exercised here.
|
|
|
|
use super::model::NodeKind;
|
|
use super::*;
|
|
|
|
fn submit<F: FnOnce(&Job)>(q: &JobQueue, spec: DagSpec<F>) -> u64 {
|
|
q.submit(spec).expect("valid spec")
|
|
}
|
|
|
|
/// A spec recipe with its concrete closure type erased. **Test-only** — the
|
|
/// module used to export this alias for the executor's growth path too, which
|
|
/// is exactly what a node declaring onto its own builder removed: nothing in
|
|
/// production stores a recipe to replay later, so nothing needs to box one.
|
|
type ErasedRecipe = Box<dyn FnOnce(&Job) + Send>;
|
|
|
|
/// Erase a spec's recipe so specs of *different* shapes can share one type —
|
|
/// e.g. a table of `(name, spec)` cases.
|
|
///
|
|
/// Production never needs this: each submit path builds one spec and hands it
|
|
/// straight to `submit`, so the concrete closure type is known end to end. A
|
|
/// test table is the one case where several shapes must be one type.
|
|
fn erase<F: FnOnce(&Job) + Send + 'static>(spec: DagSpec<F>) -> DagSpec<ErasedRecipe> {
|
|
DagSpec {
|
|
source: spec.source,
|
|
reason: spec.reason,
|
|
declare: Box::new(spec.declare),
|
|
}
|
|
}
|
|
|
|
fn ident(s: &str) -> hive_types::Ident {
|
|
hive_types::Ident::parse(s).expect("valid test ident")
|
|
}
|
|
|
|
fn rebuild(agent: &str, reason: &str) -> DagSpec<impl FnOnce(&Job) + use<>> {
|
|
templates::rebuild(agent, Source::Manual, reason.to_owned(), true)
|
|
}
|
|
|
|
/// Restart DAG spec 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<impl FnOnce(&Job) + use<>> {
|
|
let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect();
|
|
submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned())
|
|
}
|
|
|
|
/// Stop DAG spec with every agent treated as **running** — the online shape
|
|
/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`).
|
|
fn stop_online(
|
|
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();
|
|
submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned())
|
|
}
|
|
|
|
/// What a test observes about a node the settle loop just started: its id, its
|
|
/// DAG, and its payload.
|
|
///
|
|
/// **Test-only, and deliberately not a production type.** `exec::run_node`
|
|
/// takes `(NodeId, &NodeKind)` and derives the DAG id on the two arms that
|
|
/// actually want it — nothing in production needs a claim snapshot to exist.
|
|
/// The assertions here are about *which* node the graph let run, which does
|
|
/// need the payload next to the id.
|
|
#[derive(Debug, Clone)]
|
|
struct Claimed {
|
|
dag_id: u64,
|
|
node_id: NodeId,
|
|
kind: NodeKind,
|
|
agent: String,
|
|
}
|
|
|
|
/// Drive one settle wave and report every node that started.
|
|
///
|
|
/// An **extension trait rather than a method on [`JobQueue`]**: production
|
|
/// claims one node at a time ([`hive_jobq::scheduler::Scheduler::claim_next`])
|
|
/// and has no use for a whole wave, so this must not be reachable from
|
|
/// non-test code. `settle()` is that same claim primitive in a loop, so a test
|
|
/// driving it here exercises the production path.
|
|
trait ClaimReady {
|
|
fn claim_ready(&self) -> Vec<Claimed>;
|
|
}
|
|
|
|
impl ClaimReady for JobQueue {
|
|
fn claim_ready(&self) -> Vec<Claimed> {
|
|
let mut sched = self.sched().lock().expect("job_queue mutex poisoned");
|
|
let started = sched.settle();
|
|
started
|
|
.into_iter()
|
|
.filter_map(|node_id| {
|
|
let kind = sched.graph().node(node_id)?.payload.clone();
|
|
Some(Claimed {
|
|
dag_id: sched.graph().root_of(node_id)?.get(),
|
|
node_id,
|
|
agent: kind.agent().to_owned(),
|
|
kind,
|
|
})
|
|
})
|
|
.collect()
|
|
}
|
|
}
|
|
|
|
/// Drive a node terminal by hand.
|
|
///
|
|
/// Also an extension trait, for the same reason as [`ClaimReady`]: production
|
|
/// completes a node **inside** the future
|
|
/// [`hive_jobq::scheduler::Scheduler::claim_next`] hands back, so "run the node,
|
|
/// then remember to complete it" is not an expressible sequence there — which
|
|
/// was the whole point of the seam. These tests need to express it, because
|
|
/// they exercise the graph without running any executor.
|
|
trait CompleteNode {
|
|
fn complete_node(&self, node_id: NodeId, result: Result<(), String>);
|
|
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job);
|
|
fn new_job(&self) -> Job;
|
|
}
|
|
|
|
impl CompleteNode for JobQueue {
|
|
/// Mint a builder to declare growth into.
|
|
///
|
|
/// Test-only for the same reason as the rest of this trait: production
|
|
/// never mints one, because `claim_next` hands each running node its
|
|
/// builder and takes it back. That leaves `Scheduler::new_job` with no
|
|
/// non-test caller either — see the note on that fn.
|
|
fn new_job(&self) -> Job {
|
|
self.sched()
|
|
.lock()
|
|
.expect("job_queue mutex poisoned")
|
|
.new_job()
|
|
}
|
|
|
|
fn complete_node(&self, node_id: NodeId, result: Result<(), String>) {
|
|
self.sched()
|
|
.lock()
|
|
.expect("job_queue mutex poisoned")
|
|
.complete(node_id, outcome_of(result));
|
|
self.notify.notify_one();
|
|
}
|
|
|
|
fn complete_node_growing(&self, node_id: NodeId, result: Result<(), String>, grown: Job) {
|
|
// A rejected grown job is logged, not propagated: the node's own work
|
|
// already ran, and refusing to complete it here would both misreport
|
|
// that and wedge the DAG on a node stuck `Running`.
|
|
if let Err(e) = self
|
|
.sched()
|
|
.lock()
|
|
.expect("job_queue mutex poisoned")
|
|
.complete_growing(node_id, outcome_of(result), grown)
|
|
{
|
|
tracing::error!(
|
|
node = node_id.get(),
|
|
error = %e,
|
|
"job_queue: work grown by a completing node was rejected"
|
|
);
|
|
}
|
|
self.notify.notify_one();
|
|
}
|
|
}
|
|
|
|
/// One node's **declared** shape: what it is, what it hangs under, and what it
|
|
/// waits for — all by kind, since ids are not stable across runs.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
struct Declared {
|
|
kind: &'static str,
|
|
/// Parent kind, or `None` when the node hangs directly under the DAG
|
|
/// container (i.e. it is a group root).
|
|
parent: Option<&'static str>,
|
|
/// Kinds this node declared a node-dep on, in declaration order, each with
|
|
/// the outcome set that satisfies it.
|
|
///
|
|
/// The outcome set is **not** decoration: a template emits its tails as a
|
|
/// pair edged on the same upstream nodes, and the *only* thing telling the
|
|
/// ok-tail from the fail-tail is which outcomes each accepts. Without it
|
|
/// two structurally different nodes read as identical.
|
|
after: Vec<(&'static str, String)>,
|
|
}
|
|
|
|
/// Render a dep's outcome set as the outcomes it actually accepts.
|
|
///
|
|
/// ⚠️ Spelled out rather than bucketed into `ok` / `any` / other. The first
|
|
/// version of this did bucket, and a template's three `ResolveApproval` tails —
|
|
/// which differ ONLY in their accepted outcomes — all rendered as `"other"`.
|
|
/// A helper that prints two structurally different nodes identically turns an
|
|
/// assertion into a tautology.
|
|
fn when_tag(when: hive_jobq::DepWhen) -> String {
|
|
[
|
|
(TerminalState::Done, "done"),
|
|
(TerminalState::Failed, "failed"),
|
|
(TerminalState::Cancelled, "cancelled"),
|
|
(TerminalState::Skipped, "skipped"),
|
|
]
|
|
.into_iter()
|
|
.filter(|(outcome, _)| when.accepts(*outcome))
|
|
.map(|(_, name)| name)
|
|
.collect::<Vec<_>>()
|
|
.join("|")
|
|
}
|
|
|
|
/// Every work node under `dag`, in insertion order, as its declared shape.
|
|
///
|
|
/// **This is what the template tests are actually about.** A template's output
|
|
/// is fully determined the moment `submit` returns: the kinds, the parent
|
|
/// nesting and the dep edges are all sitting in the graph. Reading them here
|
|
/// keeps the assertion on c0re's own product. Whether the scheduler then
|
|
/// *honours* those edges — runs a chain serially, holds a grant across a
|
|
/// subtree — is `hive_jobq`'s property and is tested in `hive_jobq`.
|
|
fn declared_shape(q: &JobQueue, dag: u64) -> Vec<Declared> {
|
|
let sched = q.sched().lock().expect("job_queue mutex poisoned");
|
|
let graph = sched.graph();
|
|
let root = graph.resolve_id(dag).expect("dag id is a real node id");
|
|
let kind_of = |id: NodeId| graph.node(id).map(|n| n.payload.as_str());
|
|
graph
|
|
.nodes()
|
|
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root))
|
|
.map(|n| Declared {
|
|
kind: n.payload.as_str(),
|
|
parent: n.parent.filter(|p| *p != root).and_then(kind_of),
|
|
after: n
|
|
.deps
|
|
.iter()
|
|
.filter_map(|dep| match dep {
|
|
hive_jobq::Dep::Node { id, when } => kind_of(*id).map(|k| (k, when_tag(*when))),
|
|
hive_jobq::Dep::Resource { .. } => None,
|
|
})
|
|
.collect(),
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// Shorthand for one expected row, so the tables below read as a shape.
|
|
fn row(
|
|
kind: &'static str,
|
|
parent: Option<&'static str>,
|
|
after: &[(&'static str, &str)],
|
|
) -> Declared {
|
|
Declared {
|
|
kind,
|
|
parent,
|
|
after: after.iter().map(|(k, w)| (*k, (*w).to_owned())).collect(),
|
|
}
|
|
}
|
|
|
|
/// Claim helper asserting exactly one node comes back.
|
|
fn claim_one(q: &JobQueue) -> Claimed {
|
|
let mut claims = q.claim_ready();
|
|
assert_eq!(
|
|
claims.len(),
|
|
1,
|
|
"expected exactly one claim, got {claims:?}"
|
|
);
|
|
claims.pop().expect("one claim")
|
|
}
|
|
|
|
/// Claim an approval DAG's `ResolveApproval` tail and complete it, asserting it
|
|
/// is the one built for `expect`.
|
|
///
|
|
/// A template emits one tail per outcome and the graph runs exactly one, so the
|
|
/// assertion is on *which node was claimed* — that alone says what the approval
|
|
/// row is about to be resolved as. Nothing computes it.
|
|
fn settle_approval_tail(q: &JobQueue, approval_id: i64, expect: TerminalState) {
|
|
let tail = claim_one(q);
|
|
assert!(
|
|
matches!(
|
|
tail.kind,
|
|
NodeKind::ResolveApproval { approval_id: got, outcome }
|
|
if got == approval_id && outcome == expect
|
|
),
|
|
"expected the {expect:?} ResolveApproval tail for #{approval_id}, got {:?}",
|
|
tail.kind
|
|
);
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
}
|
|
|
|
/// The `EmitRebuilt` counterpart of [`settle_approval_tail`] — claim the tail the
|
|
/// graph let run and assert it's the `ok` one expected.
|
|
fn settle_rebuild_tail(q: &JobQueue, agent: &str, expect_ok: bool) {
|
|
let tail = claim_one(q);
|
|
assert!(
|
|
matches!(&tail.kind, NodeKind::EmitRebuilt { agent: a, ok } if a == agent && *ok == expect_ok),
|
|
"expected the ok={expect_ok} EmitRebuilt tail for {agent}, got {:?}",
|
|
tail.kind
|
|
);
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
}
|
|
|
|
/// The resources a node **declared**, read off its graph edges.
|
|
///
|
|
/// The declaration is the thing under test now that construction sites state
|
|
/// their own holdings: asking the `NodeKind` what it "should" need would just
|
|
/// re-run the derivation this module removed, and would pass even if the
|
|
/// construction site declared nothing.
|
|
fn declared_resources(q: &JobQueue, node_id: hive_jobq::NodeId) -> Vec<Resource> {
|
|
let inner = q.lock();
|
|
inner
|
|
.graph()
|
|
.node(node_id)
|
|
.expect("node exists")
|
|
.deps
|
|
.iter()
|
|
.filter_map(|dep| match dep {
|
|
hive_jobq::Dep::Resource { name, .. } => Some(name.clone()),
|
|
hive_jobq::Dep::Node { .. } => None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn state_of(q: &JobQueue, dag_id: u64) -> State {
|
|
// A DAG whose nodes have all settled `Done` or `Skipped` drops out of the
|
|
// snapshot — absence is the completion signal, so map it to `Done`.
|
|
// Otherwise derive the roll-up from the node set, exactly as every wire
|
|
// consumer does.
|
|
q.snapshot()
|
|
.iter()
|
|
.find(|d| d.id == dag_id)
|
|
.map_or(State::Done, DagView::rollup_state)
|
|
}
|
|
|
|
// ---- submit (dedup removed — every submit is a fresh DAG) ----
|
|
|
|
#[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);
|
|
assert_eq!(q.snapshot().len(), 2);
|
|
}
|
|
|
|
/// Submit-time dedup was removed with the agent-per-node refactor (a
|
|
/// multi-agent DAG has no single agent to key a dedup on), so an identical
|
|
/// resubmit — same template + agent, still queued — now enqueues a distinct
|
|
/// DAG instead of collapsing into the pending one. Whether any dedup needs
|
|
/// reintroducing is tracked as a follow-up.
|
|
#[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");
|
|
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);
|
|
assert_eq!(q.snapshot().len(), 3);
|
|
}
|
|
|
|
#[test]
|
|
fn resubmit_while_running_is_new_dag() {
|
|
let q = JobQueue::new(1);
|
|
let a = submit(&q, rebuild("agent-a", "first"));
|
|
let claim = claim_one(&q); // Prebuild running
|
|
assert_eq!(claim.dag_id, a);
|
|
// While the original runs, re-submit is legitimate new work.
|
|
let again = submit(&q, rebuild("agent-a", "config bumped during build"));
|
|
assert_ne!(a, again);
|
|
assert_eq!(q.snapshot().len(), 2);
|
|
}
|
|
|
|
// ---- malformed specs: no longer expressible ----
|
|
//
|
|
// Three tests lived here — a dependency cycle, a dependency on a node that
|
|
// does not exist, and an out-of-range parent index — each asserting that
|
|
// `submit` refused the spec. All three built their spec by hand out of
|
|
// positional indices, which is exactly the representation that made those
|
|
// shapes possible: an index can name a node that isn't there, or one that
|
|
// comes later.
|
|
//
|
|
// A job is now declared against handles that only exist for nodes already
|
|
// declared, so there is no index to put out of range, and every edge points
|
|
// backwards — a cycle needs a forward edge. The guard those tests covered was
|
|
// deleted along with the failure mode. What remains — a handle used against a
|
|
// builder that never issued it — is `hive_jobq`'s to reject, and its builder
|
|
// tests cover it (`a_forward_edge_is_rejected_by_name`,
|
|
// `a_forward_parent_is_rejected_by_name`, `graph_rejection_surfaces_as_is`).
|
|
|
|
// ---- dependency order within a DAG ----
|
|
|
|
#[test]
|
|
fn rebuild_chain_is_declared_serial() {
|
|
// Was `rebuild_chain_claims_in_dep_order`, which drove the whole DAG to
|
|
// observe an order that is fully declared the moment `submit` returns.
|
|
//
|
|
// ⚠️ The old name was also wrong about the mechanism, and reading it rather
|
|
// than the graph is how you'd stay wrong: **only half this chain is dep
|
|
// edges.** `stop_for_update` and `swap` declare no deps at all — they are
|
|
// ordered by *parent nesting* ("a node's sub-nodes run after its own
|
|
// 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"));
|
|
assert_eq!(
|
|
declared_shape(&q, id),
|
|
vec![
|
|
row("meta_sync", None, &[]),
|
|
row("prebuild", None, &[("meta_sync", "done")]),
|
|
// No dep: ordered by hanging under `prebuild`.
|
|
row("stop_for_update", Some("prebuild"), &[]),
|
|
row("swap", Some("stop_for_update"), &[]),
|
|
row("post_swap", Some("stop_for_update"), &[("swap", "done")]),
|
|
row("reconcile", None, &[("prebuild", "done|failed|skipped")]),
|
|
// The tail pair. The ok tail needs every root to succeed; the !ok
|
|
// tail hangs off the ok tail's *elimination* (`skipped`), which is
|
|
// what makes exactly one of them run.
|
|
row(
|
|
"emit_rebuilt",
|
|
None,
|
|
&[
|
|
("meta_sync", "done"),
|
|
("prebuild", "done"),
|
|
("reconcile", "done"),
|
|
],
|
|
),
|
|
row(
|
|
"emit_rebuilt",
|
|
None,
|
|
&[
|
|
("emit_rebuilt", "skipped"),
|
|
("meta_sync", "done|failed|skipped"),
|
|
("prebuild", "done|failed|skipped"),
|
|
("reconcile", "done|failed|skipped"),
|
|
],
|
|
),
|
|
]
|
|
);
|
|
}
|
|
|
|
/// The boot sweep's graceful shape: the agent gets `Signal` → `Drain` to
|
|
/// finish its turn before `StopForUpdate` takes the container down. `Signal`
|
|
/// *parents* the rest of the stop rather than sitting beside it, so the agent
|
|
/// lease is held continuously across the whole bounce — as siblings, each of
|
|
/// `Signal` / `Drain` / `StopForUpdate` would acquire the lease separately and
|
|
/// leave a window for another DAG to claim the agent mid-stop.
|
|
#[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,
|
|
);
|
|
}),
|
|
},
|
|
);
|
|
for expected in [
|
|
"meta_sync",
|
|
"prebuild",
|
|
"signal",
|
|
"drain",
|
|
"stop_for_update",
|
|
"swap",
|
|
"post_swap",
|
|
"reconcile",
|
|
] {
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.dag_id, id);
|
|
assert_eq!(c.kind.as_str(), expected);
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"chain must serialize: nothing ready while {expected} runs"
|
|
);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
/// The non-graceful shape is the default everywhere except the boot sweep:
|
|
/// a manual rebuild, a meta-update cascade child and a deploy must NOT spend a
|
|
/// drain window, so `StopForUpdate` still hangs straight off `Prebuild`.
|
|
#[test]
|
|
fn non_graceful_rebuild_has_no_signal_or_drain() {
|
|
// Read the shape off the queue rather than out of a node list: a declared
|
|
// 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 mut kinds = Vec::new();
|
|
for _ in 0..6 {
|
|
let c = claim_one(&q);
|
|
kinds.push(c.kind.as_str().to_owned());
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
assert_eq!(
|
|
kinds,
|
|
vec![
|
|
"meta_sync",
|
|
"prebuild",
|
|
"stop_for_update",
|
|
"swap",
|
|
"post_swap",
|
|
"reconcile"
|
|
]
|
|
);
|
|
// Settled after exactly those six — nothing else was declared.
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
/// A cleanly-finished DAG leaves the snapshot even though its not-taken
|
|
/// failure branch is still in the graph as `Skipped`. Skipped nodes ride the
|
|
/// wire so the dashboard can mark them, which makes "the node list is empty"
|
|
/// and "nothing here is still worth showing" two different questions — only
|
|
/// the second one may drop the DAG. Conflating them pins every completed
|
|
/// deploy in the queue view forever.
|
|
#[test]
|
|
fn settled_dag_leaves_the_snapshot_despite_its_skipped_branch() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
for _ in 0..6 {
|
|
let c = claim_one(&q);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
settle_rebuild_tail(&q, "agent-a", true);
|
|
assert!(
|
|
q.snapshot().iter().all(|d| d.id != id),
|
|
"a fully settled DAG drops out of the snapshot"
|
|
);
|
|
}
|
|
|
|
// ---- build slots ----
|
|
|
|
#[test]
|
|
fn build_slot_serializes_nix_heavy_nodes() {
|
|
let q = JobQueue::new(1);
|
|
let a = submit(&q, rebuild("agent-a", "r"));
|
|
let b = submit(&q, rebuild("agent-b", "r"));
|
|
// The rebuild heads are `MetaSync` (slot-free, but serialized on the
|
|
// global meta window), so drive each chain's head out of the way first.
|
|
let head_a = claim_one(&q);
|
|
assert_eq!(head_a.dag_id, a);
|
|
assert_eq!(head_a.kind.as_str(), "meta_sync");
|
|
q.complete_node(head_a.node_id, Ok(()));
|
|
// a's Prebuild takes the only slot; b's MetaSync is free to run beside it
|
|
// (different resources), but b's Prebuild is not.
|
|
let claims = q.claim_ready();
|
|
let mut kinds: Vec<(u64, &str)> = claims.iter().map(|c| (c.dag_id, c.kind.as_str())).collect();
|
|
kinds.sort_unstable();
|
|
assert_eq!(kinds, vec![(a, "prebuild"), (b, "meta_sync")]);
|
|
for c in &claims {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
// Uniform hold: agent-a keeps the build slot across its whole build chain
|
|
// (Swap re-enters it), so a's StopForUpdate (lease, slot-free) runs but b's
|
|
// Prebuild must wait for a's slot-needers (through Swap) to finish.
|
|
let kinds: Vec<(u64, &str)> = q
|
|
.claim_ready()
|
|
.iter()
|
|
.map(|c| (c.dag_id, c.kind.as_str()))
|
|
.collect();
|
|
assert_eq!(kinds, vec![(a, "stop_for_update")]);
|
|
assert!(
|
|
!kinds.iter().any(|&(d, _)| d == b),
|
|
"b's build waits — slot held across a's chain"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn two_build_slots_run_two_prebuilds() {
|
|
let q = JobQueue::new(2);
|
|
submit(&q, rebuild("agent-a", "r"));
|
|
submit(&q, rebuild("agent-b", "r"));
|
|
// Each rebuild's head `MetaSync` holds the cap-1 global meta window, so the
|
|
// two heads take turns — exactly the serialization the old runtime
|
|
// `meta::exclusive()` mutex imposed inside the prebuild executor. What must
|
|
// NOT serialize is the build itself: complete only the meta heads and watch
|
|
// both prebuilds end up in flight together, neither of them completed.
|
|
let mut prebuilds = Vec::new();
|
|
for _ in 0..3 {
|
|
for c in q.claim_ready() {
|
|
if c.kind.as_str() == "meta_sync" {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
} else {
|
|
prebuilds.push(c);
|
|
}
|
|
}
|
|
}
|
|
assert_eq!(prebuilds.len(), 2, "two slots → two concurrent prebuilds");
|
|
assert!(prebuilds.iter().all(|c| c.kind.as_str() == "prebuild"));
|
|
}
|
|
|
|
#[test]
|
|
fn fifo_fairness_for_the_slot() {
|
|
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, rebuild("agent-c", "r"));
|
|
let first = claim_one(&q);
|
|
assert_eq!(first.dag_id, a, "submit order wins the slot");
|
|
q.complete_node(first.node_id, Ok(()));
|
|
// Uniform hold: the slot stays with agent-a until its Swap (the last
|
|
// slot-needer) completes. Drive a's chain; the moment its slot frees,
|
|
// submit order (b before c) wins it.
|
|
let mut freed_to = None;
|
|
for _ in 0..6 {
|
|
let claims = q.claim_ready();
|
|
if let Some(nb) = claims.iter().find(|cl| cl.dag_id == b || cl.dag_id == c) {
|
|
freed_to = Some(nb.dag_id);
|
|
break;
|
|
}
|
|
for cl in claims {
|
|
if cl.dag_id == a {
|
|
q.complete_node(cl.node_id, Ok(()));
|
|
}
|
|
}
|
|
}
|
|
assert_eq!(
|
|
freed_to,
|
|
Some(b),
|
|
"b's prebuild wins the freed slot before c's"
|
|
);
|
|
}
|
|
|
|
// ---- per-agent lease ----
|
|
|
|
#[test]
|
|
fn lease_serializes_two_lifecycle_dags_for_same_agent() {
|
|
let q = JobQueue::new(4);
|
|
let restart = submit(&q, restart_online(&["agent-a"], false, "restart"));
|
|
let stop = submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()),
|
|
);
|
|
// Restart's first node (StopForUpdate) takes the lease; stop's
|
|
// Reconcile must wait even though slots are free.
|
|
let first = claim_one(&q);
|
|
assert_eq!(first.dag_id, restart);
|
|
assert_eq!(first.kind.as_str(), "stop_for_update");
|
|
q.complete_node(first.node_id, Ok(()));
|
|
// Same DAG keeps the lease through the tail Reconcile (re-entered from the
|
|
// dep graph — no fresh acquire), since stop's Reconcile can't re-enter it.
|
|
let second = claim_one(&q);
|
|
assert_eq!(second.dag_id, restart);
|
|
assert_eq!(second.kind.as_str(), "reconcile");
|
|
q.complete_node(second.node_id, Ok(()));
|
|
// Restart's work is terminal → its lease releases, so stop's now-unblocked
|
|
// Reconcile becomes ready (a power op has no tail node, so nothing of
|
|
// restart's remains claimable).
|
|
let third = claim_one(&q);
|
|
assert_eq!(third.dag_id, stop);
|
|
assert_eq!(third.kind.as_str(), "reconcile");
|
|
q.complete_node(third.node_id, Ok(()));
|
|
assert_eq!(state_of(&q, restart), State::Done);
|
|
assert_eq!(state_of(&q, stop), State::Done);
|
|
}
|
|
|
|
#[test]
|
|
fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() {
|
|
let q = JobQueue::new(2);
|
|
submit(&q, rebuild("agent-a", "rebuild"));
|
|
submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()),
|
|
);
|
|
// Both DAGs' heads are lease-independent of each other: the rebuild's
|
|
// MetaSync (meta window) and the stop's Reconcile (agent lease).
|
|
let heads = q.claim_ready();
|
|
let head_kinds: Vec<&str> = heads.iter().map(|c| c.kind.as_str()).collect();
|
|
assert!(head_kinds.contains(&"meta_sync"));
|
|
assert!(head_kinds.contains(&"reconcile"));
|
|
let meta_sync = heads
|
|
.iter()
|
|
.find(|c| c.kind.as_str() == "meta_sync")
|
|
.expect("meta_sync claim")
|
|
.clone();
|
|
q.complete_node(meta_sync.node_id, Ok(()));
|
|
// Prebuild is lease-exempt: the stop's Reconcile keeps the lease
|
|
// and runs concurrently with the rebuild's out-of-band nix build.
|
|
let claims = q.claim_ready();
|
|
let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect();
|
|
assert!(kinds.contains(&"prebuild"));
|
|
// But the rebuild's StopForUpdate must then wait for the stop DAG
|
|
// to finish (lease).
|
|
let prebuild = claims
|
|
.iter()
|
|
.find(|c| c.kind.as_str() == "prebuild")
|
|
.expect("prebuild claim")
|
|
.clone();
|
|
q.complete_node(prebuild.node_id, Ok(()));
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"StopForUpdate blocked while stop DAG holds the lease"
|
|
);
|
|
let reconcile = heads
|
|
.iter()
|
|
.find(|c| c.kind.as_str() == "reconcile")
|
|
.expect("reconcile claim")
|
|
.clone();
|
|
q.complete_node(reconcile.node_id, Ok(()));
|
|
// stop's Reconcile done → its lease frees, so rebuild's StopForUpdate
|
|
// unblocks. (stop's DAG rolls up terminal; a power op has no tail node, so
|
|
// nothing of stop's is left in the claim set.)
|
|
let after = q.claim_ready();
|
|
let sfu = after
|
|
.iter()
|
|
.find(|c| c.kind.as_str() == "stop_for_update")
|
|
.expect("rebuild StopForUpdate unblocked once the lease frees");
|
|
assert_eq!(sfu.agent, "agent-a");
|
|
}
|
|
|
|
#[test]
|
|
fn agents_do_not_contend_on_each_others_leases() {
|
|
let q = JobQueue::new(4);
|
|
submit(&q, restart_online(&["agent-a"], false, "r"));
|
|
submit(&q, restart_online(&["agent-b"], false, "r"));
|
|
let claims = q.claim_ready();
|
|
assert_eq!(claims.len(), 2, "different agents run concurrently");
|
|
}
|
|
|
|
#[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"),
|
|
);
|
|
// 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 root, so both are claimable at once — each takes its OWN agent's
|
|
// lease (no contention across distinct agents), all inside the single DAG.
|
|
let claims = q.claim_ready();
|
|
assert!(claims.iter().all(|c| c.dag_id == id));
|
|
let mut heads: Vec<(&str, &str)> = claims
|
|
.iter()
|
|
.map(|c| (c.agent.as_str(), c.kind.as_str()))
|
|
.collect();
|
|
heads.sort_unstable();
|
|
assert_eq!(
|
|
heads,
|
|
vec![
|
|
("agent-a", "stop_for_update"),
|
|
("agent-b", "stop_for_update"),
|
|
],
|
|
"both per-agent subgraphs start concurrently, each acquiring its own lease"
|
|
);
|
|
}
|
|
|
|
/// A multi-agent DAG frees an agent's lease the moment THAT agent's
|
|
/// subgraph is terminal — not when the whole DAG finishes. So a
|
|
/// concurrent DAG wanting the finished agent can proceed while the rest
|
|
/// of the first DAG runs on.
|
|
#[test]
|
|
fn multi_agent_lease_frees_per_subgraph_not_whole_dag() {
|
|
let q = JobQueue::new(4);
|
|
let id = submit(&q, restart_online(&["agent-a", "agent-b"], false, "r"));
|
|
|
|
// Drive agent-a's ENTIRE subgraph to Done while leaving agent-b's
|
|
// head running (so agent-b keeps holding its lease).
|
|
let mut b_in_flight = false;
|
|
loop {
|
|
let mut progressed = false;
|
|
for c in q.claim_ready() {
|
|
if c.agent == "agent-a" {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
progressed = true;
|
|
} else {
|
|
b_in_flight = true; // leave agent-b's node running
|
|
}
|
|
}
|
|
if !progressed {
|
|
break;
|
|
}
|
|
}
|
|
assert!(b_in_flight, "agent-b subgraph should still be in flight");
|
|
// The DAG as a whole is NOT terminal — agent-b runs on.
|
|
assert_eq!(state_of(&q, id), State::Running);
|
|
|
|
// agent-a's lease is freed early → a concurrent agent-a DAG runs;
|
|
// an agent-b DAG still blocks on the lease agent-b's subgraph holds.
|
|
submit(&q, restart_online(&["agent-a"], false, "concurrent-a"));
|
|
submit(&q, restart_online(&["agent-b"], false, "concurrent-b"));
|
|
let claims = q.claim_ready();
|
|
let agents: Vec<&str> = claims.iter().map(|c| c.agent.as_str()).collect();
|
|
assert!(
|
|
agents.contains(&"agent-a"),
|
|
"agent-a lease freed the moment its subgraph settled"
|
|
);
|
|
assert!(
|
|
!agents.contains(&"agent-b"),
|
|
"agent-b lease still held — its subgraph is still in flight"
|
|
);
|
|
}
|
|
|
|
#[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"),
|
|
);
|
|
// A hive-wide stop is ONE DAG, not one-per-agent.
|
|
assert_eq!(q.snapshot().len(), 1);
|
|
let claims = q.claim_ready();
|
|
assert!(claims.iter().all(|c| c.dag_id == id));
|
|
let mut heads: Vec<(&str, &str)> = claims
|
|
.iter()
|
|
.map(|c| (c.agent.as_str(), c.kind.as_str()))
|
|
.collect();
|
|
heads.sort_unstable();
|
|
assert_eq!(
|
|
heads,
|
|
vec![("agent-a", "set_wanted"), ("agent-b", "set_wanted")],
|
|
"both per-agent stop subgraphs start concurrently, each on its own lease"
|
|
);
|
|
}
|
|
|
|
#[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".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);
|
|
// Both subgraph heads (SetWanted(Up)) are roots — claimable at once,
|
|
// each acquiring its own agent lease.
|
|
let heads = q.claim_ready();
|
|
assert!(
|
|
heads
|
|
.iter()
|
|
.all(|c| c.dag_id == id && c.kind.as_str() == "set_wanted")
|
|
);
|
|
let mut head_agents: Vec<&str> = heads.iter().map(|c| c.agent.as_str()).collect();
|
|
head_agents.sort_unstable();
|
|
assert_eq!(head_agents, vec!["fresh", "stale"]);
|
|
// Complete both heads; the fresh agent then reconciles directly while
|
|
// the stale agent's subgraph is the rebuild chain (meta_sync first).
|
|
for c in &heads {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
let next = q.claim_ready();
|
|
let mut kinds: Vec<(&str, &str)> = next
|
|
.iter()
|
|
.map(|c| (c.agent.as_str(), c.kind.as_str()))
|
|
.collect();
|
|
kinds.sort_unstable();
|
|
assert_eq!(
|
|
kinds,
|
|
vec![("fresh", "reconcile"), ("stale", "meta_sync")],
|
|
"fresh agent starts directly; stale agent rebuilds first, all in one DAG"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() {
|
|
// The dynamic build skips Signal/Drain/StopForUpdate for a down agent
|
|
// (nothing to quiesce/stop) but ALWAYS keeps the Reconcile tail — the
|
|
// convergence guarantee that catches a race-up between the is_running
|
|
// 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(),
|
|
),
|
|
);
|
|
// 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 shape = |id: u64| -> Vec<String> {
|
|
q.snapshot()
|
|
.iter()
|
|
.find(|d| d.id == id)
|
|
.expect("dag")
|
|
.nodes
|
|
.iter()
|
|
.map(|n| n.kind.clone())
|
|
.collect()
|
|
};
|
|
assert_eq!(
|
|
shape(stop),
|
|
vec!["set_wanted".to_owned(), "reconcile".to_owned()],
|
|
"offline graceful stop skips the signal/drain quiesce, keeps Reconcile"
|
|
);
|
|
assert_eq!(
|
|
shape(restart),
|
|
vec!["reconcile".to_owned()],
|
|
"offline restart is a lone Reconcile (no SetWanted head, nothing to stop)"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_fanned_out_start_declares_the_lease_and_re_enters_its_reconciles_grant() {
|
|
// `Start` / `Stop` / `PostSwap` were lease-exempt *as kinds*, which was only
|
|
// safe because every construction site fans them out from inside a
|
|
// lease-holding ancestor. Now they declare the lease themselves.
|
|
//
|
|
// The contract says that costs nothing — a descendant re-enters the
|
|
// ancestor's grant instead of taking a fresh unit. That is exactly the sort
|
|
// of claim that is true until a node is used from a second site, so it is
|
|
// pinned here rather than argued: the fanned-out `Start` must (a) actually
|
|
// carry the declaration, (b) still run under its parent's grant, and
|
|
// (c) not have consumed a second unit of a cap-1 lease.
|
|
let q = JobQueue::new(4);
|
|
let id = submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "converge".to_owned()),
|
|
);
|
|
// A competing DAG on the same agent, to prove the lease is genuinely held
|
|
// (and held *once*) across the fan-out.
|
|
let rival = submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "rival".to_owned()),
|
|
);
|
|
|
|
let reconcile = claim_one(&q);
|
|
assert_eq!(reconcile.dag_id, id);
|
|
assert_eq!(reconcile.kind.as_str(), "reconcile");
|
|
|
|
// What `run_reconcile` does on observing a down container with wanted=Up:
|
|
// declare into the builder it was handed, then hand it back with the
|
|
// completion. Same two calls the scheduler makes, in the same order.
|
|
let grown = q.new_job();
|
|
let kind = NodeKind::Start {
|
|
agent: "agent-a".to_owned(),
|
|
};
|
|
let lease = Resource::Agent(kind.agent().to_owned());
|
|
let _ = grown.node(kind).needs(lease);
|
|
q.complete_node_growing(reconcile.node_id, Ok(()), grown);
|
|
|
|
// (a) + (b): the child runs, under the parent that parked in `Finishing`.
|
|
let start = claim_one(&q);
|
|
assert_eq!(start.kind.as_str(), "start");
|
|
assert_eq!(
|
|
declared_resources(&q, start.node_id),
|
|
vec![Resource::Agent("agent-a".to_owned())],
|
|
"a fanned-out Start declares the lease it runs under"
|
|
);
|
|
|
|
// (c): one unit, not two. `claim_one` above already asserted the rival did
|
|
// not come back in the same pass; make the reason explicit.
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"the rival DAG's Reconcile must still be blocked — the appended Start \
|
|
borrowed the grant rather than acquiring a second unit"
|
|
);
|
|
|
|
q.complete_node(start.node_id, Ok(()));
|
|
// Subtree terminal → the grant releases and the rival finally runs.
|
|
let rival_reconcile = claim_one(&q);
|
|
assert_eq!(rival_reconcile.dag_id, rival);
|
|
q.complete_node(rival_reconcile.node_id, Ok(()));
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
assert_eq!(state_of(&q, rival), State::Done);
|
|
}
|
|
|
|
#[test]
|
|
fn boot_sweep_nodes_declare_their_own_resources() {
|
|
// Regression, and the reason it needs its own test: `workers::auto_update`
|
|
// is the only place job nodes are constructed *outside* `job_queue/`, so
|
|
// nothing in this module covered it. When resource derivation moved to the
|
|
// construction sites, this path was missed and both kinds silently declared
|
|
// nothing — dropping the agent lease a boot `Reconcile` needs to not race
|
|
// another DAG's container ops, and letting the sweep `MetaLock` land its
|
|
// meta commit inside another node's staged deploy window. Nothing failed to
|
|
// compile; only an exhaustive caller list would have caught it.
|
|
let q = JobQueue::new(4);
|
|
let _id = submit(
|
|
&q,
|
|
DagSpec {
|
|
source: Source::AutoUpdate,
|
|
reason: "boot".to_owned(),
|
|
declare: Box::new(|b: &Job| {
|
|
crate::workers::auto_update::boot_nodes(
|
|
b,
|
|
true,
|
|
vec!["stale-agent".to_owned()],
|
|
vec!["drifted-agent".to_owned()],
|
|
);
|
|
}),
|
|
},
|
|
);
|
|
|
|
// Both are independent roots on disjoint resources, so both start at once.
|
|
let claims = q.claim_ready();
|
|
let by_kind = |kind: &str| {
|
|
claims
|
|
.iter()
|
|
.find(|c| c.kind.as_str() == kind)
|
|
.unwrap_or_else(|| panic!("no {kind} claim in {claims:?}"))
|
|
};
|
|
|
|
let mut lock = declared_resources(&q, by_kind("meta_lock").node_id);
|
|
lock.sort_by_key(|r| format!("{r:?}"));
|
|
assert_eq!(
|
|
lock,
|
|
vec![Resource::BuildSlot, Resource::MetaWindow],
|
|
"the sweep MetaLock runs a nix lock bump and commits to meta"
|
|
);
|
|
|
|
assert_eq!(
|
|
declared_resources(&q, by_kind("reconcile").node_id),
|
|
vec![Resource::Agent("drifted-agent".to_owned())],
|
|
"a boot Reconcile touches the container, so it holds that agent's lease"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn grown_subgraph_roots_on_emitter_and_rebases_local_deps() {
|
|
// The startup-sweep mechanism: a `MetaLock` emitter grows one rebuild
|
|
// 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.
|
|
let q = JobQueue::new(4);
|
|
let spec = DagSpec {
|
|
source: Source::AutoUpdate,
|
|
reason: "sweep".to_owned(),
|
|
declare: Box::new(|b: &Job| {
|
|
let _lock = b.node(NodeKind::MetaLock {
|
|
sweep: true,
|
|
fanout: None,
|
|
inputs: Vec::new(),
|
|
});
|
|
}),
|
|
};
|
|
submit(&q, spec);
|
|
let emitter = claim_one(&q);
|
|
assert_eq!(emitter.kind.as_str(), "meta_lock");
|
|
// Two independent per-agent subgraphs — the REAL production shape the
|
|
// sweep MetaLock grows: root MetaSync → root Prebuild → Signal → Drain →
|
|
// StopForUpdate → Swap → Reconcile, local 0-based deps. `graceful` must
|
|
// match the sweep arm of `run_meta_lock` or this stops tracking production.
|
|
// Both subgraphs go into the emitter's own builder, exactly as
|
|
// `run_meta_lock`'s sweep arm does. Insert-before-complete is no longer the
|
|
// caller's job to remember: it is one call, and the ordering is inside it.
|
|
let grown = q.new_job();
|
|
for agent in ["a", "b"] {
|
|
templates::rebuild_nodes(
|
|
&grown,
|
|
agent,
|
|
templates::RebuildOpts {
|
|
relock: true,
|
|
graceful: true,
|
|
},
|
|
None,
|
|
);
|
|
}
|
|
q.complete_node_growing(emitter.node_id, Ok(()), grown);
|
|
// Still ONE DAG; both subgraph roots become ready once the emitter is
|
|
// Done (rooted on it), each on its own agent lease. Their `MetaSync` heads
|
|
// take turns on the cap-1 global meta window, so drain those first — what
|
|
// must be concurrent is the builds.
|
|
assert_eq!(q.snapshot().len(), 1);
|
|
let mut kinds = drain_meta_syncs(&q);
|
|
kinds.sort_unstable();
|
|
assert_eq!(
|
|
kinds,
|
|
vec![
|
|
("a".to_owned(), "prebuild".to_owned()),
|
|
("b".to_owned(), "prebuild".to_owned())
|
|
],
|
|
"both rebuild subgraphs root on the emitter and run concurrently in one DAG"
|
|
);
|
|
}
|
|
|
|
/// Complete every `MetaSync` head the queue offers (they take turns on the
|
|
/// cap-1 global meta window) and return whatever else got claimed alongside
|
|
/// them, as `(agent, kind)` pairs left in flight.
|
|
fn drain_meta_syncs(q: &JobQueue) -> Vec<(String, String)> {
|
|
let mut rest = Vec::new();
|
|
for _ in 0..3 {
|
|
for c in q.claim_ready() {
|
|
if c.kind.as_str() == "meta_sync" {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
} else {
|
|
rest.push((c.agent.clone(), c.kind.as_str().to_owned()));
|
|
}
|
|
}
|
|
}
|
|
rest
|
|
}
|
|
|
|
/// Crash-watch suppression for a cascade rebuild, which the deleted half of
|
|
/// `meta_update_grows_cascade_in_dag` used to assert via `DagSpec::transient`.
|
|
///
|
|
/// 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
|
|
/// answers for itself, so the assertion moves to the nodes a cascade actually
|
|
/// runs. Kept as its own test rather than dropped, because it is the *property*
|
|
/// that mattered, not the field that used to carry it.
|
|
#[test]
|
|
fn rebuild_chain_nodes_suppress_crash_watch() {
|
|
for kind in [
|
|
NodeKind::StopForUpdate {
|
|
agent: "a".to_owned(),
|
|
},
|
|
NodeKind::Swap {
|
|
agent: "a".to_owned(),
|
|
},
|
|
NodeKind::Drain {
|
|
agent: "a".to_owned(),
|
|
},
|
|
] {
|
|
assert!(
|
|
kind.takes_container_down(),
|
|
"{} must suppress crash-watch — a rebuild takes the container down \
|
|
on purpose",
|
|
kind.as_str()
|
|
);
|
|
}
|
|
// The counter-case, and the reason this can't be "any node in a rebuild":
|
|
// the tail brings the container back up, so a container that dies there
|
|
// really did crash.
|
|
assert!(
|
|
!NodeKind::Start {
|
|
agent: "a".to_owned()
|
|
}
|
|
.takes_container_down()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn meta_update_grows_cascade_in_dag() {
|
|
// The meta-update `MetaLock` grows one rebuild subgraph per affected
|
|
// agent into its OWN DAG (via the builder it is handed), not child DAGs.
|
|
let spec = templates::meta_update(
|
|
vec!["nixpkgs".to_owned()],
|
|
Source::Manual,
|
|
"bump".to_owned(),
|
|
None,
|
|
);
|
|
let q = JobQueue::new(4);
|
|
submit(&q, spec);
|
|
let meta_lock = claim_one(&q);
|
|
assert_eq!(meta_lock.kind.as_str(), "meta_lock");
|
|
// Simulate the executor growing the cascade in-DAG (`relock = false` — a
|
|
// cascade child must not re-lock and revert the parent's bump). Both
|
|
// agents go into the one builder the node was handed, which is what
|
|
// `run_meta_lock`'s fanout arm does.
|
|
let grown = q.new_job();
|
|
for agent in ["alice", "bob"] {
|
|
templates::rebuild_nodes(
|
|
&grown,
|
|
agent,
|
|
templates::RebuildOpts {
|
|
relock: false,
|
|
graceful: false,
|
|
},
|
|
None,
|
|
);
|
|
}
|
|
q.complete_node_growing(meta_lock.node_id, Ok(()), grown);
|
|
// Still ONE DAG — no child DAGs — and both cascade rebuild subgraphs root
|
|
// on the MetaLock, each on its own agent lease. The per-agent `MetaSync`
|
|
// heads serialize on the global meta window (they commit to the meta repo);
|
|
// the builds behind them do not.
|
|
assert_eq!(q.snapshot().len(), 1);
|
|
let mut kinds = drain_meta_syncs(&q);
|
|
kinds.sort_unstable();
|
|
assert_eq!(
|
|
kinds,
|
|
vec![
|
|
("alice".to_owned(), "prebuild".to_owned()),
|
|
("bob".to_owned(), "prebuild".to_owned())
|
|
],
|
|
"cascade rebuilds grow in the meta-update DAG, concurrent per agent"
|
|
);
|
|
}
|
|
|
|
// ---- failure: cancel-downstream + AfterAny ----
|
|
|
|
#[test]
|
|
fn failed_node_cancels_downstream_but_afterany_reconcile_runs() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
let meta_sync = claim_one(&q);
|
|
assert_eq!(meta_sync.kind.as_str(), "meta_sync");
|
|
q.complete_node(meta_sync.node_id, Ok(()));
|
|
let prebuild = claim_one(&q);
|
|
assert_eq!(prebuild.kind.as_str(), "prebuild");
|
|
q.complete_node(prebuild.node_id, Err("nix build exploded".to_owned()));
|
|
// StopForUpdate + Swap are cancelled (AfterOk on a failed chain);
|
|
// the AfterAny Reconcile still runs once Swap is terminal.
|
|
let reconcile = claim_one(&q);
|
|
assert_eq!(reconcile.kind.as_str(), "reconcile");
|
|
q.complete_node(reconcile.node_id, Ok(()));
|
|
let snap = q.snapshot();
|
|
let dag = snap.iter().find(|d| d.id == id).expect("dag");
|
|
assert_eq!(dag.rollup_state(), State::Failed, "roll-up failed");
|
|
let by_kind = |k: &str| {
|
|
dag.nodes
|
|
.iter()
|
|
.find(|n| n.kind == k)
|
|
.expect("node present")
|
|
.state
|
|
};
|
|
assert_eq!(by_kind("prebuild"), State::Failed);
|
|
// `StopForUpdate` / `Swap` / `PostSwap` were *ruled out* by the failed
|
|
// `Prebuild`. They ride the wire as `Skipped` so an operator can see which
|
|
// steps the run never reached, without them reading as failures of their
|
|
// own — the roll-up ignores `Skipped` entirely.
|
|
for ruled_out in ["stop_for_update", "swap", "post_swap"] {
|
|
assert_eq!(
|
|
by_kind(ruled_out),
|
|
State::Skipped,
|
|
"{ruled_out} was ruled out, so it is on the wire as skipped"
|
|
);
|
|
}
|
|
// The AfterAny reconcile ran (claimed + completed Ok above) → it's `Done`,
|
|
// and `Done` nodes are excluded from the wire, so it's absent here.
|
|
assert!(
|
|
dag.nodes.iter().all(|n| n.kind != "reconcile"),
|
|
"the completed (Done) reconcile is filtered off the wire"
|
|
);
|
|
assert_eq!(
|
|
dag.nodes
|
|
.iter()
|
|
.find(|n| n.kind == "prebuild")
|
|
.and_then(|n| n.error.as_deref()),
|
|
Some("nix build exploded")
|
|
);
|
|
}
|
|
|
|
/// The swap-failure recovery: `Swap` fails → the `AfterOk` `PostSwap` is
|
|
/// cancel-cascaded → its terminal state still satisfies `Reconcile`'s
|
|
/// `AfterAny(PostSwap)` edge, so recovery-start runs and brings a wanted-up
|
|
/// agent back on its old config.
|
|
#[test]
|
|
fn swap_failure_still_runs_reconcile() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
// meta_sync + prebuild + stop_for_update
|
|
for _ in 0..3 {
|
|
let c = claim_one(&q);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
let swap = claim_one(&q);
|
|
assert_eq!(swap.kind.as_str(), "swap");
|
|
q.complete_node(swap.node_id, Err("update failed".to_owned()));
|
|
// PostSwap (AfterOk on the failed Swap) is cancel-cascaded; Reconcile is
|
|
// next-claimable via its AfterAny(PostSwap) edge.
|
|
let reconcile = claim_one(&q);
|
|
assert_eq!(reconcile.kind.as_str(), "reconcile");
|
|
q.complete_node(reconcile.node_id, Ok(()));
|
|
let all_dags = q.snapshot();
|
|
let dag = all_dags.iter().find(|d| d.id == id).expect("dag");
|
|
assert_eq!(
|
|
dag.nodes
|
|
.iter()
|
|
.find(|n| n.kind == "post_swap")
|
|
.expect("post_swap node")
|
|
.state,
|
|
State::Skipped,
|
|
"PostSwap is ruled out by the failed Swap, and says so on the wire"
|
|
);
|
|
assert_eq!(
|
|
dag.nodes
|
|
.iter()
|
|
.find(|n| n.kind == "swap")
|
|
.expect("swap node")
|
|
.state,
|
|
State::Failed,
|
|
"and the failure that ruled it out is still on the wire"
|
|
);
|
|
assert_eq!(state_of(&q, id), State::Failed);
|
|
}
|
|
|
|
/// The swap-success path: `Swap` ok → the `AfterOk` `PostSwap` (bookkeeping
|
|
/// tail) runs, and only then does `Reconcile` fire — serialized behind
|
|
/// `PostSwap` (not racing it) because `Reconcile` deps `AfterAny(PostSwap)`.
|
|
#[test]
|
|
fn swap_ok_runs_post_swap_before_reconcile() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
// meta_sync + prebuild + stop_for_update
|
|
for _ in 0..3 {
|
|
let c = claim_one(&q);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
let swap = claim_one(&q);
|
|
assert_eq!(swap.kind.as_str(), "swap");
|
|
q.complete_node(swap.node_id, Ok(()));
|
|
// PostSwap runs next, and nothing else is claimable while it does — the
|
|
// tail serializes ahead of Reconcile.
|
|
let post_swap = claim_one(&q);
|
|
assert_eq!(post_swap.kind.as_str(), "post_swap");
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"Reconcile must wait for PostSwap, not race it"
|
|
);
|
|
q.complete_node(post_swap.node_id, Ok(()));
|
|
let reconcile = claim_one(&q);
|
|
assert_eq!(reconcile.kind.as_str(), "reconcile");
|
|
q.complete_node(reconcile.node_id, Ok(()));
|
|
settle_rebuild_tail(&q, "agent-a", true);
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
#[test]
|
|
fn failed_reconcile_marks_dag_failed() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "start".to_owned()),
|
|
);
|
|
let c = claim_one(&q);
|
|
q.complete_node(c.node_id, Err("start failed".to_owned()));
|
|
assert_eq!(state_of(&q, id), State::Failed);
|
|
}
|
|
|
|
// ---- cancel ----
|
|
|
|
#[test]
|
|
fn cancel_clears_queued_dag() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
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
|
|
// operator who just cancelled it (the dashboard renders this roll-up from
|
|
// the snapshot `post_rebuild_queue_cancel` emits synchronously).
|
|
assert_eq!(state_of(&q, id), State::Cancelled, "no stale Queued gap");
|
|
// Neither `EmitRebuilt` tail accepts a *dropped* dependency — the ok one is
|
|
// `AFTER_OK`, the failure one keys on elimination — so both are cancelled
|
|
// with the work and **nothing is emitted** for a rebuild that never ran.
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"a dropped rebuild reports nothing"
|
|
);
|
|
assert_eq!(state_of(&q, id), State::Cancelled);
|
|
}
|
|
|
|
/// `cancel` takes a **node** id, not a DAG id — so an interior node can be
|
|
/// dropped without touching the rest of the group.
|
|
///
|
|
/// This is the capability the DAG-scoped version couldn't express, and the
|
|
/// reason it reads naturally: a DAG id *is* its root node's id, so the
|
|
/// whole-group cancel every other test does is just this called on a root.
|
|
/// Here a hive-wide restart drops **one agent's** subgraph and the other agent
|
|
/// still runs.
|
|
#[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"));
|
|
// 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");
|
|
let a_root = dag
|
|
.nodes
|
|
.iter()
|
|
.find(|n| n.agent == "agent-a" && n.parent.is_none())
|
|
.expect("agent-a has a group root");
|
|
|
|
assert!(q.cancel(a_root.id), "an interior/group root cancels alone");
|
|
|
|
// agent-b's work is untouched and still claimable; agent-a's is not.
|
|
let claims = q.claim_ready();
|
|
assert!(
|
|
!claims.is_empty() && claims.iter().all(|c| c.agent == "agent-b"),
|
|
"only agent-b remains runnable, got {:?}",
|
|
claims.iter().map(|c| c.agent.as_str()).collect::<Vec<_>>()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn cancel_refuses_running_dag() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, rebuild("agent-a", "r"));
|
|
let _ = claim_one(&q);
|
|
assert!(!q.cancel(id));
|
|
assert_eq!(state_of(&q, id), State::Running);
|
|
}
|
|
|
|
/// A cancelled power op must run **no** compensating node — not even one that
|
|
/// carries a `SetWanted` head.
|
|
///
|
|
/// Now structural rather than a property of a hook enum: a power op emits no
|
|
/// tail node at all, so once its work nodes cancel there is simply nothing left
|
|
/// to claim. `cancel` also refuses unless every work node is still `Pending`
|
|
/// (`cancel_refuses_running_dag`), so a `Cancelled` DAG provably never executed
|
|
/// a node: its `SetWanted` never ran and the agent's intent still reads whatever
|
|
/// the operator last set. A "revert" instead writes the agent's *observed*
|
|
/// state, which for a down-but-`wanted = Up` agent (crashed, or caught
|
|
/// mid-bounce) flips the intent to `Offline` and leaves it
|
|
/// deliberately-stopped as far as reconcile and crash-watch are concerned.
|
|
#[test]
|
|
fn cancelled_power_op_runs_no_compensating_node() {
|
|
for graceful in [false, true] {
|
|
for running in [false, true] {
|
|
let targets = vec![("agent-a".to_owned(), running)];
|
|
// Erased to one boxed recipe type: three different recipe types
|
|
// have to sit in one array.
|
|
let cases = [
|
|
(
|
|
"restart",
|
|
false,
|
|
erase(submit::restart_spec(
|
|
&targets,
|
|
graceful,
|
|
Source::Manual,
|
|
"bounce".to_owned(),
|
|
)),
|
|
),
|
|
(
|
|
"stop",
|
|
true,
|
|
erase(submit::stop_spec(
|
|
&targets,
|
|
graceful,
|
|
Source::Manual,
|
|
"stop".to_owned(),
|
|
)),
|
|
),
|
|
(
|
|
"start",
|
|
true,
|
|
erase(submit::start_spec(
|
|
&[("agent-a".to_owned(), running, false)],
|
|
Source::Manual,
|
|
"start".to_owned(),
|
|
)),
|
|
),
|
|
];
|
|
for (name, writes_intent, spec) in cases {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, spec);
|
|
// Read the intent head off the submitted DAG rather than out of
|
|
// the spec: a declared job holds its own nodes and inserts them.
|
|
assert_eq!(
|
|
q.snapshot()
|
|
.iter()
|
|
.find(|d| d.id == id)
|
|
.expect("submitted dag")
|
|
.nodes
|
|
.iter()
|
|
.any(|n| n.kind == "set_wanted"),
|
|
writes_intent,
|
|
"{name} intent head (graceful={graceful}, running={running})"
|
|
);
|
|
assert!(q.cancel(id), "cancelled while queued");
|
|
assert_eq!(state_of(&q, id), State::Cancelled);
|
|
assert!(
|
|
q.claim_ready().is_empty(),
|
|
"cancelled {name} (graceful={graceful}, running={running}) must \
|
|
leave nothing to run — a power op emits no tail node"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- terminal reporting + lease release ----
|
|
|
|
#[test]
|
|
fn dag_settles_terminal_and_releases_lease_after_work() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, restart_online(&["agent-a"], false, "r"));
|
|
// restart = StopForUpdate → Reconcile.
|
|
let stop = claim_one(&q);
|
|
assert_eq!(stop.kind.as_str(), "stop_for_update");
|
|
q.complete_node(stop.node_id, Ok(()));
|
|
let rec = claim_one(&q);
|
|
assert_eq!(rec.kind.as_str(), "reconcile");
|
|
// Completing the last work node rolls the container up terminal. A power op
|
|
// has no tail node, so nothing is left to claim.
|
|
q.complete_node(rec.node_id, Ok(()));
|
|
assert!(q.claim_ready().is_empty(), "no tail node to claim");
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
// Lease released when the work chain settled: a new DAG for the agent claims
|
|
// immediately.
|
|
let next = submit(
|
|
&q,
|
|
templates::reconcile_only("agent-a", Source::Manual, "stop".to_owned()),
|
|
);
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.dag_id, next);
|
|
}
|
|
|
|
/// A DAG cancelled while fully queued must still **run its tail**, or a queued
|
|
/// approval DAG cancelled by the operator would dangle its approval forever.
|
|
///
|
|
/// This is the load-bearing case for sparing tails in [`JobQueue::cancel`]: the
|
|
/// work nodes all cancel, but `ResolveApproval` is weak-edged, so a `Cancelled`
|
|
/// dep satisfies its edge and it becomes claimable instead of being cancelled
|
|
/// along with everything else. It reads `Cancelled` off its own deps and resolves
|
|
/// the approval as "cancelled before completion".
|
|
#[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()),
|
|
);
|
|
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 claiming it *is*
|
|
// the assertion that the approval gets resolved as cancelled.
|
|
settle_approval_tail(&q, 7, TerminalState::Cancelled);
|
|
assert_eq!(state_of(&q, id), State::Cancelled);
|
|
// Unrelated later activity doesn't disturb the settled DAG.
|
|
let other = submit(&q, rebuild("agent-b", "r"));
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.dag_id, other);
|
|
q.complete_node(c.node_id, Err("boom".to_owned()));
|
|
assert_eq!(state_of(&q, id), State::Cancelled);
|
|
}
|
|
|
|
// ---- approval deploy subtree ----
|
|
|
|
/// The config-PR deploy is a subtree, not one opaque node. The
|
|
/// resource-holding root completes immediately (its `Finishing` state is the
|
|
/// parent gate that releases the children), then the phases run strictly in
|
|
/// order — and the `AfterAny` tail still runs when the irreversible half fails,
|
|
/// because it's the node that compensates for it.
|
|
#[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 root = claim_one(&q);
|
|
assert!(
|
|
matches!(root.kind, NodeKind::DeployWindow { .. }),
|
|
"root claims first: it holds the meta window for the whole subtree"
|
|
);
|
|
q.complete_node(root.node_id, Ok(()));
|
|
|
|
let verify = claim_one(&q);
|
|
assert!(matches!(verify.kind, NodeKind::MergeVerify { .. }));
|
|
q.complete_node(verify.node_id, Ok(()));
|
|
|
|
let apply = claim_one(&q);
|
|
assert!(matches!(apply.kind, NodeKind::DeployApply { .. }));
|
|
q.complete_node(apply.node_id, Err("nixos-container update blew up".into()));
|
|
|
|
let tail = claim_one(&q);
|
|
assert!(
|
|
matches!(tail.kind, NodeKind::DeployTail { .. }),
|
|
"AfterAny tail runs on a failed apply — that's the whole point of it"
|
|
);
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
|
|
settle_approval_tail(&q, 7, TerminalState::Failed);
|
|
assert_eq!(
|
|
state_of(&q, id),
|
|
State::Failed,
|
|
"an Ok tail must not launder a failed deploy into a success"
|
|
);
|
|
}
|
|
|
|
/// The deploy's happy path: `DeployApply` does not build. It grows the ordinary
|
|
/// rebuild chain into the live DAG under itself, and `FinalizeDeploy` — gated on
|
|
/// that graft finishing — plants the deploy tag last.
|
|
///
|
|
/// The queue is built with **one** build slot on purpose. `DeployWindow` already
|
|
/// holds that slot (and the meta window) for the whole subtree, so the grafted
|
|
/// `Prebuild` can only ever claim by *re-entering* its ancestor's hold. If the
|
|
/// graft were rooted anywhere outside `DeployWindow`'s subtree it would block on
|
|
/// a resource its own DAG owns and deadlock — this test is what pins that down.
|
|
#[test]
|
|
fn deploy_apply_grows_rebuild_subgraph_and_finalizes_after_it() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
templates::approval_deploy("agent-a", 11, "approval 11".to_owned()),
|
|
);
|
|
|
|
let root = claim_one(&q);
|
|
assert!(matches!(root.kind, NodeKind::DeployWindow { .. }));
|
|
q.complete_node(root.node_id, Ok(()));
|
|
let verify = claim_one(&q);
|
|
q.complete_node(verify.node_id, Ok(()));
|
|
|
|
let apply = claim_one(&q);
|
|
assert!(matches!(apply.kind, NodeKind::DeployApply { .. }));
|
|
// Mirrors the scheduler. The graft lands BEFORE the emitting node settles,
|
|
// and that ordering is now structural rather than a rule this call site has
|
|
// to follow: completing first would settle the apply node `Done` with
|
|
// nothing under it, opening the tail's `AfterAny` gate immediately and
|
|
// letting the deploy "finish" before it had built.
|
|
let grown = q.new_job();
|
|
templates::deploy_rebuild_nodes(&grown, "agent-a", 11);
|
|
q.complete_node_growing(apply.node_id, Ok(()), grown);
|
|
|
|
// The grafted chain runs in rebuild order. `claim_one` asserts exactly one
|
|
// claimable node at each step, which also proves the `AfterAny` tail stays
|
|
// shut: `DeployApply` is `Finishing` (not terminal) while its new children
|
|
// run, and `Finishing` satisfies neither dep kind.
|
|
for expected in [
|
|
"meta_sync",
|
|
"prebuild",
|
|
"stop_for_update",
|
|
"swap",
|
|
"post_swap",
|
|
"reconcile",
|
|
] {
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.kind.as_str(), expected, "grafted phase order");
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
|
|
let finalize = claim_one(&q);
|
|
assert!(
|
|
matches!(finalize.kind, NodeKind::FinalizeDeploy { .. }),
|
|
"the deploy tag is planted only after the rebuild came up clean"
|
|
);
|
|
q.complete_node(finalize.node_id, Ok(()));
|
|
|
|
let tail = claim_one(&q);
|
|
assert!(matches!(tail.kind, NodeKind::DeployTail { .. }));
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
|
|
settle_approval_tail(&q, 11, TerminalState::Done);
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
/// A failure *inside* the grafted rebuild is the failure mode the subgraph
|
|
/// growth introduces: the deploy is already merged and the container half-swapped.
|
|
/// `FinalizeDeploy` must be cancel-cascaded (its `AfterOk` gate never opens) so
|
|
/// no `deployed/<id>` tag is planted, while the tail still runs to compensate.
|
|
/// `Reconcile` is deliberately still reached — it boots the container back up.
|
|
#[test]
|
|
fn deploy_dag_skips_finalize_but_still_tails_a_failed_graft() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
templates::approval_deploy("agent-a", 13, "approval 13".to_owned()),
|
|
);
|
|
|
|
let root = claim_one(&q);
|
|
q.complete_node(root.node_id, Ok(()));
|
|
let verify = claim_one(&q);
|
|
q.complete_node(verify.node_id, Ok(()));
|
|
let apply = claim_one(&q);
|
|
let grown = q.new_job();
|
|
templates::deploy_rebuild_nodes(&grown, "agent-a", 13);
|
|
q.complete_node_growing(apply.node_id, Ok(()), grown);
|
|
|
|
for expected in ["meta_sync", "prebuild", "stop_for_update"] {
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.kind.as_str(), expected);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
let swap = claim_one(&q);
|
|
assert_eq!(swap.kind.as_str(), "swap");
|
|
q.complete_node(swap.node_id, Err("profile swap failed".into()));
|
|
|
|
// `Reconcile` hangs off `Prebuild` with `AfterAny`, so a failed swap still
|
|
// reaches it — bringing the container back up is exactly what it's for.
|
|
let reconcile = claim_one(&q);
|
|
assert_eq!(reconcile.kind.as_str(), "reconcile");
|
|
q.complete_node(reconcile.node_id, Ok(()));
|
|
|
|
let tail = claim_one(&q);
|
|
assert!(
|
|
matches!(tail.kind, NodeKind::DeployTail { .. }),
|
|
"finalize is cancel-cascaded, so the tail is the next claimable node"
|
|
);
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
|
|
settle_approval_tail(&q, 13, TerminalState::Failed);
|
|
assert_eq!(state_of(&q, id), State::Failed);
|
|
assert_eq!(
|
|
q.first_error(id).as_deref(),
|
|
Some("profile swap failed"),
|
|
"the tail annotates failed/<id> with this — and it is also what
|
|
`exec::failure_reason` falls back to, since the tail's own dep is a
|
|
group root that rolled up Failed and so carries no error itself"
|
|
);
|
|
}
|
|
|
|
/// A pre-merge rejection (drift gate, eval failure) cancel-cascades the
|
|
/// irreversible half via its `AfterOk` edge, but the tail is still reached —
|
|
/// it owns the forge mirror, not just compensation.
|
|
#[test]
|
|
fn deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
templates::approval_deploy("agent-a", 9, "approval #9".to_owned()),
|
|
);
|
|
|
|
let root = claim_one(&q);
|
|
q.complete_node(root.node_id, Ok(()));
|
|
let verify = claim_one(&q);
|
|
q.complete_node(verify.node_id, Err("PR head drifted since review".into()));
|
|
|
|
let tail = claim_one(&q);
|
|
assert!(
|
|
matches!(tail.kind, NodeKind::DeployTail { .. }),
|
|
"apply is cancel-cascaded, so the tail is the next claimable node"
|
|
);
|
|
q.complete_node(tail.node_id, Ok(()));
|
|
|
|
settle_approval_tail(&q, 9, TerminalState::Failed);
|
|
assert_eq!(state_of(&q, id), State::Failed);
|
|
}
|
|
|
|
// ---- history ----
|
|
//
|
|
// The node → build-log link is no longer queue state: the log row carries
|
|
// `node_id` and the lookup lives in `stores::build_logs` (see
|
|
// `node_link_survives_completion_and_newest_wins` there). Nothing in the queue
|
|
// needs testing for it any more, which is the point of that move.
|
|
|
|
/// History retention is a **flat** newest-first cap over all terminal DAGs
|
|
/// (`MAX_HISTORY_DAGS`), not a per-template bucket behind a grace window.
|
|
/// The dashboard renders one recent-builds list, so one number bounds it —
|
|
/// and with no bucketing there's nothing for a burst of same-shaped DAGs to
|
|
/// evict early, which is what the grace window used to paper over.
|
|
#[test]
|
|
fn history_evicts_oldest_terminals_past_flat_cap() {
|
|
const OVERFLOW: usize = 8;
|
|
let q = JobQueue::new(1);
|
|
let mut ids = Vec::new();
|
|
for i in 0..(MAX_HISTORY_DAGS + OVERFLOW) {
|
|
let id = submit(
|
|
&q,
|
|
templates::reconcile_only(&format!("agent-{i}"), Source::Manual, "start".to_owned()),
|
|
);
|
|
let c = claim_one(&q);
|
|
// Fail the single work node so the DAG *lingers*: a fully-`Done` DAG
|
|
// drops off the wire entirely, but a `Failed` one is retained (+
|
|
// history-capped) so the operator can still triage it. Completing the
|
|
// node rolls the container up terminal.
|
|
q.complete_node(c.node_id, Err("boom".to_owned()));
|
|
ids.push(id);
|
|
}
|
|
let kept: std::collections::HashSet<u64> = q.snapshot().iter().map(|d| d.id).collect();
|
|
assert_eq!(kept.len(), MAX_HISTORY_DAGS, "flat history cap");
|
|
// Newest-first: the oldest `OVERFLOW` fall off, everything after survives.
|
|
// These DAGs settle within the same wall-clock second, so this also pins
|
|
// the `NodeId`-descending tiebreak that orders them when `finished_at` ties.
|
|
for old in &ids[..OVERFLOW] {
|
|
assert!(!kept.contains(old), "oldest terminal {old} evicted");
|
|
}
|
|
for recent in &ids[OVERFLOW..] {
|
|
assert!(kept.contains(recent), "recent terminal {recent} retained");
|
|
}
|
|
assert_eq!(q.live_count(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn error_truncation_cuts_on_a_char_boundary() {
|
|
// `truncate_error` is a pure `&str -> String`. This used to submit a DAG,
|
|
// claim its head, fail it with a long string and read the error back out of
|
|
// a snapshot — four moving parts to observe one transform, and the DAG
|
|
// round-trip is covered by its own tests either way.
|
|
//
|
|
// Testing it directly also reaches the case the round-trip never did: the
|
|
// cap is a **byte** length, so a multibyte char straddling it would panic
|
|
// the slice. That boundary scan is the only non-obvious line in the fn and
|
|
// it had no coverage at all.
|
|
assert_eq!(
|
|
truncate_error("short"),
|
|
"short",
|
|
"under the cap is untouched"
|
|
);
|
|
|
|
let ascii = truncate_error(&"x".repeat(5000));
|
|
assert!(ascii.ends_with('…'));
|
|
assert!(ascii.len() <= MAX_ERROR_LEN + '…'.len_utf8());
|
|
|
|
// 'é' is 2 bytes, so the 2000-byte cap lands mid-char.
|
|
let multibyte = truncate_error(&"é".repeat(5000));
|
|
assert!(multibyte.ends_with('…'));
|
|
assert!(multibyte.len() <= MAX_ERROR_LEN + '…'.len_utf8());
|
|
}
|
|
|
|
// ---- template shapes ----
|
|
|
|
#[test]
|
|
fn graceful_stop_shape_signal_drain_reconcile() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(&q, stop_online(&["agent-a"], true, "graceful"));
|
|
assert_eq!(
|
|
declared_shape(&q, id),
|
|
vec![
|
|
// The whole stop hangs under `set_wanted`: the durable intent is
|
|
// written first, and the mechanical steps are its sub-nodes.
|
|
row("set_wanted", None, &[]),
|
|
row("signal", Some("set_wanted"), &[]),
|
|
row("drain", Some("set_wanted"), &[("signal", "done")]),
|
|
row("reconcile", Some("set_wanted"), &[("drain", "done")]),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn graceful_signal_and_drain_hold_no_build_slot() {
|
|
// A whole-hive graceful stop overlaps every drain even at
|
|
// buildSlots = 1 while a rebuild hogs the slot.
|
|
let q = JobQueue::new(1);
|
|
submit(&q, rebuild("builder", "slot hog"));
|
|
submit(&q, stop_online(&["agent-a"], true, "g"));
|
|
submit(&q, stop_online(&["agent-b"], true, "g"));
|
|
// All three DAG heads are build-slot-exempt, so they run at once.
|
|
let heads = q.claim_ready();
|
|
let kinds: Vec<&str> = heads.iter().map(|c| c.kind.as_str()).collect();
|
|
assert_eq!(kinds, vec!["meta_sync", "set_wanted", "set_wanted"]);
|
|
for c in &heads {
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
// Now the rebuild's Prebuild holds the single slot — and both graceful
|
|
// stops still proceed to their Signal beside it.
|
|
let kinds: Vec<&str> = q
|
|
.claim_ready()
|
|
.iter()
|
|
.map(|c| c.kind.as_str())
|
|
.collect::<Vec<_>>();
|
|
assert_eq!(
|
|
kinds,
|
|
vec!["prebuild", "signal", "signal"],
|
|
"both agents' graceful-stop signals (build-slot-exempt) run while the \
|
|
rebuild holds the slot"
|
|
);
|
|
}
|
|
|
|
#[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()),
|
|
);
|
|
assert_eq!(
|
|
declared_shape(&q, id),
|
|
vec![
|
|
row("provision", None, &[]),
|
|
row("create", Some("provision"), &[]),
|
|
row("write_dropin", Some("create"), &[]),
|
|
row("reconcile", Some("create"), &[("write_dropin", "done")]),
|
|
// One tail per outcome, each edged to accept only that one — so
|
|
// *which* tail the graph lets run already is the answer, and
|
|
// nothing branches at runtime. The three differ **only** in their
|
|
// accepted outcome, which is why `declared_shape` spells the
|
|
// outcome set out instead of bucketing it.
|
|
row("resolve_approval", None, &[("provision", "done")]),
|
|
row("resolve_approval", None, &[("provision", "failed")]),
|
|
row("resolve_approval", None, &[("provision", "cancelled")]),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn perm_change_shape_prefixes_rebuild_chain() {
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
templates::perm_change(
|
|
"agent-a",
|
|
Source::Manual,
|
|
"perm".to_owned(),
|
|
PermPayload::Combined {
|
|
groups: Some(vec![]),
|
|
caps: None,
|
|
},
|
|
),
|
|
);
|
|
for expected in [
|
|
"write_perm_file",
|
|
"meta_sync",
|
|
"prebuild",
|
|
"stop_for_update",
|
|
"swap",
|
|
"post_swap",
|
|
"reconcile",
|
|
] {
|
|
let c = claim_one(&q);
|
|
assert_eq!(c.kind.as_str(), expected);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
}
|
|
settle_rebuild_tail(&q, "agent-a", true);
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
#[test]
|
|
fn reparent_shape_is_a_lone_agentless_meta_window_node() {
|
|
// Single-move `set-parent` shape: one node, no rebuild subgraph (no
|
|
// container rebuild needed for a parent move), agentless like
|
|
// `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 c = claim_one(&q);
|
|
assert_eq!(c.kind.as_str(), "reparent");
|
|
assert_eq!(c.agent, "", "Reparent is agentless — no per-agent lease");
|
|
assert_eq!(
|
|
declared_resources(&q, c.node_id),
|
|
vec![Resource::MetaWindow],
|
|
"a topology commit must declare the same MetaWindow as WritePermFile, \
|
|
and nothing else — no lease (agentless), no build slot (no nix work)"
|
|
);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|
|
|
|
#[test]
|
|
fn reparent_bulk_shape_carries_every_move_on_one_node() {
|
|
// `set-parent-bulk`: still ONE node (one git commit, `moves.len() > 1`),
|
|
// not one node per move — bulk atomicity across every move in the
|
|
// 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 c = claim_one(&q);
|
|
assert_eq!(c.kind.as_str(), "reparent");
|
|
let NodeKind::Reparent { moves: got } = &c.kind else {
|
|
panic!("expected a Reparent node, got {:?}", c.kind);
|
|
};
|
|
assert_eq!(got, &moves);
|
|
q.complete_node(c.node_id, Ok(()));
|
|
assert_eq!(state_of(&q, id), State::Done);
|
|
}
|