fix(runtime): admit resource reservations all-or-nothing

crates/native-sidecar/src/execution/network/http2.rs

@@ -197,17 +197,14 @@ fn reserve_http2_resources(
     resources: &ResourceLedger,
     reservations: &[(ResourceClass, usize)],
 ) -> Result<Vec<Reservation>, SidecarError> {
-    let mut admitted = Vec::with_capacity(reservations.len());
-    for (resource, amount) in reservations {
-        if *amount != 0 {
-            admitted.push(
-                resources
-                    .reserve(*resource, *amount)
-                    .map_err(|error| SidecarError::Execution(error.to_string()))?,
-            );
-        }
-    }
-    Ok(admitted)
+    let requests: Vec<_> = reservations
+        .iter()
+        .copied()
+        .filter(|&(_, amount)| amount != 0)
+        .collect();
+    resources
+        .reserve_all(&requests)
+        .map_err(|error| SidecarError::Execution(error.to_string()))
 }
 
 fn http2_event_bytes(event: &Http2BridgeEvent) -> usize {

crates/runtime/src/accounting.rs

@@ -242,81 +242,96 @@ impl ResourceLedger {
         resource: ResourceClass,
         amount: usize,
     ) -> Result<Reservation, LimitError> {
-        let mut allocations = Vec::with_capacity(if self.parent.is_some() { 2 } else { 1 });
-        if let Some(parent) = &self.parent {
-            parent.reserve_into(resource, amount, &mut allocations)?;
-        }
-        if let Err(error) = self.reserve_local(resource, amount) {
-            release_allocations(&mut allocations);
-            return Err(error);
-        }
-        if amount != 0 {
-            allocations.push(Allocation {
-                ledger: Arc::clone(&self.inner),
-                resource,
-                amount,
-            });
-        }
-        Ok(Reservation {
-            resource,
-            amount,
-            allocations,
-        })
+        let mut reservations = self.reserve_all(&[(resource, amount)])?;
+        Ok(reservations
+            .pop()
+            .expect("one admission request yields one reservation"))
     }
 
-    fn reserve_into(
+    /// Reserve several resources in every scope as one admission. Either every
+    /// request fits in every scope and all are charged together, or nothing
+    /// changes. A rejected admission never charges, rolls back, or notifies, so
+    /// a waiter cannot wake itself through its own failed attempt.
+    pub fn reserve_all(
         &self,
-        resource: ResourceClass,
-        amount: usize,
-        allocations: &mut Vec<Allocation>,
-    ) -> Result<(), LimitError> {
-        if let Some(parent) = &self.parent {
-            parent.reserve_into(resource, amount, allocations)?;
-        }
-        if let Err(error) = self.reserve_local(resource, amount) {
-            release_allocations(allocations);
-            return Err(error);
-        }
-        if amount != 0 {
-            allocations.push(Allocation {
-                ledger: Arc::clone(&self.inner),
-                resource,
-                amount,
-            });
+        requests: &[(ResourceClass, usize)],
+    ) -> Result<Vec<Reservation>, LimitError> {
+        // Lock root first, so every admission takes ledger locks in one order.
+        let mut scopes = Vec::with_capacity(2);
+        let mut ledger = Some(self);
+        while let Some(current) = ledger {
+            scopes.push(&current.inner);
+            ledger = current.parent.as_deref();
+        }
+        scopes.reverse();
+        let mut states: Vec<_> = scopes
+            .iter()
+            .map(|inner| lock_state(inner, "admission"))
+            .collect();
+
+        // A resource can appear more than once in one admission; check its sum.
+        let mut requested = BTreeMap::<ResourceClass, usize>::new();
+        for &(resource, amount) in requests {
+            let total = requested.entry(resource).or_default();
+            *total = total.saturating_add(amount);
+        }
+        for (inner, state) in scopes.iter().zip(&states) {
+            for (&resource, &amount) in &requested {
+                let Some(limit) = inner.limits.get(&resource) else {
+                    continue;
+                };
+                let used = state
+                    .counters
+                    .get(&resource)
+                    .map_or(0, |counter| counter.used);
+                if used
+                    .checked_add(amount)
+                    .is_none_or(|total| total > limit.maximum)
+                {
+                    return Err(LimitError {
+                        scope: inner.scope.clone(),
+                        resource,
+                        used,
+                        requested: amount,
+                        limit: limit.maximum,
+                        config_path: limit.config_path.clone(),
+                    });
+                }
+            }
         }
-        Ok(())
-    }
 
-    fn reserve_local(&self, resource: ResourceClass, amount: usize) -> Result<(), LimitError> {
-        if amount == 0 {
-            return Ok(());
-        }
-        let mut state = self.inner.state.lock().unwrap_or_else(|poisoned| {
-            eprintln!(
-                "ERR_AGENTOS_RESOURCE_LEDGER_POISONED: recovering scope={} resource={}",
-                self.inner.scope,
-                resource.name()
-            );
-            poisoned.into_inner()
-        });
-        let counter = state.counters.entry(resource).or_default();
-        let requested_total = counter.used.checked_add(amount);
-        if let Some(limit) = self.inner.limits.get(&resource) {
-            if requested_total.is_none_or(|total| total > limit.maximum) {
-                return Err(LimitError {
-                    scope: self.inner.scope.clone(),
-                    resource,
-                    used: counter.used,
-                    requested: amount,
-                    limit: limit.maximum,
-                    config_path: limit.config_path.clone(),
-                });
+        for (inner, state) in scopes.iter().zip(&mut states) {
+            for &(resource, amount) in requests {
+                if amount == 0 {
+                    continue;
+                }
+                let counter = state.counters.entry(resource).or_default();
+                counter.used = counter.used.saturating_add(amount);
+                maybe_warn(inner, resource, counter);
+                observe_usage(inner, resource, counter.used);
             }
         }
-        counter.used = requested_total.unwrap_or(usize::MAX);
-        maybe_warn(&self.inner, resource, counter);
-        observe_usage(&self.inner, resource, counter.used);
-        Ok(())
+        drop(states);
+
+        Ok(requests
+            .iter()
+            .map(|&(resource, amount)| Reservation {
+                resource,
+                amount,
+                allocations: if amount == 0 {
+                    Vec::new()
+                } else {
+                    scopes
+                        .iter()
+                        .map(|inner| Allocation {
+                            ledger: Arc::clone(inner),
+                            resource,
+                            amount,
+                        })
+                        .collect()
+                },
+            })
+            .collect())
     }
 
     pub fn usage(&self, resource: ResourceClass) -> ResourceUsage {
@@ -465,6 +480,16 @@ struct Allocation {
     amount: usize,
 }
 
+fn lock_state<'a>(inner: &'a LedgerInner, context: &str) -> std::sync::MutexGuard<'a, LedgerState> {
+    inner.state.lock().unwrap_or_else(|poisoned| {
+        eprintln!(
+            "ERR_AGENTOS_RESOURCE_LEDGER_POISONED: recovering {context} scope={}",
+            inner.scope
+        );
+        poisoned.into_inner()
+    })
+}
+
 fn release_allocation(allocation: &Allocation) {
     let mut state = allocation.ledger.state.lock().unwrap_or_else(|poisoned| {
         eprintln!(
@@ -690,6 +715,76 @@ mod tests {
         )]
     }
 
+    #[test]
+    fn waiter_on_full_child_scope_parks_until_capacity_is_released() {
+        let process = Arc::new(ResourceLedger::root("process", limit(10)));
+        let vm = Arc::new(ResourceLedger::child(
+            "vm-1",
+            limit(2),
+            Arc::clone(&process),
+        ));
+        let held = vm
+            .reserve(ResourceClass::BufferedBytes, 2)
+            .expect("fill child");
+
+        let waiting_vm = Arc::clone(&vm);
+        let admitted = crate::test_support::assert_parks_until_released(
+            async move {
+                waiting_vm
+                    .reserve_when_available(ResourceClass::BufferedBytes, 1)
+                    .await
+            },
+            move || drop(held),
+        )
+        .expect("released capacity admits the waiter");
+
+        assert_eq!(process.usage(ResourceClass::BufferedBytes).used, 1);
+        assert_eq!(vm.usage(ResourceClass::BufferedBytes).used, 1);
+        drop(admitted);
+        assert!(process.is_zero() && vm.is_zero());
+    }
+
+    #[test]
+    fn rejected_admission_charges_no_scope() {
+        let limits = [
+            (
+                ResourceClass::BufferedBytes,
+                ResourceLimit::new(10, "runtime.resources.maxSocketBufferedBytes"),
+            ),
+            (
+                ResourceClass::TlsBytes,
+                ResourceLimit::new(10, "runtime.resources.maxTlsBytes"),
+            ),
+        ];
+        let process = Arc::new(ResourceLedger::root("process", limits.clone()));
+        let vm = ResourceLedger::child(
+            "vm-1",
+            [
+                limits[0].clone(),
+                (
+                    ResourceClass::TlsBytes,
+                    ResourceLimit::new(1, "limits.tls.maxBufferedBytes"),
+                ),
+            ],
+            Arc::clone(&process),
+        );
+
+        let error = vm
+            .reserve_all(&[
+                (ResourceClass::BufferedBytes, 4),
+                (ResourceClass::TlsBytes, 4),
+            ])
+            .expect_err("the VM TLS limit rejects the admission");
+
+        assert_eq!(error.scope, "vm-1");
+        assert_eq!(error.resource, ResourceClass::TlsBytes);
+        assert!(
+            process.is_zero(),
+            "the process scope keeps no partial charge"
+        );
+        assert!(vm.is_zero(), "the VM scope keeps no partial charge");
+    }
+
     #[test]
     fn child_reservation_charges_and_releases_both_scopes() {
         let process = Arc::new(ResourceLedger::root("process", limit(10)));

crates/runtime/src/capability.rs

@@ -218,19 +218,21 @@ impl CapabilityRegistry {
                 return Err(CapabilityError::RegistryClosed);
             }
         }
-        let capability = self.inner.ledger.reserve(ResourceClass::Capabilities, 1)?;
         // One possible ready-set entry is reserved with every admitted
         // capability. Readiness for a live handle therefore cannot fail after
-        // the backend has already become observable.
-        let ready_handle = self.inner.ledger.reserve(ResourceClass::ReadyHandles, 1)?;
-        let socket = kind
-            .is_socket()
-            .then(|| self.inner.ledger.reserve(ResourceClass::Sockets, 1))
-            .transpose()?;
-        let connection = kind
-            .is_connection()
-            .then(|| self.inner.ledger.reserve(ResourceClass::Connections, 1))
-            .transpose()?;
+        // the backend has already become observable. All counts are admitted
+        // together, so a full scope leaves no partial charge to roll back.
+        let mut requests = vec![
+            (ResourceClass::Capabilities, 1),
+            (ResourceClass::ReadyHandles, 1),
+        ];
+        if kind.is_socket() {
+            requests.push((ResourceClass::Sockets, 1));
+        }
+        if kind.is_connection() {
+            requests.push((ResourceClass::Connections, 1));
+        }
+        let reservations = self.inner.ledger.reserve_all(&requests)?;
         {
             let mut state = self
                 .inner
@@ -248,10 +250,7 @@ impl CapabilityRegistry {
         Ok(PendingCapability {
             registry: Arc::clone(&self.inner),
             kind,
-            reservations: vec![Some(capability), Some(ready_handle), socket, connection]
-                .into_iter()
-                .flatten()
-                .collect(),
+            reservations,
             counted: true,
         })
     }
@@ -631,6 +630,44 @@ mod tests {
         (ledger, registry)
     }
 
+    #[test]
+    fn socket_waiter_parks_until_a_socket_is_released() {
+        // Capabilities still fit, but the only socket is taken. A partial
+        // admission would release the capability count it took and wake its
+        // own waiter, so the waiter must not charge anything until all fit.
+        let ledger = Arc::new(ResourceLedger::root(
+            "vm-1",
+            [
+                (
+                    ResourceClass::Capabilities,
+                    ResourceLimit::new(2, "runtime.capabilities.maxPerVm"),
+                ),
+                (
+                    ResourceClass::Sockets,
+                    ResourceLimit::new(1, "runtime.resources.maxSockets"),
+                ),
+            ],
+        ));
+        let registry = CapabilityRegistry::new(7, Arc::clone(&ledger));
+        let held = registry
+            .reserve(CapabilityKind::TcpListener)
+            .expect("take the only socket");
+
+        let waiting_registry = registry.clone();
+        let admitted = crate::test_support::assert_parks_until_released(
+            async move {
+                waiting_registry
+                    .reserve_when_available(CapabilityKind::TcpListener)
+                    .await
+            },
+            move || drop(held),
+        )
+        .expect("a released socket admits the waiter");
+
+        drop(admitted);
+        assert!(ledger.is_zero());
+    }
+
     #[test]
     fn admission_precedes_commit_and_drop_reconciles() {
         let (ledger, registry) = registry(1);

crates/runtime/src/lib.rs

@@ -29,6 +29,8 @@ pub mod fairness;
 pub mod metrics;
 pub mod readiness;
 pub mod supervision;
+#[cfg(test)]
+mod test_support;
 
 pub use supervision::{
     TaskClass, TaskClassSnapshot, TaskOwner, TaskSpawnError, TaskSupervisor, TaskTerminalReason,