diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index e392237a..6427a0a1 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::dag_meta`] + -/// the `hive_jobq::Graph` accessors). One shared crate [`Graph`] holds every DAG. +/// are graph queries ([`QueueInner::container`] / [`QueueInner::subtree`] / +/// [`QueueInner::dag_meta`]). 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.sched.graph().root_of(id) else { + let Some(container) = inner.dag_of(id) else { continue; }; let Some(meta) = inner.dag_meta(container) else { @@ -371,9 +371,30 @@ impl JobQueue { let Some(container) = inner.container(dag_id) else { return false; }; - if !inner.sched.cancel_node(container) { + 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 { 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 @@ -382,9 +403,7 @@ 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.sched.graph().root_of(node_id).map(NodeId::get) != Some(dag_id) - || !inner.node_running(node_id) - { + if inner.dag_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); @@ -419,11 +438,7 @@ impl JobQueue { pub fn first_error(&self, dag_id: u64) -> Option { let inner = self.lock(); let container = inner.container(dag_id)?; - inner - .sched - .graph() - .first_error(container) - .map(ToOwned::to_owned) + inner.dag_first_error(container) } /// The `(dag_id, agent, kind)` triples for every per-agent lease currently @@ -441,7 +456,7 @@ impl JobQueue { let Resource::Agent(agent) = res else { return None; }; - let container = inner.sched.graph().root_of(holder)?; + let container = inner.dag_of(holder)?; let kind = inner.dag_meta(container)?.transient?; Some((container.get(), agent, kind)) }) @@ -490,6 +505,21 @@ 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. @@ -515,6 +545,26 @@ 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 + .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 @@ -537,8 +587,10 @@ 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 node in self.sched.graph().descendants(container) { - let id = node.id; + for id in self.subtree(container) { + let Some(node) = self.sched.graph().node(id) else { + continue; + }; if let Some(s) = node.started_at { started.push(s); } @@ -616,9 +668,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.sched - .graph() - .descendants(container) + self.subtree(container) + .iter() + .filter_map(|id| self.sched.graph().node(*id)) .filter_map(|n| n.finished_at) .map(|t| t.timestamp()) .max() diff --git a/hive-jobq/src/scheduler.rs b/hive-jobq/src/scheduler.rs index e61dd64d..f18437d5 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, TerminalState}; +use crate::{Dep, Graph, GraphError, NodeId, State}; /// The result of a node's own execution, reported to [`Scheduler::complete`]. /// @@ -233,31 +233,15 @@ impl Scheduler { .all(|n| n.parent != Some(id) || n.state.is_terminal()) } - /// 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 - } + /// 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)) } /// Transition a node whose own logic just *succeeded* to its resulting state: @@ -270,10 +254,12 @@ 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) { - self.rolled_up_state(id) - } else { + let state = if !self.all_children_terminal(id) { State::Finishing + } else if self.any_child_failed(id) { + State::Failed + } else { + State::Done }; self.graph.set_state(id, state); if state.is_terminal() { @@ -293,7 +279,11 @@ impl Scheduler { { break; } - let state = self.rolled_up_state(a); + let state = if self.any_child_failed(a) { + State::Failed + } else { + State::Done + }; self.graph.set_state(a, state); // Any terminal outcome can rule a dependent out — see `settle_terminal`. self.cascade_cancel(a); @@ -301,65 +291,25 @@ impl Scheduler { } } - /// 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. + /// 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). pub fn cancel_node(&mut self, id: NodeId) -> bool { - // 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; + 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 } - 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 @@ -974,82 +924,6 @@ 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 — - // 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 - /// 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);