refactor(#2500): encapsulate the jobq lock, drop the dead borrowed-guard layer

Make the resource lock unmisusable from outside the crate: the public
surface is now purely declarative (build a Graph with Dep::Resource edges,
configure capacities, run the Scheduler), and the scheduler owns every
acquire/release — a consumer never holds a guard, so it cannot hold the
lock wrong.

- `guard` module + `ResourceTable::try_acquire_all`/`release_all` +
  `Graph::set_state` are now `pub(crate)`.
- Remove the dead borrowed-guard layer (`ResourceGuard::borrowed`,
  `Acq::Borrowed`, `is_owning`): the scheduler tracks re-entrancy via its
  own single borrow slot per (holder, resource) and never constructs a
  borrowed guard, so re-entrancy lives in exactly one place. `Acq`
  collapses into the owning `ResourceGuard` struct.
- `#[must_use]` on `Scheduler::settle` — ignoring its ids silently drops
  runnable work.
- `SharedResources::with` (test-only table observability) is `#[cfg(test)]`.
- Drop the moot borrowed-guard tests; retained owning tests are black-box,
  and the redundant `set_state` test helper is gone.

hive-jobq/src/guard.rs

@@ -1,15 +1,13 @@
-//! RAII guard objects over [`ResourceTable`] — the recursive-lock layer.
+//! RAII guard objects over [`ResourceTable`] — owning resource grants.
 //!
 //! A running node acquires its resources through [`SharedResources::acquire`],
-//! which hands back a [`ResourceGuard`]. Dropping the guard releases exactly
+//! which hands back a [`ResourceGuard`] owning those units. Dropping the guard
-//! what it acquired, so a node's resources are freed when its grant goes out of
+//! releases exactly what it acquired, so a node's resources are freed when its
-//! scope — there is no explicit release call to forget.
+//! grant goes out of scope — there is no explicit release call to forget.
 //!
-//! Re-entrancy within a node group is expressed with
+//! Re-entrancy (a sub-node reusing a resource its ancestor group already holds)
-//! [`ResourceGuard::borrowed`]: a sub-node that reuses a resource its ancestor
+//! is not expressed here: the scheduler tracks it with a single borrow slot per
-//! group already holds gets a guard that owns no units and releases nothing on
+//! `(holder, resource)` and never re-acquires, so these guards are always owning.
-//! drop, so the shared unit is released exactly once — when the owning group's
-//! guard drops — never double-counted or freed early.
 //!
 //! Single-owner by design: the scheduler drives one settle loop, so the shared
 //! table is `Rc<RefCell<…>>` (single-threaded interior mutability), not
@@ -50,76 +48,43 @@ impl<R: Clone + Eq + Hash> SharedResources<R> {
     #[must_use]
     pub fn acquire(&self, reqs: Vec<(R, u32)>) -> Option<ResourceGuard<R>> {
         if self.0.borrow_mut().try_acquire_all(&reqs) {
-            Some(ResourceGuard(Acq::Owned {
+            Some(ResourceGuard {
                 table: self.clone(),
                 reqs,
-            }))
+            })
         } else {
             None
         }
     }
 
-    /// Compute something from the underlying table (e.g. query `available`).
+    /// Observe the underlying table — test-only (the scheduler's tests assert
-    ///
+    /// on resource availability). Gated `#[cfg(test)]` so it is compiled out of
-    /// Keep the closure short: it holds a shared borrow, so calling
+    /// the shipped crate: no consumer can reach the raw table through it.
-    /// [`SharedResources::acquire`] (a mutable borrow) from inside it would
+    #[cfg(test)]
-    /// panic on the overlapping `RefCell` borrow.
     pub fn with<T>(&self, f: impl FnOnce(&ResourceTable<R>) -> T) -> T {
         f(&self.0.borrow())
     }
 }
 
-/// How a [`ResourceGuard`] relates to the units it represents.
+/// An RAII grant of resources: dropping it releases exactly the units it
+/// acquired back into the shared table.
 #[derive(Debug)]
-enum Acq<R> {
+pub struct ResourceGuard<R: Clone + Eq + Hash> {
-    /// Owns real units; drop releases them back into the shared table.
+    table: SharedResources<R>,
-    Owned {
+    reqs: Vec<(R, u32)>,
-        table: SharedResources<R>,
-        reqs: Vec<(R, u32)>,
-    },
-    /// Re-entrant reuse of a resource an ancestor group already holds; drop
-    /// releases nothing.
-    Borrowed,
 }
 
-/// An RAII grant of resources. Dropping it releases exactly what was acquired
-/// (nothing, for a borrowed re-entrant guard).
-#[derive(Debug)]
-pub struct ResourceGuard<R: Clone + Eq + Hash>(Acq<R>);
-
 impl<R: Clone + Eq + Hash> ResourceGuard<R> {
-    /// A borrowed (re-entrant) guard that owns no units and releases nothing on
+    /// The `(name, count)` units this guard releases on drop.
-    /// drop. The scheduler hands one to a sub-node that depends on a resource
-    /// its ancestor group already holds, so the shared unit is released once —
-    /// when the owning group's guard drops — never twice or early.
-    #[must_use]
-    pub fn borrowed() -> Self {
-        Self(Acq::Borrowed)
-    }
-
-    /// The `(name, count)` units this guard releases on drop — empty when it is
-    /// a borrowed re-entrant guard.
     #[must_use]
     pub fn held(&self) -> &[(R, u32)] {
-        match &self.0 {
+        &self.reqs
-            Acq::Owned { reqs, .. } => reqs,
-            Acq::Borrowed => &[],
-        }
-    }
-
-    /// Whether this guard owns real units (`true`) or is a borrowed re-entrant
-    /// guard (`false`).
-    #[must_use]
-    pub fn is_owning(&self) -> bool {
-        matches!(self.0, Acq::Owned { .. })
     }
 }
 
 impl<R: Clone + Eq + Hash> Drop for ResourceGuard<R> {
     fn drop(&mut self) {
-        if let Acq::Owned { table, reqs } = &self.0 {
+        self.table.0.borrow_mut().release_all(&self.reqs);
-            table.0.borrow_mut().release_all(reqs);
-        }
     }
 }
 
@@ -143,54 +108,23 @@ mod tests {
         let slot = res("build-slot");
         {
             let g = sr.acquire(vec![(slot.clone(), 2)]).expect("fits");
-            assert!(g.is_owning());
             assert_eq!(g.held(), &[(slot.clone(), 2)]);
-            sr.with(|t| assert_eq!(t.available(&slot), 0));
+            // Both units held → any further acquire fails.
-        } // guard dropped here
+            assert!(sr.acquire(vec![(slot.clone(), 1)]).is_none());
-        sr.with(|t| assert_eq!(t.available(&slot), 2));
+        } // guard dropped here → its units are released
+        // Full capacity is available again.
+        assert!(sr.acquire(vec![(slot.clone(), 2)]).is_some());
     }
 
     #[test]
     fn acquire_returns_none_and_leaves_table_untouched_when_it_does_not_fit() {
         let sr = shared_with(1);
         let slot = res("build-slot");
-        let _held = sr.acquire(vec![(slot.clone(), 1)]).expect("first fits");
+        let held = sr.acquire(vec![(slot.clone(), 1)]).expect("first fits");
         assert!(sr.acquire(vec![(slot.clone(), 1)]).is_none());
-        // The failed acquire took nothing extra.
+        // The failed acquire took nothing extra: dropping the one real grant
-        sr.with(|t| assert_eq!(t.held(&slot), 1));
+        // frees exactly one unit, so a single-unit acquire then fits.
-    }
+        drop(held);
-
+        assert!(sr.acquire(vec![(slot.clone(), 1)]).is_some());
-    #[test]
-    fn borrowed_guard_releases_nothing_on_drop() {
-        let sr = shared_with(1);
-        let agent = res("agent/foo");
-        // An owning guard holds the single agent-lock unit.
-        let _owner = sr.acquire(vec![(agent.clone(), 1)]).expect("fits");
-        sr.with(|t| assert_eq!(t.available(&agent), 0));
-        {
-            let b = ResourceGuard::<String>::borrowed();
-            assert!(!b.is_owning());
-            assert!(b.held().is_empty());
-        } // borrowed drop is a no-op
-        // Still held by the owner — the borrow did not release it.
-        sr.with(|t| assert_eq!(t.available(&agent), 0));
-    }
-
-    #[test]
-    fn nested_group_lock_released_once_when_owner_drops() {
-        let sr = shared_with(1);
-        let agent = res("agent/foo");
-        {
-            let _group = sr
-                .acquire(vec![(agent.clone(), 1)])
-                .expect("group takes the lock");
-            {
-                // A sub-node reuses the group's lock: borrowed, no re-acquire.
-                let _sub = ResourceGuard::<String>::borrowed();
-                sr.with(|t| assert_eq!(t.available(&agent), 0));
-            } // sub-node done — must NOT free the shared lock
-            sr.with(|t| assert_eq!(t.available(&agent), 0));
-        } // group done — frees it exactly once
-        sr.with(|t| assert_eq!(t.available(&agent), 1));
     }
 }

hive-jobq/src/lib.rs

@@ -27,9 +27,9 @@
 //! completion via guard objects, recursive within a group.
 //!
 //! The [`scheduler`] settle loop drives execution; the resource machinery
-//! lives in [`resources`] and the RAII lock guards over it in [`guard`].
+//! lives in [`resources`] and the RAII lock guards over it in `guard`.
 
-pub mod guard;
+pub(crate) mod guard;
 pub mod resources;
 pub mod scheduler;
 
@@ -298,7 +298,7 @@ impl<N, R> Graph<N, R> {
     /// 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 {
+    pub(crate) 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
@@ -381,11 +381,6 @@ mod tests {
         assert_eq!(g.node(a).unwrap().parent, None);
     }
 
-    fn set_state<N, R>(g: &mut Graph<N, R>, id: NodeId, state: State) {
-        let idx = g.nodes.iter().position(|n| n.id == id).unwrap();
-        g.nodes[idx].state = state;
-    }
-
     #[test]
     fn group_terminal_requires_the_group_node_and_all_children_terminal() {
         let mut g: Graph<&str, String> = Graph::new();
@@ -395,10 +390,10 @@ mod tests {
         assert!(!g.group_terminal(group));
         // Child done, but the group node itself is still pending → NOT terminal:
         // the group node's own state is load-bearing, not just its children.
-        set_state(&mut g, child, State::Done);
+        g.set_state(child, State::Done);
         assert!(!g.group_terminal(group));
         // Group node terminal too → the whole group is terminal.
-        set_state(&mut g, group, State::Done);
+        g.set_state(group, State::Done);
         assert!(g.group_terminal(group));
     }
 
@@ -408,10 +403,10 @@ mod tests {
         // not read as terminal just because its child set is currently empty.
         let mut g: Graph<&str, String> = Graph::new();
         let group = g.insert("group", vec![], None).unwrap();
-        set_state(&mut g, group, State::Running);
+        g.set_state(group, State::Running);
         assert!(!g.group_terminal(group));
         // Once it finishes (having grown no children), it is terminal.
-        set_state(&mut g, group, State::Done);
+        g.set_state(group, State::Done);
         assert!(g.group_terminal(group));
     }
 

hive-jobq/src/resources.rs

@@ -95,7 +95,7 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
     /// `false` and leaves the table completely untouched. A request for more
     /// units than a name's capacity can therefore never succeed — the scheduler
     /// should reject such a node at insert time so it does not wait forever.
-    pub fn try_acquire_all(&mut self, reqs: &[(R, u32)]) -> bool {
+    pub(crate) fn try_acquire_all(&mut self, reqs: &[(R, u32)]) -> bool {
         let wanted = aggregate(reqs);
         // All-or-nothing: bail before mutating if any request cannot be met.
         for (name, &count) in &wanted {
@@ -113,7 +113,7 @@ impl<R: Clone + Eq + Hash> ResourceTable<R> {
     ///
     /// Duplicate names are summed. Releasing more than is held saturates at zero
     /// rather than underflowing, so a double release is harmless.
-    pub fn release_all(&mut self, reqs: &[(R, u32)]) {
+    pub(crate) fn release_all(&mut self, reqs: &[(R, u32)]) {
         for (name, count) in aggregate(reqs) {
             if let Some(h) = self.held.get_mut(name) {
                 *h = h.saturating_sub(count);

hive-jobq/src/scheduler.rs

@@ -90,6 +90,7 @@ impl<N, R: Clone + Eq + Hash> Scheduler<N, R> {
     /// 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.
+    #[must_use]
     pub fn settle(&mut self) -> Vec<NodeId> {
         let pending: Vec<NodeId> = self
             .graph