refactor(#2802): cancelling a DAG is a scheduler operation

`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.
This commit is contained in:
atlas 2026-07-27 20:24:33 +02:00
commit 5c5c8776d2
2 changed files with 138 additions and 52 deletions

View file

@ -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<String> {
self.sched

View file

@ -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<N, R: Clone + Eq + Hash> Scheduler<N, R> {
}
}
/// 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<NodeId> = 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);