From 339e2269e330d1acebb1dcea00e37184dec570fb Mon Sep 17 00:00:00 2001 From: Sree Narayanan Date: Wed, 28 Oct 2026 18:40:16 +0400 Subject: [PATCH] fix(runtime): admit resource reservations all-or-nothing A rejected reservation charged the process ledger, then undid that charge when the VM ledger was full. The undo stored a notify_one permit, so reserve_when_available woke itself and retried without ever returning Pending. With one runtime worker the task that frees VM capacity never ran, and a large guest HTTPS download made the sidecar unresponsive. ResourceLedger::reserve_all checks every request in every scope under ordered locks and charges only when all fit, so a rejected admission has no side effects. reserve, CapabilityRegistry::reserve, and HTTP/2 stream admission use it. Fixes #2083. --- .../src/execution/network/http2.rs | 19 +- crates/runtime/src/accounting.rs | 231 ++++++++++++------ crates/runtime/src/capability.rs | 67 +++-- crates/runtime/src/lib.rs | 2 + crates/runtime/src/test_support.rs | 36 +++ 5 files changed, 261 insertions(+), 94 deletions(-) create mode 100644 crates/runtime/src/test_support.rs diff --git a/crates/native-sidecar/src/execution/network/http2.rs b/crates/native-sidecar/src/execution/network/http2.rs index 3089cdda58..89dff3884e 100644 --- a/crates/native-sidecar/src/execution/network/http2.rs +++ b/crates/native-sidecar/src/execution/network/http2.rs @@ -197,17 +197,14 @@ fn reserve_http2_resources( resources: &ResourceLedger, reservations: &[(ResourceClass, usize)], ) -> Result, 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 { diff --git a/crates/runtime/src/accounting.rs b/crates/runtime/src/accounting.rs index 5e4fa7f62f..c3a9dadbbc 100644 --- a/crates/runtime/src/accounting.rs +++ b/crates/runtime/src/accounting.rs @@ -242,81 +242,96 @@ impl ResourceLedger { resource: ResourceClass, amount: usize, ) -> Result { - 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, - ) -> 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, 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(¤t.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::::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))); diff --git a/crates/runtime/src/capability.rs b/crates/runtime/src/capability.rs index 52ef50c58b..fbacf5d10a 100644 --- a/crates/runtime/src/capability.rs +++ b/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); diff --git a/crates/runtime/src/lib.rs b/crates/runtime/src/lib.rs index 2489253797..bb2576bd87 100644 --- a/crates/runtime/src/lib.rs +++ b/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, diff --git a/crates/runtime/src/test_support.rs b/crates/runtime/src/test_support.rs new file mode 100644 index 0000000000..e786451258 --- /dev/null +++ b/crates/runtime/src/test_support.rs @@ -0,0 +1,36 @@ +use std::future::Future; +use std::sync::mpsc; +use std::task::{Context, Poll, Waker}; +use std::thread; +use std::time::Duration; + +/// Poll `waiter` on its own thread with no runtime. A waiter that never returns +/// `Pending` while capacity is full would hold a single runtime worker forever; +/// here it fails the test instead of hanging it. After the first poll parks, +/// `release` frees capacity and the next poll must admit. +pub(crate) fn assert_parks_until_released(waiter: F, release: impl FnOnce()) -> T +where + F: Future + Send + 'static, + T: Send + 'static, +{ + let (polled_tx, polled_rx) = mpsc::channel(); + let (released_tx, released_rx) = mpsc::channel::<()>(); + let poller = thread::spawn(move || { + let mut context = Context::from_waker(Waker::noop()); + let mut waiter = Box::pin(waiter); + let parked = waiter.as_mut().poll(&mut context).is_pending(); + polled_tx.send(parked).expect("test thread is waiting"); + released_rx.recv().expect("test thread releases capacity"); + match waiter.as_mut().poll(&mut context) { + Poll::Ready(value) => value, + Poll::Pending => panic!("a released capacity must admit the waiter"), + } + }); + let parked = polled_rx + .recv_timeout(Duration::from_secs(2)) + .expect("the first poll must return instead of retrying its own failed admission"); + assert!(parked, "the waiter must park while capacity is full"); + release(); + released_tx.send(()).expect("poller thread is waiting"); + poller.join().expect("poller thread") +}