feat(#2500): add hive-jobq ResourceTable with atomic all-or-nothing acquire

Named counting-semaphore resources — the Dep::Resource side of the v2
model. A ResourceTable tracks per-name capacity + held counts; unconfigured
names default to capacity 1 (created lazily). try_acquire_all grants every
requested unit or none, leaving the table untouched on failure — so a node
never holds one resource while waiting for another, which is what makes the
scheduler deadlock-free without cycle detection. Duplicate names in a
request are summed; over-capacity requests can never acquire. release_all
saturates rather than underflowing. Runtime scheduler state, not persisted:
held counts are rederived from running nodes on restart. Guard objects
(RAII release, recursive re-entrancy) wrap this in a follow-up.

hive-jobq/src/lib.rs

@@ -26,8 +26,10 @@
 //! node kind. Resources are held by the acquiring node and released on
 //! completion via guard objects, recursive within a group.
 //!
-//! The resource-acquisition machinery, the guards, and the scheduler loop are
-//! follow-ups; this is the data model they build on.
+//! The guards and the scheduler loop are follow-ups; the named-counter resource
+//! machinery lives in [`resources`], the base this data model builds on.
+
+pub mod resources;
 
 /// Opaque, stable, monotonic node identifier.
 ///

hive-jobq/src/resources.rs

@@ -0,0 +1,212 @@
+//! Named counting-semaphore resources — the [`Dep::Resource`] side of the model.
+//!
+//! A [`ResourceTable`] tracks a set of named counters. Each name has a
+//! *capacity* (how many units exist) and a *held* count (how many are currently
+//! in use); a name that was never configured is assumed to have capacity 1 and
+//! is created lazily on first use — mara's "unknown resources assumed
+//! available-once". Typical names: `build-slot` (capacity = number of build
+//! slots), `agent/<name>` (capacity 1 — the per-agent lifecycle lock).
+//!
+//! The one operation that matters is [`ResourceTable::try_acquire_all`]: it
+//! takes *all* of a node's resource requests and either grants every one or
+//! grants none, touching nothing on failure. Because a node acquires all its
+//! resources atomically at start (never holds one while waiting for another),
+//! there is no hold-and-wait and therefore no deadlock and no cycle detection.
+//!
+//! This table is runtime scheduler state, not persisted: on restart the held
+//! counts are rederived from the graph's currently-running nodes, which are the
+//! source of truth. Guard objects (RAII release, recursive re-entrancy within a
+//! group) wrap this table in a follow-up; here it is the raw counter machinery.
+//!
+//! [`Dep::Resource`]: crate::Dep::Resource
+
+use crate::ResourceName;
+use std::collections::HashMap;
+
+/// A set of named counting semaphores.
+///
+/// Configure known capacities with [`ResourceTable::set_capacity`]; any name
+/// left unconfigured has the default capacity (1). Acquire and release move
+/// units atomically via [`ResourceTable::try_acquire_all`] /
+/// [`ResourceTable::release_all`].
+#[derive(Debug, Clone)]
+pub struct ResourceTable {
+    /// Configured capacities, keyed by name. Missing ⇒ `default_capacity`.
+    capacities: HashMap<ResourceName, u32>,
+    /// Units currently held, keyed by name. Missing ⇒ 0.
+    held: HashMap<ResourceName, u32>,
+    /// Capacity assumed for a name with no configured entry.
+    default_capacity: u32,
+}
+
+impl Default for ResourceTable {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ResourceTable {
+    /// An empty table whose unconfigured names default to capacity 1.
+    #[must_use]
+    pub fn new() -> Self {
+        Self {
+            capacities: HashMap::new(),
+            held: HashMap::new(),
+            default_capacity: 1,
+        }
+    }
+
+    /// Configure the capacity of a named resource (e.g. `build-slot` = N).
+    ///
+    /// Overwrites any previous capacity for that name. Lowering capacity below
+    /// the currently-held count is allowed — the table simply reports zero
+    /// available until enough is released; it never rejects a config change.
+    pub fn set_capacity(&mut self, name: ResourceName, capacity: u32) {
+        self.capacities.insert(name, capacity);
+    }
+
+    /// The capacity of a name — its configured value, or the default (1) if it
+    /// was never configured.
+    #[must_use]
+    pub fn capacity(&self, name: &ResourceName) -> u32 {
+        self.capacities
+            .get(name)
+            .copied()
+            .unwrap_or(self.default_capacity)
+    }
+
+    /// Units of a name currently held (0 if none).
+    #[must_use]
+    pub fn held(&self, name: &ResourceName) -> u32 {
+        self.held.get(name).copied().unwrap_or(0)
+    }
+
+    /// Units of a name available to acquire right now (`capacity - held`).
+    #[must_use]
+    pub fn available(&self, name: &ResourceName) -> u32 {
+        self.capacity(name).saturating_sub(self.held(name))
+    }
+
+    /// Atomically acquire every requested `(name, count)` or none of them.
+    ///
+    /// Duplicate names in `reqs` are summed, so `[(a, 1), (a, 1)]` asks for two
+    /// units of `a`. Returns `true` and records the units as held only if every
+    /// aggregated request fits in what is available *now*; otherwise returns
+    /// `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: &[(ResourceName, u32)]) -> bool {
+        let wanted = aggregate(reqs);
+        // All-or-nothing: bail before mutating if any request cannot be met.
+        for (name, &count) in &wanted {
+            if count > self.available(name) {
+                return false;
+            }
+        }
+        for (name, count) in wanted {
+            *self.held.entry(name.clone()).or_insert(0) += count;
+        }
+        true
+    }
+
+    /// Release every requested `(name, count)`, returning the units to the pool.
+    ///
+    /// 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: &[(ResourceName, u32)]) {
+        for (name, count) in aggregate(reqs) {
+            if let Some(h) = self.held.get_mut(name) {
+                *h = h.saturating_sub(count);
+            }
+        }
+    }
+}
+
+/// Sum a request list into per-name totals so duplicate names are one entry.
+fn aggregate(reqs: &[(ResourceName, u32)]) -> HashMap<&ResourceName, u32> {
+    let mut wanted: HashMap<&ResourceName, u32> = HashMap::new();
+    for (name, count) in reqs {
+        *wanted.entry(name).or_insert(0) += *count;
+    }
+    wanted
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn res(name: &str) -> ResourceName {
+        ResourceName(name.to_owned())
+    }
+
+    #[test]
+    fn unconfigured_name_defaults_to_capacity_one() {
+        let t = ResourceTable::new();
+        let a = res("agent/foo");
+        assert_eq!(t.capacity(&a), 1);
+        assert_eq!(t.available(&a), 1);
+        assert_eq!(t.held(&a), 0);
+    }
+
+    #[test]
+    fn set_capacity_overrides_default() {
+        let mut t = ResourceTable::new();
+        let slot = res("build-slot");
+        t.set_capacity(slot.clone(), 3);
+        assert_eq!(t.capacity(&slot), 3);
+        assert_eq!(t.available(&slot), 3);
+    }
+
+    #[test]
+    fn acquire_all_or_nothing_leaves_table_untouched_on_failure() {
+        let mut t = ResourceTable::new();
+        let slot = res("build-slot");
+        let agent = res("agent/foo");
+        t.set_capacity(slot.clone(), 2);
+        // First node grabs both slots and the agent lock — fits exactly.
+        assert!(t.try_acquire_all(&[(slot.clone(), 2), (agent.clone(), 1)]));
+        assert_eq!(t.held(&slot), 2);
+        assert_eq!(t.held(&agent), 1);
+        // Second node wants a slot (none free) + the agent lock (held): must
+        // fail AND must not have grabbed the agent lock on the way.
+        assert!(!t.try_acquire_all(&[(slot.clone(), 1), (agent.clone(), 1)]));
+        assert_eq!(t.held(&slot), 2);
+        assert_eq!(t.held(&agent), 1);
+    }
+
+    #[test]
+    fn release_returns_units_to_the_pool() {
+        let mut t = ResourceTable::new();
+        let slot = res("build-slot");
+        t.set_capacity(slot.clone(), 2);
+        assert!(t.try_acquire_all(&[(slot.clone(), 2)]));
+        assert_eq!(t.available(&slot), 0);
+        t.release_all(&[(slot.clone(), 2)]);
+        assert_eq!(t.available(&slot), 2);
+        // Over-release saturates, never underflows.
+        t.release_all(&[(slot.clone(), 5)]);
+        assert_eq!(t.held(&slot), 0);
+    }
+
+    #[test]
+    fn request_over_capacity_never_acquires() {
+        let mut t = ResourceTable::new();
+        let slot = res("build-slot");
+        t.set_capacity(slot.clone(), 2);
+        assert!(!t.try_acquire_all(&[(slot.clone(), 3)]));
+        assert_eq!(t.held(&slot), 0);
+    }
+
+    #[test]
+    fn duplicate_names_in_one_request_are_summed() {
+        let mut t = ResourceTable::new();
+        let a = res("thing");
+        // Default capacity 1: two units in one request cannot both fit.
+        assert!(!t.try_acquire_all(&[(a.clone(), 1), (a.clone(), 1)]));
+        assert_eq!(t.held(&a), 0);
+        // Raise capacity to 2 and the same aggregated request fits.
+        t.set_capacity(a.clone(), 2);
+        assert!(t.try_acquire_all(&[(a.clone(), 1), (a.clone(), 1)]));
+        assert_eq!(t.held(&a), 2);
+    }
+}