feat(#2500): hive-jobq scheduler settle loop (owned resources + subtree-hold)

hive-jobq/src/lib.rs

@@ -26,11 +26,12 @@
 //! node kind. Resources are held by the acquiring node and released on
 //! completion via guard objects, recursive within a group.
 //!
-//! The scheduler loop is a follow-up; the resource machinery lives in
-//! [`resources`] and the RAII lock guards over it in [`guard`].
+//! The [`scheduler`] settle loop drives execution; the resource machinery
+//! lives in [`resources`] and the RAII lock guards over it in [`guard`].
 
 pub mod guard;
 pub mod resources;
+pub mod scheduler;
 
 /// Opaque, stable, monotonic node identifier.
 ///
@@ -291,6 +292,24 @@ impl<N> Graph<N> {
         self.nodes.iter().filter(move |n| n.parent == Some(id))
     }
 
+    /// Every node in the graph, in insertion order. The scheduler iterates
+    /// this to find runnable pending nodes.
+    pub fn nodes(&self) -> impl Iterator<Item = &Node<N>> {
+        self.nodes.iter()
+    }
+
+    /// Set a node's lifecycle state, returning `false` for an unknown id. The
+    /// scheduler drives every state transition — nothing else mutates state,
+    /// which is what keeps the resource guards + terminality in sync.
+    pub fn set_state(&mut self, id: NodeId, state: State) -> bool {
+        if let Some(node) = self.nodes.iter_mut().find(|n| n.id == id) {
+            node.state = state;
+            true
+        } else {
+            false
+        }
+    }
+
     /// A group is terminal once the group node itself is terminal *and* every
     /// node inside it (recursively) is terminal. The node's own state matters:
     /// a group node still `Pending`/`Running` is not terminal even with no

hive-jobq/src/scheduler.rs

@@ -0,0 +1,249 @@
+//! The settle loop — drives a [`Graph`] to completion over the resource pool.
+//!
+//! [`Scheduler::settle`] claims every currently-runnable pending node (its
+//! [`Dep::Node`] edges satisfied *and* all its [`Dep::Resource`] units acquired
+//! atomically), marks it `Running`, holds its resource guards, and returns the
+//! newly-started ids for the caller's runner to execute. The runner reports each
+//! node's result back with [`Scheduler::complete`]; a running node may grow its
+//! own sub-group first via [`Scheduler::append`]. Concurrency is emergent from
+//! resource capacity — there is no separate active-node cap.
+//!
+//! A resource is held for the acquiring node's *entire subtree* lifetime: the
+//! owned guard is released only when that node and every descendant is terminal
+//! (`group_terminal`), not when the node's own work finishes. Single-owner and
+//! synchronous — the caller drives `settle` / `complete`; no async or locking
+//! lives here (that's the runner's job, one layer up).
+//!
+//! Recursive-lock re-entrancy (a sub-node reusing an ancestor group's lock) and
+//! the eager `AfterOk` failure cascade are layered on top of this owned core.
+
+use std::collections::HashMap;
+
+use crate::guard::{ResourceGuard, SharedResources};
+use crate::resources::ResourceTable;
+use crate::{Dep, DepWhen, Graph, GraphError, NodeId, ResourceName, State};
+
+/// The result of a node's own execution, reported to [`Scheduler::complete`].
+///
+/// `Cancelled` is not an outcome a runner reports — it is scheduler-driven (an
+/// `AfterOk` dependency failed), so a runner only ever says `Done` or `Failed`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Outcome {
+    /// The node's work succeeded.
+    Done,
+    /// The node's work failed.
+    Failed,
+}
+
+/// Drives a [`Graph`] over a shared resource pool: claim runnable nodes, hold
+/// their resources for the subtree's lifetime, release on subtree-terminal.
+pub struct Scheduler<N> {
+    graph: Graph<N>,
+    resources: SharedResources,
+    /// Owned resource guards, keyed by the node that acquired them. Dropped
+    /// (releasing the units) when that node's whole subtree is terminal.
+    owned: HashMap<NodeId, Vec<ResourceGuard>>,
+}
+
+impl<N> Scheduler<N> {
+    /// A scheduler over `graph` with `resources` as the capacity pool.
+    #[must_use]
+    pub fn new(graph: Graph<N>, resources: ResourceTable) -> Self {
+        Self {
+            graph,
+            resources: SharedResources::new(resources),
+            owned: HashMap::new(),
+        }
+    }
+
+    /// The graph, for inspection (state, hierarchy, UI rendering).
+    #[must_use]
+    pub fn graph(&self) -> &Graph<N> {
+        &self.graph
+    }
+
+    /// Append a node — e.g. a running node growing its own sub-group. Delegates
+    /// to [`Graph::insert`]; call [`Scheduler::settle`] afterwards to start it
+    /// once it is runnable.
+    ///
+    /// # Errors
+    /// Propagates [`GraphError`] for a dangling parent or dependency id.
+    pub fn append(
+        &mut self,
+        payload: N,
+        deps: Vec<Dep>,
+        parent: Option<NodeId>,
+    ) -> Result<NodeId, GraphError> {
+        self.graph.insert(payload, deps, parent)
+    }
+
+    /// Claim every currently-runnable pending node and start it: node-deps
+    /// satisfied and all resource-deps acquired atomically (all-or-nothing).
+    /// Each claimed node is marked `Running`, its owned guards held, and its id
+    /// returned for the runner to execute. A single pass suffices — a node
+    /// started here is `Running`, not terminal, so it cannot satisfy another
+    /// node's dependency in the same pass; it only consumes resources.
+    pub fn settle(&mut self) -> Vec<NodeId> {
+        let pending: Vec<NodeId> = self
+            .graph
+            .nodes()
+            .filter(|n| n.state == State::Pending)
+            .map(|n| n.id)
+            .collect();
+        let mut started = Vec::new();
+        for id in pending {
+            if !self.node_deps_satisfied(id) {
+                continue;
+            }
+            if let Some(guard) = self.resources.acquire(self.resource_reqs(id)) {
+                self.graph.set_state(id, State::Running);
+                self.owned.entry(id).or_default().push(guard);
+                started.push(id);
+            }
+        }
+        started
+    }
+
+    /// Report a running node's own execution result. Sets its state, then
+    /// releases the owned guards of every node whose whole subtree has become
+    /// terminal — a parent keeps its lock until its last descendant finishes.
+    /// Call [`Scheduler::settle`] again afterwards to start newly-unblocked work.
+    pub fn complete(&mut self, id: NodeId, outcome: Outcome) {
+        let state = match outcome {
+            Outcome::Done => State::Done,
+            Outcome::Failed => State::Failed,
+        };
+        self.graph.set_state(id, state);
+        self.release_settled_subtrees();
+    }
+
+    /// Drop the owned guards of every holder whose subtree is now terminal.
+    fn release_settled_subtrees(&mut self) {
+        let holders: Vec<NodeId> = self.owned.keys().copied().collect();
+        for holder in holders {
+            if self.graph.group_terminal(holder) {
+                self.owned.remove(&holder); // drops guards → releases the units
+            }
+        }
+    }
+
+    /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource`
+    /// edges are handled by the atomic acquire in [`Scheduler::settle`], not here.
+    fn node_deps_satisfied(&self, id: NodeId) -> bool {
+        let Some(node) = self.graph.node(id) else {
+            return false;
+        };
+        node.deps.iter().all(|dep| match dep {
+            Dep::Resource { .. } => true,
+            Dep::Node { id, when } => self.dep_node_satisfied(*id, *when),
+        })
+    }
+
+    /// Whether a node/group dependency `id` satisfies edge kind `when`. A group
+    /// is depended on as a whole: `AfterAny` needs its subtree terminal (any
+    /// outcome), `AfterOk` needs its whole subtree to have succeeded.
+    fn dep_node_satisfied(&self, id: NodeId, when: DepWhen) -> bool {
+        match when {
+            DepWhen::AfterAny => self.graph.group_terminal(id),
+            DepWhen::AfterOk => self.subtree_all_done(id),
+        }
+    }
+
+    /// Whether `id` and every descendant reached [`State::Done`] — the success
+    /// condition for an `AfterOk` edge onto a (possibly group) node.
+    fn subtree_all_done(&self, id: NodeId) -> bool {
+        let Some(node) = self.graph.node(id) else {
+            return false;
+        };
+        node.state == State::Done && self.graph.children(id).all(|c| self.subtree_all_done(c.id))
+    }
+
+    /// The `(name, count)` resource units `id` must hold to run.
+    fn resource_reqs(&self, id: NodeId) -> Vec<(ResourceName, u32)> {
+        let Some(node) = self.graph.node(id) else {
+            return Vec::new();
+        };
+        node.deps
+            .iter()
+            .filter_map(|dep| match dep {
+                Dep::Resource { name, count } => Some((name.clone(), *count)),
+                Dep::Node { .. } => None,
+            })
+            .collect()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn res(name: &str) -> ResourceName {
+        ResourceName(name.to_owned())
+    }
+
+    /// A graph + a resource table with `build-slot` set to `slots`.
+    fn scheduler_with_slots(slots: u32) -> Scheduler<&'static str> {
+        let mut table = ResourceTable::new();
+        table.set_capacity(res("build-slot"), slots);
+        Scheduler::new(Graph::new(), table)
+    }
+
+    fn slot_dep() -> Vec<Dep> {
+        vec![Dep::Resource {
+            name: res("build-slot"),
+            count: 1,
+        }]
+    }
+
+    #[test]
+    fn resource_node_starts_then_releases_on_complete() {
+        let mut s = scheduler_with_slots(1);
+        let n = s.append("build", slot_dep(), None).expect("insert");
+        // settle claims it (a slot is free) and marks it Running.
+        assert_eq!(s.settle(), vec![n]);
+        assert_eq!(s.graph().node(n).unwrap().state, State::Running);
+        // Slot is held.
+        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 0));
+        // Completing it releases the slot (subtree is just this node).
+        s.complete(n, Outcome::Done);
+        assert_eq!(s.graph().node(n).unwrap().state, State::Done);
+        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
+    }
+
+    #[test]
+    fn build_slot_cap_limits_concurrency_and_release_unblocks() {
+        let mut s = scheduler_with_slots(2);
+        let a = s.append("a", slot_dep(), None).expect("a");
+        let b = s.append("b", slot_dep(), None).expect("b");
+        let c = s.append("c", slot_dep(), None).expect("c");
+        // cap 2 → a + b start, c blocks on the exhausted slot.
+        let started = s.settle();
+        assert_eq!(started, vec![a, b]);
+        assert_eq!(s.graph().node(c).unwrap().state, State::Pending);
+        // a finishes → its slot frees → c can now start.
+        s.complete(a, Outcome::Done);
+        assert_eq!(s.settle(), vec![c]);
+        assert_eq!(s.graph().node(c).unwrap().state, State::Running);
+    }
+
+    #[test]
+    fn parent_holds_resource_until_child_subtree_done() {
+        let mut s = scheduler_with_slots(1);
+        // Parent grabs the single build-slot and runs.
+        let parent = s.append("parent", slot_dep(), None).expect("parent");
+        assert_eq!(s.settle(), vec![parent]);
+        // Parent grows a child (no resource dep of its own) and finishes its
+        // OWN work — but its subtree is not terminal, so it keeps the slot.
+        let child = s.append("child", vec![], Some(parent)).expect("child");
+        s.complete(parent, Outcome::Done);
+        assert!(
+            s.resources.with(|t| t.available(&res("build-slot")) == 0),
+            "parent must keep its lock while a child is still pending/running"
+        );
+        // The child starts and completes → now the whole subtree is terminal →
+        // the parent's slot is released exactly once.
+        assert_eq!(s.settle(), vec![child]);
+        s.complete(child, Outcome::Done);
+        assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
+    }
+}