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

c0re's queue tests no longer drive the scheduler

Browse source 
The last two claim-driven tests were both arranging node states to observe
something that never needed a run:

`settled_dag_leaves_the_snapshot_despite_its_skipped_branch` completed all
seven nodes of a rebuild to assert the DAG left the snapshot. That is one
predicate over a list of states. `shown_on_wire` is it, split out of
`dag_view`, and the cases can now be named rather than arranged — including
the empty set, the one input where "any" and "all" disagree. It takes
states rather than projected nodes so the caller skips projecting what it
is about to discard; a `NodeView` costs a `build_logs` lookup.

`failed_node_cancels_downstream_but_afterany_reconcile_runs` asserted three
unrelated things from one arranged failure: the cascade (hive_jobq's, and
already tested there), the wire filter (now `shown_on_wire`), and the
roll-up. `DagView::rollup_state` lives in hive-host-sock, which had no
tests at all — it does now, next to the invariant, covering the ordering
its own doc comment says has silently disagreed with the frontend before.

With nothing left claiming, `Claimed` / `ClaimReady` / `CompleteNode` /
`claim_one` / `settle_rebuild_tail` are deleted. Claim/complete sites in
`job_queue/tests.rs`: 109 -> 0.

jobq narrows to match: `settle` is gone (it was a `claim_one` loop
returning a Vec, and its only callers were tests — it lives in the test
module now), `claim_one` is private, and `complete_growing` is
`pub(crate)`. `claim_next` is the whole run-loop surface.

`complete` stays `pub` for one caller, noted at the definition: `submit`
completes a group root with no logic of its own so it parks in `Finishing`
and its children unblock. That is a statement about the node, not an event
to report, and it wants to be expressible at insert time.
This commit is contained in:
atlas 2026-08-02 21:44:33 +02:00 • committed by mara
commit e646656c92
4 changed files with 334 additions and 266 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -404,32 +404,27 @@ fn dag_meta(sched: &Sched, container: NodeId) -> Option<DagMeta> {
/// aged out).
fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
let meta = dag_meta(sched, container)?;
let mut nodes = Vec::new();
// Whether anything in this DAG still has an outcome worth showing.
// Kept separate from `nodes` being non-empty: skipped nodes ride the
// wire so the dashboard can mark the branches that weren't taken, but
// they must not by themselves hold a finished DAG in the snapshot.
let mut any_unsettled = false;
let all: Vec<_> = sched.graph().descendants(container).collect();
// DAG-level timestamps are taken over *all* subtree nodes (including the
// `Done` ones excluded from the wire) — the client can't derive them
// from a `Done`-filtered node set, so the host computes them here.
let mut started: Vec<DateTime<Utc>> = Vec::new();
let mut finished: Vec<DateTime<Utc>> = Vec::new();
for node in sched.graph().descendants(container) {
let id = node.id;
for node in &all {
if let Some(s) = node.started_at {
started.push(s);
}
if let Some(f) = node.finished_at {
finished.push(f);
}
// `Done` nodes drop off the wire — a finished step isn't
// interesting. `Skipped` ones stay: which branch a run *didn't*
// take is the readable half of an outcome-branched DAG.
if matches!(node.state, State::Done) {
continue;
}
any_unsettled |= !matches!(node.state, State::Skipped);
}
// Decide which nodes ride the wire *before* projecting any of them: a
// `NodeView` costs a `build_logs` lookup, so building one for a node
// that's about to be dropped would be a query per finished step.
let shown = shown_on_wire(&all.iter().map(|n| n.state).collect::<Vec<_>>())?;
let mut nodes = Vec::new();
for node in shown.into_iter().map(|i| all[i]) {
let id = node.id;
let deps: Vec<u64> = node
.deps
.iter()
@ -480,9 +475,6 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
parent,
});
}
if !any_unsettled {
return None;
}
let is_terminal = sched.graph().is_settled(container) == Some(true);
Some(DagView {
id: container.get(),
@ -495,6 +487,39 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option<DagView> {
})
}
/// Which of a DAG's work nodes ride the wire, by index into `states` — or
/// `None` when the DAG has nothing left worth showing and drops out of the
/// snapshot entirely.
///
/// Two separate decisions, and conflating them pins every completed deploy in
/// the queue view forever:
/// - **`Done` drops off the wire.** A finished step isn't interesting.
/// `Skipped` stays: which branch a run *didn't* take is the readable half of
/// an outcome-branched DAG.
/// - **`Skipped` alone doesn't hold a DAG in the snapshot.** So "the node list
/// is non-empty" and "there's still something here worth showing" are
/// different questions, and only the second one may drop the DAG.
///
/// Takes states rather than projected nodes so the caller can skip the work of
/// projecting what it's about to discard, and so this is testable without a
/// graph — the states it keys on are ones only a run can produce.
fn shown_on_wire(states: &[State]) -> Option<Vec<usize>> {
let worth_showing = states
.iter()
.any(|s| !matches!(s, State::Done | State::Skipped));
if !worth_showing {
return None;
}
Some(
states
.iter()
.enumerate()
.filter(|(_, s)| !matches!(s, State::Done))
.map(|(i, _)| i)
.collect(),
)
}
/// When a DAG's work node finishes on `finished_at` — the max over its
/// subtree (read off the graph `Node`, as unix seconds), for the history
/// cap ordering.

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

@ -1,10 +1,18 @@
//! 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.
//! Queue-core unit tests: what c0re's templates **declare** — node kinds,
//! parent nesting, dep edges with the outcomes that satisfy them, and the
//! resources each construction site states it holds — plus the read layer over
//! that graph (wire projection, history retention, error truncation).
//!
//! **Nothing here runs a node.** Everything a template declares is in the graph
//! the moment `submit` returns, so the assertions read it there. Whether the
//! scheduler then honours those declarations — cascade, roll-up, grant
//! borrow/release, fairness, the `Finishing` gate — is `hive_jobq`'s property
//! and is tested in `hive_jobq`, against its own primitives rather than through
//! this module's templates.
//!
//! That split is why this file can't claim or complete: those are not part of
//! c0re's surface. A test helper that reached for them was reaching across the
//! boundary the two crates exist to draw.
use super::model::NodeKind;
use super::*;
@ -45,66 +53,13 @@ fn stop_online(
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 {
node_id: NodeId,
kind: NodeKind,
}
/// 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 { node_id, 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>);
}
impl CompleteNode for JobQueue {
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();
}
}
// `Claimed` / `ClaimReady` / `CompleteNode` lived here: a claim snapshot type
// and two extension traits that let this module start and finish nodes by
// hand. Nothing in this file drives the scheduler any more, so they are gone
// — which is the point. 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. A test
// helper that re-expressed it was a hole in exactly the seam it was testing.
/// 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.
@ -273,29 +228,6 @@ fn row(
}
}
/// 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 the `EmitRebuilt` 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
@ -580,25 +512,48 @@ fn non_graceful_rebuild_has_no_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.
/// Which nodes ride the wire, and whether a DAG is still worth showing.
///
/// This used to submit a rebuild, claim and complete all seven of its nodes,
/// then assert the DAG was absent from `snapshot()` — the scheduler, the
/// roll-up and the whole projection standing in for one predicate over a list
/// of states. `shown_on_wire` is that predicate, so the cases can be named
/// instead of arranged.
#[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"
fn skipped_nodes_ride_the_wire_but_do_not_hold_a_settled_dag_in_the_snapshot() {
// Nothing left worth showing → the DAG drops out of the snapshot, and its
// absence is what signals completion.
assert_eq!(shown_on_wire(&[State::Done, State::Done]), None);
// The case the old test was built around: a clean run whose not-taken
// failure branch is still in the graph as `Skipped`. "The node list is
// non-empty" and "there's something here worth showing" are different
// questions, and conflating them pins every completed deploy in the queue
// view forever.
assert_eq!(
shown_on_wire(&[State::Done, State::Skipped, State::Done]),
None,
"a skipped branch does not keep a finished DAG alive"
);
// A `Failed` DAG lingers — and takes its skipped branches with it, which is
// how an operator sees which steps the run never reached.
assert_eq!(
shown_on_wire(&[State::Done, State::Failed, State::Skipped, State::Done]),
Some(vec![1, 2]),
"done nodes drop off, the failure and what it ruled out stay"
);
// Live DAGs keep everything but their finished steps.
assert_eq!(
shown_on_wire(&[State::Done, State::Running, State::Pending]),
Some(vec![1, 2])
);
assert_eq!(
shown_on_wire(&[State::Cancelled, State::Skipped]),
Some(vec![0, 1]),
"a cancelled DAG is still worth showing to whoever cancelled it"
);
// An empty DAG has nothing worth showing either — no special case needed,
// but it is the one input where "any" and "all" disagree, so it is pinned.
assert_eq!(shown_on_wire(&[]), None);
}
// ---- build slots ----
@ -608,8 +563,8 @@ fn rebuild_chain_declares_the_slot_where_the_nix_work_is() {
// Was `fifo_fairness_for_the_slot`, which submitted three rebuilds and
// drove one to completion to watch the freed slot go to the earlier
// waiter. **That fairness guarantee is hive_jobq's**, and it had no test
// there at all — `claim_one` scans nodes in insertion order and takes the
// first satisfiable one, and nothing pinned that. It does now:
// there at all — its claim primitive scans nodes in insertion order and
// takes the first satisfiable one, and nothing pinned that. It does now:
// `a_contended_resource_goes_to_the_oldest_waiter`.
//
// What is c0re's is *which* nodes contend for the slot in the first place,
@ -900,58 +855,26 @@ fn rebuild_chain_nodes_suppress_crash_watch() {
}
// ---- 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")
);
}
//
// `failed_node_cancels_downstream_but_afterany_reconcile_runs` lived here. It
// drove a rebuild to a failed `Prebuild` and then asserted three unrelated
// things at once, which is why it needed a running scheduler at all:
//
// 1. the cascade — a failed node cancels its `AfterOk` dependants while the
// `AfterAny` reconcile still runs. That is hive_jobq's rule, and it owns
// the test: `failed_after_ok_dep_cancels_dependents_but_after_any_runs`.
// c0re's declaration of *which* edge is which is asserted in
// `rebuild_chain_is_declared_serial` — the reconcile's accepted outcome
// set is right there in the shape table.
// 2. the wire projection — `Done` off, `Skipped` on. That is
// `shown_on_wire`, tested directly above.
// 3. the roll-up — a DAG with a failed node reads `Failed`, and `Skipped`
// contributes nothing. That is `DagView::rollup_state`, which lives in
// hive-host-sock and is now tested there, next to the invariant.
//
// Reconstructing all three from one arranged run made none of them
// individually legible, and the arrangement was the only reason this module
// needed to claim and complete nodes.
/// The swap-success path: `Swap` ok → the `AfterOk` `PostSwap` (bookkeeping
/// tail) runs, and only then does `Reconcile` fire — serialized behind

103
hive-host-sock/src/jobs.rs
View file

@ -206,3 +206,106 @@ impl DagView {
}
}
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use super::{DagView, NodeView, Source, State};
/// A node set carrying nothing but the states — the only input
/// `rollup_state` reads.
fn dag(states: &[State]) -> DagView {
DagView {
id: 1,
source: Source::Manual,
reason: "test".to_owned(),
created_at: Utc::now(),
started_at: None,
finished_at: None,
nodes: states
.iter()
.enumerate()
.map(|(i, &state)| NodeView {
id: i as u64,
agent: "a".to_owned(),
kind: "reconcile".to_owned(),
deps: Vec::new(),
state,
started_at: None,
finished_at: None,
error: None,
approval_id: None,
inputs: Vec::new(),
build_log_id: None,
parent: None,
})
.collect(),
}
}
#[test]
fn a_failure_outranks_everything_and_skipped_counts_for_nothing() {
// The case this replaces used to be arranged in hive-c0re by running a
// rebuild until its Prebuild failed. Only the states ever mattered.
assert_eq!(
dag(&[State::Done, State::Failed, State::Skipped]).rollup_state(),
State::Failed
);
// A failure wins even against a node still going — the DAG's verdict
// is already decided.
assert_eq!(
dag(&[State::Running, State::Failed]).rollup_state(),
State::Failed
);
// Skipped is an expected part of a healthy run: an outcome-branched
// DAG always leaves one branch untaken, so counting it would make
// every successful DAG roll up non-Done.
assert_eq!(
dag(&[State::Done, State::Skipped]).rollup_state(),
State::Done
);
}
#[test]
fn cancelled_outranks_running_and_pending() {
// A cancelled DAG still has its weak-edged tail node to run, so
// Pending-then-Running would flicker back at the operator who just
// cancelled it and read as "the cancel didn't take".
assert_eq!(
dag(&[State::Cancelled, State::Pending]).rollup_state(),
State::Cancelled
);
assert_eq!(
dag(&[State::Cancelled, State::Running]).rollup_state(),
State::Cancelled
);
}
#[test]
fn finishing_still_counts_as_running() {
// The node's own work is done but its sub-nodes are still going, so
// the DAG is in flight. A parent parked in Finishing is the normal
// shape of a subtree mid-run, not an edge case.
assert_eq!(
dag(&[State::Finishing, State::Pending]).rollup_state(),
State::Running
);
assert_eq!(
dag(&[State::Running, State::Pending]).rollup_state(),
State::Running
);
assert_eq!(
dag(&[State::Done, State::Pending]).rollup_state(),
State::Pending
);
}
#[test]
fn an_empty_node_set_reads_done() {
// Every node Done means every node is filtered off the wire, so this
// is what a finished DAG actually looks like to a consumer that has
// one in hand at all.
assert_eq!(dag(&[]).rollup_state(), State::Done);
}
}

191
hive-jobq/src/scheduler.rs
View file

@ -1,13 +1,14 @@
//! The settle loop — drives a [`Graph`] to completion over a resource pool the
//! scheduler owns directly.
//!
//! [`Scheduler::settle`] claims every currently-runnable pending node (its
//! [`Scheduler::claim_next`] claims one currently-runnable pending node (its
//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
//! atomically), marks it `Running`, records the units it holds, and returns the
//! newly-started ids for the caller's runner to execute. The runner reports each
//! node's result back with [`Scheduler::complete`]; a running node may grow more
//! work first via [`Scheduler::append`]. Concurrency is emergent from resource
//! capacity — there is no separate active-node cap.
//! atomically), marks it `Running`, records the units it holds, and hands back
//! a future that executes the node **and completes it**, so "forgot to finish
//! the node" is not expressible. One at a time is the primitive on purpose: it
//! lets the caller choose between claiming again and backing off, which a batch
//! return can't express. A running node may grow more work by declaring into
//! the builder it was handed. Concurrency is emergent from resource capacity.
//!
//! Single-threaded by design: the scheduler is the only driver, holds the
//! [`ResourceTable`] as a plain owned field, mutating it through `&mut self` —
@ -88,8 +89,8 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
}
/// Append a node under `parent` — e.g. a running node growing more work into
/// its own subtree. Delegates to [`Graph::insert`]; call [`Scheduler::settle`]
/// afterwards to start it once it is runnable.
/// its own subtree. Delegates to [`Graph::insert`]; claim again afterwards
/// to start it once it is runnable.
///
/// # Errors
/// Propagates [`GraphError`] for a dangling dependency or parent id.
@ -116,8 +117,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// [`Graph::insert_unchecked`]: [`crate::builder::check_job_shape`] has
/// already decided every rejection the graph could raise, so re-validating
/// per node could only report a problem *after* the earlier nodes were
/// inserted. Call [`Scheduler::settle`] afterwards to start whatever became
/// runnable.
/// inserted. Claim again afterwards to start whatever became runnable.
///
/// **Atomic in the job's own shape.** A forward edge, a forward parent, or
/// a request for a handle this job never declared is rejected *before* the
@ -146,11 +146,11 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// returned for the caller to execute. `None` means nothing is runnable
/// right now — which is a different statement from "nothing is pending".
///
/// One-at-a-time is the primitive on purpose: it lets the caller decide
/// between claiming again immediately and backing off, a choice a batch
/// return can't express. [`Self::settle`] is this in a loop.
/// **Private**: [`Self::claim_next`] is the only way out of this crate.
/// Claiming without the future that completes the node is the sequence the
/// seam exists to make inexpressible, so the primitive stays in here.
#[must_use]
pub fn claim_one(&mut self) -> Option<NodeId> {
fn claim_one(&mut self) -> Option<NodeId> {
let pending: Vec<NodeId> = self
.graph
.nodes()
@ -216,24 +216,6 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
})
}
/// Claim every currently-runnable pending node. Equivalent to calling
/// [`Self::claim_one`] until it yields `None`: a node started by an earlier
/// iteration is `Running`, not terminal, so it cannot satisfy another
/// node's dependency here — it only consumes resources.
///
/// ⚠️ Each iteration rescans the pending set, so this is O(n²) in the
/// number of nodes claimed where the old single-pass version was O(n). The
/// graph is bounded by history retention, so that is affordable; it is
/// stated rather than left to be discovered.
#[must_use]
pub fn settle(&mut self) -> Vec<NodeId> {
let mut started = Vec::new();
while let Some(id) = self.claim_one() {
started.push(id);
}
started
}
/// Try to start node `id`. For each resource it needs, decide per the parent
/// tree (see the module docs): acquire fresh units (owner), acquire an extra
/// unit (grant lent elsewhere), or borrow an ancestor's grant. The fresh set
@ -321,8 +303,16 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// (every child `Done`) or [`State::Failed`] (any child `Failed`/`Cancelled`).
/// On failure it is `Failed` at once and its pending sub-nodes are cancelled
/// (gated on a `Finishing` the parent never reached). Terminality then
/// propagates up the parent chain. Call [`Scheduler::settle`] again afterwards
/// to start newly-unblocked work.
/// propagates up the parent chain. Claim again afterwards to start
/// newly-unblocked work.
///
/// ⚠️ **Still `pub` for one caller**, and that caller is the last hole in
/// this wall: a host inserting a group root with no logic of its own
/// completes it immediately so it parks in `Finishing` and its children
/// become runnable. That is a statement about the *node* ("this one has no
/// work"), not an event to report, and it wants to be expressible at
/// insert time so completion can go `pub(crate)` alongside
/// [`Self::complete_growing`].
pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
match outcome {
Outcome::Failed(error) => {
@ -361,7 +351,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// would misreport that, and leaving it `Running` forever would wedge the
/// DAG. So the error is returned for the caller to log, not used to abort
/// the completion. This crate has no logger of its own; the caller does.
pub fn complete_growing(
pub(crate) fn complete_growing(
&mut self,
id: NodeId,
outcome: Outcome,
@ -622,7 +612,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
/// done" signal) supplies parent→child ordering; `Dep::Node` edges (which the
/// graph restricts to the same parent group) supply sibling ordering.
/// `Dep::Resource` edges are handled by the atomic acquire in
/// [`Scheduler::settle`], not here.
/// [`Scheduler::try_start`], not here.
fn node_deps_satisfied(&self, id: NodeId) -> bool {
let Some(node) = self.graph.node(id) else {
return false;
@ -664,6 +654,25 @@ mod tests {
name.to_owned()
}
/// Claim every currently-runnable node, as arrangement for the assertions
/// below. Equivalent to calling [`Scheduler::claim_one`] until it yields
/// `None`: a node started by an earlier iteration is `Running`, not
/// terminal, so it cannot satisfy another node's dependency here — it only
/// consumes resources.
///
/// **Was `Scheduler::settle`, a public method.** It was a `claim_one` loop
/// returning a `Vec`, and production never wanted the batch: the run loop
/// takes one node at a time through [`Scheduler::claim_next`] so it can
/// choose between claiming again and backing off, which a batch return
/// can't express. The only callers were tests, so it lives with them.
fn settle<N, R: Clone + Eq + Hash>(s: &mut Scheduler<N, R>) -> Vec<NodeId> {
let mut started = Vec::new();
while let Some(id) = s.claim_one() {
started.push(id);
}
started
}
/// A graph + a resource table with `build-slot` set to `slots`.
fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str, String> {
let mut table = ResourceTable::new();
@ -706,7 +715,7 @@ mod tests {
fn a_completing_node_grows_the_work_it_declared() {
let mut s = scheduler_with_slots(1);
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(s.settle(), vec![n]);
assert_eq!(settle(&mut s), vec![n]);
let grown = JobBuilder::new();
grown.node("child-a");
@ -732,7 +741,7 @@ mod tests {
fn a_failed_node_grows_nothing() {
let mut s = scheduler_with_slots(1);
let n = s.append("emitter", vec![], None).expect("insert");
assert_eq!(s.settle(), vec![n]);
assert_eq!(settle(&mut s), vec![n]);
let grown = JobBuilder::new();
grown.node("never-runs");
@ -766,12 +775,12 @@ mod tests {
let b = s.append("b", res_dep("build-slot"), None).expect("b");
let c = s.append("c", res_dep("build-slot"), None).expect("c");
assert_eq!(s.settle(), vec![a], "cap 1: only the first can start");
assert_eq!(settle(&mut s), vec![a], "cap 1: only the first can start");
s.complete(a, Outcome::Done);
// b and c are both satisfiable now; b was inserted first.
assert_eq!(s.settle(), vec![b], "the freed unit goes to b, not c");
assert_eq!(settle(&mut s), vec![b], "the freed unit goes to b, not c");
s.complete(b, Outcome::Done);
assert_eq!(s.settle(), vec![c]);
assert_eq!(settle(&mut s), vec![c]);
}
#[test]
@ -780,7 +789,7 @@ mod tests {
let n = s
.append("build", res_dep("build-slot"), None)
.expect("insert");
assert_eq!(s.settle(), vec![n]);
assert_eq!(settle(&mut s), vec![n]);
assert_eq!(s.graph().node(n).unwrap().state, State::Running);
assert_eq!(avail(&s, "build-slot"), 0);
// No children → completing it goes straight to Done (skips Finishing).
@ -800,7 +809,7 @@ mod tests {
assert!(s.graph().node(ok).unwrap().started_at.is_none());
assert!(s.graph().node(ok).unwrap().finished_at.is_none());
let started = s.settle();
let started = settle(&mut s);
assert!(started.contains(&ok) && started.contains(&bad));
// Running → started_at stamped, finished_at still none.
assert!(s.graph().node(ok).unwrap().started_at.is_some());
@ -836,11 +845,11 @@ mod tests {
let b = s.append("b", res_dep("build-slot"), None).expect("b");
let c = s.append("c", res_dep("build-slot"), None).expect("c");
// cap 2 → a + b start, c blocks on the exhausted slot.
assert_eq!(s.settle(), vec![a, b]);
assert_eq!(settle(&mut s), vec![a, b]);
assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
// a finishes → its slot frees → c can now start.
s.complete(a, Outcome::Done);
assert_eq!(s.settle(), vec![c]);
assert_eq!(settle(&mut s), vec![c]);
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
}
@ -852,16 +861,16 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let c1 = s.append("c1", vec![], Some(root)).expect("c1");
let c2 = s.append("c2", vec![], Some(root)).expect("c2");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
// Children can't start yet — parent still Running (logic not done).
assert!(s.settle().is_empty(), "children gated on parent logic");
assert!(settle(&mut s).is_empty(), "children gated on parent logic");
s.complete(root, Outcome::Done);
assert_eq!(
s.graph().node(root).unwrap().state,
State::Finishing,
"logic done, children pending → Finishing"
);
let mut started = s.settle();
let mut started = settle(&mut s);
started.sort();
let mut expected = vec![c1, c2];
expected.sort();
@ -885,9 +894,9 @@ mod tests {
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let root = s.append("root", vec![], None).expect("root");
let child = s.append("child", vec![], Some(root)).expect("child");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.settle(), vec![child]);
assert_eq!(settle(&mut s), vec![child]);
s.complete(child, Outcome::Failed(String::new()));
assert_eq!(
s.graph().node(root).unwrap().state,
@ -905,10 +914,10 @@ mod tests {
let r = s.append("R", res_dep("build-slot"), None).expect("R");
let c1 = s.append("c1", res_dep("build-slot"), Some(r)).expect("c1");
let c2 = s.append("c2", vec![after_ok(c1)], Some(r)).expect("c2");
assert_eq!(s.settle(), vec![r]);
assert_eq!(settle(&mut s), vec![r]);
s.complete(r, Outcome::Done); // → Finishing (children pending)
assert_eq!(avail(&s, "build-slot"), 0, "held: subtree not terminal");
assert_eq!(s.settle(), vec![c1], "c1 borrows R's slot");
assert_eq!(settle(&mut s), vec![c1], "c1 borrows R's slot");
assert_eq!(avail(&s, "build-slot"), 0, "borrow reuses R's unit");
s.complete(c1, Outcome::Done);
assert_eq!(
@ -916,7 +925,7 @@ mod tests {
0,
"still held: c2 pending in subtree"
);
assert_eq!(s.settle(), vec![c2]);
assert_eq!(settle(&mut s), vec![c2]);
s.complete(c2, Outcome::Done);
assert_eq!(
avail(&s, "build-slot"),
@ -934,14 +943,14 @@ mod tests {
let owner = s
.append("owner", res_dep("agent/foo"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
assert_eq!(avail(&s, "agent/foo"), 0);
let child = s
.append("child", res_dep("agent/foo"), Some(owner))
.expect("child");
s.complete(owner, Outcome::Done); // → Finishing
assert_eq!(avail(&s, "agent/foo"), 0, "held while a borrower pends");
assert_eq!(s.settle(), vec![child]);
assert_eq!(settle(&mut s), vec![child]);
assert_eq!(avail(&s, "agent/foo"), 0, "borrow reuses the one unit");
s.complete(child, Outcome::Done);
assert_eq!(avail(&s, "agent/foo"), 1);
@ -964,13 +973,13 @@ mod tests {
let great = s
.append("great", res_dep("agent/foo"), Some(grand))
.expect("great");
assert_eq!(s.settle(), vec![r]);
assert_eq!(settle(&mut s), vec![r]);
s.complete(r, Outcome::Done);
assert_eq!(s.settle(), vec![child], "child borrows R's grant");
assert_eq!(settle(&mut s), vec![child], "child borrows R's grant");
s.complete(child, Outcome::Done);
assert_eq!(s.settle(), vec![grand], "grand covered, no deadlock");
assert_eq!(settle(&mut s), vec![grand], "grand covered, no deadlock");
s.complete(grand, Outcome::Done);
assert_eq!(s.settle(), vec![great], "great covered too");
assert_eq!(settle(&mut s), vec![great], "great covered too");
assert_eq!(avail(&s, "agent/foo"), 0, "held across the whole nest");
s.complete(great, Outcome::Done);
assert_eq!(s.graph().node(r).unwrap().state, State::Done, "R rolled up");
@ -984,10 +993,14 @@ mod tests {
let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new());
let a = s.append("a", res_dep("agent/foo"), None).expect("a");
let b = s.append("b", res_dep("agent/foo"), None).expect("b");
assert_eq!(s.settle(), vec![a], "only a acquires; b can't borrow it");
assert_eq!(
settle(&mut s),
vec![a],
"only a acquires; b can't borrow it"
);
assert_eq!(s.graph().node(b).unwrap().state, State::Pending);
s.complete(a, Outcome::Done);
assert_eq!(s.settle(), vec![b]);
assert_eq!(settle(&mut s), vec![b]);
assert_eq!(s.graph().node(b).unwrap().state, State::Running);
}
@ -1000,7 +1013,7 @@ mod tests {
let owner = s
.append("owner", res_dep("agent/foo"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
let c1 = s
.append("c1", res_dep("agent/foo"), Some(owner))
.expect("c1");
@ -1008,10 +1021,10 @@ mod tests {
.append("c2", res_dep("agent/foo"), Some(owner))
.expect("c2");
s.complete(owner, Outcome::Done); // → Finishing
assert_eq!(s.settle(), vec![c1], "c1 borrows; c2 can't (cap 1)");
assert_eq!(settle(&mut s), vec![c1], "c1 borrows; c2 can't (cap 1)");
assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
s.complete(c1, Outcome::Done);
assert_eq!(s.settle(), vec![c2], "borrow returned → c2 borrows");
assert_eq!(settle(&mut s), vec![c2], "borrow returned → c2 borrows");
assert_eq!(avail(&s, "agent/foo"), 0, "still just the owner's unit");
}
@ -1023,7 +1036,7 @@ mod tests {
let owner = s
.append("owner", res_dep("build-slot"), None)
.expect("owner");
assert_eq!(s.settle(), vec![owner]);
assert_eq!(settle(&mut s), vec![owner]);
assert_eq!(avail(&s, "build-slot"), 1, "owner took one of two");
let c1 = s
.append("c1", res_dep("build-slot"), Some(owner))
@ -1032,7 +1045,7 @@ mod tests {
.append("c2", res_dep("build-slot"), Some(owner))
.expect("c2");
s.complete(owner, Outcome::Done); // → Finishing
let mut started = s.settle();
let mut started = settle(&mut s);
started.sort();
let mut expected = vec![c1, c2];
expected.sort();
@ -1062,11 +1075,11 @@ mod tests {
None,
)
.expect("weak");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(strong1).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(strong2).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![weak]);
assert_eq!(settle(&mut s), vec![weak]);
}
/// The direction only a *set* edge can express: a branch that runs solely on
@ -1086,7 +1099,7 @@ mod tests {
None,
)
.expect("compensate");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(
s.graph().node(on_fail).unwrap().state,
@ -1094,7 +1107,7 @@ mod tests {
"a Failed-only branch is unsatisfiable once its dep succeeds — and it is \
`Skipped`, not `Cancelled`, so the parent roll-up ignores it"
);
assert!(s.settle().is_empty(), "and nothing is left runnable");
assert!(settle(&mut s).is_empty(), "and nothing is left runnable");
}
/// The mirror: the same branch is exactly what *does* run on failure, while
@ -1116,10 +1129,10 @@ mod tests {
None,
)
.expect("on_fail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed("boom".to_owned()));
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![on_fail]);
assert_eq!(settle(&mut s), vec![on_fail]);
}
/// A weak edge accepts a dependency that was *ruled out*, so a tail still
@ -1140,11 +1153,11 @@ mod tests {
None,
)
.expect("tail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed("boom".to_owned()));
assert_eq!(s.graph().node(mid).unwrap().state, State::Skipped);
assert_eq!(
s.settle(),
settle(&mut s),
vec![tail],
"the tail runs off a cancelled dependency"
);
@ -1181,22 +1194,26 @@ mod tests {
// Everything succeeds: the ok branch runs, the failure branch is ruled out.
let (mut s, a, b, on_ok, on_fail) = build();
assert_eq!(s.settle(), vec![a, b], "both roots start; neither tail can");
assert_eq!(
settle(&mut s),
vec![a, b],
"both roots start; neither tail can"
);
s.complete(a, Outcome::Done);
s.complete(b, Outcome::Done);
assert_eq!(s.settle(), vec![on_ok]);
assert_eq!(settle(&mut s), vec![on_ok]);
s.complete(on_ok, Outcome::Done);
assert_eq!(s.graph().node(on_fail).unwrap().state, State::Skipped);
assert!(s.settle().is_empty());
assert!(settle(&mut s).is_empty());
// One of them fails: the ok branch is ruled out, which is precisely the
// signal the failure branch waits on.
let (mut s, a, b, on_ok, on_fail) = build();
assert_eq!(s.settle(), vec![a, b]);
assert_eq!(settle(&mut s), vec![a, b]);
s.complete(a, Outcome::Failed("boom".to_owned()));
s.complete(b, Outcome::Done);
assert_eq!(s.graph().node(on_ok).unwrap().state, State::Skipped);
assert_eq!(s.settle(), vec![on_fail]);
assert_eq!(settle(&mut s), vec![on_fail]);
}
#[test]
@ -1207,7 +1224,7 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let child = s.append("child", vec![], Some(root)).expect("child");
let grandchild = s.append("gc", vec![], Some(child)).expect("gc");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Failed(String::new()));
assert_eq!(s.graph().node(child).unwrap().state, State::Skipped);
assert_eq!(s.graph().node(grandchild).unwrap().state, State::Skipped);
@ -1223,7 +1240,7 @@ mod tests {
assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled);
assert_eq!(s.graph().node(b).unwrap().state, State::Skipped);
let c = s.append("c", vec![], None).expect("c");
assert_eq!(s.settle(), vec![c]);
assert_eq!(settle(&mut s), vec![c]);
assert!(!s.cancel_node(c));
assert_eq!(s.graph().node(c).unwrap().state, State::Running);
}
@ -1239,7 +1256,7 @@ mod tests {
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
// The root runs first and parks in `Finishing` while its children are
// outstanding — the state a group root is actually in when cancelled.
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.graph().node(root).unwrap().state, State::Finishing);
@ -1260,9 +1277,9 @@ mod tests {
let root = s.append("root", vec![], None).expect("root");
let a = s.append("a", vec![], Some(root)).expect("a");
let b = s.append("b", vec![after_ok(a)], Some(root)).expect("b");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert_eq!(s.settle(), vec![a], "a is claimed and running");
assert_eq!(settle(&mut s), vec![a], "a is claimed and running");
assert!(!s.cancel_node(root), "refused while a runs");
assert_eq!(s.graph().node(a).unwrap().state, State::Running);
@ -1291,7 +1308,7 @@ mod tests {
Some(root),
)
.expect("tail");
assert_eq!(s.settle(), vec![root]);
assert_eq!(settle(&mut s), vec![root]);
s.complete(root, Outcome::Done);
assert!(s.cancel_node(root));
@ -1301,7 +1318,7 @@ mod tests {
State::Pending,
"spared, and now runnable since its dep is Cancelled"
);
assert_eq!(s.settle(), vec![tail], "the tail still gets to report");
assert_eq!(settle(&mut s), vec![tail], "the tail still gets to report");
}
#[test]
@ -1309,7 +1326,7 @@ mod tests {
let mut s = scheduler_with_slots(1);
let g = s.append("g", res_dep("agent/foo"), None).expect("g");
let b = s.append("b", res_dep("build-slot"), None).expect("b");
assert_eq!(s.settle().len(), 2);
assert_eq!(settle(&mut s).len(), 2);
let state = s.resource_state();
assert!(state.contains(&(res("agent/foo"), g)));
assert!(state.contains(&(res("build-slot"), b)));