From 0a67ae64455e59dd81ef55447ae931ccce333b2e Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:42:27 -0500 Subject: [PATCH 1/3] fix(runtime-core): bound the stop-path disk-handle scan at 30 s and make it injectable (HACK-1328) `down`, `recover` and host-filesystem inspection bounded the lsof disk-handle scan at 5 s, below scans measured on loaded hosts (macOS 5.4-6.5 s; Linux up to 5.3 s with concurrent scans at load ~120). A slow scan made an ordinary `down` end provider_timeout_uncertain without the clean-stop marker, so the next restart fell back to the fresh disk budget. - Move the scan into lifecycle/disk_handles.rs behind a HandleProbe and a Clock; the result is interpreted the same way for every probe (only exit 1 with no output proves the disks closed). - Raise STOP_HANDLE_WAIT to 30 s, matching RELEASE_HANDLE_WAIT; document both. - Split process::capture into spawn + Pending::poll/expire + wait_within so the bounded wait runs on an injected clock; production behavior is unchanged. - Fake-clock tests cover both bounds (slow scan accepted, hung or late scan expired as uncertain, finite), killed by mutation; real-lsof tests cover a held disk and closed disks. - Serialize real lsof scans across test processes with a test-only flock so concurrent whole-host scans no longer time out the lsof-dependent tests. - Add scripts/measure-lsof-scan.ts to measure scan times on a host. Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/README.md | 7 +- .../runtime-core/src/provider/lifecycle.rs | 27 +- .../src/provider/lifecycle/admission_pool.rs | 11 +- .../src/provider/lifecycle/disk_handles.rs | 474 ++++++++++++++++++ packages/runtime-core/src/provider/process.rs | 200 ++++++-- scripts/measure-lsof-scan.ts | 136 +++++ 6 files changed, 790 insertions(+), 65 deletions(-) create mode 100644 packages/runtime-core/src/provider/lifecycle/disk_handles.rs create mode 100644 scripts/measure-lsof-scan.ts diff --git a/packages/runtime-core/README.md b/packages/runtime-core/README.md index c4fe69adf..ac977d979 100644 --- a/packages/runtime-core/README.md +++ b/packages/runtime-core/README.md @@ -1514,7 +1514,12 @@ clean `down` keeps it clean. Pools stopped before this marker existed qualify af next clean `down`. Clean also requires a creation receipt, rootfs pin, both disk identities and declared sizes, the recorded executable and user, a dead recorded provider and no pending owner update; these receipt checks are read-only, take no -provider lock and run around every sample, including `runtime probe`. +provider lock and run around every sample, including `runtime probe`. The disk-handle +scan (`lsof` over both disks) that proves the provider's release is bounded at 30 s on +the stop paths (`down`, `recover`) and in release checks, well above the slowest scan +measured on a loaded host. A scan that still does not finish ends the stop as +`provider_timeout_uncertain`, without the marker. `scripts/measure-lsof-scan.ts` +re-measures a host. The reserve is used only when a start needs it. Once the startup lease is held, a start whose every qualifying sample, including the lease recheck, also fits the 58 GiB diff --git a/packages/runtime-core/src/provider/lifecycle.rs b/packages/runtime-core/src/provider/lifecycle.rs index 41458e9a3..efddc86f8 100644 --- a/packages/runtime-core/src/provider/lifecycle.rs +++ b/packages/runtime-core/src/provider/lifecycle.rs @@ -1,4 +1,5 @@ mod admission_pool; +mod disk_handles; pub mod host_filesystem; mod interrupted; mod operating_guard; @@ -1997,16 +1998,14 @@ impl Drop for VmLock { } } -/// How long `down`, `recover` and other stop paths wait for `lsof` over the owned disks. -const STOP_HANDLE_WAIT: Duration = Duration::from_secs(5); - /// Keep this guard through alias restoration and the stopped receipt. fn lock_absent_disks(candidate: &Candidate, owner: &Owner) -> Result { - lock_absent_disks_within(candidate, owner, STOP_HANDLE_WAIT) + lock_absent_disks_within(candidate, owner, disk_handles::STOP_HANDLE_WAIT) } -/// [`lock_absent_disks`] with a caller-chosen bound on the `lsof` disk-handle scan. A scan -/// that does not finish within `handle_wait` fails as `provider_timeout_uncertain`. +/// [`lock_absent_disks`] with a caller-chosen bound on the disk-handle scan +/// ([`disk_handles::disks_closed`]). A scan that does not finish within `handle_wait` fails +/// as `provider_timeout_uncertain`. fn lock_absent_disks_within( candidate: &Candidate, owner: &Owner, @@ -2033,20 +2032,12 @@ fn lock_absent_disks_within( )); } let vm_lock = VmLock(vm_lock); - let handles = process::capture( - process::clean_command(Path::new("/usr/sbin/lsof")) - .args(["-n", "-P", "-t", "--"]) - .arg(directory.join("storage.raw")) - .arg(directory.join("overlay.raw")), + disk_handles::disks_closed( + &disk_handles::Lsof::host(), + &process::MonotonicClock::new(), + &[directory.join("storage.raw"), directory.join("overlay.raw")], handle_wait, )?; - if handles.status.code() != Some(1) || !handles.stdout.is_empty() || !handles.stderr.is_empty() - { - return Err(CandidateError::new( - "stop_uncertain", - "Cannot confirm that all handles to owned disks are closed.", - )); - } verify_disks(candidate, owner)?; Ok(vm_lock) } diff --git a/packages/runtime-core/src/provider/lifecycle/admission_pool.rs b/packages/runtime-core/src/provider/lifecycle/admission_pool.rs index dc7c26b83..e1ec90a86 100644 --- a/packages/runtime-core/src/provider/lifecycle/admission_pool.rs +++ b/packages/runtime-core/src/provider/lifecycle/admission_pool.rs @@ -29,20 +29,13 @@ //! needs a [`StartPermit`], which only the pre-boot admission ([`admit_boot`]) followed by //! the final leased recheck ([`BootAdmission::release`]) issues. use super::{ - Owner, admission, binary, io, lock_absent_disks_within, root, sockets_absent, + Owner, admission, binary, disk_handles, io, lock_absent_disks_within, root, sockets_absent, verify_disk_allocation, verify_disks, verify_live, }; use crate::provider::{Profile, config_audit, identity, state}; use crate::{Candidate, CandidateError}; use serde_json::Value; use std::fs; -use std::time::Duration; - -/// How long a release check waits for `lsof` over the owned disks. `lsof` scans every -/// process, so it is slow on a loaded host (5.4–6.5 s at about 1,850 processes and load -/// 30–80). The wait is longer than `down`'s because only a start that needs the reserve runs -/// it, under the startup lease; a scan that still does not finish refuses the reserve. -const RELEASE_HANDLE_WAIT: Duration = Duration::from_secs(30); #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum Basis { @@ -251,7 +244,7 @@ fn released( ) -> Result<(), StoppedRefusal> { let check = |error: CandidateError| StoppedRefusal::from_check(&error); lease.verify_path(&root(candidate)).map_err(check)?; - lock_absent_disks_within(candidate, owner, RELEASE_HANDLE_WAIT).map_err(check)?; + lock_absent_disks_within(candidate, owner, disk_handles::RELEASE_HANDLE_WAIT).map_err(check)?; sockets_absent(owner).map_err(check)?; config_audit::verify(candidate, owner).map_err(check) } diff --git a/packages/runtime-core/src/provider/lifecycle/disk_handles.rs b/packages/runtime-core/src/provider/lifecycle/disk_handles.rs new file mode 100644 index 000000000..179f459f7 --- /dev/null +++ b/packages/runtime-core/src/provider/lifecycle/disk_handles.rs @@ -0,0 +1,474 @@ +//! Bounded scan for open handles on a pool's owned disks. +//! +//! A stop (`down`, `recover`, host-filesystem inspection) and a reserve release check both +//! prove the provider let its disks go by asking `lsof` which processes hold them. `lsof` +//! walks every process's descriptors, so its time grows with the process count and with CPU +//! contention. A scan that does not finish within its bound fails as +//! `provider_timeout_uncertain`: `down` then writes no clean-stop marker and the next start +//! falls back to the fresh disk budget. Both bounds therefore sit well above the slowest scan +//! measured on a loaded host, and stay finite so a stuck `lsof` cannot hang a lifecycle +//! operation. `scripts/measure-lsof-scan.ts` re-measures a host. +//! +//! The probe ([`HandleProbe`]) and the clock are injected, so tests drive both bounds with a +//! fake clock instead of the host's `lsof`. Only the scan is replaceable; the result is +//! interpreted here the same way for every probe. +use crate::CandidateError; +use crate::provider::process::{self, Clock, Pending}; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +/// Slowest single disk-handle scan measured on a loaded host so far: macOS at about 1,850 +/// processes and load 30–80 took 5.4–6.5 s. Linux at about 890 processes, load about 120 and +/// 16 scans at once took up to 5.3 s. Larger macOS process counts are still to be measured. +#[cfg(test)] +const SLOWEST_MEASURED_SCAN: Duration = Duration::from_millis(6_500); + +/// How long `down`, `recover` and other stop paths wait for the disk-handle scan. It was +/// 5 s, below the scans measured above, so an ordinary `down` on a loaded host could end +/// uncertain and lose its clean-stop marker. 30 s keeps more than four times the slowest +/// measured scan. With `down`'s other bounded waits (15 s for the guest's shutdown +/// acknowledgement, 10 s for the provider to exit) a stop still ends within about a minute. +pub(super) const STOP_HANDLE_WAIT: Duration = Duration::from_secs(30); + +/// How long a reserve release check waits for the disk-handle scan. Only a start that needs +/// the host reserve runs it, under the startup lease; a scan that still does not finish +/// refuses the reserve and the start keeps the fresh budget. +pub(super) const RELEASE_HANDLE_WAIT: Duration = Duration::from_secs(30); + +/// The fixed host `lsof` the scan runs on both platforms. +const LSOF: &str = "/usr/sbin/lsof"; + +/// Starts one scan for processes that hold any of the given disks open. +pub(super) trait HandleProbe { + fn start(&self, disks: &[PathBuf]) -> Result, CandidateError>; +} + +/// The host's `lsof`, with a clean environment. +pub(super) struct Lsof<'a> { + executable: &'a Path, +} + +impl Lsof<'static> { + pub(super) fn host() -> Self { + Self { + executable: Path::new(LSOF), + } + } +} + +impl<'a> Lsof<'a> { + /// An `lsof` at another path, for tests on hosts that install it elsewhere. + #[cfg(test)] + pub(super) fn at(executable: &'a Path) -> Self { + Self { executable } + } +} + +impl HandleProbe for Lsof<'_> { + fn start(&self, disks: &[PathBuf]) -> Result, CandidateError> { + #[cfg(test)] + let serial = tests::ScanSerial::acquire(); + let running = process::spawn( + process::clean_command(self.executable) + .args(["-n", "-P", "-t", "--"]) + .args(disks), + )?; + #[cfg(test)] + let running = serial.hold(running); + Ok(Box::new(running)) + } +} + +/// Prove no process holds any of `disks` open, waiting at most `wait` on `clock` for the +/// scan. `lsof` exits 1 with no output when it finds no holder; anything else (a listed +/// holder, a diagnostic, another status) fails as `stop_uncertain`, and a scan that does not +/// finish in time fails as `provider_timeout_uncertain`. +pub(super) fn disks_closed( + probe: &dyn HandleProbe, + clock: &dyn Clock, + disks: &[PathBuf], + wait: Duration, +) -> Result<(), CandidateError> { + let mut scan = probe.start(disks)?; + let handles = process::wait_within(scan.as_mut(), wait, clock)?; + if handles.status.code() != Some(1) || !handles.stdout.is_empty() || !handles.stderr.is_empty() + { + return Err(CandidateError::new( + "stop_uncertain", + "Cannot confirm that all handles to owned disks are closed.", + )); + } + Ok(()) +} + +#[cfg(test)] +pub(super) mod tests { + use super::*; + use crate::provider::process::Captured; + use std::cell::{Cell, RefCell}; + use std::fs::{File, OpenOptions}; + use std::os::unix::process::ExitStatusExt; + use std::rc::Rc; + + /// Real `lsof` scans in this test binary run one at a time across every test process + /// (isolated children included), and each scan's bound starts once it holds the turn. + /// Concurrent whole-host scans on a loaded host are what made these tests time out; a + /// lone scan stays far inside the bounds above. The lock is released explicitly, so a + /// copy of the descriptor in an unrelated fork cannot keep it held. + pub(in crate::provider::lifecycle) struct ScanSerial(File); + + impl ScanSerial { + pub(in crate::provider::lifecycle) fn acquire() -> Self { + use std::os::fd::AsRawFd; + let path = std::env::temp_dir() + .join(format!("hack-runtime-core-lsof-scan-{}.lock", unsafe { + libc::geteuid() + })); + let file = OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(path) + .expect("open the test lsof scan lock"); + // SAFETY: the descriptor is live for this call. + assert_eq!(unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) }, 0); + Self(file) + } + + pub(super) fn hold(self, running: process::Running) -> Serialized { + Serialized { + running, + _serial: self, + } + } + } + + impl Drop for ScanSerial { + fn drop(&mut self) { + use std::os::fd::AsRawFd; + // SAFETY: the descriptor is live through this Drop call. + unsafe { libc::flock(self.0.as_raw_fd(), libc::LOCK_UN) }; + } + } + + /// A real scan that keeps the serial turn until it completes or expires. + pub(super) struct Serialized { + running: process::Running, + _serial: ScanSerial, + } + + impl Pending for Serialized { + fn poll(&mut self) -> Result, CandidateError> { + self.running.poll() + } + fn expire(&mut self) -> CandidateError { + self.running.expire() + } + } + + /// A clock that moves only when the code under test sleeps. It refuses to pass an hour, + /// so a wait that never expires fails the test instead of spinning. + #[derive(Default)] + struct FakeClock(Cell); + + const RUNAWAY: Duration = Duration::from_secs(3600); + + impl Clock for FakeClock { + fn now(&self) -> Duration { + self.0.get() + } + fn sleep(&self, duration: Duration) { + let next = self.0.get() + duration; + assert!(next <= RUNAWAY, "the disk-handle wait never expired"); + self.0.set(next); + } + } + + /// What one fake scan reports and how long it takes on the fake clock. + #[derive(Clone)] + struct Outcome { + /// `None`: the scan never finishes on its own. + takes: Option, + code: i32, + stdout: &'static [u8], + stderr: &'static [u8], + } + + impl Outcome { + /// `lsof`'s answer when no process holds the disks. + fn no_holder(takes: Duration) -> Self { + Self { + takes: Some(takes), + code: 1, + stdout: b"", + stderr: b"", + } + } + fn hangs() -> Self { + Self { + takes: None, + ..Self::no_holder(Duration::ZERO) + } + } + } + + #[derive(Default)] + struct Record { + disks: Vec, + expired: bool, + } + + struct FakeProbe { + clock: Rc, + outcome: Outcome, + record: Rc>, + } + + struct FakeScan { + clock: Rc, + started: Duration, + outcome: Outcome, + record: Rc>, + } + + impl HandleProbe for FakeProbe { + fn start(&self, disks: &[PathBuf]) -> Result, CandidateError> { + self.record.borrow_mut().disks = disks.to_vec(); + Ok(Box::new(FakeScan { + clock: self.clock.clone(), + started: self.clock.now(), + outcome: self.outcome.clone(), + record: self.record.clone(), + })) + } + } + + impl Pending for FakeScan { + fn poll(&mut self) -> Result, CandidateError> { + let done = self + .outcome + .takes + .is_some_and(|takes| self.clock.now() - self.started >= takes); + Ok(done.then(|| Captured { + status: std::process::ExitStatus::from_raw(self.outcome.code << 8), + stdout: self.outcome.stdout.to_vec(), + stderr: self.outcome.stderr.to_vec(), + })) + } + fn expire(&mut self) -> CandidateError { + self.record.borrow_mut().expired = true; + CandidateError::new("provider_timeout_uncertain", "fake scan expired") + } + } + + struct Run { + result: Result<(), CandidateError>, + elapsed: Duration, + record: Record, + } + + fn scan(outcome: Outcome, wait: Duration) -> Run { + let clock = Rc::new(FakeClock::default()); + let record = Rc::new(RefCell::new(Record::default())); + let probe = FakeProbe { + clock: clock.clone(), + outcome, + record: record.clone(), + }; + let result = disks_closed(&probe, clock.as_ref(), &disks(Path::new("/pool")), wait); + let elapsed = clock.now(); + drop(probe); + Run { + result, + elapsed, + record: Rc::try_unwrap(record).ok().unwrap().into_inner(), + } + } + + fn disks(directory: &Path) -> Vec { + vec![directory.join("storage.raw"), directory.join("overlay.raw")] + } + + fn code(result: &Result<(), CandidateError>) -> &'static str { + result.as_ref().err().map_or("ok", |error| error.code) + } + + /// Both bounds outlast a scan several times slower than the slowest one measured, so a + /// realistic loaded host does not make `down` or a release check uncertain. + #[test] + fn both_bounds_outlast_a_slow_loaded_host_scan() { + let slow = SLOWEST_MEASURED_SCAN * 4; + for (name, wait) in [("stop", STOP_HANDLE_WAIT), ("release", RELEASE_HANDLE_WAIT)] { + let run = scan(Outcome::no_holder(slow), wait); + assert_eq!( + code(&run.result), + "ok", + "{name} bound refused a {slow:?} scan" + ); + assert!(!run.record.expired, "{name} bound expired a completed scan"); + assert!(run.elapsed >= slow && run.elapsed < slow + process::POLL_INTERVAL * 2); + assert_eq!(run.record.disks, disks(Path::new("/pool"))); + } + } + + /// Both bounds are finite: a scan that never finishes is expired (its child signalled) + /// at the bound and reported as uncertain, never as released, and well within a minute. + #[test] + fn both_bounds_expire_a_hung_scan_as_uncertain_at_the_bound() { + for (name, wait) in [("stop", STOP_HANDLE_WAIT), ("release", RELEASE_HANDLE_WAIT)] { + assert!( + wait <= Duration::from_secs(45), + "{name} bound {wait:?} is too long" + ); + let run = scan(Outcome::hangs(), wait); + assert_eq!(code(&run.result), "provider_timeout_uncertain", "{name}"); + assert!(run.record.expired, "{name} bound did not expire the scan"); + assert!( + run.elapsed >= wait && run.elapsed < wait + process::POLL_INTERVAL * 2, + "{name} bound {wait:?} expired after {:?}", + run.elapsed + ); + } + } + + /// A scan that finishes just past the bound is still uncertain: a late "no holder" is + /// not accepted. + #[test] + fn a_scan_finishing_after_the_bound_is_uncertain() { + for wait in [STOP_HANDLE_WAIT, RELEASE_HANDLE_WAIT] { + let late = Outcome::no_holder(wait + process::POLL_INTERVAL * 3); + let run = scan(late, wait); + assert_eq!(code(&run.result), "provider_timeout_uncertain"); + assert!(run.record.expired); + } + } + + /// The probe is replaceable but its answer is not trusted any less: only exit 1 with no + /// output proves the disks closed. + #[test] + fn only_a_silent_no_holder_answer_proves_the_disks_closed() { + let quick = Duration::from_millis(500); + let answers = [ + (0, &b"4242\n"[..], &b""[..]), + (1, b"4242\n", b""), + (1, b"", b"lsof: WARNING: can't stat()\n"), + (0, b"", b""), + (2, b"", b""), + ]; + for (code_value, stdout, stderr) in answers { + let outcome = Outcome { + takes: Some(quick), + code: code_value, + stdout, + stderr, + }; + let run = scan(outcome, STOP_HANDLE_WAIT); + assert_eq!(code(&run.result), "stop_uncertain"); + assert!(!run.record.expired); + } + assert_eq!( + code(&scan(Outcome::no_holder(quick), STOP_HANDLE_WAIT).result), + "ok" + ); + } + + /// The host `lsof` this test binary can run: the fixed production path, or the usual + /// Linux location where `/usr/sbin` is a separate directory. macOS must have it. + fn host_lsof() -> Option { + let found = ["/usr/sbin/lsof", "/usr/bin/lsof"] + .into_iter() + .map(PathBuf::from) + .find(|path| path.is_file()); + assert!( + found.is_some() || !cfg!(target_os = "macos"), + "macOS hosts must provide {LSOF}" + ); + if found.is_none() { + eprintln!("skipping: no lsof on this host"); + } + found + } + + /// A private scratch directory removed when the test ends. + struct Scratch(PathBuf); + + impl Scratch { + fn new() -> Self { + use std::os::unix::fs::DirBuilderExt; + let path = std::env::temp_dir().join(format!( + "hack-disk-handles-{}-{}", + std::process::id(), + crate::next_test_scratch() + )); + std::fs::DirBuilder::new() + .mode(0o700) + .create(&path) + .unwrap(); + Self(path) + } + fn path(&self) -> &Path { + &self.0 + } + } + + impl Drop for Scratch { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + fn sparse_disks(directory: &Path) -> Vec { + let disks = disks(directory); + for disk in &disks { + File::create(disk).unwrap().set_len(1 << 30).unwrap(); + } + disks + } + + /// The real `lsof` still finds a disk this process holds open, through the same + /// interpretation and the stop path's bound. + #[test] + fn host_lsof_finds_a_held_disk() { + let Some(lsof) = host_lsof() else { return }; + let directory = Scratch::new(); + let disks = sparse_disks(directory.path()); + let _held = File::open(&disks[1]).unwrap(); + let result = disks_closed( + &Lsof::at(&lsof), + &process::MonotonicClock::new(), + &disks, + STOP_HANDLE_WAIT, + ); + assert_eq!(code(&result), "stop_uncertain"); + } + + /// The real `lsof` proves closed disks closed within the stop path's bound. It runs + /// alone in a child process: in the shared test process an unrelated test's pre-exec + /// fork can briefly hold a copy of the disk descriptors this fixture just closed. + #[test] + fn host_lsof_proves_closed_disks_closed() { + use crate::provider::test_executable::{ISOLATED_HANG_GUARD, run_isolated}; + if host_lsof().is_none() { + return; + } + run_isolated( + "provider::lifecycle::disk_handles::tests::host_lsof_closed_disks_child", + ISOLATED_HANG_GUARD, + ); + } + + #[test] + #[ignore = "isolated real-lsof fixture launched by host_lsof_proves_closed_disks_closed"] + fn host_lsof_closed_disks_child() { + let lsof = host_lsof().expect("parent checked for lsof"); + let directory = Scratch::new(); + let disks = sparse_disks(directory.path()); + let result = disks_closed( + &Lsof::at(&lsof), + &process::MonotonicClock::new(), + &disks, + STOP_HANDLE_WAIT, + ); + assert_eq!(code(&result), "ok", "{:?}", result.as_ref().err()); + } +} diff --git a/packages/runtime-core/src/provider/process.rs b/packages/runtime-core/src/provider/process.rs index 72ba28784..e0c1f63db 100644 --- a/packages/runtime-core/src/provider/process.rs +++ b/packages/runtime-core/src/provider/process.rs @@ -11,14 +11,62 @@ pub struct Captured { pub stderr: Vec, } -pub fn capture(command: &mut Command, timeout: Duration) -> Result { +/// Time source for bounded waits, so a test can drive a deadline without real time. +pub trait Clock { + /// Monotonic time since an arbitrary fixed origin. + fn now(&self) -> Duration; + fn sleep(&self, duration: Duration); +} + +/// The host's monotonic clock. +pub struct MonotonicClock(Instant); + +impl MonotonicClock { + pub fn new() -> Self { + Self(Instant::now()) + } +} + +impl Default for MonotonicClock { + fn default() -> Self { + Self::new() + } +} + +impl Clock for MonotonicClock { + fn now(&self) -> Duration { + self.0.elapsed() + } + fn sleep(&self, duration: Duration) { + std::thread::sleep(duration); + } +} + +/// A started child that [`wait_within`] polls until it completes or its bound expires. +pub trait Pending { + /// `Some` once the child exited and both its output streams closed; `None` while it runs + /// or a detached helper still holds a stream open. + fn poll(&mut self) -> Result, CandidateError>; + /// The bound expired: signal and reap only the unreaped direct child (no PID-file or + /// process-group kill) and return the uncertain outcome. + fn expire(&mut self) -> CandidateError; +} + +/// A spawned child whose streams are drained continuously, keeping at most 64 KiB each. +pub struct Running { + child: std::process::Child, + status: Option, + out: Option>>, + err: Option>>, +} + +pub fn spawn(command: &mut Command) -> Result { let mut child = command .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() .map_err(|e| CandidateError::new("provider_spawn_failed", e.to_string()))?; - // Drain both streams continuously, retaining at most 64 KiB per stream. fn drain(mut pipe: impl Read) -> Vec { let mut output = Vec::new(); let mut buffer = [0; 4096]; @@ -33,44 +81,85 @@ pub fn capture(command: &mut Command, timeout: Duration) -> Result { - // Do not join: detached provider helpers may inherit pipe descriptors. - // A helper retaining output is treated as uncertain, with a bounded wait. - while (!out.is_finished() || !err.is_finished()) && start.elapsed() < timeout { - std::thread::sleep(Duration::from_millis(10)); - } - if !out.is_finished() || !err.is_finished() { - return Err(CandidateError::new( - "provider_output_uncertain", - "A provider helper retained output descriptors; inspect owned state.", - )); - } - let output = out.join().unwrap_or_default(); - let diagnostic = err.join().unwrap_or_default(); - return Ok(Captured { - status, - stdout: output, - stderr: diagnostic, - }); - } - Ok(None) if start.elapsed() < timeout => std::thread::sleep(Duration::from_millis(10)), - Ok(None) => { - // Only the unreaped direct child is signalled; no PID-file or process-group kill. - let _ = child.kill(); - let _ = child.wait(); - return Err(CandidateError::new( - "provider_timeout_uncertain", - "Provider request timed out. Detached effects may remain; inspect before retry.", - )); + Ok(Running { + child, + status: None, + out: Some(std::thread::spawn(move || drain(stdout))), + err: Some(std::thread::spawn(move || drain(stderr))), + }) +} + +impl Pending for Running { + fn poll(&mut self) -> Result, CandidateError> { + if self.status.is_none() { + match self.child.try_wait() { + Ok(Some(status)) => self.status = Some(status), + Ok(None) => return Ok(None), + Err(e) => return Err(CandidateError::new("provider_wait_failed", e.to_string())), } - Err(e) => return Err(CandidateError::new("provider_wait_failed", e.to_string())), } + // Do not join early: detached provider helpers may inherit pipe descriptors. + let finished = |stream: &Option>>| { + stream.as_ref().is_some_and(|handle| handle.is_finished()) + }; + if !finished(&self.out) || !finished(&self.err) { + return Ok(None); + } + let join = |stream: &mut Option>>| { + stream + .take() + .map(|handle| handle.join().unwrap_or_default()) + .unwrap_or_default() + }; + Ok(Some(Captured { + status: self.status.expect("exit status recorded"), + stdout: join(&mut self.out), + stderr: join(&mut self.err), + })) } + + fn expire(&mut self) -> CandidateError { + if self.status.is_some() { + // A helper retaining output is treated as uncertain, with a bounded wait. + return CandidateError::new( + "provider_output_uncertain", + "A provider helper retained output descriptors; inspect owned state.", + ); + } + let _ = self.child.kill(); + let _ = self.child.wait(); + CandidateError::new( + "provider_timeout_uncertain", + "Provider request timed out. Detached effects may remain; inspect before retry.", + ) + } +} + +/// How often [`wait_within`] polls a pending child. +pub const POLL_INTERVAL: Duration = Duration::from_millis(10); + +/// Wait for `pending` on `clock` until it completes, or expire it once `timeout` has passed +/// since this call. The bound covers both the child's exit and its streams closing. +pub fn wait_within( + pending: &mut dyn Pending, + timeout: Duration, + clock: &dyn Clock, +) -> Result { + let start = clock.now(); + loop { + if let Some(captured) = pending.poll()? { + return Ok(captured); + } + if clock.now().saturating_sub(start) >= timeout { + return Err(pending.expire()); + } + clock.sleep(POLL_INTERVAL); + } +} + +pub fn capture(command: &mut Command, timeout: Duration) -> Result { + let mut running = spawn(command)?; + wait_within(&mut running, timeout, &MonotonicClock::new()) } pub fn run(command: &mut Command, timeout: Duration) -> Result { @@ -109,6 +198,43 @@ pub fn clean_command(binary: &std::path::Path) -> Command { mod tests { use super::*; + fn shell(script: &str) -> Command { + let mut command = clean_command(std::path::Path::new("/bin/sh")); + command.args(["-c", script]); + command + } + + #[test] + fn capture_returns_the_exit_status_and_both_streams() { + let output = capture( + &mut shell("printf out; printf err >&2; exit 3"), + Duration::from_secs(30), + ) + .unwrap(); + assert_eq!(output.status.code(), Some(3)); + assert_eq!(output.stdout, b"out"); + assert_eq!(output.stderr, b"err"); + } + + #[test] + fn capture_kills_and_reaps_a_child_that_outlives_its_bound() { + let started = Instant::now(); + let error = capture(&mut shell("exec sleep 30"), Duration::from_millis(200)) + .err() + .unwrap(); + assert_eq!(error.code, "provider_timeout_uncertain"); + assert!(started.elapsed() < Duration::from_secs(20)); + } + + #[test] + fn capture_reports_a_helper_that_retains_output_as_uncertain() { + // The shell exits at once; its background child keeps stdout open past the bound. + let error = capture(&mut shell("sleep 3 & exit 0"), Duration::from_millis(300)) + .err() + .unwrap(); + assert_eq!(error.code, "provider_output_uncertain"); + } + #[test] fn private_child_mask_does_not_depend_on_or_change_parent_mask() { let output = Command::new(std::env::current_exe().unwrap()) diff --git a/scripts/measure-lsof-scan.ts b/scripts/measure-lsof-scan.ts new file mode 100644 index 000000000..abac947e2 --- /dev/null +++ b/scripts/measure-lsof-scan.ts @@ -0,0 +1,136 @@ +#!/usr/bin/env bun +/** + * Time the disk-handle scan the native stop and release paths run + * (`lsof -n -P -t -- storage.raw overlay.raw`, see + * `packages/runtime-core/src/provider/lifecycle.rs`), so the bounds + * `STOP_HANDLE_WAIT` and `RELEASE_HANDLE_WAIT` can be checked against the host. + * + * `lsof` walks every process's descriptors, so its time grows with the process + * count and with CPU contention. The scan targets two fresh sparse files in a + * temporary directory; nothing holds them, so a healthy scan exits 1 with no + * output, exactly the stop path's success case. + * + * Usage: bun scripts/measure-lsof-scan.ts [--runs N] [--concurrency C] [--burn B] + * --runs scans per round (default 20) + * --concurrency scans started at once in the concurrent round (default 8) + * --burn busy-loop processes to run during both rounds (default 0), + * to add CPU contention on an otherwise idle host + * + * Prints one JSON report: host platform, process count, load averages and + * min/median/p95/max seconds for a sequential and a concurrent round. It + * prints no paths, hostnames or process names. + */ +import { mkdtempSync, rmSync, truncateSync, writeFileSync } from "node:fs"; +import { cpus, loadavg, platform, tmpdir } from "node:os"; +import { join } from "node:path"; +import { parseArgs } from "node:util"; + +const { values } = parseArgs({ + options: { + runs: { type: "string", default: "20" }, + concurrency: { type: "string", default: "8" }, + burn: { type: "string", default: "0" }, + }, +}); +const runs = Number(values.runs); +const concurrency = Number(values.concurrency); +const burn = Number(values.burn); +for (const [name, value] of [ + ["--runs", runs], + ["--concurrency", concurrency], +] as const) { + if (!Number.isInteger(value) || value < 1) { + throw new Error(`${name} must be a positive integer`); + } +} +if (!Number.isInteger(burn) || burn < 0) { + throw new Error("--burn must be a non-negative integer"); +} + +// The runtime invokes this fixed path on both platforms. +const LSOF = "/usr/sbin/lsof"; +const directory = mkdtempSync(join(tmpdir(), "lsof-scan-")); +const disks = ["storage.raw", "overlay.raw"].map((name) => { + const path = join(directory, name); + writeFileSync(path, ""); + truncateSync(path, 1024 * 1024 * 1024); + return path; +}); + +type Scan = { seconds: number; exitCode: number | null; clean: boolean }; + +async function scan(): Promise { + const started = performance.now(); + const child = Bun.spawn([LSOF, "-n", "-P", "-t", "--", ...disks], { + env: { PATH: "/usr/bin:/bin:/usr/sbin:/sbin", LANG: "C" }, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([ + new Response(child.stdout).text(), + new Response(child.stderr).text(), + child.exited, + ]); + return { + seconds: (performance.now() - started) / 1000, + exitCode, + clean: exitCode === 1 && stdout === "" && stderr === "", + }; +} + +function summary(scans: Scan[]) { + const seconds = scans.map((s) => s.seconds).sort((a, b) => a - b); + const at = (q: number) => + seconds[Math.min(seconds.length - 1, Math.ceil(q * seconds.length) - 1)]; + const round = (n: number) => Math.round(n * 1000) / 1000; + return { + scans: scans.length, + unclean: scans.filter((s) => !s.clean).length, + min: round(seconds[0]), + median: round(at(0.5)), + p95: round(at(0.95)), + max: round(seconds.at(-1) ?? 0), + }; +} + +function processCount(): number { + const ps = Bun.spawnSync(["/bin/ps", "-A", "-o", "pid="]); + return ps.stdout.toString().split("\n").filter(Boolean).length; +} + +const burners = Array.from({ length: burn }, () => + Bun.spawn(["/bin/sh", "-c", "while :; do :; done"], { + stdout: "ignore", + stderr: "ignore", + }) +); + +try { + const before = { processes: processCount(), load: loadavg() }; + const sequential: Scan[] = []; + for (let i = 0; i < runs; i++) { + sequential.push(await scan()); + } + const concurrent: Scan[] = []; + for (let done = 0; done < runs; done += concurrency) { + const batch = Math.min(concurrency, runs - done); + concurrent.push( + ...(await Promise.all(Array.from({ length: batch }, () => scan()))) + ); + } + const report = { + platform: platform(), + cpus: cpus().length, + burn, + before, + after: { processes: processCount(), load: loadavg() }, + sequential: summary(sequential), + concurrent: { concurrency, ...summary(concurrent) }, + }; + console.log(JSON.stringify(report, null, 2)); +} finally { + for (const burner of burners) { + burner.kill(); + } + rmSync(directory, { recursive: true, force: true }); +} From 11b4215efec5a3f806085e9ed8b098d3a9286157 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:09:28 -0500 Subject: [PATCH 2/3] fix(runtime-core): refuse scans observed only after their bound; own every lsof benchmark child (HACK-1328) Review fixes for #553. - process::wait_within read the clock only when the child was still running, so a waiter that overslept past the bound and found the child finished accepted it. The clock is now read after every poll: a result observed at or after the bound is refused as provider_timeout_uncertain (the child has already been reaped, so nothing is signalled). A running child or held streams at the bound still expire as timeout or output-retention. Fake-clock tests oversleep across the bound and the completion (refused) and within the bound (accepted); restoring the old order fails the first. - scripts/measure-lsof-scan.ts registers every child (burners and scans) as it is spawned, inside the cleanup scope, and kills and awaits all of them on success, a failed spawn part-way through, a hung scan or SIGINT/SIGTERM. Each scan is bounded (--scan-timeout, default 120 s) across exit and both streams; expiry cancels the reads so a descendant holding a pipe cannot keep the run alive. tests/measure-lsof-scan.test.ts covers a clean run, a hung scan, a held pipe, a partial launch failure and an abort, each checking no child is left running; mutations removing the bound, the read cancel, the cleanup scope or stopAll each fail it. Co-Authored-By: Claude Opus 5.5 --- .../src/provider/lifecycle/disk_handles.rs | 64 +++- packages/runtime-core/src/provider/process.rs | 30 +- scripts/measure-lsof-scan.ts | 304 +++++++++++++----- tests/measure-lsof-scan.test.ts | 150 +++++++++ 4 files changed, 462 insertions(+), 86 deletions(-) create mode 100644 tests/measure-lsof-scan.test.ts diff --git a/packages/runtime-core/src/provider/lifecycle/disk_handles.rs b/packages/runtime-core/src/provider/lifecycle/disk_handles.rs index 179f459f7..6418ff98e 100644 --- a/packages/runtime-core/src/provider/lifecycle/disk_handles.rs +++ b/packages/runtime-core/src/provider/lifecycle/disk_handles.rs @@ -167,20 +167,41 @@ pub(super) mod tests { } /// A clock that moves only when the code under test sleeps. It refuses to pass an hour, - /// so a wait that never expires fails the test instead of spinning. + /// so a wait that never expires fails the test instead of spinning. With an oversleep, + /// the first sleep that starts at or after `from` wakes at `until` instead, as a waiter + /// descheduled on a loaded host would. #[derive(Default)] - struct FakeClock(Cell); + struct FakeClock { + now: Cell, + oversleep: Cell>, + } const RUNAWAY: Duration = Duration::from_secs(3600); + impl FakeClock { + fn oversleeping(from: Duration, until: Duration) -> Self { + assert!(until > from); + Self { + now: Cell::default(), + oversleep: Cell::new(Some((from, until))), + } + } + } + impl Clock for FakeClock { fn now(&self) -> Duration { - self.0.get() + self.now.get() } fn sleep(&self, duration: Duration) { - let next = self.0.get() + duration; + let mut next = self.now.get() + duration; + if let Some((from, until)) = self.oversleep.get() + && self.now.get() >= from + { + self.oversleep.set(None); + next = next.max(until); + } assert!(next <= RUNAWAY, "the disk-handle wait never expired"); - self.0.set(next); + self.now.set(next); } } @@ -268,7 +289,11 @@ pub(super) mod tests { } fn scan(outcome: Outcome, wait: Duration) -> Run { - let clock = Rc::new(FakeClock::default()); + scan_on(FakeClock::default(), outcome, wait) + } + + fn scan_on(clock: FakeClock, outcome: Outcome, wait: Duration) -> Run { + let clock = Rc::new(clock); let record = Rc::new(RefCell::new(Record::default())); let probe = FakeProbe { clock: clock.clone(), @@ -343,6 +368,33 @@ pub(super) mod tests { } } + /// A waiter that oversleeps across both the bound and the scan's completion finds a + /// finished scan only after the bound. That late "no holder" is refused as uncertain; + /// the scan already finished, so it is not expired (nothing is signalled). + #[test] + fn an_oversleep_across_the_bound_and_completion_refuses_the_late_result() { + let ms = Duration::from_millis; + for wait in [STOP_HANDLE_WAIT, RELEASE_HANDLE_WAIT] { + let clock = FakeClock::oversleeping(wait - ms(10), wait + ms(50)); + let run = scan_on(clock, Outcome::no_holder(wait + ms(20)), wait); + assert_eq!(code(&run.result), "provider_timeout_uncertain"); + assert!(!run.record.expired, "a finished scan was signalled"); + assert_eq!(run.elapsed, wait + ms(50)); + } + } + + /// An oversleep that still wakes inside the bound accepts a scan that finished in it. + #[test] + fn an_oversleep_that_wakes_inside_the_bound_accepts_the_scan() { + let ms = Duration::from_millis; + for wait in [STOP_HANDLE_WAIT, RELEASE_HANDLE_WAIT] { + let clock = FakeClock::oversleeping(wait - ms(60), wait - ms(20)); + let run = scan_on(clock, Outcome::no_holder(wait - ms(30)), wait); + assert_eq!(code(&run.result), "ok"); + assert_eq!(run.elapsed, wait - ms(20)); + } + } + /// The probe is replaceable but its answer is not trusted any less: only exit 1 with no /// output proves the disks closed. #[test] diff --git a/packages/runtime-core/src/provider/process.rs b/packages/runtime-core/src/provider/process.rs index e0c1f63db..31344fd14 100644 --- a/packages/runtime-core/src/provider/process.rs +++ b/packages/runtime-core/src/provider/process.rs @@ -140,6 +140,14 @@ pub const POLL_INTERVAL: Duration = Duration::from_millis(10); /// Wait for `pending` on `clock` until it completes, or expire it once `timeout` has passed /// since this call. The bound covers both the child's exit and its streams closing. +/// +/// A result counts only if it is observed before the bound: the clock is read after each +/// poll, so a waiter that oversleeps past the bound and then finds the child finished +/// refuses it as `provider_timeout_uncertain` rather than accepting a result it cannot +/// place inside the bound. Such a child has already exited and been reaped, so nothing is +/// signalled. A child still running, or one whose streams are still held open, at the bound +/// is expired by [`Pending::expire`], which keeps the timeout and output-retention outcomes +/// apart. pub fn wait_within( pending: &mut dyn Pending, timeout: Duration, @@ -147,16 +155,26 @@ pub fn wait_within( ) -> Result { let start = clock.now(); loop { - if let Some(captured) = pending.poll()? { - return Ok(captured); + let polled = pending.poll()?; + let within = clock.now().saturating_sub(start) < timeout; + match polled { + Some(captured) if within => return Ok(captured), + Some(_) => return Err(completed_late()), + None if within => clock.sleep(POLL_INTERVAL), + None => return Err(pending.expire()), } - if clock.now().saturating_sub(start) >= timeout { - return Err(pending.expire()); - } - clock.sleep(POLL_INTERVAL); } } +/// A child observed complete only after its bound: its result is not trusted. +fn completed_late() -> CandidateError { + CandidateError::new( + "provider_timeout_uncertain", + "Provider request finished only after its time bound; its result is not trusted. \ + Inspect before retry.", + ) +} + pub fn capture(command: &mut Command, timeout: Duration) -> Result { let mut running = spawn(command)?; wait_within(&mut running, timeout, &MonotonicClock::new()) diff --git a/scripts/measure-lsof-scan.ts b/scripts/measure-lsof-scan.ts index abac947e2..360e675ff 100644 --- a/scripts/measure-lsof-scan.ts +++ b/scripts/measure-lsof-scan.ts @@ -2,7 +2,7 @@ /** * Time the disk-handle scan the native stop and release paths run * (`lsof -n -P -t -- storage.raw overlay.raw`, see - * `packages/runtime-core/src/provider/lifecycle.rs`), so the bounds + * `packages/runtime-core/src/provider/lifecycle/disk_handles.rs`), so the bounds * `STOP_HANDLE_WAIT` and `RELEASE_HANDLE_WAIT` can be checked against the host. * * `lsof` walks every process's descriptors, so its time grows with the process @@ -11,70 +11,160 @@ * output, exactly the stop path's success case. * * Usage: bun scripts/measure-lsof-scan.ts [--runs N] [--concurrency C] [--burn B] - * --runs scans per round (default 20) - * --concurrency scans started at once in the concurrent round (default 8) - * --burn busy-loop processes to run during both rounds (default 0), - * to add CPU contention on an otherwise idle host + * [--scan-timeout SECONDS] + * --runs scans per round (default 20) + * --concurrency scans started at once in the concurrent round (default 8) + * --burn busy-loop processes to run during both rounds (default 0), + * to add CPU contention on an otherwise idle host + * --scan-timeout bound on one scan, its exit and its output (default 120) * * Prints one JSON report: host platform, process count, load averages and * min/median/p95/max seconds for a sequential and a concurrent round. It * prints no paths, hostnames or process names. + * + * Every child (scan and burner) is registered the moment it is spawned and is + * killed and awaited on every exit path: success, a failed spawn part-way + * through launching burners, a scan that exceeds its bound, or SIGINT/SIGTERM. */ import { mkdtempSync, rmSync, truncateSync, writeFileSync } from "node:fs"; import { cpus, loadavg, platform, tmpdir } from "node:os"; import { join } from "node:path"; import { parseArgs } from "node:util"; -const { values } = parseArgs({ +/** The fixed `lsof` the runtime invokes on both platforms. */ +export const HOST_LSOF = "/usr/sbin/lsof"; + +type Child = { + readonly pid: number; + readonly exited: Promise; + readonly stdout: ReadableStream | number | undefined | null; + readonly stderr: ReadableStream | number | undefined | null; + kill(signal?: number | NodeJS.Signals): void; +}; + +export type Spawn = ( + command: string[], options: { - runs: { type: "string", default: "20" }, - concurrency: { type: "string", default: "8" }, - burn: { type: "string", default: "0" }, - }, -}); -const runs = Number(values.runs); -const concurrency = Number(values.concurrency); -const burn = Number(values.burn); -for (const [name, value] of [ - ["--runs", runs], - ["--concurrency", concurrency], -] as const) { - if (!Number.isInteger(value) || value < 1) { - throw new Error(`${name} must be a positive integer`); + env?: Record; + stdout: "pipe" | "ignore"; + stderr: "pipe" | "ignore"; + } +) => Child; + +export type MeasureOptions = { + runs: number; + concurrency: number; + burn: number; + scanTimeoutMs: number; + lsof?: string; + spawn?: Spawn; +}; + +export type Scan = { + seconds: number; + exitCode: number | null; + clean: boolean; + timedOut: boolean; +}; + +/** Children this run owns; each is killed and awaited before the run returns. */ +class Owned { + readonly #children = new Set(); + readonly #spawn: Spawn; + + constructor(spawn: Spawn) { + this.#spawn = spawn; + } + + spawn(...args: Parameters): Child { + const child = this.#spawn(...args); + this.#children.add(child); + return child; + } + + async release(child: Child): Promise { + this.#children.delete(child); + await stop(child); + } + + async stopAll(): Promise { + const children = [...this.#children]; + this.#children.clear(); + await Promise.all(children.map(stop)); } -} -if (!Number.isInteger(burn) || burn < 0) { - throw new Error("--burn must be a non-negative integer"); } -// The runtime invokes this fixed path on both platforms. -const LSOF = "/usr/sbin/lsof"; -const directory = mkdtempSync(join(tmpdir(), "lsof-scan-")); -const disks = ["storage.raw", "overlay.raw"].map((name) => { - const path = join(directory, name); - writeFileSync(path, ""); - truncateSync(path, 1024 * 1024 * 1024); - return path; -}); +async function stop(child: Child): Promise { + try { + child.kill("SIGKILL"); + } catch { + // Already exited. + } + await child.exited; +} -type Scan = { seconds: number; exitCode: number | null; clean: boolean }; +/** Read a stream to its end, or stop reading when `abort` settles. */ +async function drain( + stream: Child["stdout"], + abort: Promise +): Promise { + if (!(stream instanceof ReadableStream)) { + return ""; + } + const reader = stream.getReader(); + const chunks: Uint8Array[] = []; + // Cancelling ends the pending read, so a descendant holding the pipe cannot + // keep this loop (or the run) waiting past the scan's bound. + abort.then(() => reader.cancel().catch(() => undefined)); + while (true) { + const { done, value } = await reader.read().catch(() => ({ + done: true as const, + value: undefined, + })); + if (done) { + break; + } + chunks.push(value); + } + return Buffer.concat(chunks).toString(); +} -async function scan(): Promise { +async function scan( + owned: Owned, + lsof: string, + disks: string[], + timeoutMs: number +): Promise { const started = performance.now(); - const child = Bun.spawn([LSOF, "-n", "-P", "-t", "--", ...disks], { + const child = owned.spawn([lsof, "-n", "-P", "-t", "--", ...disks], { env: { PATH: "/usr/bin:/bin:/usr/sbin:/sbin", LANG: "C" }, stdout: "pipe", stderr: "pipe", }); - const [stdout, stderr, exitCode] = await Promise.all([ - new Response(child.stdout).text(), - new Response(child.stderr).text(), - child.exited, - ]); + let timer: ReturnType | undefined; + const expired = new Promise<"timeout">((resolve) => { + timer = setTimeout(() => resolve("timeout"), timeoutMs); + }); + // The bound covers the exit and both streams closing, as in the runtime. + const stdoutRead = drain(child.stdout, expired); + const stderrRead = drain(child.stderr, expired); + const finished = Promise.all([stdoutRead, stderrRead, child.exited]); + const outcome = await Promise.race([finished, expired]); + clearTimeout(timer); + const seconds = (performance.now() - started) / 1000; + await owned.release(child); + // Expiry cancelled both reads, so this settles at once even when a + // descendant still holds a pipe; no read outlives the scan. + await Promise.allSettled([stdoutRead, stderrRead]); + if (outcome === "timeout") { + return { seconds, exitCode: null, clean: false, timedOut: true }; + } + const [stdout, stderr, exitCode] = outcome; return { - seconds: (performance.now() - started) / 1000, + seconds, exitCode, clean: exitCode === 1 && stdout === "" && stderr === "", + timedOut: false, }; } @@ -86,9 +176,10 @@ function summary(scans: Scan[]) { return { scans: scans.length, unclean: scans.filter((s) => !s.clean).length, - min: round(seconds[0]), - median: round(at(0.5)), - p95: round(at(0.95)), + timedOut: scans.filter((s) => s.timedOut).length, + min: round(seconds[0] ?? 0), + median: round(at(0.5) ?? 0), + p95: round(at(0.95) ?? 0), max: round(seconds.at(-1) ?? 0), }; } @@ -98,39 +189,104 @@ function processCount(): number { return ps.stdout.toString().split("\n").filter(Boolean).length; } -const burners = Array.from({ length: burn }, () => - Bun.spawn(["/bin/sh", "-c", "while :; do :; done"], { - stdout: "ignore", - stderr: "ignore", - }) -); - -try { - const before = { processes: processCount(), load: loadavg() }; - const sequential: Scan[] = []; - for (let i = 0; i < runs; i++) { - sequential.push(await scan()); +const defaultSpawn: Spawn = (command, options) => Bun.spawn(command, options); + +/** + * Run both rounds and return the report. `signal` aborts the run; owned + * children are stopped and the scratch directory removed on every path. + */ +export async function measure(options: MeasureOptions, signal?: AbortSignal) { + const owned = new Owned(options.spawn ?? defaultSpawn); + const lsof = options.lsof ?? HOST_LSOF; + const directory = mkdtempSync(join(tmpdir(), "lsof-scan-")); + const aborted = new Promise((_, reject) => { + signal?.addEventListener("abort", () => reject(new Error("aborted")), { + once: true, + }); + }); + aborted.catch(() => undefined); + try { + const disks = ["storage.raw", "overlay.raw"].map((name) => { + const path = join(directory, name); + writeFileSync(path, ""); + truncateSync(path, 1024 * 1024 * 1024); + return path; + }); + for (let i = 0; i < options.burn; i++) { + owned.spawn(["/bin/sh", "-c", "while :; do :; done"], { + stdout: "ignore", + stderr: "ignore", + }); + } + const one = () => + Promise.race([scan(owned, lsof, disks, options.scanTimeoutMs), aborted]); + const before = { processes: processCount(), load: loadavg() }; + const sequential: Scan[] = []; + for (let i = 0; i < options.runs; i++) { + sequential.push(await one()); + } + const concurrent: Scan[] = []; + for (let done = 0; done < options.runs; done += options.concurrency) { + const batch = Math.min(options.concurrency, options.runs - done); + concurrent.push( + ...(await Promise.all(Array.from({ length: batch }, one))) + ); + } + return { + platform: platform(), + cpus: cpus().length, + burn: options.burn, + scanTimeoutSeconds: options.scanTimeoutMs / 1000, + before, + after: { processes: processCount(), load: loadavg() }, + sequential: summary(sequential), + concurrent: { concurrency: options.concurrency, ...summary(concurrent) }, + }; + } finally { + await owned.stopAll(); + rmSync(directory, { recursive: true, force: true }); } - const concurrent: Scan[] = []; - for (let done = 0; done < runs; done += concurrency) { - const batch = Math.min(concurrency, runs - done); - concurrent.push( - ...(await Promise.all(Array.from({ length: batch }, () => scan()))) - ); +} + +function parseOptions(): MeasureOptions { + const { values } = parseArgs({ + options: { + runs: { type: "string", default: "20" }, + concurrency: { type: "string", default: "8" }, + burn: { type: "string", default: "0" }, + "scan-timeout": { type: "string", default: "120" }, + }, + }); + const positive = (name: string, value: number) => { + if (!Number.isInteger(value) || value < 1) { + throw new Error(`${name} must be a positive integer`); + } + return value; + }; + const burn = Number(values.burn); + if (!Number.isInteger(burn) || burn < 0) { + throw new Error("--burn must be a non-negative integer"); } - const report = { - platform: platform(), - cpus: cpus().length, + return { + runs: positive("--runs", Number(values.runs)), + concurrency: positive("--concurrency", Number(values.concurrency)), burn, - before, - after: { processes: processCount(), load: loadavg() }, - sequential: summary(sequential), - concurrent: { concurrency, ...summary(concurrent) }, + scanTimeoutMs: + positive("--scan-timeout", Number(values["scan-timeout"])) * 1000, }; - console.log(JSON.stringify(report, null, 2)); -} finally { - for (const burner of burners) { - burner.kill(); +} + +if (import.meta.main) { + const options = parseOptions(); + const controller = new AbortController(); + for (const name of ["SIGINT", "SIGTERM"] as const) { + process.once(name, () => controller.abort()); + } + try { + const report = await measure(options, controller.signal); + console.log(JSON.stringify(report, null, 2)); + } catch (error) { + console.error(error instanceof Error ? error.message : error); + process.exitCode = controller.signal.aborted ? 130 : 1; } - rmSync(directory, { recursive: true, force: true }); } diff --git a/tests/measure-lsof-scan.test.ts b/tests/measure-lsof-scan.test.ts new file mode 100644 index 000000000..df53fd7e2 --- /dev/null +++ b/tests/measure-lsof-scan.test.ts @@ -0,0 +1,150 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + type MeasureOptions, + measure, + type Spawn, +} from "../scripts/measure-lsof-scan.ts"; + +let scratch = ""; + +beforeEach(async () => { + scratch = await mkdtemp(join(tmpdir(), "measure-lsof-scan-test-")); +}); + +afterEach(async () => { + await rm(scratch, { recursive: true, force: true }); +}); + +/** A stand-in `lsof` running `body` as a shell script. */ +async function fakeLsof(body: string): Promise { + const path = join(scratch, "lsof"); + await writeFile(path, `#!/bin/sh\n${body}\n`); + await chmod(path, 0o700); + return path; +} + +type Started = { pid: number; command: string[] }; + +/** Spawn through Bun, recording every child; optionally fail one spawn. */ +function recordingSpawn(failAt?: number): { + spawn: Spawn; + started: Started[]; +} { + const started: Started[] = []; + let calls = 0; + const spawn: Spawn = (command, options) => { + calls += 1; + if (calls === failAt) { + throw new Error("spawn refused: resource pressure"); + } + const child = Bun.spawn(command, options); + started.push({ pid: child.pid, command }); + return child; + }; + return { spawn, started }; +} + +function alive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +const base: MeasureOptions = { + runs: 1, + concurrency: 1, + burn: 0, + scanTimeoutMs: 300, +}; + +describe("measure-lsof-scan", () => { + test("reports a clean no-holder scan and leaves no child running", async () => { + const { spawn, started } = recordingSpawn(); + const report = await measure({ + ...base, + runs: 2, + concurrency: 2, + burn: 2, + scanTimeoutMs: 10_000, + lsof: await fakeLsof("exit 1"), + spawn, + }); + expect(report.sequential).toMatchObject({ + scans: 2, + unclean: 0, + timedOut: 0, + }); + expect(report.concurrent).toMatchObject({ + scans: 2, + unclean: 0, + timedOut: 0, + }); + expect(started).toHaveLength(6); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); + + test("bounds a hung scan, kills it and still stops the burners", async () => { + const { spawn, started } = recordingSpawn(); + const begun = performance.now(); + const report = await measure({ + ...base, + burn: 2, + lsof: await fakeLsof("exec sleep 600"), + spawn, + }); + expect(performance.now() - begun).toBeLessThan(10_000); + expect(report.sequential).toMatchObject({ unclean: 1, timedOut: 1 }); + expect(report.concurrent).toMatchObject({ unclean: 1, timedOut: 1 }); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); + + test("bounds a scan whose output a descendant keeps open", async () => { + const { spawn, started } = recordingSpawn(); + // The scan exits at once; its background child holds stdout past the bound. + const begun = performance.now(); + const report = await measure({ + ...base, + lsof: await fakeLsof("sleep 5 &\nexit 1"), + spawn, + }); + // Both rounds return at their bounds, not when the descendant lets go. + expect(performance.now() - begun).toBeLessThan(4000); + expect(report.sequential).toMatchObject({ unclean: 1, timedOut: 1 }); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); + + test("stops burners already started when a later spawn fails", async () => { + const { spawn, started } = recordingSpawn(3); + await expect( + measure({ ...base, burn: 4, lsof: await fakeLsof("exit 1"), spawn }) + ).rejects.toThrow("spawn refused"); + expect(started).toHaveLength(2); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); + + test("an abort during a hung scan stops every child", async () => { + const { spawn, started } = recordingSpawn(); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 200); + await expect( + measure( + { + ...base, + burn: 1, + scanTimeoutMs: 60_000, + lsof: await fakeLsof("exec sleep 600"), + spawn, + }, + controller.signal + ) + ).rejects.toThrow("aborted"); + expect(started).toHaveLength(2); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); +}); From fe29abef154eb03b74649a5b7573be76de2fd931 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy <344842+roodboi@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:13:16 -0500 Subject: [PATCH 3/3] fix(scripts): cancel and await in-flight lsof scans on abort or failure (HACK-1328) Review fix for #553: abort raced the run against a scan without stopping it, so a scan whose pipe a descendant held kept its readers and its timer alive, and the CLI could linger up to the scan bound after SIGINT/SIGTERM. A spawn failure in the concurrent round could likewise abandon sibling scans. Each scan now takes the run's AbortSignal and ends at completion, its bound, or abort; on every path it clears its timer and listener, kills and awaits its child and cancels and awaits both reads. Teardown aborts the run, awaits all active scans, then stops every owned child. An already-aborted signal starts nothing. Adds --lsof for timing another lsof and for the CLI test. Tests: an already-aborted signal; a failed spawn in the concurrent round stops its hung sibling; SIGINT to the CLI during a scan whose pipe a descendant holds exits 130 within 4 s. Co-Authored-By: Claude Opus 5.5 --- scripts/measure-lsof-scan.ts | 92 ++++++++++++++++++++++++++------- tests/measure-lsof-scan.test.ts | 65 +++++++++++++++++++++++ 2 files changed, 137 insertions(+), 20 deletions(-) diff --git a/scripts/measure-lsof-scan.ts b/scripts/measure-lsof-scan.ts index 360e675ff..5b9b5e458 100644 --- a/scripts/measure-lsof-scan.ts +++ b/scripts/measure-lsof-scan.ts @@ -11,12 +11,13 @@ * output, exactly the stop path's success case. * * Usage: bun scripts/measure-lsof-scan.ts [--runs N] [--concurrency C] [--burn B] - * [--scan-timeout SECONDS] + * [--scan-timeout SECONDS] [--lsof PATH] * --runs scans per round (default 20) * --concurrency scans started at once in the concurrent round (default 8) * --burn busy-loop processes to run during both rounds (default 0), * to add CPU contention on an otherwise idle host * --scan-timeout bound on one scan, its exit and its output (default 120) + * --lsof the lsof to time (default /usr/sbin/lsof, the runtime's) * * Prints one JSON report: host platform, process count, load averages and * min/median/p95/max seconds for a sequential and a concurrent round. It @@ -129,12 +130,28 @@ async function drain( return Buffer.concat(chunks).toString(); } +class Aborted extends Error { + constructor() { + super("aborted"); + } +} + +/** + * One bounded scan. It ends at the first of: the scan completing (exit and + * both streams closed), its bound, or `run` aborting. On every path its timer + * and abort listener are removed, its child is killed and awaited and both + * reads are cancelled and awaited, so nothing of it outlives the call. + */ async function scan( owned: Owned, lsof: string, disks: string[], - timeoutMs: number + timeoutMs: number, + run: AbortSignal ): Promise { + if (run.aborted) { + throw new Aborted(); + } const started = performance.now(); const child = owned.spawn([lsof, "-n", "-P", "-t", "--", ...disks], { env: { PATH: "/usr/bin:/bin:/usr/sbin:/sbin", LANG: "C" }, @@ -142,20 +159,34 @@ async function scan( stderr: "pipe", }); let timer: ReturnType | undefined; - const expired = new Promise<"timeout">((resolve) => { + let onAbort: (() => void) | undefined; + const stopped = new Promise<"timeout" | "aborted">((resolve) => { timer = setTimeout(() => resolve("timeout"), timeoutMs); + onAbort = () => resolve("aborted"); + run.addEventListener("abort", onAbort, { once: true }); }); // The bound covers the exit and both streams closing, as in the runtime. - const stdoutRead = drain(child.stdout, expired); - const stderrRead = drain(child.stderr, expired); + const stdoutRead = drain(child.stdout, stopped); + const stderrRead = drain(child.stderr, stopped); const finished = Promise.all([stdoutRead, stderrRead, child.exited]); - const outcome = await Promise.race([finished, expired]); - clearTimeout(timer); - const seconds = (performance.now() - started) / 1000; - await owned.release(child); - // Expiry cancelled both reads, so this settles at once even when a - // descendant still holds a pipe; no read outlives the scan. - await Promise.allSettled([stdoutRead, stderrRead]); + let outcome: Awaited | "timeout" | "aborted" = "aborted"; + let seconds = 0; + try { + outcome = await Promise.race([finished, stopped]); + seconds = (performance.now() - started) / 1000; + } finally { + clearTimeout(timer); + if (onAbort) { + run.removeEventListener("abort", onAbort); + } + await owned.release(child); + // A stopped scan cancelled both reads, so this settles at once even when + // a descendant still holds a pipe; no read outlives the scan. + await Promise.allSettled([stdoutRead, stderrRead]); + } + if (outcome === "aborted") { + throw new Aborted(); + } if (outcome === "timeout") { return { seconds, exitCode: null, clean: false, timedOut: true }; } @@ -196,15 +227,18 @@ const defaultSpawn: Spawn = (command, options) => Bun.spawn(command, options); * children are stopped and the scratch directory removed on every path. */ export async function measure(options: MeasureOptions, signal?: AbortSignal) { + if (signal?.aborted) { + throw new Aborted(); + } const owned = new Owned(options.spawn ?? defaultSpawn); const lsof = options.lsof ?? HOST_LSOF; + // Aborted by the caller's signal, or by teardown after any failure, so a + // sibling scan never outlives the run. + const run = new AbortController(); + const forward = () => run.abort(); + signal?.addEventListener("abort", forward, { once: true }); + const active = new Set>(); const directory = mkdtempSync(join(tmpdir(), "lsof-scan-")); - const aborted = new Promise((_, reject) => { - signal?.addEventListener("abort", () => reject(new Error("aborted")), { - once: true, - }); - }); - aborted.catch(() => undefined); try { const disks = ["storage.raw", "overlay.raw"].map((name) => { const path = join(directory, name); @@ -218,8 +252,21 @@ export async function measure(options: MeasureOptions, signal?: AbortSignal) { stderr: "ignore", }); } - const one = () => - Promise.race([scan(owned, lsof, disks, options.scanTimeoutMs), aborted]); + const one = () => { + const pending = scan( + owned, + lsof, + disks, + options.scanTimeoutMs, + run.signal + ); + active.add(pending); + pending.then( + () => active.delete(pending), + () => active.delete(pending) + ); + return pending; + }; const before = { processes: processCount(), load: loadavg() }; const sequential: Scan[] = []; for (let i = 0; i < options.runs; i++) { @@ -243,6 +290,9 @@ export async function measure(options: MeasureOptions, signal?: AbortSignal) { concurrent: { concurrency: options.concurrency, ...summary(concurrent) }, }; } finally { + run.abort(); + signal?.removeEventListener("abort", forward); + await Promise.allSettled([...active]); await owned.stopAll(); rmSync(directory, { recursive: true, force: true }); } @@ -255,6 +305,7 @@ function parseOptions(): MeasureOptions { concurrency: { type: "string", default: "8" }, burn: { type: "string", default: "0" }, "scan-timeout": { type: "string", default: "120" }, + lsof: { type: "string", default: HOST_LSOF }, }, }); const positive = (name: string, value: number) => { @@ -273,6 +324,7 @@ function parseOptions(): MeasureOptions { burn, scanTimeoutMs: positive("--scan-timeout", Number(values["scan-timeout"])) * 1000, + lsof: values.lsof, }; } diff --git a/tests/measure-lsof-scan.test.ts b/tests/measure-lsof-scan.test.ts index df53fd7e2..3c80cd3c2 100644 --- a/tests/measure-lsof-scan.test.ts +++ b/tests/measure-lsof-scan.test.ts @@ -147,4 +147,69 @@ describe("measure-lsof-scan", () => { expect(started).toHaveLength(2); expect(started.filter((child) => alive(child.pid))).toEqual([]); }); + + test("an already-aborted signal starts nothing", async () => { + const { spawn, started } = recordingSpawn(); + const controller = new AbortController(); + controller.abort(); + await expect( + measure( + { ...base, burn: 2, lsof: await fakeLsof("exit 1"), spawn }, + controller.signal + ) + ).rejects.toThrow("aborted"); + expect(started).toEqual([]); + }); + + test("a failed spawn in the concurrent round stops its sibling scan", async () => { + // Spawns 1-2 are the sequential round (quick scans); spawn 3 is a sibling + // that hangs and spawn 4 fails. + const { spawn, started } = recordingSpawn(4); + const count = join(scratch, "count"); + await writeFile(count, "0"); + const lsof = await fakeLsof( + `n=$(cat '${count}'); echo $((n + 1)) > '${count}'\n` + + '[ "$n" -ge 2 ] && exec sleep 600\nexit 1' + ); + const begun = performance.now(); + await expect( + measure({ + ...base, + runs: 2, + concurrency: 2, + scanTimeoutMs: 60_000, + lsof, + spawn, + }) + ).rejects.toThrow("spawn refused"); + expect(performance.now() - begun).toBeLessThan(5000); + expect(started).toHaveLength(3); + expect(started.filter((child) => alive(child.pid))).toEqual([]); + }); + + test("SIGINT during a scan whose pipe a descendant holds exits promptly", async () => { + // The scan exits at once; its background child holds the scan's stdout + // far past the time the runner may take to exit. + const lsof = await fakeLsof("sleep 8 &\nexit 1"); + const runner = Bun.spawn( + [ + process.execPath, + join(import.meta.dir, "../scripts/measure-lsof-scan.ts"), + "--runs", + "1", + "--scan-timeout", + "60", + "--lsof", + lsof, + ], + { stdout: "pipe", stderr: "pipe" } + ); + await Bun.sleep(700); + const signalled = performance.now(); + runner.kill("SIGINT"); + const exitCode = await runner.exited; + expect(performance.now() - signalled).toBeLessThan(4000); + expect(exitCode).toBe(130); + expect(await new Response(runner.stderr).text()).toContain("aborted"); + }); });