feat(#2500): hive-jobq scheduler re-entrancy + eager AfterOk cascade

hive-jobq/src/scheduler.rs

@@ -43,6 +43,11 @@ pub struct Scheduler<N> {
     /// 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>>,
+    /// The single re-entrancy slot per `(ancestor-holder, resource)`: the id of
+    /// the descendant currently *borrowing* that ancestor's lock. Present ⇒ the
+    /// slot is taken, so no other descendant may re-enter the same lock until
+    /// the borrower's subtree is terminal — "only one node at a time within".
+    borrow_slots: HashMap<(NodeId, ResourceName), NodeId>,
 }
 
 impl<N> Scheduler<N> {
@@ -53,6 +58,7 @@ impl<N> Scheduler<N> {
             graph,
             resources: SharedResources::new(resources),
             owned: HashMap::new(),
+            borrow_slots: HashMap::new(),
         }
     }
 
@@ -92,18 +98,66 @@ impl<N> Scheduler<N> {
             .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);
+            if self.node_deps_satisfied(id) && self.try_start(id) {
                 started.push(id);
             }
         }
         started
     }
 
+    /// Try to start node `id`: classify each resource dep as *owned* (no
+    /// ancestor holds it → acquire real units) or *borrowed* (an ancestor group
+    /// already holds it → re-enter, gated by the one re-entrancy slot), then
+    /// take everything atomically or nothing. Returns whether it started.
+    fn try_start(&mut self, id: NodeId) -> bool {
+        let mut owned_reqs = Vec::new();
+        let mut borrows = Vec::new();
+        for (name, count) in self.resource_reqs(id) {
+            if let Some(ancestor) = self.ancestor_owning(id, &name) {
+                // Re-entrant reuse: allowed only if the slot is free.
+                if self.borrow_slots.contains_key(&(ancestor, name.clone())) {
+                    return false;
+                }
+                borrows.push((ancestor, name));
+            } else {
+                owned_reqs.push((name, count));
+            }
+        }
+        // Owned units are all-or-nothing; borrow slots were all confirmed free
+        // above, so this is the only fallible step. Nothing mutated until here.
+        let Some(guard) = self.resources.acquire(owned_reqs) else {
+            return false;
+        };
+        self.owned.entry(id).or_default().push(guard);
+        for slot in borrows {
+            self.borrow_slots.insert(slot, id);
+        }
+        self.graph.set_state(id, State::Running);
+        true
+    }
+
+    /// The nearest ancestor of `id` that *owns* (holds real units of) `name`,
+    /// or `None` if no ancestor holds it (⇒ `id` must own-acquire it itself).
+    fn ancestor_owning(&self, id: NodeId, name: &ResourceName) -> Option<NodeId> {
+        let mut cursor = self.graph.node(id)?.parent;
+        while let Some(ancestor) = cursor {
+            if self.node_owns(ancestor, name) {
+                return Some(ancestor);
+            }
+            cursor = self.graph.node(ancestor)?.parent;
+        }
+        None
+    }
+
+    /// Whether node `holder` holds an owned guard covering resource `name`.
+    fn node_owns(&self, holder: NodeId, name: &ResourceName) -> bool {
+        self.owned.get(&holder).is_some_and(|guards| {
+            guards
+                .iter()
+                .any(|g| g.held().iter().any(|(n, _)| n == name))
+        })
+    }
+
     /// 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.
@@ -114,10 +168,41 @@ impl<N> Scheduler<N> {
             Outcome::Failed => State::Failed,
         };
         self.graph.set_state(id, state);
+        if outcome == Outcome::Failed {
+            self.cascade_cancel(id);
+        }
         self.release_settled_subtrees();
     }
 
-    /// Drop the owned guards of every holder whose subtree is now terminal.
+    /// Eagerly cancel the transitive `AfterOk` dependents of a just-failed node:
+    /// they can never run (a strong dependency failed), so mark them `Cancelled`
+    /// now — before they could claim resources. A dependent is always still
+    /// `Pending` here (a `Running` node's `AfterOk` deps were `Done` when it
+    /// started, and `Done` is terminal), so no resources need releasing.
+    fn cascade_cancel(&mut self, failed: NodeId) {
+        let mut stack = vec![failed];
+        while let Some(dep) = stack.pop() {
+            let dependents: Vec<NodeId> = self
+                .graph
+                .nodes()
+                .filter(|n| {
+                    n.state == State::Pending
+                        && n.deps.iter().any(
+                            |d| matches!(d, Dep::Node { id, when: DepWhen::AfterOk } if *id == dep),
+                        )
+                })
+                .map(|n| n.id)
+                .collect();
+            for d in dependents {
+                self.graph.set_state(d, State::Cancelled);
+                stack.push(d);
+            }
+        }
+    }
+
+    /// Release everything whose subtree has become terminal: drop the owned
+    /// guards of any holder (→ frees its units) and free any re-entrancy slot
+    /// held by a borrower — both are held for the whole subtree lifetime.
     fn release_settled_subtrees(&mut self) {
         let holders: Vec<NodeId> = self.owned.keys().copied().collect();
         for holder in holders {
@@ -125,6 +210,8 @@ impl<N> Scheduler<N> {
                 self.owned.remove(&holder); // drops guards → releases the units
             }
         }
+        self.borrow_slots
+            .retain(|_, borrower| !self.graph.group_terminal(*borrower));
     }
 
     /// Whether every [`Dep::Node`] edge of `id` is satisfied. `Dep::Resource`
@@ -195,6 +282,13 @@ mod tests {
         }]
     }
 
+    fn resource_dep(name: &str) -> Vec<Dep> {
+        vec![Dep::Resource {
+            name: res(name),
+            count: 1,
+        }]
+    }
+
     #[test]
     fn resource_node_starts_then_releases_on_complete() {
         let mut s = scheduler_with_slots(1);
@@ -246,4 +340,84 @@ mod tests {
         s.complete(child, Outcome::Done);
         assert!(s.resources.with(|t| t.available(&res("build-slot")) == 1));
     }
+
+    #[test]
+    fn recursive_lock_serializes_re_entrant_descendants() {
+        // `agent/foo` is unconfigured → default capacity 1.
+        let mut s = Scheduler::new(Graph::new(), ResourceTable::new());
+        let agent = res("agent/foo");
+        // A group node owns agent/foo and runs.
+        let group = s
+            .append("group", resource_dep("agent/foo"), None)
+            .expect("group");
+        assert_eq!(s.settle(), vec![group]);
+        assert!(s.resources.with(|t| t.available(&agent) == 0));
+        // Two sub-nodes each need agent/foo → they re-enter the group's lock,
+        // but only ONE at a time (the single re-entrancy slot).
+        let c1 = s
+            .append("c1", resource_dep("agent/foo"), Some(group))
+            .expect("c1");
+        let c2 = s
+            .append("c2", resource_dep("agent/foo"), Some(group))
+            .expect("c2");
+        let started = s.settle();
+        assert_eq!(
+            started,
+            vec![c1],
+            "only one descendant may borrow at a time"
+        );
+        assert_eq!(s.graph().node(c2).unwrap().state, State::Pending);
+        // The lock was NOT re-acquired — still just the group's one unit held.
+        assert!(s.resources.with(|t| t.available(&agent) == 0));
+        // c1 finishes → its borrow slot frees → c2 can now re-enter.
+        s.complete(c1, Outcome::Done);
+        assert_eq!(s.settle(), vec![c2]);
+        assert_eq!(s.graph().node(c2).unwrap().state, State::Running);
+        // Still no double-acquire; the group's single unit is the only hold.
+        assert!(s.resources.with(|t| t.available(&agent) == 0));
+    }
+
+    #[test]
+    fn failed_after_ok_dep_cancels_dependents_but_after_any_still_runs() {
+        let mut s = Scheduler::new(Graph::new(), ResourceTable::new());
+        let root = s.append("root", vec![], None).expect("root");
+        let strong1 = s
+            .append(
+                "strong1",
+                vec![Dep::Node {
+                    id: root,
+                    when: DepWhen::AfterOk,
+                }],
+                None,
+            )
+            .expect("strong1");
+        let strong2 = s
+            .append(
+                "strong2",
+                vec![Dep::Node {
+                    id: strong1,
+                    when: DepWhen::AfterOk,
+                }],
+                None,
+            )
+            .expect("strong2");
+        let weak = s
+            .append(
+                "weak",
+                vec![Dep::Node {
+                    id: root,
+                    when: DepWhen::AfterAny,
+                }],
+                None,
+            )
+            .expect("weak");
+        assert_eq!(s.settle(), vec![root]);
+        s.complete(root, Outcome::Failed);
+        // The AfterOk chain strong1→strong2 is eagerly cancelled (a strong dep
+        // failed)…
+        assert_eq!(s.graph().node(strong1).unwrap().state, State::Cancelled);
+        assert_eq!(s.graph().node(strong2).unwrap().state, State::Cancelled);
+        // …but the AfterAny dependent still runs — it converges regardless.
+        assert_eq!(s.settle(), vec![weak]);
+    }
 }