From e646656c928acc26d9083cb396b1e1362b1f0d44 Mon Sep 17 00:00:00 2001 From: atlas Date: Sun, 2 Aug 2026 21:44:33 +0200 Subject: [PATCH] c0re's queue tests no longer drive the scheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- hive-c0re/src/job_queue/mod.rs | 61 +++++--- hive-c0re/src/job_queue/tests.rs | 245 +++++++++++-------------------- hive-host-sock/src/jobs.rs | 103 +++++++++++++ hive-jobq/src/scheduler.rs | 191 +++++++++++++----------- 4 files changed, 334 insertions(+), 266 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 8181995f..a812d4ec 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -404,32 +404,27 @@ fn dag_meta(sched: &Sched, container: NodeId) -> Option { /// aged out). fn dag_view(sched: &Sched, container: NodeId) -> Option { 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> = Vec::new(); let mut finished: Vec> = 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::>())?; + let mut nodes = Vec::new(); + for node in shown.into_iter().map(|i| all[i]) { + let id = node.id; let deps: Vec = node .deps .iter() @@ -480,9 +475,6 @@ fn dag_view(sched: &Sched, container: NodeId) -> Option { 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 { }) } +/// 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> { + 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. diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index ae94f24c..5ef2a0be 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -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; -} - -impl ClaimReady for JobQueue { - fn claim_ready(&self) -> Vec { - 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 diff --git a/hive-host-sock/src/jobs.rs b/hive-host-sock/src/jobs.rs index feba036f..2f567bfd 100644 --- a/hive-host-sock/src/jobs.rs +++ b/hive-host-sock/src/jobs.rs @@ -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); + } +} diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 2300a552..dadbac0d 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -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 Scheduler { } /// 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 Scheduler { /// [`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 Scheduler { /// 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 { + fn claim_one(&mut self) -> Option { let pending: Vec = self .graph .nodes() @@ -216,24 +216,6 @@ impl Scheduler { }) } - /// 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 { - 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 Scheduler { /// (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 Scheduler { /// 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 Scheduler { /// 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(s: &mut Scheduler) -> Vec { + 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)));