From 5c5c8776d268960ae69ae6a65734fda95571d89e Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 20:24:33 +0200 Subject: [PATCH 1/3] refactor(#2802): cancelling a DAG is a scheduler operation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `JobQueue::cancel` decided whether a DAG could be cancelled by reading node run-state, walked the subtree, judged per node whether that node had asked to observe cancellation, and re-ran the container's roll-up. Every one of those is a fact the scheduler owns; core was reaching across the boundary to compute them. `Scheduler::cancel_node` now takes the whole subtree: cancelling a node cancels the work under it, since a group is abandoned by abandoning its root. The existing method generalises rather than gaining a sibling — it had one production caller, which this replaces. The gate runs over the work *under* the node, not the node itself: a group root's state is its subtree's roll-up rather than a step that ran, so a container is `Finishing` and never `Pending`, and gating on it would refuse every cancel. A node with no children is its own work, which keeps the previous single-node behaviour. `observes_cancellation` moves in with it — it reads a node's declared edges and knows nothing about what the payload means. Core keeps the one genuinely domain-specific step, resolving a wire `dag_id` to its container node, and is three lines otherwise. --- hive-c0re/src/job_queue/mod.rs | 35 +------- hive-jobq/src/scheduler.rs | 155 +++++++++++++++++++++++++++++---- 2 files changed, 138 insertions(+), 52 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 6427a0a1..049e6080 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -371,30 +371,9 @@ impl JobQueue { let Some(container) = inner.container(dag_id) else { return false; }; - let work = inner.subtree(container); - let all_pending = work.iter().all(|&id| { - inner - .sched - .graph() - .node(id) - .is_some_and(|n| n.state == JobState::Pending) - }); - if !all_pending { + if !inner.sched.cancel_node(container) { return false; } - for id in work { - if inner.observes_cancellation(id) { - continue; - } - inner.sched.cancel_node(id); - } - // Re-run the container's roll-up now that its children are `Cancelled`. - // With a spared tail still `Pending` this is a deliberate no-op — the - // container has a non-terminal child, so `settle_terminal` parks it back - // in `Finishing` and it rolls up for real once the tail finishes. With no - // tail (a power op) every child *is* terminal, so it settles synchronously - // here exactly as before. - inner.sched.complete(container, Outcome::Done); drop(inner); self.notify.notify_one(); true @@ -545,18 +524,6 @@ impl QueueInner { }) } - /// Whether `id` has an edge that accepts a **dropped** dependency — i.e. the - /// node exists to report on work that may never run. Used by - /// [`JobQueue::cancel`] to decide what to spare, so the decision comes from - /// the node's own declared edges rather than a hardcoded list of kinds. - fn observes_cancellation(&self, id: NodeId) -> bool { - self.sched.graph().node(id).is_some_and(|n| { - n.deps.iter().any(|d| { - matches!(d, Dep::Node { when, .. } if when.accepts(hive_jobq::TerminalState::Cancelled)) - }) - }) - } - /// First failed work node's error, for the dashboard's DAG-level error line. fn dag_first_error(&self, container: NodeId) -> Option { self.sched diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index f18437d5..462ff3d1 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -32,7 +32,7 @@ use std::collections::HashMap; use std::hash::Hash; use crate::resources::ResourceTable; -use crate::{Dep, Graph, GraphError, NodeId, State}; +use crate::{Dep, Graph, GraphError, NodeId, State, TerminalState}; /// The result of a node's own execution, reported to [`Scheduler::complete`]. /// @@ -291,25 +291,65 @@ impl Scheduler { } } - /// Cancel a still-*pending* node (and cascade to the dependents it rules out): - /// mark it [`State::Cancelled`] and report whether it was cancellable. A - /// node that has already started (`Running`) or finished is left untouched — - /// an in-flight node's work is not interruptible. A pending node holds no - /// resources, so nothing is released here; call [`Scheduler::settle`] - /// afterwards to let now-terminal dependents advance (e.g. a weak-edge - /// terminal node observing the cancellation). + /// Cancel the not-yet-started work at `id`: mark it [`State::Cancelled`] + /// and cascade to the dependents that rules out. Cancelling a node cancels + /// what hangs under it — a group is abandoned by abandoning its root. + /// Reports whether anything was cancelled. + /// + /// **All-or-nothing.** A subtree with any node already `Running` or terminal + /// is left completely untouched: an in-flight node's work is not + /// interruptible, and cancelling only the pending half would leave the group + /// half-executed with no way to finish it. + /// + /// **A node whose edge accepts [`TerminalState::Cancelled`] is spared.** Such + /// a node is asking to run precisely when the work it follows is dropped, + /// which is what lets a reporting tail settle whatever it reports to instead + /// of leaving it dangling forever. Nothing is special-cased by payload — the + /// node's own declared edges decide. Note `DepWhen::AFTER_ANY` deliberately + /// does *not* accept `Cancelled`, so an ordinary weak-edged step is cancelled + /// along with the rest; there is nothing to do when no node ever ran. + /// + /// Cancelled nodes were pending, so they hold no resources and none are + /// released here. pub fn cancel_node(&mut self, id: NodeId) -> bool { - if self - .graph - .node(id) - .is_some_and(|n| n.state == State::Pending) - { - self.graph.set_state(id, State::Cancelled); - self.cascade_cancel(id); - true - } else { - false + // The unit is the work *under* `id`; a node with no children is its own + // work. A group root's state is its subtree's roll-up rather than a step + // that ran, so the root itself is not part of the gate. + let under: Vec = self.graph.descendants(id).map(|n| n.id).collect(); + let is_group = !under.is_empty(); + let targets = if is_group { under } else { vec![id] }; + if !targets.iter().all(|&n| { + self.graph + .node(n) + .is_some_and(|n| n.state == State::Pending) + }) { + return false; } + for n in targets { + if self.observes_cancellation(n) { + continue; + } + self.graph.set_state(n, State::Cancelled); + self.cascade_cancel(n); + } + if is_group { + // Re-run the root's roll-up now its children are terminal. With a + // spared node still pending this is a deliberate no-op: the root + // still has a non-terminal child, so it parks back in `Finishing` + // and rolls up for real once that node finishes. + self.complete(id, Outcome::Done); + } + true + } + + /// Whether `id` has an edge that accepts a **dropped** dependency — i.e. it + /// exists to report on work that may never run. + fn observes_cancellation(&self, id: NodeId) -> bool { + self.graph.node(id).is_some_and(|n| { + n.deps.iter().any( + |d| matches!(d, Dep::Node { when, .. } if when.accepts(TerminalState::Cancelled)), + ) + }) } /// Snapshot the currently-held grants as `(resource, owner)` pairs — one @@ -924,6 +964,85 @@ mod tests { assert_eq!(s.graph().node(c).unwrap().state, State::Running); } + /// Cancelling a group root abandons the work under it. The root's own state + /// is a roll-up (it parks in `Finishing`), never `Pending`, so gating on the + /// root instead of its children would refuse every group. + #[test] + fn cancel_node_cancels_the_work_under_a_group_root() { + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + 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"); + // 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]); + s.complete(root, Outcome::Done); + assert_eq!(s.graph().node(root).unwrap().state, State::Finishing); + + assert!(s.cancel_node(root)); + assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled); + assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled); + // With every child terminal the root leaves `Finishing` and rolls up. + // It rolls up `Failed`, not `Cancelled`: the roll-up counts a cancelled + // child as a non-success, and it has no separate "the whole group was + // dropped" outcome. Pre-existing and unchanged here — a caller that + // wants to show a cancel as a cancel derives that from the node states, + // not from the root's. + assert_eq!(s.graph().node(root).unwrap().state, State::Failed); + } + + /// All-or-nothing: an in-flight node's work is not interruptible, and + /// cancelling only the pending half would strand the group half-executed. + #[test] + fn cancel_node_refuses_a_group_with_anything_running() { + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + 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]); + s.complete(root, Outcome::Done); + assert_eq!(s.settle(), 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); + assert_eq!( + s.graph().node(b).unwrap().state, + State::Pending, + "the pending half is left alone too — nothing partial" + ); + } + + /// A node whose edge accepts `Cancelled` asked to run when the work it + /// follows is dropped. Sparing it is what lets a reporting tail settle the + /// thing it reports to instead of leaving it dangling forever. + #[test] + fn cancel_node_spares_a_node_that_observes_cancellation() { + let mut s: Scheduler<&str, String> = Scheduler::new(Graph::new(), ResourceTable::new()); + let root = s.append("root", vec![], None).expect("root"); + let work = s.append("work", vec![], Some(root)).expect("work"); + let tail = s + .append( + "tail", + vec![Dep::Node { + id: work, + when: DepWhen::of(&[TerminalState::Cancelled]), + }], + Some(root), + ) + .expect("tail"); + assert_eq!(s.settle(), vec![root]); + s.complete(root, Outcome::Done); + + assert!(s.cancel_node(root)); + assert_eq!(s.graph().node(work).unwrap().state, State::Cancelled); + assert_eq!( + s.graph().node(tail).unwrap().state, + State::Pending, + "spared, and now runnable since its dep is Cancelled" + ); + assert_eq!(s.settle(), vec![tail], "the tail still gets to report"); + } + #[test] fn resource_state_reports_owners() { let mut s = scheduler_with_slots(1); From 6f551334de9e0e2a3ae2149374803bf46b2a2a09 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 21:10:24 +0200 Subject: [PATCH 2/3] refactor(#2802): drop the wrappers that now only forward to the graph MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `dag_of` and `dag_first_error` had shrunk to a single delegating call once the walks moved into `hive-jobq`; their callers say what they mean without the hop. `subtree` was worse than redundant. It collected the descendant ids into a `Vec` and both callers then looked each node up again by id — `dag_view` needed a `let … else { continue }` for a lookup that could not fail. Iterating `descendants()` hands back the node directly, so the round-trip and the re-lookup both go. --- hive-c0re/src/job_queue/mod.rs | 53 +++++++++++----------------------- 1 file changed, 17 insertions(+), 36 deletions(-) diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index 049e6080..e392237a 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -104,8 +104,8 @@ struct DagMeta { /// container node** ([`NodeKind::Dag`], `parent = None`) whose subtree is the /// DAG's work — so the container's `NodeId` is the DAG id, its rolled-up state /// is the DAG state, and there are no grouping side-tables: membership + meta -/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] / -/// [`QueueInner::dag_meta`]). One shared crate [`Graph`] holds every DAG. +/// are graph queries ([`QueueInner::container`] / [`QueueInner::dag_meta`] + +/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG. struct QueueInner { sched: Scheduler, /// Per-node runtime metadata (the build-log id) — mutable after @@ -309,7 +309,7 @@ impl JobQueue { }; let kind = node.payload.clone(); let agent = node.payload.agent().to_owned(); - let Some(container) = inner.dag_of(id) else { + let Some(container) = inner.sched.graph().root_of(id) else { continue; }; let Some(meta) = inner.dag_meta(container) else { @@ -382,7 +382,9 @@ impl JobQueue { /// Link a `build_logs` row to a specific `Running` node. pub fn set_build_log_id(&self, dag_id: u64, node_id: NodeId, log_id: i64) -> bool { let mut inner = self.lock(); - if inner.dag_of(node_id).map(NodeId::get) != Some(dag_id) || !inner.node_running(node_id) { + if inner.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id) + || !inner.node_running(node_id) + { return false; } inner.node_rt.entry(node_id).or_default().build_log_id = Some(log_id); @@ -417,7 +419,11 @@ impl JobQueue { pub fn first_error(&self, dag_id: u64) -> Option { let inner = self.lock(); let container = inner.container(dag_id)?; - inner.dag_first_error(container) + inner + .sched + .graph() + .first_error(container) + .map(ToOwned::to_owned) } /// The `(dag_id, agent, kind)` triples for every per-agent lease currently @@ -435,7 +441,7 @@ impl JobQueue { let Resource::Agent(agent) = res else { return None; }; - let container = inner.dag_of(holder)?; + let container = inner.sched.graph().root_of(holder)?; let kind = inner.dag_meta(container)?.transient?; Some((container.get(), agent, kind)) }) @@ -484,21 +490,6 @@ impl QueueInner { }) } - /// The DAG container a node belongs to. A container is exactly a group root, - /// so this is the graph's own parent-chain walk. - fn dag_of(&self, id: NodeId) -> Option { - self.sched.graph().root_of(id) - } - - /// The DAG's work nodes — its `container`'s subtree, excluding the container. - fn subtree(&self, container: NodeId) -> Vec { - self.sched - .graph() - .descendants(container) - .map(|n| n.id) - .collect() - } - /// The container's carried domain metadata as an owned read-view. The data /// lives solely in the [`NodeKind::Dag`] payload — this is a derived read, /// not a stored side-table. @@ -524,14 +515,6 @@ impl QueueInner { }) } - /// First failed work node's error, for the dashboard's DAG-level error line. - fn dag_first_error(&self, container: NodeId) -> Option { - self.sched - .graph() - .first_error(container) - .map(ToOwned::to_owned) - } - /// Project a DAG into its wire [`DagView`]: a near-raw view of the /// container's work nodes, with `Done` nodes excluded. Lifecycle /// (`state` / `started_at` / `finished_at` / `error`) is read straight @@ -554,10 +537,8 @@ impl QueueInner { // 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 id in self.subtree(container) { - let Some(node) = self.sched.graph().node(id) else { - continue; - }; + for node in self.sched.graph().descendants(container) { + let id = node.id; if let Some(s) = node.started_at { started.push(s); } @@ -635,9 +616,9 @@ impl QueueInner { /// subtree (read off the graph `Node`, as unix seconds), for the history /// cap ordering. fn dag_finished_at(&self, container: NodeId) -> i64 { - self.subtree(container) - .iter() - .filter_map(|id| self.sched.graph().node(*id)) + self.sched + .graph() + .descendants(container) .filter_map(|n| n.finished_at) .map(|t| t.timestamp()) .max() From 6fd91ccf6a42b404b6fdcc9674a9396e39c25fce Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 21:22:53 +0200 Subject: [PATCH 3/3] fix(#2802): a group whose children were all dropped rolls up Cancelled MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The roll-up treated a cancelled child the same as a failed one, so a DAG the operator cancelled before it started reported `Failed` — it claimed to have failed at something when nothing under it ever ran. `Failed` still outranks `Cancelled`: a group where one step broke and the rest were dropped in response is a failure, and that is the fact worth surfacing. Only a group with no failed child at all reports the cancel. This also settles a disagreement. `DagView::rollup_state` on the wire has always ranked failed over cancelled over the rest; the graph's own roll-up had no `Cancelled` outcome to rank, so the two described the same DAG differently depending on which one you asked. --- hive-jobq/src/scheduler.rs | 59 +++++++++++++++++++++----------------- 1 file changed, 33 insertions(+), 26 deletions(-) diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index 462ff3d1..e61dd64d 100644 --- a/hive-jobq/src/scheduler.rs +++ b/hive-jobq/src/scheduler.rs @@ -233,15 +233,31 @@ impl Scheduler { .all(|n| n.parent != Some(id) || n.state.is_terminal()) } - /// Whether any direct child of `id` ended `Failed`/`Cancelled` — the roll-up - /// failure condition for the parent. - /// `Skipped` children are **not** counted: being ruled out by an edge is the - /// expected fate of every branch not taken, so counting it would make any - /// group that branches on outcome roll up failed no matter how the run went. - fn any_child_failed(&self, id: NodeId) -> bool { - self.graph - .nodes() - .any(|n| n.parent == Some(id) && matches!(n.state, State::Failed | State::Cancelled)) + /// What `id` rolls up to once all its children are terminal: `Failed` if any + /// child failed, else `Cancelled` if any was cancelled, else `Done`. + /// + /// `Failed` outranks `Cancelled` because the failure is the actionable fact — + /// a group where one step broke and the rest were dropped in response is a + /// failure, not a cancellation. A group whose children were *all* dropped + /// never failed at anything, and says so. + /// + /// `Skipped` children are ignored: being ruled out by an edge is the expected + /// fate of every branch not taken, so counting it would make any group that + /// branches on outcome roll up non-`Done` however the run went. + fn rolled_up_state(&self, id: NodeId) -> State { + let mut any_cancelled = false; + for child in self.graph.nodes().filter(|n| n.parent == Some(id)) { + match child.state { + State::Failed => return State::Failed, + State::Cancelled => any_cancelled = true, + _ => {} + } + } + if any_cancelled { + State::Cancelled + } else { + State::Done + } } /// Transition a node whose own logic just *succeeded* to its resulting state: @@ -254,12 +270,10 @@ impl Scheduler { /// its dependency succeeds, and leaving it `Pending` would wedge the subtree /// non-terminal forever. fn settle_terminal(&mut self, id: NodeId) { - let state = if !self.all_children_terminal(id) { - State::Finishing - } else if self.any_child_failed(id) { - State::Failed + let state = if self.all_children_terminal(id) { + self.rolled_up_state(id) } else { - State::Done + State::Finishing }; self.graph.set_state(id, state); if state.is_terminal() { @@ -279,11 +293,7 @@ impl Scheduler { { break; } - let state = if self.any_child_failed(a) { - State::Failed - } else { - State::Done - }; + let state = self.rolled_up_state(a); self.graph.set_state(a, state); // Any terminal outcome can rule a dependent out — see `settle_terminal`. self.cascade_cancel(a); @@ -982,13 +992,10 @@ mod tests { assert!(s.cancel_node(root)); assert_eq!(s.graph().node(a).unwrap().state, State::Cancelled); assert_eq!(s.graph().node(b).unwrap().state, State::Cancelled); - // With every child terminal the root leaves `Finishing` and rolls up. - // It rolls up `Failed`, not `Cancelled`: the roll-up counts a cancelled - // child as a non-success, and it has no separate "the whole group was - // dropped" outcome. Pre-existing and unchanged here — a caller that - // wants to show a cancel as a cancel derives that from the node states, - // not from the root's. - assert_eq!(s.graph().node(root).unwrap().state, State::Failed); + // With every child terminal the root leaves `Finishing` and rolls up — + // as `Cancelled`, not `Failed`: nothing under it failed at anything, the + // work was dropped. + assert_eq!(s.graph().node(root).unwrap().state, State::Cancelled); } /// All-or-nothing: an in-flight node's work is not interruptible, and