refactor(#2772): graph walks belong to jobq, not to its caller

hive-c0re hand-rolled four traversals over a graph it doesn't own, because
`Graph` exposed only `node()` and `nodes()`. They are generic — nothing in
them knows what a hyperhive DAG is — so they move to `hive-jobq` and core
delegates.

`Graph` gains `root_of`, `descendants`, `roots`, `is_settled` and
`first_error`; they reuse the private `is_descendant` the crate already had
for its dep-scope rule. `subtree` gets faster on the way: core walked every
node's whole parent chain to the root for every node in the graph, where
`is_descendant` stops as soon as it sees the ancestor.

`first_error` deliberately looks for the first `Failed` descendant that
*carries* an error rather than the first `Failed` one. A node that rolled
its failure up from a child holds no error of its own and sorts before that
child, so the simpler version reports `None` for the common case and the
dashboard loses the reason. The distinction has its own test.

`dag_is_terminal` is deleted rather than moved: it was already a plain
`state.is_terminal()` read, and its three call sites now ask the graph.

hive-c0re/src/job_queue/mod.rs

@@ -480,7 +480,7 @@ impl JobQueue {
         inner
             .containers()
             .into_iter()
-            .filter(|&c| !inner.dag_is_terminal(c))
+            .filter(|&c| !inner.sched.graph().is_settled(c))
             .count()
     }
 }
@@ -505,25 +505,17 @@ impl QueueInner {
         })
     }
 
-    /// The DAG container a node belongs to — walk its parent chain to the root
-    /// (`parent == None`), which is the container. Returns `id` itself for a
-    /// container node.
+    /// 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<NodeId> {
-        let mut cur = id;
-        loop {
-            match self.sched.graph().node(cur)?.parent {
-                Some(p) => cur = p,
-                None => return Some(cur),
-            }
-        }
+        self.sched.graph().root_of(id)
     }
 
     /// The DAG's work nodes — its `container`'s subtree, excluding the container.
     fn subtree(&self, container: NodeId) -> Vec<NodeId> {
         self.sched
             .graph()
-            .nodes()
-            .filter(|n| n.id != container && self.dag_of(n.id) == Some(container))
+            .descendants(container)
             .map(|n| n.id)
             .collect()
     }
@@ -553,15 +545,6 @@ impl QueueInner {
         })
     }
 
-    /// True when the DAG has settled — its container has rolled up terminal
-    /// (equivalent to every work node being terminal).
-    fn dag_is_terminal(&self, container: NodeId) -> bool {
-        self.sched
-            .graph()
-            .node(container)
-            .is_some_and(|n| n.state.is_terminal())
-    }
-
     /// 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
@@ -574,18 +557,12 @@ impl QueueInner {
         })
     }
 
-    /// First failed work node's error (read off the graph `Node`), for the
-    /// dashboard's DAG-level error line.
+    /// First failed work node's error, for the dashboard's DAG-level error line.
     fn dag_first_error(&self, container: NodeId) -> Option<String> {
-        for id in self.subtree(container) {
-            if let Some(n) = self.sched.graph().node(id)
-                && n.state == JobState::Failed
-                && let Some(e) = n.error.clone()
-            {
-                return Some(e);
-            }
-        }
-        None
+        self.sched
+            .graph()
+            .first_error(container)
+            .map(ToOwned::to_owned)
     }
 
     /// Project a DAG into its wire [`DagView`]: a near-raw view of the
@@ -675,7 +652,7 @@ impl QueueInner {
         if !any_unsettled {
             return None;
         }
-        let is_terminal = self.dag_is_terminal(container);
+        let is_terminal = self.sched.graph().is_settled(container);
         Some(DagView {
             id: container.get(),
             source: meta.source,
@@ -718,7 +695,7 @@ impl QueueInner {
         let mut live: Vec<NodeId> = Vec::new();
         let mut terminal: Vec<(NodeId, i64)> = Vec::new();
         for c in self.containers() {
-            if self.dag_is_terminal(c) {
+            if self.sched.graph().is_settled(c) {
                 terminal.push((c, self.dag_finished_at(c)));
             } else {
                 live.push(c);

hive-jobq/src/lib.rs

@@ -462,6 +462,56 @@ impl<N, R> Graph<N, R> {
         self.nodes.iter()
     }
 
+    /// Top of `id`'s [`Node::parent`] chain — the group root whose subtree `id`
+    /// lives in. Returns `id` itself when `id` is already a root, and `None`
+    /// only when `id` isn't in the graph.
+    #[must_use]
+    pub fn root_of(&self, id: NodeId) -> Option<NodeId> {
+        let mut cur = id;
+        loop {
+            match self.node(cur)?.parent {
+                Some(p) => cur = p,
+                None => return Some(cur),
+            }
+        }
+    }
+
+    /// Every node in `id`'s subtree, excluding `id` itself, in insertion order.
+    pub fn descendants(&self, id: NodeId) -> impl Iterator<Item = &Node<N, R>> {
+        self.nodes
+            .iter()
+            .filter(move |n| self.is_descendant(n.id, id))
+    }
+
+    /// Every group root — the nodes with no parent.
+    pub fn roots(&self) -> impl Iterator<Item = &Node<N, R>> {
+        self.nodes.iter().filter(|n| n.parent.is_none())
+    }
+
+    /// Whether `id` has settled. A group root's state is its subtree's roll-up,
+    /// so for a root this answers "is everything under it finished" — which is
+    /// why callers don't scan the subtree themselves. Unknown ids are not
+    /// settled.
+    #[must_use]
+    pub fn is_settled(&self, id: NodeId) -> bool {
+        self.node(id).is_some_and(|n| n.state.is_terminal())
+    }
+
+    /// Why `id`'s subtree failed: the error of the first `Failed` descendant
+    /// that carries one, in insertion order.
+    ///
+    /// Skipping the ones without an error is the point, not an optimisation. A
+    /// node that rolled up `Failed` from a child holds no error of its own, and
+    /// such a node can sort before the child that actually broke — stopping at
+    /// the first `Failed` node would report `None` while the real reason sits
+    /// further down the subtree.
+    #[must_use]
+    pub fn first_error(&self, id: NodeId) -> Option<&str> {
+        self.descendants(id)
+            .filter(|n| matches!(n.state, State::Failed))
+            .find_map(|n| n.error.as_deref())
+    }
+
     /// Whether `ancestor` lies on `node`'s [`Node::parent`] chain (i.e. `node` is
     /// in `ancestor`'s subtree). `node` is not its own ancestor.
     fn is_descendant(&self, node: NodeId, ancestor: NodeId) -> bool {
@@ -594,6 +644,74 @@ mod tests {
         ));
     }
 
+    /// Borrow a node mutably so a test can force its lifecycle state. Looks the
+    /// node up by id rather than indexing, so it doesn't quietly depend on ids
+    /// and positions coinciding.
+    fn node_mut<'g>(
+        g: &'g mut Graph<&'static str, String>,
+        id: NodeId,
+    ) -> &'g mut Node<&'static str, String> {
+        g.nodes
+            .iter_mut()
+            .find(|n| n.id == id)
+            .expect("node in graph")
+    }
+
+    /// `root_of` walks to the top of the parent chain; `descendants` is its
+    /// inverse and excludes the node itself.
+    #[test]
+    fn root_of_and_descendants_span_the_subtree() {
+        let mut g: Graph<&str, String> = Graph::new();
+        let root = g.insert("root", vec![], None).unwrap();
+        let mid = g.insert("mid", vec![], Some(root)).unwrap();
+        let leaf = g.insert("leaf", vec![], Some(mid)).unwrap();
+        let other = g.insert("other-root", vec![], None).unwrap();
+
+        assert_eq!(g.root_of(leaf), Some(root), "walks the whole chain");
+        assert_eq!(g.root_of(root), Some(root), "a root is its own root");
+        assert_eq!(g.root_of(NodeId(99)), None, "unknown id");
+
+        let mut under_root: Vec<NodeId> = g.descendants(root).map(|n| n.id).collect();
+        under_root.sort_unstable();
+        assert_eq!(under_root, vec![mid, leaf], "excludes the node itself");
+        assert_eq!(g.descendants(other).count(), 0);
+
+        let roots: Vec<NodeId> = g.roots().map(|n| n.id).collect();
+        assert_eq!(roots, vec![root, other]);
+    }
+
+    /// The reason `first_error` looks for the first failed descendant **that
+    /// carries an error** rather than simply the first failed one: a node that
+    /// rolled its `Failed` up from a child holds no error of its own, and it
+    /// sorts *before* that child. Stopping at the first `Failed` node would
+    /// report `None` and lose the real reason.
+    #[test]
+    fn first_error_skips_a_rolled_up_failure_carrying_no_error() {
+        let mut g: Graph<&str, String> = Graph::new();
+        let container = g.insert("dag", vec![], None).unwrap();
+        let rolled_up = g.insert("prebuild", vec![], Some(container)).unwrap();
+        let broke = g.insert("swap", vec![], Some(rolled_up)).unwrap();
+
+        node_mut(&mut g, rolled_up).state = State::Failed;
+        let broken = node_mut(&mut g, broke);
+        broken.state = State::Failed;
+        broken.error = Some("nix build exploded".to_owned());
+
+        assert_eq!(g.first_error(container), Some("nix build exploded"));
+    }
+
+    #[test]
+    fn is_settled_tracks_node_state() {
+        let mut g: Graph<&str, String> = Graph::new();
+        let n = g.insert("n", vec![], None).unwrap();
+        assert!(!g.is_settled(n), "Pending is not settled");
+        node_mut(&mut g, n).state = State::Finishing;
+        assert!(!g.is_settled(n), "Finishing still has children running");
+        node_mut(&mut g, n).state = State::Done;
+        assert!(g.is_settled(n));
+        assert!(!g.is_settled(NodeId(99)), "unknown id is not settled");
+    }
+
     #[test]
     fn state_terminality() {
         assert!(State::Done.is_terminal());