From 5c5c8776d268960ae69ae6a65734fda95571d89e Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 27 Jul 2026 20:24:33 +0200 Subject: [PATCH] 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);