From da5215bd317e6d3516b2922dc872640618d7d303 Mon Sep 17 00:00:00 2001 From: epicvinny Date: Tue, 18 Aug 2026 13:49:50 -0300 Subject: [PATCH] fix(server): reap orphaned runtime processes and retry RocksDB lock on startup After a hard kill (SIGKILL, OOM, force-deleted pod), firecracker and uvm-ublk-daemon children spawned with process_group(0) can outlive the server while still holding ublk devices, open image files, and inherited fds that pin the RocksDB LOCK of the persisted-sandbox store. The next server then fails to open records.db and crash-loops. Observed in staging: recovery required kubectl debug node + manually killing the leaked PIDs. Two changes make startup resilient: - src/setup/orphans.rs (wired in src/bin/server.rs before any subsystem spawns): scan /proc for processes whose executable is the configured firecracker or uvm-ublk-daemon binary - necessarily leftovers from a previous incarnation - and terminate them (SIGTERM, 5s grace, SIGKILL). Zombies are ignored. - src/local_store.rs: LocalKvStore::open now retries lock-contention errors with exponential backoff for up to 120s, covering lock holders in a different PID namespace (e.g. a previous container still terminating on the same hostPath) that the reaper cannot see. --- src/bin/server.rs | 6 + src/local_store.rs | 144 +++++++++++++++++++++-- src/setup/mod.rs | 8 ++ src/setup/orphans.rs | 264 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 414 insertions(+), 8 deletions(-) create mode 100644 src/setup/orphans.rs diff --git a/src/bin/server.rs b/src/bin/server.rs index 479d038c2..47962b89f 100644 --- a/src/bin/server.rs +++ b/src/bin/server.rs @@ -79,6 +79,12 @@ async fn main() -> anyhow::Result<()> { agentenv::privileges::require_runtime_capabilities()?; agentenv::privileges::clear_ambient_capabilities()?; + // Terminate firecracker/uvm-ublk-daemon processes orphaned by a previous + // server incarnation before they can wedge startup: they hold ublk devices + // and inherited fds that pin the RocksDB LOCK of the persisted-sandbox + // store. + agentenv::setup::reap_orphaned_runtime_processes(config).await?; + let addr = std::env::var("API_ADDR").unwrap_or_else(|_| "0.0.0.0:8000".to_string()); let identity = NodeIdentity::from_config(&config.node_identity); let p2p_transport = agentenv::p2p::transport_from_config(config, &identity).await?; diff --git a/src/local_store.rs b/src/local_store.rs index 8a7ec1385..b6c4529aa 100644 --- a/src/local_store.rs +++ b/src/local_store.rs @@ -1,8 +1,19 @@ use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use anyhow::Context; use rocksdb::{Direction, IteratorMode, Options, WriteBatch, WriteOptions, DB}; +use tokio::time::Instant; +use tracing::warn; + +/// Total time budget for retrying a RocksDB open that fails because the +/// database lock is still held by a leftover process from a previous server +/// incarnation (for example, an old container that has not finished +/// terminating while a replacement pod already started on the same hostPath). +/// Without this, a transient overlap turns into a CrashLoopBackOff even though +/// the lock is released as soon as the previous process exits. +const LOCK_RETRY_BUDGET: Duration = Duration::from_secs(120); /// Durability policy for writes made through [`LocalKvStore`]. /// @@ -80,23 +91,58 @@ impl LocalKvStore { /// /// The parent directory is created automatically. The database itself uses /// RocksDB's default column family and stores opaque byte keys and values. + /// + /// If the database lock is still held by a leftover process from a + /// previous server incarnation, the open is retried with exponential + /// backoff for up to [`LOCK_RETRY_BUDGET`] before failing. pub async fn open( path: impl Into, durability: LocalStoreDurability, ) -> anyhow::Result { - let path = path.into(); + Self::open_with_lock_retry_budget(path.into(), durability, LOCK_RETRY_BUDGET).await + } + + async fn open_with_lock_retry_budget( + path: PathBuf, + durability: LocalStoreDurability, + lock_retry_budget: Duration, + ) -> anyhow::Result { if let Some(parent) = path.parent() { tokio::fs::create_dir_all(parent) .await .with_context(|| format!("create RocksDB parent dir {}", parent.display()))?; } - let db = tokio::task::spawn_blocking(move || { - let mut options = Options::default(); - options.create_if_missing(true); - DB::open(&options, &path).with_context(|| format!("open RocksDB {}", path.display())) - }) - .await - .context("join RocksDB open task")??; + + let started = Instant::now(); + let mut backoff = Duration::from_millis(500); + let db = loop { + let attempt_path = path.clone(); + let result = tokio::task::spawn_blocking(move || { + let mut options = Options::default(); + options.create_if_missing(true); + DB::open(&options, &attempt_path) + .with_context(|| format!("open RocksDB {}", attempt_path.display())) + }) + .await + .context("join RocksDB open task")?; + + match result { + Ok(db) => break db, + Err(err) => { + let elapsed = started.elapsed(); + if !is_lock_contention(&err) || elapsed >= lock_retry_budget { + return Err(err); + } + warn!( + path = %path.display(), + error = %err, + "RocksDB lock still held by another process; retrying" + ); + tokio::time::sleep(backoff.min(lock_retry_budget - elapsed)).await; + backoff = (backoff * 2).min(Duration::from_secs(10)); + } + } + }; Ok(Self { db: Arc::new(db), @@ -236,3 +282,85 @@ impl LocalKvStore { .context("join RocksDB prefix scan task")? } } + +/// Returns `true` when `err` looks like RocksDB failing to acquire the +/// database `LOCK` file because another process still holds it. RocksDB +/// surfaces this as an IO error such as +/// `lock hold by current process ... /LOCK: No locks available` (same host) +/// or `While lock file: /LOCK: Resource temporarily unavailable`. +fn is_lock_contention(err: &anyhow::Error) -> bool { + err.chain().any(|cause| { + let message = cause.to_string().to_ascii_lowercase(); + message.contains("lock") + && (message.contains("no locks available") + || message.contains("resource temporarily unavailable")) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + #[test] + fn lock_contention_detection_matches_rocksdb_lock_errors() { + let lock_err = anyhow::anyhow!( + "IO error: While lock file: /var/lib/aenv/env/persisted-sandboxes/records.db/LOCK: Resource temporarily unavailable" + ); + assert!(is_lock_contention(&lock_err)); + + let same_host_err = anyhow::anyhow!( + "IO error: lock hold by current process, acquire time 1700000000 acquiring thread 42: /var/lib/aenv/env/persisted-sandboxes/records.db/LOCK: No locks available" + ); + assert!(is_lock_contention(&same_host_err)); + + let other_err = anyhow::anyhow!("IO error: No such file or directory"); + assert!(!is_lock_contention(&other_err)); + } + + #[tokio::test] + async fn open_retries_until_lock_is_released() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let db_path = temp.path().join("records.db"); + let holder = LocalKvStore::open(&db_path, LocalStoreDurability::Memory).await?; + + let release = tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(300)).await; + drop(holder); + }); + let reopened = LocalKvStore::open_with_lock_retry_budget( + db_path, + LocalStoreDurability::Memory, + Duration::from_secs(30), + ) + .await?; + release.await?; + + reopened.put(b"key".to_vec(), b"value".to_vec()).await?; + assert_eq!( + reopened.get(b"key".to_vec()).await?, + Some(b"value".to_vec()) + ); + Ok(()) + } + + #[tokio::test] + async fn open_fails_after_lock_retry_budget_is_exhausted() -> anyhow::Result<()> { + let temp = TempDir::new()?; + let db_path = temp.path().join("records.db"); + let _holder = LocalKvStore::open(&db_path, LocalStoreDurability::Memory).await?; + + let started = Instant::now(); + let err = LocalKvStore::open_with_lock_retry_budget( + db_path, + LocalStoreDurability::Memory, + Duration::from_millis(1200), + ) + .await + .expect_err("open should keep failing while the lock is held"); + + assert!(is_lock_contention(&err)); + assert!(started.elapsed() >= Duration::from_millis(1200)); + Ok(()) + } +} diff --git a/src/setup/mod.rs b/src/setup/mod.rs index ac79dfe4e..4bbaeb9c2 100644 --- a/src/setup/mod.rs +++ b/src/setup/mod.rs @@ -1,6 +1,7 @@ mod deps; mod kvm; mod network_capacity; +mod orphans; pub mod overlaybd; mod packages; mod ublk; @@ -28,6 +29,13 @@ pub async fn ensure_provisioning(config: &AppConfig) -> Result<()> { deps::write_generated_overlaybd_global_configs(config, None) } +/// Terminate leftover `firecracker`/`uvm-ublk-daemon` processes orphaned by a +/// previous server incarnation before they can wedge startup (ublk socket +/// conflicts, RocksDB `LOCK` contention on the persisted-sandbox store). +pub async fn reap_orphaned_runtime_processes(config: &AppConfig) -> Result<()> { + orphans::reap(config).await +} + /// Provision machine-wide prerequisites for the configured runtime account. /// This mode is intentionally separate from `--setup-only`, which is also used /// while assembling release artifacts on unprivileged builders. diff --git a/src/setup/orphans.rs b/src/setup/orphans.rs new file mode 100644 index 000000000..00c044afd --- /dev/null +++ b/src/setup/orphans.rs @@ -0,0 +1,264 @@ +//! Reaping of orphaned runtime processes left by a previous server incarnation. +//! +//! The node agent spawns `firecracker` and `uvm-ublk-daemon` as detached child +//! processes (each in its own process group). When the agent is hard-killed — +//! SIGKILL from the container runtime, an OOM kill, a node reboot — those +//! children can outlive it while still holding sandbox resources: ublk devices, +//! network namespaces, open image files, and inherited file descriptors that +//! pin the RocksDB `LOCK` of the persisted-sandbox store. The next server +//! process then fails to spawn its daemon or open its record database and +//! crash-loops. +//! +//! [`reap`] runs early in server startup, before the ublk daemon is spawned +//! and before the persisted-sandbox store is opened, and terminates any process +//! in this PID namespace whose executable is one of the configured runtime +//! binaries. Any match necessarily belongs to a previous incarnation: the +//! current process has not launched its own children yet. + +use std::collections::HashSet; +use std::fs; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use anyhow::Result; +use nix::sys::signal::{kill, Signal}; +use nix::unistd::Pid; +use tokio::time::Instant; +use tracing::{debug, info, warn}; + +use crate::cfg::AppConfig; + +/// How long to wait for SIGTERM to take effect before escalating to SIGKILL. +const TERM_GRACE: Duration = Duration::from_secs(5); +/// Extra wait for SIGKILLed processes to disappear from `/proc`. +const KILL_GRACE: Duration = Duration::from_secs(2); +/// `readlink` on `/proc//exe` reports the original path with this suffix +/// when the binary was replaced on disk after the process started. +const DELETED_SUFFIX: &str = " (deleted)"; + +/// Terminate leftover `firecracker`/`uvm-ublk-daemon` processes from a previous +/// server incarnation. Best-effort: per-process failures are logged, not +/// propagated, so an uncooperative orphan never blocks startup. +pub async fn reap(config: &AppConfig) -> Result<()> { + let mut targets = HashSet::new(); + targets.insert(config.resolved_firecracker_binary_path()); + if let Some(daemon_binary) = config + .ublk + .daemon_binary_path + .clone() + .or_else(|| which::which("uvm-ublk-daemon").ok()) + { + targets.insert(daemon_binary); + } + + let mut orphans = find_processes_by_exe(Path::new("/proc"), &targets); + if orphans.is_empty() { + return Ok(()); + } + orphans.sort_unstable(); + + warn!( + count = orphans.len(), + ?orphans, + "found orphaned runtime processes from a previous incarnation; terminating them" + ); + for pid in &orphans { + signal(*pid, Signal::SIGTERM); + } + let survivors = wait_for_exit(&orphans, TERM_GRACE).await; + if !survivors.is_empty() { + warn!( + ?survivors, + "orphaned runtime processes ignored SIGTERM; sending SIGKILL" + ); + for pid in &survivors { + signal(*pid, Signal::SIGKILL); + } + let stuck = wait_for_exit(&survivors, KILL_GRACE).await; + if !stuck.is_empty() { + warn!( + ?stuck, + "orphaned runtime processes still present after SIGKILL (likely uninterruptible sleep)" + ); + } + } + + info!(count = orphans.len(), "orphaned runtime processes reaped"); + Ok(()) +} + +fn signal(pid: u32, sig: Signal) { + let raw = match i32::try_from(pid) { + Ok(raw) => raw, + Err(err) => { + debug!(pid, error = %err, "orphaned process pid does not fit in i32"); + return; + } + }; + if let Err(err) = kill(Pid::from_raw(raw), sig) { + debug!(pid, %sig, error = %err, "failed to signal orphaned process"); + } +} + +/// Wait until none of `pids` is a running process anymore, or the grace period +/// elapses. Returns the pids still running. +async fn wait_for_exit(pids: &[u32], grace: Duration) -> Vec { + let deadline = Instant::now() + grace; + loop { + let survivors: Vec = pids + .iter() + .copied() + .filter(|pid| process_running(*pid)) + .collect(); + if survivors.is_empty() || Instant::now() >= deadline { + return survivors; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +/// Returns `true` if `pid` exists and is not a zombie. Zombies hold no +/// resources — their parent just has not reaped them yet. +fn process_running(pid: u32) -> bool { + match fs::read_to_string(format!("/proc/{pid}/stat")) { + Ok(stat) => !stat_is_zombie(&stat), + Err(_) => false, + } +} + +/// Parse the state field of `/proc//stat`. The state is the first field +/// after the closing parenthesis of `comm`, which may itself contain spaces +/// and parentheses, so split at the *last* `)`. +fn stat_is_zombie(stat: &str) -> bool { + stat.rfind(')') + .and_then(|close| stat[close + 1..].split_whitespace().next()) + .is_some_and(|state| state == "Z") +} + +/// Scan a `/proc`-like directory for process IDs whose `exe` link resolves to +/// one of `targets`. Entries that are not numeric, have no `exe` link, or +/// cannot be read are skipped. +fn find_processes_by_exe(proc_root: &Path, targets: &HashSet) -> Vec { + if targets.is_empty() { + return Vec::new(); + } + let normalized_targets: HashSet = targets.iter().map(|t| normalize_exe(t)).collect(); + + let entries = match fs::read_dir(proc_root) { + Ok(entries) => entries, + Err(err) => { + warn!( + proc_root = %proc_root.display(), + error = %err, + "failed to scan for orphaned runtime processes" + ); + return Vec::new(); + } + }; + + let mut pids = Vec::new(); + for entry in entries.flatten() { + let Ok(pid) = entry.file_name().to_string_lossy().parse::() else { + continue; + }; + let Ok(exe) = fs::read_link(entry.path().join("exe")) else { + continue; + }; + if normalized_targets.contains(&normalize_exe(&exe)) { + pids.push(pid); + } + } + pids +} + +/// Canonicalize an executable path so symlinked spawn paths and `/proc` `exe` +/// links compare equal, tolerating missing files and the `(deleted)` suffix +/// reported for replaced binaries. +fn normalize_exe(path: &Path) -> PathBuf { + let stripped = path + .to_str() + .and_then(|raw| raw.strip_suffix(DELETED_SUFFIX)) + .map_or_else(|| path.to_path_buf(), PathBuf::from); + fs::canonicalize(&stripped).unwrap_or(stripped) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + /// Build a fake `/proc` tree where `entries` maps pid -> exe link target. + fn fake_proc(entries: &[(u32, &Path)]) -> TempDir { + let temp = TempDir::new().expect("create fake proc root"); + for (pid, exe_target) in entries { + let dir = temp.path().join(pid.to_string()); + fs::create_dir_all(&dir).expect("create fake pid dir"); + std::os::unix::fs::symlink(exe_target, dir.join("exe")).expect("symlink fake exe"); + } + temp + } + + #[test] + fn find_processes_matches_only_configured_binaries() { + let temp = TempDir::new().unwrap(); + let firecracker = temp.path().join("firecracker"); + let daemon = temp.path().join("uvm-ublk-daemon"); + let unrelated = temp.path().join("unrelated"); + for path in [&firecracker, &daemon, &unrelated] { + fs::write(path, b"binary").unwrap(); + } + + let proc = fake_proc(&[(1234, &firecracker), (5678, &daemon), (4321, &unrelated)]); + let mut targets = HashSet::new(); + targets.insert(firecracker.clone()); + targets.insert(daemon.clone()); + + let mut pids = find_processes_by_exe(proc.path(), &targets); + pids.sort_unstable(); + assert_eq!(pids, vec![1234, 5678]); + } + + #[test] + fn find_processes_matches_binary_replaced_on_disk() { + let temp = TempDir::new().unwrap(); + let firecracker = temp.path().join("firecracker"); + + // Simulate readlink reporting " (deleted)" for a process whose + // binary was replaced after it started: a dangling symlink carrying + // the suffix, while the target itself no longer exists. + let proc = fake_proc(&[( + 1234, + Path::new(&format!("{} (deleted)", firecracker.display())), + )]); + let mut targets = HashSet::new(); + targets.insert(firecracker.clone()); + + let pids = find_processes_by_exe(proc.path(), &targets); + assert_eq!(pids, vec![1234]); + } + + #[test] + fn find_processes_skips_non_numeric_entries_and_missing_exe() { + let temp = TempDir::new().unwrap(); + let firecracker = temp.path().join("firecracker"); + fs::write(&firecracker, b"binary").unwrap(); + + let proc = fake_proc(&[(1234, &firecracker)]); + fs::create_dir_all(proc.path().join("not-a-pid")).unwrap(); + fs::create_dir_all(proc.path().join("5678")).unwrap(); // no exe link + + let mut targets = HashSet::new(); + targets.insert(firecracker); + + let pids = find_processes_by_exe(proc.path(), &targets); + assert_eq!(pids, vec![1234]); + } + + #[test] + fn stat_zombie_detection_uses_state_after_last_paren() { + assert!(stat_is_zombie("1234 (firecracker) Z 1 2 3")); + assert!(stat_is_zombie("1234 (weird) name) Z 1 2 3")); + assert!(!stat_is_zombie("1234 (firecracker) S 1 2 3")); + assert!(!stat_is_zombie("garbage")); + } +}