7d1709cfc5
deploy_dag_runs_phases_in_order_and_tails_a_failed_apply and deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails differed only in where they injected the failure -- apply in one, verify in the other -- and each drove the whole DAG to watch the compensation tail run anyway. Both follow from a single declared edge. The tail accepts done|failed|skipped on apply, and skipped is exactly the state apply lands in when verify failed and it never ran. Asserting that edge covers both cases without running anything. The runtime halves are hive-jobq's and tested there: a failed dep cancels its AfterOk dependents while the AfterAny one still runs, and a parent rolls up Failed from a failed child -- which is what stops an Ok tail laundering a failed deploy into a success. Mutation-checked: turning the tail's after_any(apply) into after_ok(apply) fails the surviving test on that edge alone.
1998 lines
77 KiB
Rust
1998 lines
77 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")
|
|
}
|
|
|
|
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()
|
|
}
|
|
|
|
/// The id of the one node of `kind` under `dag`, for the resource assertions.
|
|
///
|
|
/// Panics unless there is exactly one — every caller is about a shape where the
|
|
/// kind is unique, so two would mean the assertion had quietly stopped being
|
|
/// about the node the test names.
|
|
fn node_of(q: &JobQueue, dag: u64, kind: &str) -> hive_jobq::NodeId {
|
|
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 mut found: Vec<_> = graph
|
|
.nodes()
|
|
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.payload.as_str() == kind)
|
|
.map(|n| n.id)
|
|
.collect();
|
|
assert_eq!(
|
|
found.len(),
|
|
1,
|
|
"expected exactly one {kind} node in the dag"
|
|
);
|
|
found.pop().expect("checked above")
|
|
}
|
|
|
|
/// Kinds of every node under `dag` still `Pending` — the nodes that could yet
|
|
/// run. Stronger than asking the scheduler what is *ready right now*: a node
|
|
/// blocked on a dep is not ready but is very much still alive.
|
|
fn pending_kinds(q: &JobQueue, dag: u64) -> Vec<&'static str> {
|
|
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");
|
|
graph
|
|
.nodes()
|
|
.filter(|n| n.id != root && graph.root_of(n.id) == Some(root) && n.state == State::Pending)
|
|
.map(|n| n.payload.as_str())
|
|
.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,
|
|
);
|
|
}),
|
|
},
|
|
);
|
|
assert_eq!(
|
|
declared_shape(&q, id)
|
|
.iter()
|
|
.map(|d| d.kind)
|
|
.collect::<Vec<_>>(),
|
|
vec![
|
|
"meta_sync",
|
|
"prebuild",
|
|
"stop_for_update",
|
|
"swap",
|
|
"post_swap",
|
|
"reconcile"
|
|
],
|
|
"exactly six nodes, and neither of them is signal or drain"
|
|
);
|
|
}
|
|
|
|
/// 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
|
|
// group root with no deps, so nothing orders them against each other; and
|
|
// each declares only its OWN agent's lease, so nothing makes them contend.
|
|
// Those two declared facts are what "they run concurrently" *means* here —
|
|
// that a scheduler then does run independent, resource-disjoint roots at
|
|
// once is hive_jobq's property, tested there.
|
|
let heads: Vec<_> = declared_shape(&q, id)
|
|
.into_iter()
|
|
.filter(|d| d.kind == "stop_for_update")
|
|
.collect();
|
|
assert_eq!(
|
|
heads,
|
|
vec![
|
|
row("stop_for_update", None, &[]),
|
|
row("stop_for_update", None, &[]),
|
|
],
|
|
"both per-agent heads are independent group roots"
|
|
);
|
|
let sched = q.sched().lock().expect("job_queue mutex poisoned");
|
|
let graph = sched.graph();
|
|
let root = graph.resolve_id(id).expect("dag id");
|
|
let mut leases: Vec<String> = graph
|
|
.nodes()
|
|
.filter(|n| graph.root_of(n.id) == Some(root) && n.payload.as_str() == "stop_for_update")
|
|
.flat_map(|n| {
|
|
n.deps.iter().filter_map(|dep| match dep {
|
|
hive_jobq::Dep::Resource { name, .. } => Some(format!("{name:?}")),
|
|
hive_jobq::Dep::Node { .. } => None,
|
|
})
|
|
})
|
|
.collect();
|
|
leases.sort();
|
|
assert_eq!(
|
|
leases,
|
|
vec![
|
|
format!("{:?}", Resource::Agent("agent-a".to_owned())),
|
|
format!("{:?}", Resource::Agent("agent-b".to_owned())),
|
|
],
|
|
"each head declares only its own agent's lease — disjoint, so no contention"
|
|
);
|
|
}
|
|
|
|
/// 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() {
|
|
/// Submit-cancel-assert for one power op. Taking the already-submitted DAG
|
|
/// id is what removes the need to put three differently-typed recipes in
|
|
/// one array: each caller submits its own spec, so no closure type has to
|
|
/// be erased to a boxed one.
|
|
fn assert_cancels_clean(q: &JobQueue, id: u64, writes_intent: bool, case: &str) {
|
|
// Read the intent head off the submitted DAG rather than out of the
|
|
// spec: a declared job holds its own nodes and inserts them.
|
|
let has_intent = declared_shape(q, id).iter().any(|d| d.kind == "set_wanted");
|
|
assert_eq!(has_intent, writes_intent, "{case}: intent head");
|
|
assert!(q.cancel(id), "{case}: cancelled while queued");
|
|
assert_eq!(state_of(q, id), State::Cancelled);
|
|
// Nothing is left that *could* run. Asserting on the pending set rather
|
|
// than on "what is ready this instant" also covers a node that is alive
|
|
// but blocked — which is exactly what a leftover compensating node
|
|
// would look like.
|
|
assert_eq!(
|
|
pending_kinds(q, id),
|
|
Vec::<&str>::new(),
|
|
"{case}: a power op emits no tail node, so a cancelled one leaves nothing"
|
|
);
|
|
}
|
|
|
|
for graceful in [false, true] {
|
|
for running in [false, true] {
|
|
let targets = vec![("agent-a".to_owned(), running)];
|
|
let case = format!("graceful={graceful} running={running}");
|
|
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
submit::restart_spec(&targets, graceful, Source::Manual, "bounce".to_owned()),
|
|
);
|
|
assert_cancels_clean(&q, id, false, &format!("restart {case}"));
|
|
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
submit::stop_spec(&targets, graceful, Source::Manual, "stop".to_owned()),
|
|
);
|
|
assert_cancels_clean(&q, id, true, &format!("stop {case}"));
|
|
|
|
let q = JobQueue::new(1);
|
|
let id = submit(
|
|
&q,
|
|
submit::start_spec(
|
|
&[("agent-a".to_owned(), running, false)],
|
|
Source::Manual,
|
|
"start".to_owned(),
|
|
),
|
|
);
|
|
assert_cancels_clean(&q, id, true, &format!("start {case}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- 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()),
|
|
);
|
|
|
|
assert_eq!(
|
|
declared_shape(&q, id),
|
|
vec![
|
|
// The window is the group root and holds the meta window for the
|
|
// whole subtree; the three phases are its sub-nodes.
|
|
row("deploy_window", None, &[]),
|
|
row("merge_verify", Some("deploy_window"), &[]),
|
|
// Apply only on a clean verify — a failed verify cancel-cascades
|
|
// it, which is what leaves the forge and the applied repo untouched.
|
|
row(
|
|
"deploy_apply",
|
|
Some("deploy_window"),
|
|
&[("merge_verify", "done")]
|
|
),
|
|
// The compensation tail accepts every terminal outcome of apply,
|
|
// *including `skipped`* — which is the state apply lands in when
|
|
// verify failed and it never ran. That one edge is the entire
|
|
// "still tails a failed apply / a failed verify" behaviour, and it
|
|
// is why two separate DAG-driving tests collapsed into this table.
|
|
row(
|
|
"deploy_tail",
|
|
Some("deploy_window"),
|
|
&[("deploy_apply", "done|failed|skipped")]
|
|
),
|
|
// One approval tail per outcome, gated on the window's roll-up.
|
|
row("resolve_approval", None, &[("deploy_window", "done")]),
|
|
row("resolve_approval", None, &[("deploy_window", "failed")]),
|
|
row("resolve_approval", None, &[("deploy_window", "cancelled")]),
|
|
]
|
|
);
|
|
}
|
|
|
|
/// 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"
|
|
);
|
|
}
|
|
|
|
// `deploy_dag_skips_apply_but_still_runs_tail_when_verify_fails` lived here.
|
|
//
|
|
// A pre-merge rejection (drift gate, eval failure) cancel-cascades the
|
|
// irreversible half via its `AfterOk` edge, while the tail is still reached —
|
|
// it owns the forge mirror, not just compensation. That test and
|
|
// `deploy_dag_runs_phases_in_order_and_tails_a_failed_apply` differed only in
|
|
// *where* they injected the failure, and each drove the whole DAG to watch the
|
|
// tail run anyway.
|
|
//
|
|
// Both outcomes follow from one declared edge, which the surviving test now
|
|
// asserts directly: the tail accepts `done|failed|skipped` on apply, and
|
|
// `skipped` is exactly the state apply lands in when verify failed and it never
|
|
// ran. The runtime halves are hive_jobq's and tested there —
|
|
// `failed_after_ok_dep_cancels_dependents_but_after_any_still_runs` and
|
|
// `failed_child_rolls_parent_up_to_failed` (an Ok tail cannot launder a failed
|
|
// deploy into a success).
|
|
|
|
// ---- 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(),
|
|
),
|
|
);
|
|
assert_eq!(
|
|
declared_shape(&q, id),
|
|
vec![row("reparent", None, &[])],
|
|
"one node, no rebuild subgraph"
|
|
);
|
|
let node = node_of(&q, id, "reparent");
|
|
assert_eq!(
|
|
declared_resources(&q, node),
|
|
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)"
|
|
);
|
|
}
|
|
|
|
#[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);
|
|
}
|