From ee28f6cb581a545a161e4b2ecfb087291f67a5c1 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Mon, 17 Aug 2026 21:21:37 -0300 Subject: [PATCH 1/8] fix(pool): discard warm firecracker entries whose process died while parked Parked warm processes run with oom_score_adj=1000, so they are the first OOM-kill candidates and can exit while idle in the pool. The acquire path handed whatever it popped straight to snapshot resume, which then failed on the dead process. Add FirecrackerInstance::is_process_running() and make try_acquire pop until a live entry is found. Dead entries skip the graceful-stop path (the process is already gone): the instance is dropped and the network slot is released synchronously, keeping the acquire path free of runtime block_on calls so it stays safe in async context. --- src/sandbox/firecracker/instance.rs | 12 ++++++++ src/sandbox/firecracker/pool.rs | 46 ++++++++++++++++++++++++++++- 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/src/sandbox/firecracker/instance.rs b/src/sandbox/firecracker/instance.rs index fd6e4746..c05c13c5 100644 --- a/src/sandbox/firecracker/instance.rs +++ b/src/sandbox/firecracker/instance.rs @@ -61,6 +61,18 @@ impl FirecrackerInstance { Ok(Pid::from_raw(raw_pid)) } + /// Returns true while the firecracker process is still running. + /// + /// Warm pool entries park with `oom_score_adj=1000`, so they are the first + /// OOM-kill candidates and can die while idle. Callers must check this + /// before handing a parked instance to snapshot resume. + pub fn is_process_running(&mut self) -> bool { + match self.process.as_mut() { + Some(child) => matches!(child.try_wait(), Ok(None)), + None => false, + } + } + pub async fn spawn_with_netns( &mut self, firecracker_binary: &Path, diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index eb0c3f1a..6758c0b8 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -136,13 +136,57 @@ impl FirecrackerPool { } pub(crate) fn try_acquire(&self) -> Option { - let warm = self.pool.try_acquire()?; + let warm = self.acquire_live_warm()?; if self.pool.len() < self.pool.config().low_watermark { self.pool.request_maintenance(); } Some(warm) } + /// Pop warm entries until a live one is found. Parked warm processes run + /// with `oom_score_adj=1000`, so they are the first OOM-kill candidates and + /// can die while idle; handing a dead process to snapshot resume would fail + /// the resume, so dead entries are discarded here instead. + fn acquire_live_warm(&self) -> Option { + loop { + let mut warm = self.pool.try_acquire()?; + if warm.fc_instance.is_process_running() { + return Some(warm); + } + self.discard_dead_warm(warm); + } + } + + /// Discard a warm entry whose firecracker process already exited. + /// + /// The process is known dead, so skip the graceful-stop path: dropping the + /// instance best-effort kills any residual handle, and the network slot is + /// released synchronously. This keeps the acquire path free of runtime + /// `block_on` calls so it stays safe to call from async context. + fn discard_dead_warm(&self, warm: WarmFirecracker) { + let WarmFirecracker { + slot, + fc_instance, + work_dir, + } = warm; + let slot_idx = slot.idx; + warn!( + slot = slot_idx, + "firecracker pool: warm process exited while parked; discarding entry" + ); + + drop(fc_instance); + drop(work_dir); + + if let Err(err) = NetworkManager::global().cleanup_allocated_slot(slot, false) { + warn!( + slot = slot_idx, + error = %err, + "firecracker pool: cleanup of dead warm network slot failed" + ); + } + } + pub fn warm_len(&self) -> usize { self.pool.len() } From 0b20901c26eaf57a392fca7d29835e0c44b8ea8c Mon Sep 17 00:00:00 2001 From: epicvinny Date: Mon, 17 Aug 2026 22:13:36 -0300 Subject: [PATCH 2/8] fix(pool): propagate liveness probe errors and defer dead warm cleanup FirecrackerInstance::is_process_running collapsed every try_wait() I/O error into "process not running", so the pool would tear down an entry (and release its network slot) on an inconclusive probe. Return std::io::Result instead; on Err the pool logs the failure, returns the entry to the pool keeping its slot, and reports a regular miss. try_acquire() also ran full network teardown for dead entries inline, from the async snapshot-resume path; cleanup_allocated_slot does blocking netlink/ip work that can stall a runtime worker thread. Dead entries now go onto a dead_entries queue drained by the maintenance worker at the start of each cycle, next to the teardown it already owns. Shutdown paths drain the queue too, and with maintenance disabled the cleanup falls back to inline execution since no worker exists. --- src/sandbox/firecracker/instance.rs | 10 ++- src/sandbox/firecracker/pool.rs | 122 ++++++++++++++++++++++------ 2 files changed, 103 insertions(+), 29 deletions(-) diff --git a/src/sandbox/firecracker/instance.rs b/src/sandbox/firecracker/instance.rs index c05c13c5..f9cdc416 100644 --- a/src/sandbox/firecracker/instance.rs +++ b/src/sandbox/firecracker/instance.rs @@ -66,10 +66,14 @@ impl FirecrackerInstance { /// Warm pool entries park with `oom_score_adj=1000`, so they are the first /// OOM-kill candidates and can die while idle. Callers must check this /// before handing a parked instance to snapshot resume. - pub fn is_process_running(&mut self) -> bool { + /// + /// An I/O error from the probe means the process state is unknown, not + /// that the child exited, so it is propagated instead of being collapsed + /// into `false`. + pub fn is_process_running(&mut self) -> std::io::Result { match self.process.as_mut() { - Some(child) => matches!(child.try_wait(), Ok(None)), - None => false, + Some(child) => child.try_wait().map(|status| status.is_none()), + None => Ok(false), } } diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index 6758c0b8..23bb55b9 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -5,7 +5,7 @@ //! entry to skip process spawn and API socket polling on the critical path. use std::path::{Path, PathBuf}; -use std::sync::OnceLock; +use std::sync::{Mutex, OnceLock}; use std::time::{Duration, Instant}; use anyhow::{anyhow, Context, Result}; @@ -50,6 +50,11 @@ pub(crate) struct WarmFirecracker { pub struct FirecrackerPool { pool: WarmPool, + /// Warm entries whose firecracker process died while parked, waiting for + /// network-slot teardown on the maintenance thread. The acquire path runs + /// in async snapshot-resume context and must not block on netlink/`ip` + /// teardown, so cleanup is deferred here. + dead_entries: Mutex>, binary: PathBuf, socket_timeout: Duration, socket_poll_interval: Duration, @@ -118,6 +123,7 @@ impl FirecrackerPool { Self { pool: WarmPool::new(pool_config.pool), + dead_entries: Mutex::new(Vec::new()), binary, socket_timeout, socket_poll_interval, @@ -150,41 +156,73 @@ impl FirecrackerPool { fn acquire_live_warm(&self) -> Option { loop { let mut warm = self.pool.try_acquire()?; - if warm.fc_instance.is_process_running() { - return Some(warm); + match warm.fc_instance.is_process_running() { + Ok(true) => return Some(warm), + Ok(false) => self.enqueue_dead_warm(warm), + Err(err) => { + // The probe failed, so the process state is unknown: an + // I/O error does not prove the child exited. Keep the + // entry (and its network slot) and report a pool miss + // instead of tearing down a process that may be alive. + warn!( + slot = warm.slot.idx, + error = %err, + "firecracker pool: warm process state probe failed; returning entry to pool" + ); + if let Err(warm) = self.pool.release(warm) { + // Shutdown reclaimed the entry; queue it for cleanup. + self.enqueue_dead_warm(warm); + } + return None; + } } - self.discard_dead_warm(warm); } } - /// Discard a warm entry whose firecracker process already exited. + /// Queue a warm entry whose firecracker process already exited for + /// cleanup by the maintenance worker. /// - /// The process is known dead, so skip the graceful-stop path: dropping the - /// instance best-effort kills any residual handle, and the network slot is - /// released synchronously. This keeps the acquire path free of runtime - /// `block_on` calls so it stays safe to call from async context. - fn discard_dead_warm(&self, warm: WarmFirecracker) { - let WarmFirecracker { - slot, - fc_instance, - work_dir, - } = warm; - let slot_idx = slot.idx; + /// The acquire path is called from the async snapshot-resume path, so it + /// must not run network teardown inline: `cleanup_allocated_slot` does + /// blocking netlink/`ip` work that could stall a runtime worker thread. + /// The maintenance thread already owns all other pool teardown, so dead + /// entries are deferred to it and the miss is returned immediately. + fn enqueue_dead_warm(&self, warm: WarmFirecracker) { warn!( - slot = slot_idx, - "firecracker pool: warm process exited while parked; discarding entry" + slot = warm.slot.idx, + "firecracker pool: warm process exited while parked; queueing entry for cleanup" ); - drop(fc_instance); - drop(work_dir); + if !self.pool.config().maintenance_enabled { + // Without a maintenance worker nothing consumes the dead-entry + // queue, so clean up inline. + if let Err(err) = cleanup_dead_warm(warm) { + warn!( + error = %err, + "firecracker pool: cleanup of dead warm entry failed" + ); + } + return; + } - if let Err(err) = NetworkManager::global().cleanup_allocated_slot(slot, false) { - warn!( - slot = slot_idx, - error = %err, - "firecracker pool: cleanup of dead warm network slot failed" - ); + self.dead_entries.lock().unwrap().push(warm); + self.pool.request_maintenance(); + } + + /// Clean up queued dead entries on the maintenance thread. + fn cleanup_dead_warm_entries(&self) -> Result<()> { + let dead = std::mem::take(&mut *self.dead_entries.lock().unwrap()); + let mut failures = Vec::new(); + for warm in dead { + if let Err(err) = cleanup_dead_warm(warm) { + warn!( + error = %err, + "firecracker pool: cleanup of dead warm entry failed" + ); + failures.push(err.to_string()); + } } + firecracker_pool_cleanup_result(failures) } pub fn warm_len(&self) -> usize { @@ -199,6 +237,9 @@ impl FirecrackerPool { failures.push(err.to_string()); } } + if let Err(err) = self.cleanup_dead_warm_entries() { + failures.push(err.to_string()); + } firecracker_pool_cleanup_result(failures) } @@ -215,6 +256,9 @@ impl FirecrackerPool { failures.push(err.to_string()); } } + if let Err(err) = self.cleanup_dead_warm_entries() { + failures.push(err.to_string()); + } firecracker_pool_cleanup_result(failures) } @@ -272,6 +316,10 @@ impl FirecrackerPool { } fn run_maintenance_cycle(&self) -> Result<()> { + // Dead entries deferred from the acquire path are cleaned up here, on + // the maintenance thread, where blocking network teardown is safe. + self.cleanup_dead_warm_entries()?; + match self.pool.compute_maintenance_action(self.pool.len()) { PoolMaintenanceAction::Fill(to_fill) => { self.runtime.block_on(self.fill_warm_entries(to_fill))?; @@ -457,6 +505,28 @@ fn firecracker_pool_cleanup_result(failures: Vec) -> Result<()> { } } +/// Tear down a warm entry whose firecracker process already exited. +/// +/// The process is known dead, so skip the graceful-stop path: dropping the +/// instance best-effort kills any residual handle, and the network slot is +/// released directly. Runs on the pool maintenance thread (or inline when +/// maintenance is disabled), never on the acquire path. +fn cleanup_dead_warm(warm: WarmFirecracker) -> Result<()> { + let WarmFirecracker { + slot, + fc_instance, + work_dir, + } = warm; + let slot_idx = slot.idx; + + drop(fc_instance); + drop(work_dir); + + NetworkManager::global() + .cleanup_allocated_slot(slot, false) + .with_context(|| format!("firecracker pool: cleanup dead warm network slot {slot_idx}")) +} + pub(crate) fn warm_stdout_path(work_dir: &Path) -> PathBuf { work_dir.join("firecracker-stdout.log") } From 0a88bc2b123ed21c44d054145b4a67c568c22d15 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Mon, 17 Aug 2026 23:18:14 -0300 Subject: [PATCH 3/8] fix(pool): harden dead warm entry cleanup against shutdown races and failures Address follow-up review on the deferred dead-entry cleanup: - The maintenance-disabled fallback ran blocking teardown inline on the async acquire path; it now spawns a detached cleanup thread instead. - Enqueue was not coordinated with shutdown: an acquire racing drain_all could queue an entry after shutdown had already drained the queue, leaving it without a consumer. The queue now carries a closed flag checked under the same lock; shutdown closes it right after drain_all and late enqueues clean up inline. - cleanup_dead_warm_entries took every entry out of the queue before cleanup and only logged failures, losing track of stale host network state while the slot index was already back in the allocation bitmap. Failed entries are now retained in the queue and retried on later cycles. Retries go through the new NetworkManager::cleanup_slot_resources which redoes only the resource teardown: the allocation bit is released on the first attempt and must never be released twice, since the index may have been reallocated to a live sandbox. cleanup_allocated_slot now borrows the Slot so failed teardown can keep the entry alive for retry; Slot::drop remains the last-resort cleanup. --- src/sandbox/firecracker/pool.rs | 241 +++++++++++++++++++++++++------- src/sandbox/network/manager.rs | 15 +- 2 files changed, 202 insertions(+), 54 deletions(-) diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index 23bb55b9..48c8db85 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -48,13 +48,91 @@ pub(crate) struct WarmFirecracker { pub work_dir: TempDir, } +/// A dead warm entry waiting for network-slot teardown. +enum DeadWarmEntry { + /// Full entry on the first teardown attempt. + Full(WarmFirecracker), + /// Retry after a failed first attempt. The allocation bit was already + /// released, so retries must redo only the resource teardown and never + /// touch the allocation bitmap: the index may have been reallocated to + /// a live sandbox. + SlotOnly(Slot), +} + +impl DeadWarmEntry { + fn slot_idx(&self) -> u32 { + match self { + DeadWarmEntry::Full(warm) => warm.slot.idx, + DeadWarmEntry::SlotOnly(slot) => slot.idx, + } + } + + /// Tear down the entry. The process is known dead, so skip the + /// graceful-stop path: dropping the instance best-effort kills any + /// residual handle before the network slot is released. On failure the + /// entry is returned so the caller can retain it for retry instead of + /// losing track of stale host network state. + fn attempt_cleanup(self) -> std::result::Result<(), DeadWarmCleanupError> { + let slot_idx = self.slot_idx(); + let is_retry = matches!(self, DeadWarmEntry::SlotOnly(_)); + let slot = match self { + DeadWarmEntry::Full(warm) => { + let WarmFirecracker { + slot, + fc_instance, + work_dir, + } = warm; + drop(fc_instance); + drop(work_dir); + slot + } + DeadWarmEntry::SlotOnly(slot) => slot, + }; + + let network = NetworkManager::global(); + // The first attempt also releases the allocation bit; + // `Slot::cleanup` re-arms itself on failure, so retries redo only + // the teardown and never touch the bitmap again. + let result = if is_retry { + network.cleanup_slot_resources(&slot, false) + } else { + network.cleanup_allocated_slot(&slot, false) + }; + result.map_err(|error| DeadWarmCleanupError { + entry: DeadWarmEntry::SlotOnly(slot), + error: error.context(format!( + "firecracker pool: cleanup dead warm network slot {slot_idx}" + )), + }) + } +} + +/// Error from dead-entry teardown. Keeps the entry so the caller can retain +/// it for retry. +struct DeadWarmCleanupError { + entry: DeadWarmEntry, + error: anyhow::Error, +} + +/// Dead warm entries queued for teardown, plus the shutdown state the +/// enqueue path checks under the same lock. +#[derive(Default)] +struct DeadWarmQueue { + /// Set once shutdown starts; late entries must be cleaned up inline by + /// the caller because no consumer will run again. + closed: bool, + entries: Vec, +} + pub struct FirecrackerPool { pool: WarmPool, /// Warm entries whose firecracker process died while parked, waiting for /// network-slot teardown on the maintenance thread. The acquire path runs /// in async snapshot-resume context and must not block on netlink/`ip` - /// teardown, so cleanup is deferred here. - dead_entries: Mutex>, + /// teardown, so cleanup is deferred here. The queue lock also serializes + /// enqueue against shutdown: once `closed` is set, a late enqueue cleans + /// up inline instead of queueing an entry no consumer will ever see. + dead_entries: Mutex, binary: PathBuf, socket_timeout: Duration, socket_poll_interval: Duration, @@ -123,7 +201,7 @@ impl FirecrackerPool { Self { pool: WarmPool::new(pool_config.pool), - dead_entries: Mutex::new(Vec::new()), + dead_entries: Mutex::new(DeadWarmQueue::default()), binary, socket_timeout, socket_poll_interval, @@ -188,57 +266,134 @@ impl FirecrackerPool { /// The maintenance thread already owns all other pool teardown, so dead /// entries are deferred to it and the miss is returned immediately. fn enqueue_dead_warm(&self, warm: WarmFirecracker) { + let slot_idx = warm.slot.idx; warn!( - slot = warm.slot.idx, + slot = slot_idx, "firecracker pool: warm process exited while parked; queueing entry for cleanup" ); - if !self.pool.config().maintenance_enabled { - // Without a maintenance worker nothing consumes the dead-entry - // queue, so clean up inline. - if let Err(err) = cleanup_dead_warm(warm) { - warn!( - error = %err, - "firecracker pool: cleanup of dead warm entry failed" - ); + let entry = DeadWarmEntry::Full(warm); + { + let mut queue = self.dead_entries.lock().unwrap(); + if queue.closed { + // Shutdown already drained the queue: no consumer will run + // again, so clean up inline. This is the shutdown path's own + // blocking teardown, consistent with the rest of shutdown + // cleanup. + drop(queue); + if let Err(err) = entry.attempt_cleanup() { + warn!( + slot = slot_idx, + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + } + return; + } + if self.pool.config().maintenance_enabled { + queue.entries.push(entry); + drop(queue); + self.pool.request_maintenance(); + return; } - return; } - self.dead_entries.lock().unwrap().push(warm); - self.pool.request_maintenance(); + // Maintenance is disabled: no worker consumes the queue. Run the + // blocking teardown on a detached thread so the async acquire path + // never stalls on netlink/`ip` work. If the attempt fails, dropping + // the returned entry retries the teardown once more via `Slot::drop`. + if let Err(err) = std::thread::Builder::new() + .name("firecracker-pool-dead-cleanup".to_string()) + .spawn(move || { + if let Err(err) = entry.attempt_cleanup() { + warn!( + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed" + ); + } + }) + { + warn!( + error = %err, + "firecracker pool: failed to spawn dead-entry cleanup thread" + ); + } } - /// Clean up queued dead entries on the maintenance thread. + /// Clean up queued dead entries. Runs on the maintenance thread. + /// + /// Entries whose teardown fails are retained in the queue so a later + /// cycle retries them; losing them would leave stale host network state + /// behind while the slot index is already back in the allocation bitmap. fn cleanup_dead_warm_entries(&self) -> Result<()> { - let dead = std::mem::take(&mut *self.dead_entries.lock().unwrap()); + let dead = { + let mut queue = self.dead_entries.lock().unwrap(); + std::mem::take(&mut queue.entries) + }; let mut failures = Vec::new(); - for warm in dead { - if let Err(err) = cleanup_dead_warm(warm) { + let mut failed = Vec::new(); + for entry in dead { + if let Err(err) = entry.attempt_cleanup() { warn!( - error = %err, - "firecracker pool: cleanup of dead warm entry failed" + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed; retaining for retry" ); - failures.push(err.to_string()); + failures.push(err.error.to_string()); + failed.push(err.entry); + } + } + if !failed.is_empty() { + let mut queue = self.dead_entries.lock().unwrap(); + if queue.closed { + // Shutdown already closed the queue and no consumer remains; + // dropping retries the teardown once more via `Slot::drop`. + warn!( + count = failed.len(), + "firecracker pool: dropping dead warm entries that failed cleanup during shutdown" + ); + } else { + queue.entries.append(&mut failed); } } firecracker_pool_cleanup_result(failures) } + /// Mark the dead-entry queue closed and take its entries. An acquire + /// that popped its entry just before `drain_all` either enqueues before + /// this take, or observes `closed` under the same lock and cleans up + /// inline, so no entry is ever left queued without a consumer. + fn close_dead_queue(&self) -> Vec { + let mut queue = self.dead_entries.lock().unwrap(); + queue.closed = true; + std::mem::take(&mut queue.entries) + } + pub fn warm_len(&self) -> usize { self.pool.len() } pub async fn shutdown(&self) -> Result<()> { let drained = self.pool.drain_all(); + // Close the dead-entry queue right after drain_all so a concurrent + // acquire that popped its entry just before shutdown either lands in + // this take, or observes `closed` and cleans up inline. + let dead = self.close_dead_queue(); let mut failures = Vec::new(); for warm in drained { if let Err(err) = self.cleanup_warm_async(warm).await { failures.push(err.to_string()); } } - if let Err(err) = self.cleanup_dead_warm_entries() { - failures.push(err.to_string()); + for entry in dead { + if let Err(err) = entry.attempt_cleanup() { + warn!( + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + failures.push(err.error.to_string()); + } } firecracker_pool_cleanup_result(failures) @@ -250,14 +405,22 @@ impl FirecrackerPool { fn shutdown_blocking(&self, sync_network_cleanup: bool) -> Result<()> { let drained = self.pool.drain_all(); + let dead = self.close_dead_queue(); let mut failures = Vec::new(); for warm in drained { if let Err(err) = self.cleanup_warm_blocking(warm, sync_network_cleanup) { failures.push(err.to_string()); } } - if let Err(err) = self.cleanup_dead_warm_entries() { - failures.push(err.to_string()); + for entry in dead { + if let Err(err) = entry.attempt_cleanup() { + warn!( + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + failures.push(err.error.to_string()); + } } firecracker_pool_cleanup_result(failures) @@ -466,7 +629,7 @@ impl FirecrackerPool { drop(work_dir); NetworkManager::global() - .cleanup_allocated_slot(slot, sync_network_cleanup) + .cleanup_allocated_slot(&slot, sync_network_cleanup) .context("firecracker pool: cleanup warm network slot") } @@ -489,7 +652,7 @@ impl FirecrackerPool { drop(work_dir); NetworkManager::global() - .cleanup_allocated_slot(slot, false) + .cleanup_allocated_slot(&slot, false) .context("firecracker pool: cleanup warm network slot") } } @@ -505,28 +668,6 @@ fn firecracker_pool_cleanup_result(failures: Vec) -> Result<()> { } } -/// Tear down a warm entry whose firecracker process already exited. -/// -/// The process is known dead, so skip the graceful-stop path: dropping the -/// instance best-effort kills any residual handle, and the network slot is -/// released directly. Runs on the pool maintenance thread (or inline when -/// maintenance is disabled), never on the acquire path. -fn cleanup_dead_warm(warm: WarmFirecracker) -> Result<()> { - let WarmFirecracker { - slot, - fc_instance, - work_dir, - } = warm; - let slot_idx = slot.idx; - - drop(fc_instance); - drop(work_dir); - - NetworkManager::global() - .cleanup_allocated_slot(slot, false) - .with_context(|| format!("firecracker pool: cleanup dead warm network slot {slot_idx}")) -} - pub(crate) fn warm_stdout_path(work_dir: &Path) -> PathBuf { work_dir.join("firecracker-stdout.log") } diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index fce7136a..e4b0a5b3 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -199,10 +199,10 @@ impl NetworkManager { } fn cleanup_slot_and_release_bit(&self, slot: Slot) -> Result<()> { - self.cleanup_slot_and_release_bit_inner(slot, false) + self.cleanup_slot_and_release_bit_inner(&slot, false) } - fn cleanup_slot_and_release_bit_inner(&self, slot: Slot, sync_cleanup: bool) -> Result<()> { + fn cleanup_slot_and_release_bit_inner(&self, slot: &Slot, sync_cleanup: bool) -> Result<()> { let idx = slot.idx; let cleanup_result = slot.cleanup(sync_cleanup); let bitset_result = self.release_slot_bit(idx); @@ -217,10 +217,17 @@ impl NetworkManager { } } - pub(crate) fn cleanup_allocated_slot(&self, slot: Slot, sync_cleanup: bool) -> Result<()> { + pub(crate) fn cleanup_allocated_slot(&self, slot: &Slot, sync_cleanup: bool) -> Result<()> { self.cleanup_slot_and_release_bit_inner(slot, sync_cleanup) } + /// Redo only the resource teardown for a slot whose allocation bit was + /// already released. Never touches the allocation bitmap: the index may + /// have been reallocated to a live sandbox since the failed cleanup. + pub(crate) fn cleanup_slot_resources(&self, slot: &Slot, sync_cleanup: bool) -> Result<()> { + slot.cleanup(sync_cleanup).map_err(Into::into) + } + /// Find and allocate the next available slot. /// Slot 0 is reserved at init, so returned indices are always >= 1. /// @@ -395,7 +402,7 @@ impl NetworkManager { ); for slot in drained_slots { let idx = slot.idx; - if let Err(err) = self.cleanup_slot_and_release_bit_inner(slot, sync_cleanup) { + if let Err(err) = self.cleanup_slot_and_release_bit_inner(&slot, sync_cleanup) { failures.push(format!("slot {idx} cleanup failed: {err}")); } } From 10e4cd8755002ac82363980d0ea7d1c3f52afc52 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 12:55:32 -0300 Subject: [PATCH 4/8] fix(pool): keep dead warm slot allocated until teardown completes Address remaining review threads on dead warm entry cleanup: - A failed first cleanup released the allocation bit while the retained SlotOnly entry could still retry index-derived teardown (veth-) after the index was reallocated to a live sandbox. The bit now stays allocated until teardown succeeds via NetworkManager::cleanup_allocated_slot_retain_bit_on_failure, so every retry still owns the resources it deletes. - The detached cleanup thread (maintenance-disabled mode) was untracked and unjoined, and a spawn failure dropped the entry on the async acquire path, forcing synchronous Slot::drop cleanup. A managed cleanup worker now drains the queue, is stored in the queue and joined by close_dead_queue, sleeps in bounded slices so shutdown join stays fast, and a spawn failure leaves entries queued (non-blocking) for the next enqueue or for shutdown to drain inline. - Retained failed entries propagated their error out of the maintenance cycle, blocking fill/drain and hot-looping the worker on the same failing teardown. Dead-entry cleanup and watermark maintenance now run as independent work with errors aggregated afterwards, and failed teardowns are requeued with exponential backoff (100ms doubling, capped at 30s) instead of being retried every cycle. --- src/sandbox/firecracker/pool.rs | 472 +++++++++++++++++++++++++++----- src/sandbox/network/manager.rs | 21 +- 2 files changed, 413 insertions(+), 80 deletions(-) diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index 48c8db85..e686d001 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -5,7 +5,7 @@ //! entry to skip process spawn and API socket polling on the critical path. use std::path::{Path, PathBuf}; -use std::sync::{Mutex, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::{Duration, Instant}; use anyhow::{anyhow, Context, Result}; @@ -23,6 +23,24 @@ use crate::sandbox::network::{NetworkManager, Slot}; const POOL_FIRECRACKER_STOP_TIMEOUT: Duration = Duration::from_secs(2); const POOL_PRIME_POLL_INTERVAL: Duration = Duration::from_millis(20); +/// Base delay before retrying a failed dead-entry teardown; doubles per +/// attempt up to `DEAD_WARM_CLEANUP_RETRY_MAX`. +const DEAD_WARM_CLEANUP_RETRY_BASE: Duration = Duration::from_millis(100); +const DEAD_WARM_CLEANUP_RETRY_MAX: Duration = Duration::from_secs(30); +/// Upper bound for one backoff sleep slice in the dead-cleanup worker, so a +/// shutdown `join` is never blocked for longer than this. +const DEAD_WARM_CLEANUP_WORKER_POLL: Duration = Duration::from_secs(1); + +/// Exponential backoff between dead-entry teardown retries. Keeps a retained +/// entry whose cleanup keeps failing from being retried on every maintenance +/// cycle. +fn dead_warm_cleanup_backoff(failed_attempts: u32) -> Duration { + // Shift capped at 9: 100ms << 9 = 51.2s, already past the 30s cap, so + // higher attempt counts saturate at `DEAD_WARM_CLEANUP_RETRY_MAX`. + let shift = failed_attempts.saturating_sub(1).min(9); + let millis = (DEAD_WARM_CLEANUP_RETRY_BASE.as_millis() as u64).saturating_mul(1u64 << shift); + Duration::from_millis(millis).min(DEAD_WARM_CLEANUP_RETRY_MAX) +} static POOL: OnceLock> = OnceLock::new(); @@ -49,71 +67,114 @@ pub(crate) struct WarmFirecracker { } /// A dead warm entry waiting for network-slot teardown. -enum DeadWarmEntry { - /// Full entry on the first teardown attempt. - Full(WarmFirecracker), - /// Retry after a failed first attempt. The allocation bit was already - /// released, so retries must redo only the resource teardown and never - /// touch the allocation bitmap: the index may have been reallocated to - /// a live sandbox. +/// +/// The allocation bit stays set until teardown completes, so the slot index +/// can never be reallocated to a live sandbox while the entry is retained for +/// retry. Every attempt therefore still owns the index-derived resources +/// (`veth-`) it tears down. Failed attempts are retried with exponential +/// backoff (`not_before`) instead of on every maintenance cycle. +struct DeadWarmEntry { + inner: DeadWarmEntryInner, + /// Failed teardown attempts so far; drives the retry backoff. + failed_attempts: u32, + /// Earliest time the next teardown attempt may run. + not_before: Instant, +} + +enum DeadWarmEntryInner { + /// Full entry on the first teardown attempt. Boxed to keep the enum + /// small (clippy::large_enum_variant): `SlotOnly` is the common case for + /// retained retries. + Full(Box), + /// Retained after a failed attempt. The process handle and work dir are + /// already gone; only the network slot (still allocated) needs teardown. SlotOnly(Slot), } impl DeadWarmEntry { + fn new(warm: WarmFirecracker) -> Self { + Self { + inner: DeadWarmEntryInner::Full(Box::new(warm)), + failed_attempts: 0, + not_before: Instant::now(), + } + } + fn slot_idx(&self) -> u32 { - match self { - DeadWarmEntry::Full(warm) => warm.slot.idx, - DeadWarmEntry::SlotOnly(slot) => slot.idx, + match &self.inner { + DeadWarmEntryInner::Full(warm) => warm.slot.idx, + DeadWarmEntryInner::SlotOnly(slot) => slot.idx, } } + fn due(&self, now: Instant) -> bool { + self.not_before <= now + } + /// Tear down the entry. The process is known dead, so skip the /// graceful-stop path: dropping the instance best-effort kills any /// residual handle before the network slot is released. On failure the - /// entry is returned so the caller can retain it for retry instead of - /// losing track of stale host network state. - fn attempt_cleanup(self) -> std::result::Result<(), DeadWarmCleanupError> { + /// entry is returned (with an updated backoff) so the caller can retain + /// it for retry instead of losing track of stale host network state. The + /// allocation bit is released only after teardown succeeds + /// (`cleanup_allocated_slot_retain_bit_on_failure`), so a retry can never + /// race a reallocation of the same index. + fn attempt_cleanup( + self, + network: &NetworkManager, + ) -> std::result::Result<(), DeadWarmCleanupError> { let slot_idx = self.slot_idx(); - let is_retry = matches!(self, DeadWarmEntry::SlotOnly(_)); - let slot = match self { - DeadWarmEntry::Full(warm) => { + let failed_attempts = self.failed_attempts; + let slot = match self.inner { + DeadWarmEntryInner::Full(warm) => { let WarmFirecracker { slot, fc_instance, work_dir, - } = warm; + } = *warm; drop(fc_instance); drop(work_dir); slot } - DeadWarmEntry::SlotOnly(slot) => slot, + DeadWarmEntryInner::SlotOnly(slot) => slot, }; - let network = NetworkManager::global(); - // The first attempt also releases the allocation bit; - // `Slot::cleanup` re-arms itself on failure, so retries redo only - // the teardown and never touch the bitmap again. - let result = if is_retry { - network.cleanup_slot_resources(&slot, false) - } else { - network.cleanup_allocated_slot(&slot, false) - }; - result.map_err(|error| DeadWarmCleanupError { - entry: DeadWarmEntry::SlotOnly(slot), - error: error.context(format!( - "firecracker pool: cleanup dead warm network slot {slot_idx}" - )), + let result = network.cleanup_allocated_slot_retain_bit_on_failure(&slot, false); + result.map_err(|error| { + let failed_attempts = failed_attempts + 1; + DeadWarmCleanupError { + entry: Box::new(DeadWarmEntry { + inner: DeadWarmEntryInner::SlotOnly(slot), + failed_attempts, + not_before: Instant::now() + dead_warm_cleanup_backoff(failed_attempts), + }), + error: error.context(format!( + "firecracker pool: cleanup dead warm network slot {slot_idx}" + )), + } }) } } /// Error from dead-entry teardown. Keeps the entry so the caller can retain -/// it for retry. +/// it for retry. The entry is boxed so the `Err` variant stays small +/// (clippy::result_large_err): `DeadWarmEntry` can hold a full warm +/// Firecracker entry. struct DeadWarmCleanupError { - entry: DeadWarmEntry, + entry: Box, error: anyhow::Error, } +impl std::fmt::Debug for DeadWarmCleanupError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("DeadWarmCleanupError") + .field("slot", &self.entry.slot_idx()) + .field("failed_attempts", &self.entry.failed_attempts) + .field("error", &self.error) + .finish() + } +} + /// Dead warm entries queued for teardown, plus the shutdown state the /// enqueue path checks under the same lock. #[derive(Default)] @@ -122,6 +183,11 @@ struct DeadWarmQueue { /// the caller because no consumer will run again. closed: bool, entries: Vec, + /// Managed cleanup worker used when pool maintenance is disabled (no + /// maintenance worker consumes the queue in that mode). The handle is + /// joined by `close_dead_queue` so shutdown never reports completion + /// while teardown is still running. + worker: Option>, } pub struct FirecrackerPool { @@ -132,7 +198,7 @@ pub struct FirecrackerPool { /// teardown, so cleanup is deferred here. The queue lock also serializes /// enqueue against shutdown: once `closed` is set, a late enqueue cleans /// up inline instead of queueing an entry no consumer will ever see. - dead_entries: Mutex, + dead_entries: Arc>, binary: PathBuf, socket_timeout: Duration, socket_poll_interval: Duration, @@ -201,7 +267,7 @@ impl FirecrackerPool { Self { pool: WarmPool::new(pool_config.pool), - dead_entries: Mutex::new(DeadWarmQueue::default()), + dead_entries: Arc::new(Mutex::new(DeadWarmQueue::default())), binary, socket_timeout, socket_poll_interval, @@ -272,7 +338,7 @@ impl FirecrackerPool { "firecracker pool: warm process exited while parked; queueing entry for cleanup" ); - let entry = DeadWarmEntry::Full(warm); + let entry = DeadWarmEntry::new(warm); { let mut queue = self.dead_entries.lock().unwrap(); if queue.closed { @@ -281,7 +347,7 @@ impl FirecrackerPool { // blocking teardown, consistent with the rest of shutdown // cleanup. drop(queue); - if let Err(err) = entry.attempt_cleanup() { + if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( slot = slot_idx, error = %err.error, @@ -290,70 +356,156 @@ impl FirecrackerPool { } return; } + queue.entries.push(entry); if self.pool.config().maintenance_enabled { - queue.entries.push(entry); drop(queue); self.pool.request_maintenance(); return; } + // Maintenance is disabled: no pool worker consumes the queue. Run + // the blocking teardown on a managed cleanup worker so the async + // acquire path never stalls on netlink/`ip` work. The worker + // handle is stored in the queue and joined by shutdown. If + // spawning fails the entry stays queued (non-blocking fallback): + // the next enqueue retries the spawn, and shutdown drains the + // queue inline. + Self::ensure_dead_cleanup_worker( + &mut queue, + &self.dead_entries, + NetworkManager::global(), + ); } + } - // Maintenance is disabled: no worker consumes the queue. Run the - // blocking teardown on a detached thread so the async acquire path - // never stalls on netlink/`ip` work. If the attempt fails, dropping - // the returned entry retries the teardown once more via `Slot::drop`. - if let Err(err) = std::thread::Builder::new() + /// Spawn the dead-entry cleanup worker if none is running. Called with + /// the queue lock held; never blocks the caller on teardown. + fn ensure_dead_cleanup_worker( + queue: &mut DeadWarmQueue, + shared: &Arc>, + network: &'static NetworkManager, + ) { + if let Some(handle) = queue.worker.take() { + if handle.is_finished() { + // Reap the finished worker; join cannot block here. + let _ = handle.join(); + } else { + queue.worker = Some(handle); + return; + } + } + + let shared = Arc::clone(shared); + match std::thread::Builder::new() .name("firecracker-pool-dead-cleanup".to_string()) - .spawn(move || { - if let Err(err) = entry.attempt_cleanup() { + .spawn(move || Self::dead_cleanup_worker_loop(shared, network)) + { + Ok(handle) => queue.worker = Some(handle), + Err(err) => { + warn!( + error = %err, + "firecracker pool: failed to spawn dead-entry cleanup worker; entries stay queued" + ); + } + } + } + + /// Drain the dead-entry queue on a dedicated blocking thread. Failed + /// teardowns are requeued with backoff; the worker sleeps in short slices + /// while waiting for the next due entry so shutdown's `join` stays + /// bounded, and exits once the queue is drained or closed. + fn dead_cleanup_worker_loop( + shared: Arc>, + network: &'static NetworkManager, + ) { + loop { + let next = { + let mut queue = shared.lock().unwrap(); + if queue.closed { + return; + } + let now = Instant::now(); + match queue.entries.iter().position(|entry| entry.due(now)) { + Some(pos) => queue.entries.remove(pos), + None if queue.entries.is_empty() => return, + None => { + // All retained entries are backing off. Sleep at most + // DEAD_WARM_CLEANUP_WORKER_POLL at a time so shutdown + // never waits long when joining this worker. + let wait = queue + .entries + .iter() + .map(|entry| entry.not_before) + .min() + .unwrap_or(now) + .saturating_duration_since(now) + .min(DEAD_WARM_CLEANUP_WORKER_POLL); + drop(queue); + std::thread::sleep(wait); + continue; + } + } + }; + if let Err(err) = next.attempt_cleanup(network) { + warn!( + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed; retaining for retry" + ); + let mut queue = shared.lock().unwrap(); + if queue.closed { + // Shutdown drained the queue and no consumer remains; + // dropping retries the teardown once more via `Slot::drop`, + // here on the worker thread rather than the async acquire + // path. warn!( - error = %err.error, - "firecracker pool: cleanup of dead warm entry failed" + slot = err.entry.slot_idx(), + "firecracker pool: dropping dead warm entry that failed cleanup during shutdown" ); + } else { + queue.entries.push(*err.entry); } - }) - { - warn!( - error = %err, - "firecracker pool: failed to spawn dead-entry cleanup thread" - ); + } } } /// Clean up queued dead entries. Runs on the maintenance thread. /// - /// Entries whose teardown fails are retained in the queue so a later - /// cycle retries them; losing them would leave stale host network state - /// behind while the slot index is already back in the allocation bitmap. + /// Entries whose teardown fails are retained in the queue with an + /// exponential backoff so a later cycle retries them; losing them would + /// leave stale host network state behind, and retrying without delay + /// would hot-loop the maintenance worker on a persistently failing + /// teardown. Entries still backing off are left queued untouched. fn cleanup_dead_warm_entries(&self) -> Result<()> { - let dead = { + let now = Instant::now(); + let (due, mut retained): (Vec, Vec) = { let mut queue = self.dead_entries.lock().unwrap(); std::mem::take(&mut queue.entries) + .into_iter() + .partition(|entry| entry.due(now)) }; let mut failures = Vec::new(); - let mut failed = Vec::new(); - for entry in dead { - if let Err(err) = entry.attempt_cleanup() { + for entry in due { + if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( slot = err.entry.slot_idx(), error = %err.error, "firecracker pool: cleanup of dead warm entry failed; retaining for retry" ); failures.push(err.error.to_string()); - failed.push(err.entry); + retained.push(*err.entry); } } - if !failed.is_empty() { + if !retained.is_empty() { let mut queue = self.dead_entries.lock().unwrap(); if queue.closed { // Shutdown already closed the queue and no consumer remains; // dropping retries the teardown once more via `Slot::drop`. warn!( - count = failed.len(), + count = retained.len(), "firecracker pool: dropping dead warm entries that failed cleanup during shutdown" ); } else { - queue.entries.append(&mut failed); + queue.entries.append(&mut retained); } } firecracker_pool_cleanup_result(failures) @@ -363,10 +515,25 @@ impl FirecrackerPool { /// that popped its entry just before `drain_all` either enqueues before /// this take, or observes `closed` under the same lock and cleans up /// inline, so no entry is ever left queued without a consumer. + /// + /// A managed cleanup worker (maintenance-disabled mode) may be + /// mid-teardown; it is joined before returning so shutdown never reports + /// completion while teardown is still running. fn close_dead_queue(&self) -> Vec { - let mut queue = self.dead_entries.lock().unwrap(); - queue.closed = true; - std::mem::take(&mut queue.entries) + let (entries, worker) = { + let mut queue = self.dead_entries.lock().unwrap(); + queue.closed = true; + (std::mem::take(&mut queue.entries), queue.worker.take()) + }; + if let Some(handle) = worker { + if let Err(err) = handle.join() { + warn!( + ?err, + "firecracker pool: dead-entry cleanup worker panicked during join" + ); + } + } + entries } pub fn warm_len(&self) -> usize { @@ -386,7 +553,7 @@ impl FirecrackerPool { } } for entry in dead { - if let Err(err) = entry.attempt_cleanup() { + if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( slot = err.entry.slot_idx(), error = %err.error, @@ -413,7 +580,7 @@ impl FirecrackerPool { } } for entry in dead { - if let Err(err) = entry.attempt_cleanup() { + if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( slot = err.entry.slot_idx(), error = %err.error, @@ -479,26 +646,40 @@ impl FirecrackerPool { } fn run_maintenance_cycle(&self) -> Result<()> { + // Dead-entry cleanup and watermark maintenance are independent work: + // dead entries are deliberately retained on failure, so propagating a + // teardown error early would block every later maintenance action — + // and with the pool depleted, the outstanding Fill would hot-loop the + // worker on the same failing cleanup without ever refilling. Attempt + // both and aggregate the errors afterwards. + let mut failures = Vec::new(); + // Dead entries deferred from the acquire path are cleaned up here, on // the maintenance thread, where blocking network teardown is safe. - self.cleanup_dead_warm_entries()?; + if let Err(err) = self.cleanup_dead_warm_entries() { + failures.push(format!("{err:#}")); + } match self.pool.compute_maintenance_action(self.pool.len()) { PoolMaintenanceAction::Fill(to_fill) => { - self.runtime.block_on(self.fill_warm_entries(to_fill))?; + if let Err(err) = self.runtime.block_on(self.fill_warm_entries(to_fill)) { + failures.push(format!("{err:#}")); + } } PoolMaintenanceAction::Drain(to_drain) => { for _ in 0..to_drain { let Some(warm) = self.pool.try_drain_one() else { break; }; - self.cleanup_warm_blocking(warm, false)?; + if let Err(err) = self.cleanup_warm_blocking(warm, false) { + failures.push(format!("{err:#}")); + } } } PoolMaintenanceAction::Idle => {} } - Ok(()) + firecracker_pool_cleanup_result(failures) } async fn fill_warm_entries(&self, to_fill: usize) -> Result<()> { @@ -675,3 +856,144 @@ pub(crate) fn warm_stdout_path(work_dir: &Path) -> PathBuf { pub(crate) fn warm_stderr_path(work_dir: &Path) -> PathBuf { work_dir.join("firecracker-stderr.log") } + +#[cfg(test)] +mod tests { + use super::*; + + fn test_manager() -> NetworkManager { + NetworkManager::new(false, 0, 0) + } + + fn slot_only_entry(slot: Slot) -> DeadWarmEntry { + DeadWarmEntry { + inner: DeadWarmEntryInner::SlotOnly(slot), + failed_attempts: 0, + not_before: Instant::now(), + } + } + + fn wait_until(timeout: Duration, mut predicate: impl FnMut() -> bool) -> bool { + let deadline = Instant::now() + timeout; + loop { + if predicate() { + return true; + } + if Instant::now() >= deadline { + return false; + } + std::thread::sleep(Duration::from_millis(10)); + } + } + + #[test] + fn dead_warm_cleanup_backoff_grows_and_caps() { + assert_eq!(dead_warm_cleanup_backoff(1), DEAD_WARM_CLEANUP_RETRY_BASE); + assert_eq!(dead_warm_cleanup_backoff(2), Duration::from_millis(200)); + assert_eq!(dead_warm_cleanup_backoff(3), Duration::from_millis(400)); + assert_eq!(dead_warm_cleanup_backoff(4), Duration::from_millis(800)); + assert_eq!(dead_warm_cleanup_backoff(64), DEAD_WARM_CLEANUP_RETRY_MAX); + } + + #[test] + fn attempt_cleanup_success_releases_allocation_bit() { + let manager = test_manager(); + let slot = manager.allocate_slot(43).unwrap(); + let idx = slot.idx; + let work_dir = TempDir::new().unwrap(); + let warm = WarmFirecracker { + slot, + fc_instance: FirecrackerInstance::new(work_dir.path().to_path_buf()), + work_dir, + }; + + DeadWarmEntry::new(warm).attempt_cleanup(&manager).unwrap(); + + // Teardown succeeded, so the bit was released and the index can be + // allocated again. + let slot = manager.allocate_slot(idx).unwrap(); + drop(slot); + } + + #[test] + fn attempt_cleanup_failure_retains_entry_with_backoff_and_bit_held() { + let manager = test_manager(); + let slot = manager.allocate_slot(44).unwrap(); + // Release the bit up front: the entry's teardown step succeeds (the + // slot never created kernel resources), but releasing the bit again + // fails, so the entry must be retained. + manager.cleanup_allocated_slot(&slot, false).unwrap(); + + let err = slot_only_entry(slot).attempt_cleanup(&manager).unwrap_err(); + + assert_eq!(err.entry.failed_attempts, 1); + assert!(err.entry.due(Instant::now() + DEAD_WARM_CLEANUP_RETRY_MAX)); + assert!(!err.entry.due(Instant::now())); + assert!(matches!(err.entry.inner, DeadWarmEntryInner::SlotOnly(_))); + } + + #[test] + fn dead_cleanup_worker_drains_queue_and_shutdown_joins_it() { + let manager: &'static NetworkManager = Box::leak(Box::new(test_manager())); + let shared = Arc::new(Mutex::new(DeadWarmQueue::default())); + let slot = manager.allocate_slot(45).unwrap(); + let idx = slot.idx; + + { + let mut queue = shared.lock().unwrap(); + queue.entries.push(slot_only_entry(slot)); + FirecrackerPool::ensure_dead_cleanup_worker(&mut queue, &shared, manager); + assert!(queue.worker.is_some()); + } + + let drained = wait_until(Duration::from_secs(5), || { + shared.lock().unwrap().entries.is_empty() + }); + assert!(drained, "cleanup worker did not drain the queue"); + + // Close + join, mirroring close_dead_queue: the join must complete. + let worker = { + let mut queue = shared.lock().unwrap(); + queue.closed = true; + queue.worker.take() + }; + worker.unwrap().join().unwrap(); + + // Teardown ran and released the bit. + let slot = manager.allocate_slot(idx).unwrap(); + drop(slot); + } + + #[test] + fn dead_cleanup_worker_join_is_bounded_with_backing_off_entries() { + let manager: &'static NetworkManager = Box::leak(Box::new(test_manager())); + let shared = Arc::new(Mutex::new(DeadWarmQueue::default())); + let slot = manager.allocate_slot(46).unwrap(); + + { + let mut entry = slot_only_entry(slot); + // Back off far into the future so the worker parks in its sleep + // slice instead of tearing the slot down. + entry.not_before = Instant::now() + DEAD_WARM_CLEANUP_RETRY_MAX; + let mut queue = shared.lock().unwrap(); + queue.entries.push(entry); + FirecrackerPool::ensure_dead_cleanup_worker(&mut queue, &shared, manager); + } + + // Let the worker reach its sleep slice. + std::thread::sleep(Duration::from_millis(50)); + + let close_started = Instant::now(); + let worker = { + let mut queue = shared.lock().unwrap(); + queue.closed = true; + queue.worker.take() + }; + worker.unwrap().join().unwrap(); + + assert!( + close_started.elapsed() < DEAD_WARM_CLEANUP_WORKER_POLL + Duration::from_secs(4), + "shutdown join was blocked by a backing-off entry" + ); + } +} diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index e4b0a5b3..ec786b6d 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -221,11 +221,22 @@ impl NetworkManager { self.cleanup_slot_and_release_bit_inner(slot, sync_cleanup) } - /// Redo only the resource teardown for a slot whose allocation bit was - /// already released. Never touches the allocation bitmap: the index may - /// have been reallocated to a live sandbox since the failed cleanup. - pub(crate) fn cleanup_slot_resources(&self, slot: &Slot, sync_cleanup: bool) -> Result<()> { - slot.cleanup(sync_cleanup).map_err(Into::into) + /// Tear down an allocated slot, releasing the allocation bit only after + /// the resource teardown succeeds. + /// + /// Unlike `cleanup_allocated_slot`, a teardown failure keeps the bit set: + /// the index then cannot be reallocated while the caller retains the slot + /// for retry, so a later retry still owns the index-derived resources + /// (`veth-`) it tears down and can never delete networking that was + /// recreated for a different sandbox. + pub(crate) fn cleanup_allocated_slot_retain_bit_on_failure( + &self, + slot: &Slot, + sync_cleanup: bool, + ) -> Result<()> { + slot.cleanup(sync_cleanup) + .map_err(Into::::into)?; + self.release_slot_bit(slot.idx) } /// Find and allocate the next available slot. From 90cd557d7c5d9ae02bd1bbf8c3b0610db58b0221 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 13:39:08 -0300 Subject: [PATCH 5/8] fix(pool): harden dead-entry cleanup against overflow, poisoning, and leaks Address the second round of review threads: - The retry counter overflowed after u32::MAX failed attempts, panicking the cleanup worker in debug builds; it now saturates (the backoff already caps at 30s). - All dead-queue locks recover from mutex poisoning via into_inner(), so a panicked cleanup worker cannot turn into a pool-wide panic on the next lock. - Entries whose teardown failed while close_dead_queue raced in-flight cleanup were dropped: Slot::drop re-ran resource cleanup but never released the allocation bit, leaking the index permanently, and the failure never reached shutdown's result. A shared finalize_closed_dead_entry helper now retries teardown once on the calling blocking thread, releases the bit explicitly as a last resort, and reports the failure; the worker records it in the queue so close_dead_queue returns it to shutdown. - The shutdown join was not actually bounded because the synchronous 'ip link del' fallback ran Command::output() without a timeout. The fallback now enforces a 10s timeout (spawn + try_wait + kill), and close_dead_queue joins the worker with a bounded polling wait (DEAD_WARM_CLEANUP_JOIN_TIMEOUT) instead of an unbounded join. --- src/sandbox/firecracker/pool.rs | 255 ++++++++++++++++++++++++++------ src/sandbox/network/manager.rs | 2 +- src/sandbox/network/slot.rs | 35 ++++- 3 files changed, 240 insertions(+), 52 deletions(-) diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index e686d001..df541769 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -30,6 +30,10 @@ const DEAD_WARM_CLEANUP_RETRY_MAX: Duration = Duration::from_secs(30); /// Upper bound for one backoff sleep slice in the dead-cleanup worker, so a /// shutdown `join` is never blocked for longer than this. const DEAD_WARM_CLEANUP_WORKER_POLL: Duration = Duration::from_secs(1); +/// Upper bound for joining the dead-cleanup worker during shutdown. The +/// worker sleeps in short slices and the synchronous `ip link del` fallback +/// has its own timeout, but a stuck teardown must not hang shutdown forever. +const DEAD_WARM_CLEANUP_JOIN_TIMEOUT: Duration = Duration::from_secs(15); /// Exponential backoff between dead-entry teardown retries. Keeps a retained /// entry whose cleanup keeps failing from being retried on every maintenance @@ -44,6 +48,12 @@ fn dead_warm_cleanup_backoff(failed_attempts: u32) -> Duration { static POOL: OnceLock> = OnceLock::new(); +/// Lock the dead-entry queue, recovering from poisoning: a panicked cleanup +/// worker must not turn into a pool-wide panic on the next lock. +fn lock_dead_queue(queue: &Mutex) -> std::sync::MutexGuard<'_, DeadWarmQueue> { + queue.lock().unwrap_or_else(|poisoned| poisoned.into_inner()) +} + extern "C" fn firecracker_pool_exit_hook() { let _ = std::panic::catch_unwind(|| { if let Some(Some(pool)) = POOL.get() { @@ -141,7 +151,10 @@ impl DeadWarmEntry { let result = network.cleanup_allocated_slot_retain_bit_on_failure(&slot, false); result.map_err(|error| { - let failed_attempts = failed_attempts + 1; + // Saturate: a permanently failing slot is retried indefinitely, + // and an overflowing counter would panic the cleanup worker in + // debug builds. The backoff saturates at the cap anyway. + let failed_attempts = failed_attempts.saturating_add(1); DeadWarmCleanupError { entry: Box::new(DeadWarmEntry { inner: DeadWarmEntryInner::SlotOnly(slot), @@ -175,6 +188,46 @@ impl std::fmt::Debug for DeadWarmCleanupError { } } +/// Finish teardown of an entry whose queue was closed while its cleanup was +/// in flight. The entry cannot be requeued (no consumer remains), and simply +/// dropping it would re-run `Slot::cleanup` via `Slot::drop` without ever +/// releasing the allocation bit, leaking the index permanently. The teardown +/// is retried once on the calling (blocking) thread; if it still fails, the +/// bit is released explicitly as a last resort so the index is not leaked. +/// Returns the failure message so shutdown can report it. +fn finalize_closed_dead_entry( + network: &NetworkManager, + entry: DeadWarmEntry, + error: anyhow::Error, +) -> String { + let slot_idx = entry.slot_idx(); + let mut message = error.to_string(); + match entry.attempt_cleanup(network) { + Ok(()) => { + // Teardown completed on the final attempt; the allocation bit + // was released by `attempt_cleanup`. + } + Err(err) => { + message = format!( + "{message}; final teardown during shutdown also failed: {}", + err.error + ); + // Dropping the entry re-runs `Slot::cleanup` best-effort via + // `Slot::drop`, which never touches the bitmap; release the bit + // explicitly so the index is not leaked. + drop(err); + if let Err(bit_err) = network.release_slot_bit(slot_idx) { + warn!( + slot = slot_idx, + error = %bit_err, + "firecracker pool: failed to release dead warm slot bit during shutdown" + ); + } + } + } + message +} + /// Dead warm entries queued for teardown, plus the shutdown state the /// enqueue path checks under the same lock. #[derive(Default)] @@ -188,6 +241,9 @@ struct DeadWarmQueue { /// joined by `close_dead_queue` so shutdown never reports completion /// while teardown is still running. worker: Option>, + /// Failures finalized by the cleanup worker after the queue was closed; + /// collected by `close_dead_queue` so shutdown can report them. + failures: Vec, } pub struct FirecrackerPool { @@ -340,17 +396,20 @@ impl FirecrackerPool { let entry = DeadWarmEntry::new(warm); { - let mut queue = self.dead_entries.lock().unwrap(); + let mut queue = lock_dead_queue(&self.dead_entries); if queue.closed { // Shutdown already drained the queue: no consumer will run - // again, so clean up inline. This is the shutdown path's own - // blocking teardown, consistent with the rest of shutdown - // cleanup. + // again, so finish the teardown inline on this thread and + // make sure the allocation bit is released even on failure + // (dropping the entry would only re-run `Slot::cleanup` via + // `Slot::drop` and leak the bit). drop(queue); if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { + let message = + finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); warn!( slot = slot_idx, - error = %err.error, + error = %message, "firecracker pool: cleanup of dead warm entry failed during shutdown" ); } @@ -419,7 +478,7 @@ impl FirecrackerPool { ) { loop { let next = { - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); if queue.closed { return; } @@ -451,16 +510,24 @@ impl FirecrackerPool { error = %err.error, "firecracker pool: cleanup of dead warm entry failed; retaining for retry" ); - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); if queue.closed { - // Shutdown drained the queue and no consumer remains; - // dropping retries the teardown once more via `Slot::drop`, - // here on the worker thread rather than the async acquire - // path. + // Shutdown closed the queue while this teardown was in + // flight; the entry cannot be requeued. Finish it on this + // worker thread and release the allocation bit explicitly + // on failure: dropping the entry would only re-run + // `Slot::cleanup` via `Slot::drop`, leaking the bit, and + // the failure would never reach shutdown's result. + let DeadWarmCleanupError { entry, error } = err; + let slot_idx = entry.slot_idx(); + drop(queue); + let message = finalize_closed_dead_entry(network, *entry, error); warn!( - slot = err.entry.slot_idx(), - "firecracker pool: dropping dead warm entry that failed cleanup during shutdown" + slot = slot_idx, + error = %message, + "firecracker pool: finalized dead warm entry after queue close" ); + lock_dead_queue(&shared).failures.push(message); } else { queue.entries.push(*err.entry); } @@ -478,12 +545,15 @@ impl FirecrackerPool { fn cleanup_dead_warm_entries(&self) -> Result<()> { let now = Instant::now(); let (due, mut retained): (Vec, Vec) = { - let mut queue = self.dead_entries.lock().unwrap(); + let mut queue = lock_dead_queue(&self.dead_entries); std::mem::take(&mut queue.entries) .into_iter() .partition(|entry| entry.due(now)) }; let mut failures = Vec::new(); + // Errors of retained entries, kept aligned with `retained` so the + // shutdown path can finalize teardown with the original error. + let mut retained_errors: Vec = Vec::new(); for entry in due { if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( @@ -491,20 +561,28 @@ impl FirecrackerPool { error = %err.error, "firecracker pool: cleanup of dead warm entry failed; retaining for retry" ); - failures.push(err.error.to_string()); retained.push(*err.entry); + retained_errors.push(err.error); } } if !retained.is_empty() { - let mut queue = self.dead_entries.lock().unwrap(); + let mut queue = lock_dead_queue(&self.dead_entries); if queue.closed { - // Shutdown already closed the queue and no consumer remains; - // dropping retries the teardown once more via `Slot::drop`. - warn!( - count = retained.len(), - "firecracker pool: dropping dead warm entries that failed cleanup during shutdown" - ); + // Shutdown already closed the queue and no consumer remains. + // Finish each teardown inline (this is the blocking + // maintenance thread) and release the allocation bit + // explicitly on failure: dropping the entries would only + // re-run `Slot::cleanup` via `Slot::drop` and leak the bits. + drop(queue); + for (entry, error) in retained.into_iter().zip(retained_errors) { + failures.push(finalize_closed_dead_entry( + NetworkManager::global(), + entry, + error, + )); + } } else { + failures.extend(retained_errors.iter().map(|error| error.to_string())); queue.entries.append(&mut retained); } } @@ -517,23 +595,44 @@ impl FirecrackerPool { /// inline, so no entry is ever left queued without a consumer. /// /// A managed cleanup worker (maintenance-disabled mode) may be - /// mid-teardown; it is joined before returning so shutdown never reports - /// completion while teardown is still running. - fn close_dead_queue(&self) -> Vec { + /// mid-teardown; it is joined (with a bounded wait) before returning so + /// shutdown only reports completion once in-flight teardown is done. + /// Failures the worker finalized after the queue closed are collected + /// and returned so shutdown can report them. + fn close_dead_queue(&self) -> (Vec, Vec) { let (entries, worker) = { - let mut queue = self.dead_entries.lock().unwrap(); + let mut queue = lock_dead_queue(&self.dead_entries); queue.closed = true; (std::mem::take(&mut queue.entries), queue.worker.take()) }; if let Some(handle) = worker { - if let Err(err) = handle.join() { - warn!( - ?err, - "firecracker pool: dead-entry cleanup worker panicked during join" - ); + // Bounded join: the worker sleeps in short slices between + // attempts and the synchronous `ip link del` fallback has its + // own timeout, but a stuck teardown must not hang shutdown + // forever. + let deadline = Instant::now() + DEAD_WARM_CLEANUP_JOIN_TIMEOUT; + loop { + if handle.is_finished() { + if let Err(err) = handle.join() { + warn!( + ?err, + "firecracker pool: dead-entry cleanup worker panicked during join" + ); + } + break; + } + if Instant::now() >= deadline { + warn!( + timeout = ?DEAD_WARM_CLEANUP_JOIN_TIMEOUT, + "firecracker pool: dead-entry cleanup worker did not finish before the join timeout; shutdown continues without it" + ); + break; + } + std::thread::sleep(Duration::from_millis(20)); } } - entries + let failures = std::mem::take(&mut lock_dead_queue(&self.dead_entries).failures); + (entries, failures) } pub fn warm_len(&self) -> usize { @@ -545,8 +644,8 @@ impl FirecrackerPool { // Close the dead-entry queue right after drain_all so a concurrent // acquire that popped its entry just before shutdown either lands in // this take, or observes `closed` and cleans up inline. - let dead = self.close_dead_queue(); - let mut failures = Vec::new(); + let (dead, dead_failures) = self.close_dead_queue(); + let mut failures = dead_failures; for warm in drained { if let Err(err) = self.cleanup_warm_async(warm).await { failures.push(err.to_string()); @@ -554,12 +653,15 @@ impl FirecrackerPool { } for entry in dead { if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { + let slot = err.entry.slot_idx(); + let message = + finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); warn!( - slot = err.entry.slot_idx(), - error = %err.error, + slot, + error = %message, "firecracker pool: cleanup of dead warm entry failed during shutdown" ); - failures.push(err.error.to_string()); + failures.push(message); } } @@ -572,8 +674,8 @@ impl FirecrackerPool { fn shutdown_blocking(&self, sync_network_cleanup: bool) -> Result<()> { let drained = self.pool.drain_all(); - let dead = self.close_dead_queue(); - let mut failures = Vec::new(); + let (dead, dead_failures) = self.close_dead_queue(); + let mut failures = dead_failures; for warm in drained { if let Err(err) = self.cleanup_warm_blocking(warm, sync_network_cleanup) { failures.push(err.to_string()); @@ -581,12 +683,15 @@ impl FirecrackerPool { } for entry in dead { if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { + let slot = err.entry.slot_idx(); + let message = + finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); warn!( - slot = err.entry.slot_idx(), - error = %err.error, + slot, + error = %message, "firecracker pool: cleanup of dead warm entry failed during shutdown" ); - failures.push(err.error.to_string()); + failures.push(message); } } @@ -940,20 +1045,20 @@ mod tests { let idx = slot.idx; { - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); queue.entries.push(slot_only_entry(slot)); FirecrackerPool::ensure_dead_cleanup_worker(&mut queue, &shared, manager); assert!(queue.worker.is_some()); } let drained = wait_until(Duration::from_secs(5), || { - shared.lock().unwrap().entries.is_empty() + lock_dead_queue(&shared).entries.is_empty() }); assert!(drained, "cleanup worker did not drain the queue"); // Close + join, mirroring close_dead_queue: the join must complete. let worker = { - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); queue.closed = true; queue.worker.take() }; @@ -975,7 +1080,7 @@ mod tests { // Back off far into the future so the worker parks in its sleep // slice instead of tearing the slot down. entry.not_before = Instant::now() + DEAD_WARM_CLEANUP_RETRY_MAX; - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); queue.entries.push(entry); FirecrackerPool::ensure_dead_cleanup_worker(&mut queue, &shared, manager); } @@ -985,7 +1090,7 @@ mod tests { let close_started = Instant::now(); let worker = { - let mut queue = shared.lock().unwrap(); + let mut queue = lock_dead_queue(&shared); queue.closed = true; queue.worker.take() }; @@ -996,4 +1101,58 @@ mod tests { "shutdown join was blocked by a backing-off entry" ); } + + #[test] + fn failed_attempts_counter_saturates_instead_of_overflowing() { + let manager = test_manager(); + let slot = manager.allocate_slot(47).unwrap(); + // Force the teardown to fail at the bit-release step so the entry is + // retained; the counter must saturate rather than overflow (which + // would panic the cleanup worker in debug builds). + manager.cleanup_allocated_slot(&slot, false).unwrap(); + let mut entry = slot_only_entry(slot); + entry.failed_attempts = u32::MAX; + + let err = entry.attempt_cleanup(&manager).unwrap_err(); + + assert_eq!(err.entry.failed_attempts, u32::MAX); + } + + #[test] + fn finalize_closed_entry_releases_bit_and_reports_failure() { + let manager = test_manager(); + let slot = manager.allocate_slot(48).unwrap(); + let idx = slot.idx; + // Make teardown fail at the bit-release step (bit already released): + // finalize must still leave the index allocatable and report the + // failure instead of silently dropping the entry. + manager.cleanup_allocated_slot(&slot, false).unwrap(); + + let message = finalize_closed_dead_entry( + &manager, + slot_only_entry(slot), + anyhow!("in-flight teardown failed"), + ); + + assert!(message.contains("in-flight teardown failed")); + // The index was not leaked: it can be allocated again. + let slot = manager.allocate_slot(idx).unwrap(); + drop(slot); + } + + #[test] + fn dead_queue_lock_recovers_from_poisoning() { + let shared = Arc::new(Mutex::new(DeadWarmQueue::default())); + let shared_in_thread = Arc::clone(&shared); + let handle = std::thread::spawn(move || { + let _guard = shared_in_thread.lock().unwrap(); + panic!("intentional poisoning"); + }); + let _ = handle.join(); + + // A panicked cleanup worker must not turn into a pool-wide panic on + // the next lock. + lock_dead_queue(&shared).closed = true; + assert!(lock_dead_queue(&shared).closed); + } } diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index ec786b6d..dae42609 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -185,7 +185,7 @@ impl NetworkManager { } /// Release only the bitmap bit for a slot index. - fn release_slot_bit(&self, idx: u32) -> Result<()> { + pub(crate) fn release_slot_bit(&self, idx: u32) -> Result<()> { if idx == 0 || idx as usize >= MAX_SLOTS { return Err(anyhow!("Slot index {} out of range", idx)); } diff --git a/src/sandbox/network/slot.rs b/src/sandbox/network/slot.rs index bd5327d4..6bacb6e0 100644 --- a/src/sandbox/network/slot.rs +++ b/src/sandbox/network/slot.rs @@ -711,10 +711,39 @@ impl Slot { let output = crate::privileges::run_with_scoped_capabilities( &[crate::privileges::CAP_NET_ADMIN], || { - Command::new("ip") + // Hard timeout: this runs on shutdown/exit cleanup paths that + // are joined synchronously (e.g. the firecracker pool's dead + // entry cleanup worker), so a stuck `ip link del` must not + // block the caller forever. + const IP_LINK_DEL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + let mut child = Command::new("ip") .args(["link", "del", &veth_name]) - .output() - .context("Failed to execute ip link del") + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .spawn() + .context("Failed to spawn ip link del")?; + let deadline = std::time::Instant::now() + IP_LINK_DEL_TIMEOUT; + loop { + if child + .try_wait() + .context("Failed to poll ip link del")? + .is_some() + { + return child + .wait_with_output() + .context("Failed to collect ip link del output"); + } + if std::time::Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + return Err(anyhow!( + "ip link del {} timed out after {:?}", + veth_name, + IP_LINK_DEL_TIMEOUT + )); + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } }, )?; From ce5d2ff19525cc33cff2cee998aac69798f43af0 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 13:56:44 -0300 Subject: [PATCH 6/8] fix(pool): close shutdown races in dead-entry cleanup finalization Address the third round of review threads: - cleanup_dead_warm_entries paired failed entries with errors via a misaligned zip: backing-off entries could be dropped on shutdown and leak their allocation bits. Backing-off and failed entries are now tracked separately, and on queue close every retained entry is finalized (failed ones with their own error, backing-off ones with a plain teardown attempt that reports only if it actually fails). finalize_closed_dead_entry now takes Option and returns Option accordingly. - A worker spawn failure in maintenance-disabled mode left entries retained until the next enqueue or shutdown; the spawn is now retried once via an unnamed Builder before falling back to the queued path. - A join timeout no longer detaches the cleanup worker: the handle is kept managed in the queue for a later close to retry, and an explicit incomplete-shutdown failure is reported instead of silent completion. - The async shutdown() ran the bounded join, sleeps, and blocking netlink/ip teardown inline on a Tokio worker; that portion now runs in tokio::task::spawn_blocking via shutdown_dead_entries_blocking. - The 'ip link del' timeout path could still block: kill() errors are propagated and the child is reaped with bounded try_wait polling (1s grace) instead of a blocking wait(). - Tests now inject a genuine teardown failure (fail_slot_teardown hook) while the allocation bit stays held, asserting the index cannot be reallocated until a successful retry releases it. --- src/sandbox/firecracker/pool.rs | 299 ++++++++++++++++++++++---------- src/sandbox/network/manager.rs | 13 ++ src/sandbox/network/slot.rs | 19 +- 3 files changed, 233 insertions(+), 98 deletions(-) diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index df541769..f3110f1c 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -192,26 +192,27 @@ impl std::fmt::Debug for DeadWarmCleanupError { /// in flight. The entry cannot be requeued (no consumer remains), and simply /// dropping it would re-run `Slot::cleanup` via `Slot::drop` without ever /// releasing the allocation bit, leaking the index permanently. The teardown -/// is retried once on the calling (blocking) thread; if it still fails, the -/// bit is released explicitly as a last resort so the index is not leaked. -/// Returns the failure message so shutdown can report it. +/// is attempted once on the calling (blocking) thread; if it fails, the bit +/// is released explicitly as a last resort so the index is not leaked. +/// Returns the failure message to report, if any: a failed teardown always +/// reports, while a successful teardown only reports a prior failure. fn finalize_closed_dead_entry( network: &NetworkManager, entry: DeadWarmEntry, - error: anyhow::Error, -) -> String { + prior_error: Option, +) -> Option { let slot_idx = entry.slot_idx(); - let mut message = error.to_string(); match entry.attempt_cleanup(network) { Ok(()) => { - // Teardown completed on the final attempt; the allocation bit - // was released by `attempt_cleanup`. + // Teardown completed; the allocation bit was released by + // `attempt_cleanup`. Report only a prior failure, if any. + prior_error.map(|error| error.to_string()) } Err(err) => { - message = format!( - "{message}; final teardown during shutdown also failed: {}", - err.error - ); + let mut message = err.error.to_string(); + if let Some(prior) = prior_error { + message = format!("{prior:#}; final teardown during shutdown also failed: {message}"); + } // Dropping the entry re-runs `Slot::cleanup` best-effort via // `Slot::drop`, which never touches the bitmap; release the bit // explicitly so the index is not leaked. @@ -223,9 +224,9 @@ fn finalize_closed_dead_entry( "firecracker pool: failed to release dead warm slot bit during shutdown" ); } + Some(message) } } - message } /// Dead warm entries queued for teardown, plus the shutdown state the @@ -405,13 +406,17 @@ impl FirecrackerPool { // `Slot::drop` and leak the bit). drop(queue); if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - let message = - finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); - warn!( - slot = slot_idx, - error = %message, - "firecracker pool: cleanup of dead warm entry failed during shutdown" - ); + if let Some(message) = finalize_closed_dead_entry( + NetworkManager::global(), + *err.entry, + Some(err.error), + ) { + warn!( + slot = slot_idx, + error = %message, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + } } return; } @@ -454,16 +459,33 @@ impl FirecrackerPool { } let shared = Arc::clone(shared); + let shared_retry = Arc::clone(&shared); match std::thread::Builder::new() .name("firecracker-pool-dead-cleanup".to_string()) .spawn(move || Self::dead_cleanup_worker_loop(shared, network)) { Ok(handle) => queue.worker = Some(handle), Err(err) => { + // Retry once with a plain spawn: a builder failure is often + // transient (resource exhaustion), and with maintenance + // disabled no other consumer retries until the next enqueue + // or shutdown — the slot and its allocation bit would stay + // retained indefinitely. warn!( error = %err, - "firecracker pool: failed to spawn dead-entry cleanup worker; entries stay queued" + "firecracker pool: failed to spawn dead-entry cleanup worker; retrying with plain spawn" ); + match std::thread::Builder::new().spawn(move || { + Self::dead_cleanup_worker_loop(shared_retry, network) + }) { + Ok(handle) => queue.worker = Some(handle), + Err(retry_err) => { + warn!( + error = %retry_err, + "firecracker pool: cleanup worker spawn retry failed; entries stay queued until the next enqueue or shutdown drains them inline" + ); + } + } } } } @@ -521,13 +543,16 @@ impl FirecrackerPool { let DeadWarmCleanupError { entry, error } = err; let slot_idx = entry.slot_idx(); drop(queue); - let message = finalize_closed_dead_entry(network, *entry, error); - warn!( - slot = slot_idx, - error = %message, - "firecracker pool: finalized dead warm entry after queue close" - ); - lock_dead_queue(&shared).failures.push(message); + if let Some(message) = + finalize_closed_dead_entry(network, *entry, Some(error)) + { + warn!( + slot = slot_idx, + error = %message, + "firecracker pool: finalized dead warm entry after queue close" + ); + lock_dead_queue(&shared).failures.push(message); + } } else { queue.entries.push(*err.entry); } @@ -544,16 +569,16 @@ impl FirecrackerPool { /// teardown. Entries still backing off are left queued untouched. fn cleanup_dead_warm_entries(&self) -> Result<()> { let now = Instant::now(); - let (due, mut retained): (Vec, Vec) = { + // Backing-off entries are kept separate from due entries so each + // failed entry always stays paired with its own error. + let (due, mut backoff): (Vec, Vec) = { let mut queue = lock_dead_queue(&self.dead_entries); std::mem::take(&mut queue.entries) .into_iter() .partition(|entry| entry.due(now)) }; let mut failures = Vec::new(); - // Errors of retained entries, kept aligned with `retained` so the - // shutdown path can finalize teardown with the original error. - let mut retained_errors: Vec = Vec::new(); + let mut failed: Vec<(DeadWarmEntry, anyhow::Error)> = Vec::new(); for entry in due { if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { warn!( @@ -561,29 +586,40 @@ impl FirecrackerPool { error = %err.error, "firecracker pool: cleanup of dead warm entry failed; retaining for retry" ); - retained.push(*err.entry); - retained_errors.push(err.error); + failed.push((*err.entry, err.error)); } } - if !retained.is_empty() { + if !backoff.is_empty() || !failed.is_empty() { let mut queue = lock_dead_queue(&self.dead_entries); if queue.closed { // Shutdown already closed the queue and no consumer remains. - // Finish each teardown inline (this is the blocking + // Finish every retained entry inline (this is the blocking // maintenance thread) and release the allocation bit // explicitly on failure: dropping the entries would only // re-run `Slot::cleanup` via `Slot::drop` and leak the bits. drop(queue); - for (entry, error) in retained.into_iter().zip(retained_errors) { - failures.push(finalize_closed_dead_entry( + for (entry, error) in failed { + if let Some(message) = finalize_closed_dead_entry( NetworkManager::global(), entry, - error, - )); + Some(error), + ) { + failures.push(message); + } + } + for entry in backoff { + if let Some(message) = + finalize_closed_dead_entry(NetworkManager::global(), entry, None) + { + failures.push(message); + } } } else { - failures.extend(retained_errors.iter().map(|error| error.to_string())); - queue.entries.append(&mut retained); + failures.extend(failed.iter().map(|(_, error)| error.to_string())); + queue.entries.append(&mut backoff); + queue + .entries + .extend(failed.into_iter().map(|(entry, _)| entry)); } } firecracker_pool_cleanup_result(failures) @@ -605,33 +641,50 @@ impl FirecrackerPool { queue.closed = true; (std::mem::take(&mut queue.entries), queue.worker.take()) }; + let mut failures = Vec::new(); if let Some(handle) = worker { // Bounded join: the worker sleeps in short slices between // attempts and the synchronous `ip link del` fallback has its // own timeout, but a stuck teardown must not hang shutdown // forever. let deadline = Instant::now() + DEAD_WARM_CLEANUP_JOIN_TIMEOUT; + // `join` consumes the handle, so keep it in an Option and only + // take it once `is_finished` guarantees the join cannot block. + let mut handle = Some(handle); loop { - if handle.is_finished() { - if let Err(err) = handle.join() { - warn!( - ?err, - "firecracker pool: dead-entry cleanup worker panicked during join" - ); + if handle.as_ref().map(|h| h.is_finished()).unwrap_or(true) { + if let Some(h) = handle.take() { + if let Err(err) = h.join() { + warn!( + ?err, + "firecracker pool: dead-entry cleanup worker panicked during join" + ); + } } break; } if Instant::now() >= deadline { + // Do not detach a still-running worker: keep the handle + // managed in the queue so a later close can retry the + // join, and report the incomplete shutdown instead of + // silently reporting completion while teardown may still + // be running. warn!( timeout = ?DEAD_WARM_CLEANUP_JOIN_TIMEOUT, - "firecracker pool: dead-entry cleanup worker did not finish before the join timeout; shutdown continues without it" + "firecracker pool: dead-entry cleanup worker did not finish before the join timeout" ); + failures.push(format!( + "firecracker pool: dead-entry cleanup worker still running after {DEAD_WARM_CLEANUP_JOIN_TIMEOUT:?} join timeout" + )); + lock_dead_queue(&self.dead_entries).worker = handle.take(); break; } std::thread::sleep(Duration::from_millis(20)); } } - let failures = std::mem::take(&mut lock_dead_queue(&self.dead_entries).failures); + failures.extend(std::mem::take( + &mut lock_dead_queue(&self.dead_entries).failures, + )); (entries, failures) } @@ -639,31 +692,58 @@ impl FirecrackerPool { self.pool.len() } + /// Blocking portion of shutdown that touches the dead-entry queue: + /// close the queue, join the cleanup worker (bounded), and tear down + /// queued entries. Kept separate so the async shutdown path can run it + /// on a blocking thread instead of stalling a Tokio worker. + fn shutdown_dead_entries_blocking(&self) -> Vec { + let (dead, mut failures) = self.close_dead_queue(); + for entry in dead { + if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { + let slot = err.entry.slot_idx(); + if let Some(message) = + finalize_closed_dead_entry(NetworkManager::global(), *err.entry, Some(err.error)) + { + warn!( + slot, + error = %message, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + failures.push(message); + } + } + } + failures + } + pub async fn shutdown(&self) -> Result<()> { let drained = self.pool.drain_all(); - // Close the dead-entry queue right after drain_all so a concurrent - // acquire that popped its entry just before shutdown either lands in - // this take, or observes `closed` and cleans up inline. - let (dead, dead_failures) = self.close_dead_queue(); + // The dead-queue join and dead-entry teardown do blocking network + // teardown (netlink/`ip`) plus bounded sleeps; run them on a + // blocking thread so this awaited shutdown never stalls a Tokio + // worker and delays unrelated shutdown futures. The queue lock + // serializes with acquire exactly as before: a concurrent enqueue + // either lands before the queue is closed or observes `closed` and + // cleans up inline. `Self::global()` re-fetches the 'static pool + // reference for the blocking closure. + let dead_failures = match tokio::task::spawn_blocking(|| { + Self::global() + .map(|pool| pool.shutdown_dead_entries_blocking()) + .unwrap_or_default() + }) + .await + { + Ok(failures) => failures, + Err(err) => vec![format!( + "firecracker pool: blocking dead-entry shutdown task failed: {err}" + )], + }; let mut failures = dead_failures; for warm in drained { if let Err(err) = self.cleanup_warm_async(warm).await { failures.push(err.to_string()); } } - for entry in dead { - if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - let slot = err.entry.slot_idx(); - let message = - finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); - warn!( - slot, - error = %message, - "firecracker pool: cleanup of dead warm entry failed during shutdown" - ); - failures.push(message); - } - } firecracker_pool_cleanup_result(failures) } @@ -674,26 +754,12 @@ impl FirecrackerPool { fn shutdown_blocking(&self, sync_network_cleanup: bool) -> Result<()> { let drained = self.pool.drain_all(); - let (dead, dead_failures) = self.close_dead_queue(); - let mut failures = dead_failures; + let mut failures = self.shutdown_dead_entries_blocking(); for warm in drained { if let Err(err) = self.cleanup_warm_blocking(warm, sync_network_cleanup) { failures.push(err.to_string()); } } - for entry in dead { - if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - let slot = err.entry.slot_idx(); - let message = - finalize_closed_dead_entry(NetworkManager::global(), *err.entry, err.error); - warn!( - slot, - error = %message, - "firecracker pool: cleanup of dead warm entry failed during shutdown" - ); - failures.push(message); - } - } firecracker_pool_cleanup_result(failures) } @@ -1024,10 +1090,13 @@ mod tests { fn attempt_cleanup_failure_retains_entry_with_backoff_and_bit_held() { let manager = test_manager(); let slot = manager.allocate_slot(44).unwrap(); - // Release the bit up front: the entry's teardown step succeeds (the - // slot never created kernel resources), but releasing the bit again - // fails, so the entry must be retained. - manager.cleanup_allocated_slot(&slot, false).unwrap(); + let idx = slot.idx; + // Inject a genuine teardown failure while the bit is still allocated: + // the entry must be retained and the index must NOT be reallocatable + // until a successful retry releases it. + manager + .fail_slot_teardown + .store(true, std::sync::atomic::Ordering::Relaxed); let err = slot_only_entry(slot).attempt_cleanup(&manager).unwrap_err(); @@ -1035,6 +1104,16 @@ mod tests { assert!(err.entry.due(Instant::now() + DEAD_WARM_CLEANUP_RETRY_MAX)); assert!(!err.entry.due(Instant::now())); assert!(matches!(err.entry.inner, DeadWarmEntryInner::SlotOnly(_))); + // The production invariant: the bit is still held, so the index + // cannot be reallocated to a live sandbox while the entry is + // retained for retry. + assert!(manager.allocate_slot(idx).is_err()); + + // The injected failure is consumed; the retry succeeds and releases + // the bit. + err.entry.attempt_cleanup(&manager).unwrap(); + let slot = manager.allocate_slot(idx).unwrap(); + drop(slot); } #[test] @@ -1106,10 +1185,12 @@ mod tests { fn failed_attempts_counter_saturates_instead_of_overflowing() { let manager = test_manager(); let slot = manager.allocate_slot(47).unwrap(); - // Force the teardown to fail at the bit-release step so the entry is - // retained; the counter must saturate rather than overflow (which - // would panic the cleanup worker in debug builds). - manager.cleanup_allocated_slot(&slot, false).unwrap(); + // Inject a genuine teardown failure so the entry is retained; the + // counter must saturate rather than overflow (which would panic the + // cleanup worker in debug builds). + manager + .fail_slot_teardown + .store(true, std::sync::atomic::Ordering::Relaxed); let mut entry = slot_only_entry(slot); entry.failed_attempts = u32::MAX; @@ -1123,16 +1204,19 @@ mod tests { let manager = test_manager(); let slot = manager.allocate_slot(48).unwrap(); let idx = slot.idx; - // Make teardown fail at the bit-release step (bit already released): - // finalize must still leave the index allocatable and report the - // failure instead of silently dropping the entry. - manager.cleanup_allocated_slot(&slot, false).unwrap(); + // Inject a genuine teardown failure while the bit is held: finalize + // must still leave the index allocatable (explicit bit release) and + // report the failure instead of silently dropping the entry. + manager + .fail_slot_teardown + .store(true, std::sync::atomic::Ordering::Relaxed); let message = finalize_closed_dead_entry( &manager, slot_only_entry(slot), - anyhow!("in-flight teardown failed"), - ); + Some(anyhow!("in-flight teardown failed")), + ) + .expect("failed teardown must be reported"); assert!(message.contains("in-flight teardown failed")); // The index was not leaked: it can be allocated again. @@ -1140,6 +1224,29 @@ mod tests { drop(slot); } + #[test] + fn finalize_closed_entry_success_reports_only_prior_failure() { + let manager = test_manager(); + let slot = manager.allocate_slot(49).unwrap(); + let idx = slot.idx; + + // Backing-off entry (no prior failure) whose teardown succeeds at + // shutdown: nothing to report, bit released by attempt_cleanup. + let message = finalize_closed_dead_entry(&manager, slot_only_entry(slot), None); + assert!(message.is_none()); + let slot = manager.allocate_slot(idx).unwrap(); + drop(slot); + + // Same success, but with a prior in-flight failure: reported. + let slot = manager.allocate_slot(50).unwrap(); + let message = finalize_closed_dead_entry( + &manager, + slot_only_entry(slot), + Some(anyhow!("first attempt failed")), + ); + assert_eq!(message.as_deref(), Some("first attempt failed")); + } + #[test] fn dead_queue_lock_recovers_from_poisoning() { let shared = Arc::new(Mutex::new(DeadWarmQueue::default())); diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index dae42609..8ac80ea0 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -69,6 +69,13 @@ pub(crate) struct NetworkManager { /// Rejects new allocations once shutdown cleanup starts. shutting_down: AtomicBool, + + /// Test hook: when set, the next + /// `cleanup_allocated_slot_retain_bit_on_failure` call fails before + /// touching slot resources, so tests can exercise a genuine teardown + /// failure while the allocation bit stays held. + #[cfg(test)] + pub(crate) fail_slot_teardown: AtomicBool, } impl NetworkManager { @@ -128,6 +135,8 @@ impl NetworkManager { address_plan: config.address_plan, netns_dir: config.netns_dir, shutting_down: AtomicBool::new(false), + #[cfg(test)] + fail_slot_teardown: AtomicBool::new(false), }; // Reserve slot 0 (invalid for IP addresses) @@ -234,6 +243,10 @@ impl NetworkManager { slot: &Slot, sync_cleanup: bool, ) -> Result<()> { + #[cfg(test)] + if self.fail_slot_teardown.swap(false, Ordering::AcqRel) { + return Err(anyhow!("injected slot teardown failure")); + } slot.cleanup(sync_cleanup) .map_err(Into::::into)?; self.release_slot_bit(slot.idx) diff --git a/src/sandbox/network/slot.rs b/src/sandbox/network/slot.rs index 6bacb6e0..e77fb91f 100644 --- a/src/sandbox/network/slot.rs +++ b/src/sandbox/network/slot.rs @@ -734,8 +734,23 @@ impl Slot { .context("Failed to collect ip link del output"); } if std::time::Instant::now() >= deadline { - let _ = child.kill(); - let _ = child.wait(); + child.kill().context("Failed to kill timed-out ip link del")?; + // `kill` only sends the termination request; reap + // with bounded polling instead of a blocking `wait` + // so a process stuck in uninterruptible sleep cannot + // hang this shutdown path beyond the grace period. + let reap_deadline = std::time::Instant::now() + + std::time::Duration::from_secs(1); + while std::time::Instant::now() < reap_deadline { + if child + .try_wait() + .context("Failed to reap timed-out ip link del")? + .is_some() + { + break; + } + std::thread::sleep(std::time::Duration::from_millis(20)); + } return Err(anyhow!( "ip link del {} timed out after {:?}", veth_name, From e7ef3cb9bad7f2587b200134fda3078b22a5e5dd Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 14:19:57 -0300 Subject: [PATCH 7/8] fix(pool): close dead-entry lost wakeup and schedule backoff retries --- crates/warm-pool/src/lib.rs | 117 +++++++++++++++++++++++++++++++- src/sandbox/firecracker/pool.rs | 54 +++++++++++++-- src/sandbox/network/manager.rs | 7 ++ src/sandbox/network/slot.rs | 19 +++++- 4 files changed, 188 insertions(+), 9 deletions(-) diff --git a/crates/warm-pool/src/lib.rs b/crates/warm-pool/src/lib.rs index c288b46e..107aeb27 100644 --- a/crates/warm-pool/src/lib.rs +++ b/crates/warm-pool/src/lib.rs @@ -12,6 +12,7 @@ use std::collections::VecDeque; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Condvar, Mutex}; +use std::time::{Duration, Instant}; /// Action computed by watermark logic for the maintenance worker. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -31,6 +32,9 @@ struct PoolMaintenanceSignal { pending: bool, /// Worker should exit. stop: bool, + /// Earliest instant the worker must wake even without a `pending` + /// signal, for work scheduled in the future (e.g. backoff retries). + wake_at: Option, } /// Configuration for a warm pool. @@ -206,6 +210,31 @@ impl WarmPool { self.maintenance_cv.notify_one(); } + /// Request the maintenance worker to wake up after `delay`. + /// + /// Used when work becomes due at a future time (e.g. backoff retries + /// scheduled by the cycle callback): the worker sleeps on the condvar + /// with a timeout instead of waiting for unrelated pool activity. The + /// earliest requested instant wins. + pub fn request_maintenance_after(&self, delay: Duration) { + if !self.config.maintenance_enabled || self.is_shutting_down() { + return; + } + + let mut signal = self.maintenance_signal.lock().unwrap(); + if signal.stop { + return; + } + let Some(wake_at) = Instant::now().checked_add(delay) else { + return; + }; + signal.wake_at = Some(match signal.wake_at { + Some(existing) => existing.min(wake_at), + None => wake_at, + }); + self.maintenance_cv.notify_one(); + } + /// Try to acquire a resource from the pool (fast path). /// /// Returns `Some(resource)` if one is available, `None` if the pool is empty. @@ -342,8 +371,27 @@ impl WarmPool { loop { if !has_immediate_work { let mut signal = self.maintenance_signal.lock().unwrap(); - while !signal.stop && !signal.pending { - signal = self.maintenance_cv.wait(signal).unwrap(); + loop { + if signal.stop || signal.pending { + break; + } + match signal.wake_at { + Some(wake_at) => { + let now = Instant::now(); + if wake_at <= now { + // Scheduled future work is due. + signal.wake_at = None; + break; + } + let wait = wake_at - now; + let (guard, _) = + self.maintenance_cv.wait_timeout(signal, wait).unwrap(); + signal = guard; + } + None => { + signal = self.maintenance_cv.wait(signal).unwrap(); + } + } } if signal.stop { break; @@ -497,6 +545,71 @@ mod tests { assert!(pool.maintenance_signal.lock().unwrap().pending); } + #[test] + fn request_maintenance_after_keeps_earliest_wake() { + let pool = WarmPool::::new(PoolConfig { + low_watermark: 2, + high_watermark: 10, + maintenance_enabled: true, + startup_prewarm: false, + }); + + pool.request_maintenance_after(Duration::from_millis(50)); + let first = pool.maintenance_signal.lock().unwrap().wake_at; + assert!(first.is_some()); + + // A later instant must not push the scheduled wake back. + pool.request_maintenance_after(Duration::from_secs(60)); + assert_eq!(pool.maintenance_signal.lock().unwrap().wake_at, first); + + // An earlier instant wins. + pool.request_maintenance_after(Duration::from_millis(10)); + let earlier = pool.maintenance_signal.lock().unwrap().wake_at; + assert!(earlier.is_some()); + assert!(earlier < first); + } + + #[test] + fn request_maintenance_after_ignored_when_maintenance_disabled() { + let pool = WarmPool::::new(PoolConfig { + low_watermark: 2, + high_watermark: 10, + maintenance_enabled: false, + startup_prewarm: false, + }); + + pool.request_maintenance_after(Duration::from_millis(1)); + assert_eq!(pool.maintenance_signal.lock().unwrap().wake_at, None); + } + + #[test] + fn maintenance_worker_wakes_for_scheduled_work() { + let pool: &'static WarmPool = Box::leak(Box::new(WarmPool::new(PoolConfig { + low_watermark: 0, + high_watermark: 10, + maintenance_enabled: true, + startup_prewarm: false, + }))); + let cycles: &'static std::sync::atomic::AtomicUsize = + Box::leak(Box::new(std::sync::atomic::AtomicUsize::new(0))); + pool.start_maintenance_worker(move || { + cycles.fetch_add(1, Ordering::Relaxed); + }); + + // No pending signal: the worker must wake solely from the scheduled + // instant instead of waiting for unrelated pool activity. + pool.request_maintenance_after(Duration::from_millis(50)); + let deadline = Instant::now() + Duration::from_secs(5); + while cycles.load(Ordering::Relaxed) < 2 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + // One cycle runs at worker start; the second must come from the + // scheduled wake. + assert!(cycles.load(Ordering::Relaxed) >= 2); + + pool.drain_all(); + } + #[test] fn compute_maintenance_action_drains_above_high() { let pool = WarmPool::::new(PoolConfig { diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index f3110f1c..39d7c0e1 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -355,7 +355,15 @@ impl FirecrackerPool { /// can die while idle; handing a dead process to snapshot resume would fail /// the resume, so dead entries are discarded here instead. fn acquire_live_warm(&self) -> Option { + // Bound the scan: probe errors release the entry back into the pool, + // so an unhealthy pool could otherwise be scanned forever — and a + // single failing probe must not mask healthy entries behind it. + let mut attempts_left = self.pool.len() + 1; loop { + if attempts_left == 0 { + return None; + } + attempts_left -= 1; let mut warm = self.pool.try_acquire()?; match warm.fc_instance.is_process_running() { Ok(true) => return Some(warm), @@ -363,8 +371,9 @@ impl FirecrackerPool { Err(err) => { // The probe failed, so the process state is unknown: an // I/O error does not prove the child exited. Keep the - // entry (and its network slot) and report a pool miss - // instead of tearing down a process that may be alive. + // entry (and its network slot) and move on to the next + // entry instead of tearing down a process that may be + // alive or masking healthy entries behind this one. warn!( slot = warm.slot.idx, error = %err, @@ -374,7 +383,7 @@ impl FirecrackerPool { // Shutdown reclaimed the entry; queue it for cleanup. self.enqueue_dead_warm(warm); } - return None; + continue; } } } @@ -507,7 +516,15 @@ impl FirecrackerPool { let now = Instant::now(); match queue.entries.iter().position(|entry| entry.due(now)) { Some(pos) => queue.entries.remove(pos), - None if queue.entries.is_empty() => return, + None if queue.entries.is_empty() => { + // Clear the worker slot under the same lock before + // exiting: an enqueue that observed this handle as + // not-yet-finished would otherwise skip spawning a + // new worker, leaving the entry queued until the next + // enqueue or shutdown (lost wakeup). + queue.worker = None; + return; + } None => { // All retained entries are backing off. Sleep at most // DEAD_WARM_CLEANUP_WORKER_POLL at a time so shutdown @@ -620,6 +637,17 @@ impl FirecrackerPool { queue .entries .extend(failed.into_iter().map(|(entry, _)| entry)); + // The maintenance worker sleeps on a condvar between cycles, + // so retained backoff entries must schedule their own wakeup: + // without it they would only be retried when unrelated pool + // activity happens to trigger maintenance again. + let next_wake = queue.entries.iter().map(|entry| entry.not_before).min(); + drop(queue); + if let Some(not_before) = next_wake { + self.pool.request_maintenance_after( + not_before.saturating_duration_since(Instant::now()), + ); + } } } firecracker_pool_cleanup_result(failures) @@ -1135,13 +1163,27 @@ mod tests { }); assert!(drained, "cleanup worker did not drain the queue"); - // Close + join, mirroring close_dead_queue: the join must complete. + // Once the queue is empty the worker clears its own slot before + // exiting, so the next enqueue can spawn a fresh worker (no lost + // wakeup against a stale not-yet-finished handle). + let cleared = wait_until(Duration::from_secs(5), || { + lock_dead_queue(&shared).worker.is_none() + }); + assert!( + cleared, + "cleanup worker did not release its worker slot on exit" + ); + + // Close + join, mirroring close_dead_queue: if the worker had not + // exited yet when the slot was checked, the join must still complete. let worker = { let mut queue = lock_dead_queue(&shared); queue.closed = true; queue.worker.take() }; - worker.unwrap().join().unwrap(); + if let Some(worker) = worker { + worker.join().unwrap(); + } // Teardown ran and released the bit. let slot = manager.allocate_slot(idx).unwrap(); diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index 8ac80ea0..bf995af5 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -238,6 +238,13 @@ impl NetworkManager { /// for retry, so a later retry still owns the index-derived resources /// (`veth-`) it tears down and can never delete networking that was /// recreated for a different sandbox. + /// + /// The caller must hold exclusive ownership of the slot: every path that + /// reaches here consumes the `Slot` by value (release and dead-entry + /// cleanup alike), so the `cleanup_armed` flag inside `Slot::cleanup` + /// only provides idempotency against a repeated teardown by the same + /// owner — it is not a guard against concurrent owners, which the type's + /// ownership rules already exclude. pub(crate) fn cleanup_allocated_slot_retain_bit_on_failure( &self, slot: &Slot, diff --git a/src/sandbox/network/slot.rs b/src/sandbox/network/slot.rs index e77fb91f..5a2869f8 100644 --- a/src/sandbox/network/slot.rs +++ b/src/sandbox/network/slot.rs @@ -718,7 +718,12 @@ impl Slot { const IP_LINK_DEL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); let mut child = Command::new("ip") .args(["link", "del", &veth_name]) - .stdout(std::process::Stdio::piped()) + // The polling loop below never drains the pipes, so a + // child that filled its stdout buffer would block forever + // on write. stdout carries nothing useful here; stderr + // stays piped for the error message (it is a single + // short line). + .stdout(std::process::Stdio::null()) .stderr(std::process::Stdio::piped()) .spawn() .context("Failed to spawn ip link del")?; @@ -741,16 +746,28 @@ impl Slot { // hang this shutdown path beyond the grace period. let reap_deadline = std::time::Instant::now() + std::time::Duration::from_secs(1); + let mut reaped = false; while std::time::Instant::now() < reap_deadline { if child .try_wait() .context("Failed to reap timed-out ip link del")? .is_some() { + reaped = true; break; } std::thread::sleep(std::time::Duration::from_millis(20)); } + if !reaped { + // A child stuck in uninterruptible sleep must not + // be dropped without `wait`: dropping `Child` does + // not reap it and would leak a zombie. Hand it to + // a detached reaper thread that only blocks on + // `wait` and owns no pool state. + std::thread::spawn(move || { + let _ = child.wait(); + }); + } return Err(anyhow!( "ip link del {} timed out after {:?}", veth_name, From ddf73770920842b664bb58b37f5541c496360dcf Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 14:45:23 -0300 Subject: [PATCH 8/8] fix(pool): type dead-entry teardown failures and harden ip link del fallback --- crates/warm-pool/src/lib.rs | 43 ++++++ src/sandbox/firecracker/pool.rs | 237 ++++++++++++++++++++++---------- src/sandbox/network/manager.rs | 32 ++++- src/sandbox/network/mod.rs | 2 +- src/sandbox/network/slot.rs | 111 ++++++++++++--- 5 files changed, 325 insertions(+), 100 deletions(-) diff --git a/crates/warm-pool/src/lib.rs b/crates/warm-pool/src/lib.rs index 107aeb27..757a4bd3 100644 --- a/crates/warm-pool/src/lib.rs +++ b/crates/warm-pool/src/lib.rs @@ -397,6 +397,14 @@ impl WarmPool { break; } signal.pending = false; + // This cycle also consumes any already-due scheduled wake: + // an expired instant left behind would survive to a later + // `request_maintenance_after`, whose earliest-wins `min` + // would then fire an immediate extra cycle and defeat the + // new backoff. + if signal.wake_at.is_some_and(|wake_at| wake_at <= Instant::now()) { + signal.wake_at = None; + } } if self.is_shutting_down() { @@ -610,6 +618,41 @@ mod tests { pool.drain_all(); } + #[test] + fn pending_cycle_clears_expired_scheduled_wake() { + let pool: &'static WarmPool = Box::leak(Box::new(WarmPool::new(PoolConfig { + low_watermark: 0, + high_watermark: 10, + maintenance_enabled: true, + startup_prewarm: false, + }))); + let cycles: &'static std::sync::atomic::AtomicUsize = + Box::leak(Box::new(std::sync::atomic::AtomicUsize::new(0))); + pool.start_maintenance_worker(move || { + cycles.fetch_add(1, Ordering::Relaxed); + }); + + // Wait for the startup cycle, then leave an already-expired wake + // behind and trigger an immediate cycle: the cycle must consume the + // stale instant instead of letting a later earliest-wins `min` + // resurrect it as an immediate extra cycle. + let deadline = Instant::now() + Duration::from_secs(5); + while cycles.load(Ordering::Relaxed) < 1 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + pool.maintenance_signal.lock().unwrap().wake_at = + Some(Instant::now() - Duration::from_millis(1)); + pool.request_maintenance(); + let deadline = Instant::now() + Duration::from_secs(5); + while cycles.load(Ordering::Relaxed) < 2 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(10)); + } + assert!(cycles.load(Ordering::Relaxed) >= 2); + assert_eq!(pool.maintenance_signal.lock().unwrap().wake_at, None); + + pool.drain_all(); + } + #[test] fn compute_maintenance_action_drains_above_high() { let pool = WarmPool::::new(PoolConfig { diff --git a/src/sandbox/firecracker/pool.rs b/src/sandbox/firecracker/pool.rs index 39d7c0e1..bf123265 100644 --- a/src/sandbox/firecracker/pool.rs +++ b/src/sandbox/firecracker/pool.rs @@ -19,7 +19,7 @@ use warm_pool::{PoolMaintenanceAction, WarmPool}; use super::config::create_firecracker_work_dir; use super::FirecrackerInstance; use crate::cfg::{ConfigManager, ResolvedFirecrackerPoolConfig}; -use crate::sandbox::network::{NetworkManager, Slot}; +use crate::sandbox::network::{NetworkManager, Slot, SlotTeardownError}; const POOL_FIRECRACKER_STOP_TIMEOUT: Duration = Duration::from_secs(2); const POOL_PRIME_POLL_INTERVAL: Duration = Duration::from_millis(20); @@ -123,16 +123,20 @@ impl DeadWarmEntry { /// Tear down the entry. The process is known dead, so skip the /// graceful-stop path: dropping the instance best-effort kills any - /// residual handle before the network slot is released. On failure the - /// entry is returned (with an updated backoff) so the caller can retain - /// it for retry instead of losing track of stale host network state. The - /// allocation bit is released only after teardown succeeds + /// residual handle before the network slot is released. On a teardown + /// failure the entry is returned (with an updated backoff) so the caller + /// can retain it for retry instead of losing track of stale host network + /// state. The allocation bit is released only after teardown succeeds /// (`cleanup_allocated_slot_retain_bit_on_failure`), so a retry can never - /// race a reallocation of the same index. + /// race a reallocation of the same index. A bit-release failure after a + /// successful teardown is terminal (`DeadWarmCleanupOutcome::Terminal`): + /// the slot is already disarmed and the bit already clear, so retaining + /// and retrying could release a bit that meanwhile belongs to a + /// different sandbox. fn attempt_cleanup( self, network: &NetworkManager, - ) -> std::result::Result<(), DeadWarmCleanupError> { + ) -> std::result::Result { let slot_idx = self.slot_idx(); let failed_attempts = self.failed_attempts; let slot = match self.inner { @@ -149,26 +153,50 @@ impl DeadWarmEntry { DeadWarmEntryInner::SlotOnly(slot) => slot, }; - let result = network.cleanup_allocated_slot_retain_bit_on_failure(&slot, false); - result.map_err(|error| { - // Saturate: a permanently failing slot is retried indefinitely, - // and an overflowing counter would panic the cleanup worker in - // debug builds. The backoff saturates at the cap anyway. - let failed_attempts = failed_attempts.saturating_add(1); - DeadWarmCleanupError { - entry: Box::new(DeadWarmEntry { - inner: DeadWarmEntryInner::SlotOnly(slot), - failed_attempts, - not_before: Instant::now() + dead_warm_cleanup_backoff(failed_attempts), - }), - error: error.context(format!( - "firecracker pool: cleanup dead warm network slot {slot_idx}" - )), + match network.cleanup_allocated_slot_retain_bit_on_failure(&slot, false) { + Ok(()) => Ok(DeadWarmCleanupOutcome::Clean), + Err(SlotTeardownError::BitRelease(error)) => { + // Terminal: teardown already disarmed the slot and the bit is + // already clear, so the index is reallocatable right away. A + // retry would be a no-op cleanup followed by + // `release_slot_bit` against an index that can now belong to + // a different sandbox — report instead of retaining. + Ok(DeadWarmCleanupOutcome::Terminal(format!( + "firecracker pool: cleanup dead warm network slot {slot_idx}: teardown completed but releasing the allocation bit failed: {error:#}" + ))) } - }) + Err(SlotTeardownError::Teardown(error)) => { + // Saturate: a permanently failing slot is retried + // indefinitely, and an overflowing counter would panic the + // cleanup worker in debug builds. The backoff saturates at + // the cap anyway. + let failed_attempts = failed_attempts.saturating_add(1); + Err(DeadWarmCleanupError { + entry: Box::new(DeadWarmEntry { + inner: DeadWarmEntryInner::SlotOnly(slot), + failed_attempts, + not_before: Instant::now() + dead_warm_cleanup_backoff(failed_attempts), + }), + error: error.context(format!( + "firecracker pool: cleanup dead warm network slot {slot_idx}" + )), + }) + } + } } } +/// Outcome of a dead-entry teardown attempt that is not retained for retry. +#[derive(Debug)] +enum DeadWarmCleanupOutcome { + /// Teardown and bit release both succeeded. + Clean, + /// Teardown completed but releasing the allocation bit failed. Terminal, + /// not retryable (see `SlotTeardownError::BitRelease`); the message must + /// be surfaced in the caller's failure reporting. + Terminal(String), +} + /// Error from dead-entry teardown. Keeps the entry so the caller can retain /// it for retry. The entry is boxed so the `Err` variant stays small /// (clippy::result_large_err): `DeadWarmEntry` can hold a full warm @@ -203,11 +231,20 @@ fn finalize_closed_dead_entry( ) -> Option { let slot_idx = entry.slot_idx(); match entry.attempt_cleanup(network) { - Ok(()) => { + Ok(DeadWarmCleanupOutcome::Clean) => { // Teardown completed; the allocation bit was released by // `attempt_cleanup`. Report only a prior failure, if any. prior_error.map(|error| error.to_string()) } + Ok(DeadWarmCleanupOutcome::Terminal(message)) => { + // Teardown finished; only the bitmap release failed, which is + // terminal (not retryable). Surface it alongside any prior + // failure. + Some(match prior_error { + Some(prior) => format!("{prior:#}; {message}"), + None => message, + }) + } Err(err) => { let mut message = err.error.to_string(); if let Some(prior) = prior_error { @@ -414,18 +451,28 @@ impl FirecrackerPool { // (dropping the entry would only re-run `Slot::cleanup` via // `Slot::drop` and leak the bit). drop(queue); - if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - if let Some(message) = finalize_closed_dead_entry( - NetworkManager::global(), - *err.entry, - Some(err.error), - ) { + match entry.attempt_cleanup(NetworkManager::global()) { + Ok(DeadWarmCleanupOutcome::Clean) => {} + Ok(DeadWarmCleanupOutcome::Terminal(message)) => { warn!( slot = slot_idx, error = %message, - "firecracker pool: cleanup of dead warm entry failed during shutdown" + "firecracker pool: terminal cleanup failure of dead warm entry during shutdown" ); } + Err(err) => { + if let Some(message) = finalize_closed_dead_entry( + NetworkManager::global(), + *err.entry, + Some(err.error), + ) { + warn!( + slot = slot_idx, + error = %message, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + } + } } return; } @@ -543,35 +590,47 @@ impl FirecrackerPool { } } }; - if let Err(err) = next.attempt_cleanup(network) { - warn!( - slot = err.entry.slot_idx(), - error = %err.error, - "firecracker pool: cleanup of dead warm entry failed; retaining for retry" - ); - let mut queue = lock_dead_queue(&shared); - if queue.closed { - // Shutdown closed the queue while this teardown was in - // flight; the entry cannot be requeued. Finish it on this - // worker thread and release the allocation bit explicitly - // on failure: dropping the entry would only re-run - // `Slot::cleanup` via `Slot::drop`, leaking the bit, and - // the failure would never reach shutdown's result. - let DeadWarmCleanupError { entry, error } = err; - let slot_idx = entry.slot_idx(); - drop(queue); - if let Some(message) = - finalize_closed_dead_entry(network, *entry, Some(error)) - { - warn!( - slot = slot_idx, - error = %message, - "firecracker pool: finalized dead warm entry after queue close" - ); - lock_dead_queue(&shared).failures.push(message); + match next.attempt_cleanup(network) { + Ok(DeadWarmCleanupOutcome::Clean) => {} + Ok(DeadWarmCleanupOutcome::Terminal(message)) => { + // Terminal (bit release after a successful teardown): + // not retryable; record it so shutdown can report it. + warn!( + error = %message, + "firecracker pool: terminal cleanup failure of dead warm entry" + ); + lock_dead_queue(&shared).failures.push(message); + } + Err(err) => { + warn!( + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed; retaining for retry" + ); + let mut queue = lock_dead_queue(&shared); + if queue.closed { + // Shutdown closed the queue while this teardown was in + // flight; the entry cannot be requeued. Finish it on this + // worker thread and release the allocation bit explicitly + // on failure: dropping the entry would only re-run + // `Slot::cleanup` via `Slot::drop`, leaking the bit, and + // the failure would never reach shutdown's result. + let DeadWarmCleanupError { entry, error } = err; + let slot_idx = entry.slot_idx(); + drop(queue); + if let Some(message) = + finalize_closed_dead_entry(network, *entry, Some(error)) + { + warn!( + slot = slot_idx, + error = %message, + "firecracker pool: finalized dead warm entry after queue close" + ); + lock_dead_queue(&shared).failures.push(message); + } + } else { + queue.entries.push(*err.entry); } - } else { - queue.entries.push(*err.entry); } } } @@ -597,13 +656,21 @@ impl FirecrackerPool { let mut failures = Vec::new(); let mut failed: Vec<(DeadWarmEntry, anyhow::Error)> = Vec::new(); for entry in due { - if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - warn!( - slot = err.entry.slot_idx(), - error = %err.error, - "firecracker pool: cleanup of dead warm entry failed; retaining for retry" - ); - failed.push((*err.entry, err.error)); + match entry.attempt_cleanup(NetworkManager::global()) { + Ok(DeadWarmCleanupOutcome::Clean) => {} + Ok(DeadWarmCleanupOutcome::Terminal(message)) => { + // Terminal (bit release after a successful teardown): + // report it, but do not retain the entry for retry. + failures.push(message); + } + Err(err) => { + warn!( + slot = err.entry.slot_idx(), + error = %err.error, + "firecracker pool: cleanup of dead warm entry failed; retaining for retry" + ); + failed.push((*err.entry, err.error)); + } } } if !backoff.is_empty() || !failed.is_empty() { @@ -727,18 +794,30 @@ impl FirecrackerPool { fn shutdown_dead_entries_blocking(&self) -> Vec { let (dead, mut failures) = self.close_dead_queue(); for entry in dead { - if let Err(err) = entry.attempt_cleanup(NetworkManager::global()) { - let slot = err.entry.slot_idx(); - if let Some(message) = - finalize_closed_dead_entry(NetworkManager::global(), *err.entry, Some(err.error)) - { + match entry.attempt_cleanup(NetworkManager::global()) { + Ok(DeadWarmCleanupOutcome::Clean) => {} + Ok(DeadWarmCleanupOutcome::Terminal(message)) => { warn!( - slot, error = %message, - "firecracker pool: cleanup of dead warm entry failed during shutdown" + "firecracker pool: terminal cleanup failure of dead warm entry during shutdown" ); failures.push(message); } + Err(err) => { + let slot = err.entry.slot_idx(); + if let Some(message) = finalize_closed_dead_entry( + NetworkManager::global(), + *err.entry, + Some(err.error), + ) { + warn!( + slot, + error = %message, + "firecracker pool: cleanup of dead warm entry failed during shutdown" + ); + failures.push(message); + } + } } } failures @@ -1126,11 +1205,19 @@ mod tests { .fail_slot_teardown .store(true, std::sync::atomic::Ordering::Relaxed); + let before = Instant::now(); let err = slot_only_entry(slot).attempt_cleanup(&manager).unwrap_err(); assert_eq!(err.entry.failed_attempts, 1); - assert!(err.entry.due(Instant::now() + DEAD_WARM_CLEANUP_RETRY_MAX)); - assert!(!err.entry.due(Instant::now())); + // Validate the retry delay on both sides against the reference + // instant captured before the attempt: not immediate, not a very + // short delay (rules out a zero/incorrect backoff), and within the + // expected first backoff plus slack for slow test execution. + let first_backoff = dead_warm_cleanup_backoff(1); + assert!(!err.entry.due(before)); + assert!(!err.entry.due(before + Duration::from_millis(10))); + assert!(err.entry.due(before + first_backoff * 2)); + assert!(err.entry.due(before + DEAD_WARM_CLEANUP_RETRY_MAX * 2)); assert!(matches!(err.entry.inner, DeadWarmEntryInner::SlotOnly(_))); // The production invariant: the bit is still held, so the index // cannot be reallocated to a live sandbox while the entry is diff --git a/src/sandbox/network/manager.rs b/src/sandbox/network/manager.rs index bf995af5..8b1985ba 100644 --- a/src/sandbox/network/manager.rs +++ b/src/sandbox/network/manager.rs @@ -78,6 +78,24 @@ pub(crate) struct NetworkManager { pub(crate) fail_slot_teardown: AtomicBool, } +/// Failure of `cleanup_allocated_slot_retain_bit_on_failure`, typed so the +/// caller can tell a retryable teardown failure from a terminal +/// post-teardown bitmap error. +pub(crate) enum SlotTeardownError { + /// `Slot::cleanup` failed: the allocation bit is still held, so the + /// caller must retain the slot and retry — the index cannot be + /// reallocated in between, and the retry still owns the index-derived + /// resources (`veth-`) it tears down. + Teardown(anyhow::Error), + /// Teardown succeeded but releasing the allocation bit failed: the slot + /// is already disarmed and the bit is already clear, so the index may be + /// reallocated immediately. Retrying would be a no-op cleanup followed + /// by `release_slot_bit` against an index that can now belong to a + /// different sandbox, so this failure is terminal: report it, never + /// retain the slot for retry. + BitRelease(anyhow::Error), +} + impl NetworkManager { /// Global network manager for slot allocation across all sandboxes. pub fn global() -> &'static Self { @@ -245,18 +263,26 @@ impl NetworkManager { /// only provides idempotency against a repeated teardown by the same /// owner — it is not a guard against concurrent owners, which the type's /// ownership rules already exclude. + /// + /// The error is typed (`SlotTeardownError`) so callers can distinguish a + /// retryable teardown failure (bit still held) from a terminal + /// post-teardown bitmap error (slot already disarmed, bit already + /// clear), which must be reported but never retained for retry. pub(crate) fn cleanup_allocated_slot_retain_bit_on_failure( &self, slot: &Slot, sync_cleanup: bool, - ) -> Result<()> { + ) -> std::result::Result<(), SlotTeardownError> { #[cfg(test)] if self.fail_slot_teardown.swap(false, Ordering::AcqRel) { - return Err(anyhow!("injected slot teardown failure")); + return Err(SlotTeardownError::Teardown(anyhow!( + "injected slot teardown failure" + ))); } slot.cleanup(sync_cleanup) - .map_err(Into::::into)?; + .map_err(|e| SlotTeardownError::Teardown(Into::::into(e)))?; self.release_slot_bit(slot.idx) + .map_err(SlotTeardownError::BitRelease) } /// Find and allocate the next available slot. diff --git a/src/sandbox/network/mod.rs b/src/sandbox/network/mod.rs index a2e183ef..6d34869a 100644 --- a/src/sandbox/network/mod.rs +++ b/src/sandbox/network/mod.rs @@ -9,7 +9,7 @@ use std::path::Path; use anyhow::Context; pub(crate) use address_plan::NetworkAddressPlan; -pub(crate) use manager::NetworkManager; +pub(crate) use manager::{NetworkManager, SlotTeardownError}; pub use policy::{BaseSandboxNetworkPolicy, SandboxNetworkEgressPolicy, SandboxNetworkPolicy}; pub(crate) use slot::Slot; diff --git a/src/sandbox/network/slot.rs b/src/sandbox/network/slot.rs index 5a2869f8..9ad25cb9 100644 --- a/src/sandbox/network/slot.rs +++ b/src/sandbox/network/slot.rs @@ -4,7 +4,7 @@ use std::os::fd::{AsFd, AsRawFd, BorrowedFd, OwnedFd}; use std::path::PathBuf; use std::process::Command; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::OnceLock; +use std::sync::{Condvar, Mutex, Once, OnceLock}; use std::thread; use std::time::Duration; @@ -35,6 +35,63 @@ use super::{NetworkAddressPlan, NetworkError, HOST_VETH_PREFIX, MAX_SLOTS, NETNS /// All subsequent slot creations move host-side interfaces back to this namespace. static HOST_NS_FD: OnceLock = OnceLock::new(); +/// Children that could not be reaped within the grace period after `kill`. +/// +/// A single process-lifetime reaper thread polls them with `try_wait`, so a +/// persistently stuck `ip` child never accumulates one blocked thread per +/// timeout, and dropping a `Child` (which never reaps) cannot leak a zombie. +static UNREAPED_CHILDREN: Mutex> = Mutex::new(Vec::new()); +static REAPER_CV: Condvar = Condvar::new(); +static REAPER_ONCE: Once = Once::new(); + +/// Hand a killed-but-unreapable child to the centralized reaper thread. +/// +/// The reaper is the only thread that ever blocks on these children; it owns +/// no pool or slot state, and there is at most one reaper for the process +/// lifetime regardless of how many teardown attempts time out. +fn hand_off_unreaped_child(child: std::process::Child) { + REAPER_ONCE.call_once(|| { + let spawned = std::thread::Builder::new() + .name("network-slot-child-reaper".to_string()) + .spawn(|| { + let mut pending: Vec = Vec::new(); + loop { + if pending.is_empty() { + let mut guard = UNREAPED_CHILDREN + .lock() + .unwrap_or_else(|err| err.into_inner()); + while guard.is_empty() { + guard = REAPER_CV.wait(guard).unwrap_or_else(|err| err.into_inner()); + } + pending.append(&mut guard); + drop(guard); + } + // `try_wait` reaps exited children; errored polls are + // retried on the next pass. + pending.retain_mut(|child| !matches!(child.try_wait(), Ok(Some(_)))); + if pending.is_empty() { + continue; + } + thread::sleep(Duration::from_millis(20)); + // Pick up children handed off while polling. + let mut guard = UNREAPED_CHILDREN + .lock() + .unwrap_or_else(|err| err.into_inner()); + pending.append(&mut guard); + drop(guard); + } + }); + if let Err(err) = spawned { + warn!(error = %err, "failed to spawn child reaper thread"); + } + }); + UNREAPED_CHILDREN + .lock() + .unwrap_or_else(|err| err.into_inner()) + .push(child); + REAPER_CV.notify_one(); +} + const ARP_RETRANS_TIME_MS: &str = "100"; const NEIGH_SYSCTL_RETRIES: usize = 5; const NEIGH_SYSCTL_RETRY_DELAY_MS: u64 = 20; @@ -708,7 +765,7 @@ impl Slot { /// Tokio context may already be unavailable. fn delete_host_veth_interface_sync(idx: u32) -> Result<()> { let veth_name = Self::host_veth_name(idx); - let output = crate::privileges::run_with_scoped_capabilities( + let (status, stderr_bytes) = crate::privileges::run_with_scoped_capabilities( &[crate::privileges::CAP_NET_ADMIN], || { // Hard timeout: this runs on shutdown/exit cleanup paths that @@ -716,27 +773,31 @@ impl Slot { // entry cleanup worker), so a stuck `ip link del` must not // block the caller forever. const IP_LINK_DEL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + // stderr goes to a temp file, not a pipe: the polling loop + // below never drains pipes, so a child whose diagnostics fill + // a pipe buffer would block on write and be misclassified as + // a timeout. A file is an unbounded sink, so diagnostics can + // never stall the child. + let stderr_path = std::env::temp_dir().join(format!( + "agentenv-ip-link-del-{}-{veth_name}.stderr", + std::process::id() + )); + let stderr_file = File::create(&stderr_path).with_context(|| { + format!("Failed to create {}", stderr_path.display()) + })?; let mut child = Command::new("ip") .args(["link", "del", &veth_name]) - // The polling loop below never drains the pipes, so a - // child that filled its stdout buffer would block forever - // on write. stdout carries nothing useful here; stderr - // stays piped for the error message (it is a single - // short line). .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::piped()) + .stderr(std::process::Stdio::from(stderr_file)) .spawn() .context("Failed to spawn ip link del")?; let deadline = std::time::Instant::now() + IP_LINK_DEL_TIMEOUT; - loop { - if child + let status = loop { + if let Some(status) = child .try_wait() .context("Failed to poll ip link del")? - .is_some() { - return child - .wait_with_output() - .context("Failed to collect ip link del output"); + break status; } if std::time::Instant::now() >= deadline { child.kill().context("Failed to kill timed-out ip link del")?; @@ -762,12 +823,12 @@ impl Slot { // A child stuck in uninterruptible sleep must not // be dropped without `wait`: dropping `Child` does // not reap it and would leak a zombie. Hand it to - // a detached reaper thread that only blocks on - // `wait` and owns no pool state. - std::thread::spawn(move || { - let _ = child.wait(); - }); + // the centralized reaper (a single thread for the + // process lifetime) instead of spawning an + // unbounded thread per timed-out attempt. + hand_off_unreaped_child(child); } + let _ = fs::remove_file(&stderr_path); return Err(anyhow!( "ip link del {} timed out after {:?}", veth_name, @@ -775,15 +836,23 @@ impl Slot { )); } std::thread::sleep(std::time::Duration::from_millis(20)); + }; + // Bounded read of the diagnostics captured in the temp file. + let mut stderr_bytes = Vec::new(); + if let Ok(file) = File::open(&stderr_path) { + let mut limited = std::io::Read::take(file, 64 * 1024); + let _ = std::io::Read::read_to_end(&mut limited, &mut stderr_bytes); } + let _ = fs::remove_file(&stderr_path); + Ok((status, stderr_bytes)) }, )?; - if output.status.success() { + if status.success() { return Ok(()); } - let stderr = String::from_utf8_lossy(&output.stderr); + let stderr = String::from_utf8_lossy(&stderr_bytes); let stderr_lower = stderr.to_lowercase(); if stderr_lower.contains("cannot find device") || stderr_lower.contains("no such device")