diff --git a/packages/runtime-core/src/node/journal.rs b/packages/runtime-core/src/node/journal.rs index b37917db0..fe8c284dd 100644 --- a/packages/runtime-core/src/node/journal.rs +++ b/packages/runtime-core/src/node/journal.rs @@ -1124,4 +1124,39 @@ mod tests { drop(inspect); std::fs::remove_dir_all(root).unwrap(); } + + #[test] + fn a_rollback_journal_deleted_between_check_and_open_reads_as_absent() { + use super::super::test_support::BeforeOpen; + let root = test_root("journal-race"); + drop(Store::open(&root, true).unwrap()); + let journal = root.join("journal.sqlite-journal"); + let rollback_journal = || { + std::fs::write(&journal, b"").unwrap(); + std::fs::set_permissions( + &journal, + std::os::unix::fs::PermissionsExt::from_mode(0o600), + ) + .unwrap(); + }; + // A running supervisor's commit deletes its rollback journal after any earlier + // check saw it and just before the restarted node's open (HACK-1418). + rollback_journal(); + let race = BeforeOpen::arm(&journal, |path| std::fs::remove_file(path).unwrap()); + let store = Store::open(&root, false).unwrap(); + assert!(race.fired()); + assert!(journal.symlink_metadata().is_err()); + drop((store, race)); + // The same instant cannot swap in an alias: the O_NOFOLLOW open still refuses. + rollback_journal(); + let outside = root.join("outside"); + let race = BeforeOpen::arm(&journal, move |path| { + std::fs::remove_file(path).unwrap(); + std::os::unix::fs::symlink(&outside, path).unwrap(); + }); + assert!(Store::open(&root, false).is_err()); + assert!(race.fired()); + drop(race); + std::fs::remove_dir_all(root).unwrap(); + } } diff --git a/packages/runtime-core/src/node/mod.rs b/packages/runtime-core/src/node/mod.rs index 90110a444..dc61f8b0a 100644 --- a/packages/runtime-core/src/node/mod.rs +++ b/packages/runtime-core/src/node/mod.rs @@ -66,6 +66,8 @@ pub(crate) fn existing_private_file(path: &Path) -> Result> { } fn open_private_file(path: &Path, create: bool) -> std::io::Result { + #[cfg(test)] + test_support::before_open(path); OpenOptions::new() .read(true) .write(true) @@ -103,6 +105,85 @@ pub(crate) fn try_lock(path: &Path) -> Result> { } } +#[cfg(test)] +pub(crate) mod test_support { + use super::{Receipt, Store}; + use std::cell::RefCell; + use std::path::{Path, PathBuf}; + + type Hook = (PathBuf, Box); + thread_local! { + static BEFORE_OPEN: RefCell> = const { RefCell::new(None) }; + } + + /// Runs this thread's one-shot hook when the node opens `path`, just before the + /// `open` system call: the instant another writer (SQLite committing) can act. + pub(crate) fn before_open(path: &Path) { + let hook = BEFORE_OPEN.with(|cell| { + let mut slot = cell.borrow_mut(); + if slot.as_ref().is_some_and(|(target, _)| target == path) { + slot.take() + } else { + None + } + }); + if let Some((target, hook)) = hook { + hook(&target); + } + } + + /// Arms a one-shot hook for `path` on this thread; dropping it disarms the hook. + pub(crate) struct BeforeOpen; + impl BeforeOpen { + pub(crate) fn arm(path: &Path, hook: impl FnOnce(&Path) + 'static) -> Self { + BEFORE_OPEN.with(|cell| cell.replace(Some((path.to_owned(), Box::new(hook))))); + Self + } + pub(crate) fn fired(&self) -> bool { + BEFORE_OPEN.with(|cell| cell.borrow().is_none()) + } + } + impl Drop for BeforeOpen { + fn drop(&mut self) { + BEFORE_OPEN.with(|cell| cell.replace(None)); + } + } + + /// Writes one non-terminal job receipt directly, bypassing admission, so supervisor + /// and recovery tests can start from an exact journal state. + pub(crate) fn job(store: &Store, id: &str, state: &str, timeout_ms: u64) -> Receipt { + let receipt = Receipt { + job_id: id.into(), + fixture: "tree".into(), + source: None, + state: state.into(), + accepted_at_ms: super::now(), + queue_deadline_ms: super::now() + 60_000, + execution_timeout_ms: timeout_ms, + starts: 0, + supervisor_pid: None, + child_pid: None, + supervisor_identity: None, + child_identity: None, + cancel_requested: false, + stdout: String::new(), + stderr: String::new(), + truncated: false, + exit_code: None, + signal: None, + detail: None, + }; + store + .connection + .execute( + "INSERT INTO jobs VALUES (?,?)", + rusqlite::params![id, serde_json::to_string(&receipt).unwrap()], + ) + .unwrap(); + receipt + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/packages/runtime-core/src/node/ownership.rs b/packages/runtime-core/src/node/ownership.rs index c9df92af3..74cd98e53 100644 --- a/packages/runtime-core/src/node/ownership.rs +++ b/packages/runtime-core/src/node/ownership.rs @@ -56,29 +56,120 @@ pub fn inspect(pid: i32) -> Result { return Err(error("Native job process identity unavailable.")); } let info = unsafe { info.assume_init() }; - let mut boot: libc::timeval = unsafe { std::mem::zeroed() }; - let mut length = std::mem::size_of_val(&boot); - // SAFETY: read-only sysctl fills a correctly sized timeval. - if unsafe { - libc::sysctlbyname( - c"kern.boottime".as_ptr(), - (&mut boot as *mut libc::timeval).cast(), - &mut length, - std::ptr::null_mut(), - 0, - ) - } != 0 - { - return Err(error("Host boot identity unavailable.")); - } Ok(Observed::Present(ProcessIdentity { pid, uid: info.pbi_uid, process_group: info.pbi_pgid as i32, + // XNU snapshots p_start from calendar time once, at fork, and never rewrites + // it: a later clock step leaves a live process's start unchanged, and only a + // new process (a reused PID) gets a new value. start: info.pbi_start_tvsec * 1_000_000 + info.pbi_start_tvusec, - boot: format!("{}:{}", boot.tv_sec, boot.tv_usec), + boot: boot_session()?, })) } + +/// This boot's incarnation: `kern.bootsessionuuid`, the Darwin peer of Linux `boot_id`. +/// +/// `kern.boottime` is calendar time, and XNU moves it whenever the wall clock is +/// stepped (`timed` calls `settimeofday` routinely), so it never names a boot. An +/// unavailable or malformed session is an error; there is no calendar fallback. +#[cfg(target_os = "macos")] +fn boot_session() -> Result { + let mut buffer = [0u8; 37]; + let length = host::sysctl(c"kern.bootsessionuuid", &mut buffer) + .map_err(|_| error("Host boot identity unavailable."))?; + // XNU publishes uppercase UUID text with its terminating NUL. + match &buffer[..length] { + [text @ .., 0] => std::str::from_utf8(text).ok(), + _ => None, + } + .map(str::to_ascii_lowercase) + .filter(|text| canonical_boot(text)) + .ok_or_else(|| error("Host boot identity unavailable.")) +} + +/// Native host reads behind one seam, so tests can simulate host changes (a clock +/// step, a new boot session, a refused sysctl) without touching the host. +#[cfg(target_os = "macos")] +mod host { + /// Reads one sysctl into `buffer` and returns the byte length written. + pub(super) fn sysctl(name: &std::ffi::CStr, buffer: &mut [u8]) -> std::io::Result { + #[cfg(test)] + if let Some(value) = test::value(name) { + let value = value.ok_or_else(|| std::io::Error::from_raw_os_error(libc::ENOENT))?; + if value.len() > buffer.len() { + return Err(std::io::Error::from_raw_os_error(libc::ENOMEM)); + } + buffer[..value.len()].copy_from_slice(&value); + return Ok(value.len()); + } + let mut length = buffer.len(); + // SAFETY: the read-only sysctl writes at most this live buffer's length; its + // size pointer is live, newp is null, and no pointer is retained. + if unsafe { + libc::sysctlbyname( + name.as_ptr(), + buffer.as_mut_ptr().cast(), + &mut length, + std::ptr::null_mut(), + 0, + ) + } != 0 + { + return Err(std::io::Error::last_os_error()); + } + Ok(length) + } + + #[cfg(test)] + pub(super) mod test { + use std::{cell::RefCell, collections::BTreeMap}; + type Values = BTreeMap>>; + thread_local! { + static VALUES: RefCell> = const { RefCell::new(None) }; + } + pub(in super::super) fn value(name: &std::ffi::CStr) -> Option>> { + VALUES.with(|cell| cell.borrow().as_ref()?.get(name.to_str().ok()?).cloned()) + } + /// This thread's sysctl overrides; names never set still read the real host. + pub(in super::super) struct Host(Option); + impl Host { + pub(in super::super) fn new() -> Self { + Self(VALUES.with(|cell| cell.replace(Some(Values::new())))) + } + /// `None` makes the sysctl fail as if the host refused it. + pub(in super::super) fn set(&self, name: &str, value: Option<&[u8]>) { + VALUES.with(|cell| { + cell.borrow_mut() + .as_mut() + .expect("override active") + .insert(name.into(), value.map(<[u8]>::to_vec)) + }); + } + } + impl Drop for Host { + fn drop(&mut self) { + VALUES.with(|cell| cell.replace(self.0.take())); + } + } + } +} + +/// A canonical lowercase, nonzero UUID: the only boot form either platform records. +fn canonical_boot(value: &str) -> bool { + let bytes = value.as_bytes(); + bytes.len() == 36 + && bytes.iter().enumerate().all(|(i, byte)| { + if matches!(i, 8 | 13 | 18 | 23) { + *byte == b'-' + } else { + byte.is_ascii_digit() || (b'a'..=b'f').contains(byte) + } + }) + && bytes + .iter() + .any(|byte| matches!(byte, b'1'..=b'9' | b'a'..=b'f')) +} #[cfg(target_os = "linux")] pub fn inspect(pid: i32) -> Result { use std::os::unix::fs::MetadataExt; @@ -116,7 +207,7 @@ pub fn inspect(pid: i32) -> Result { .map_err(error)? .trim() .to_owned(); - if boot.is_empty() { + if !canonical_boot(&boot) { return Err(error("Host boot identity unavailable.")); } Ok(Observed::Present(ProcessIdentity { @@ -128,7 +219,19 @@ pub fn inspect(pid: i32) -> Result { })) } +/// Exact match of a launch identity against a fresh observation of the same PID. +/// +/// A recorded boot that is not a canonical boot UUID is refused first. Builds before +/// HACK-1429 recorded Darwin boots as `kern.boottime` text (`seconds:micros`), which +/// names no stable boot, so such an identity never authorizes a signal, not even +/// against an observation in the same retired format. Receipts keep that value only +/// as diagnostic history; nothing migrates it. pub fn verify(recorded: &ProcessIdentity, observed: &ProcessIdentity) -> Result<()> { + if !canonical_boot(&recorded.boot) { + return Err(error( + "Owned child identity uses a retired boot format; no signal sent.", + )); + } if recorded != observed || recorded.pid <= 1 || recorded.start == 0 @@ -210,12 +313,12 @@ mod tests { uid: unsafe { libc::geteuid() }, process_group: 40000, start: 100, - boot: "boot-a".into(), + boot: "0f1e2d3c-4b5a-4968-8776-655443322110".into(), }; verify(&identity, &identity).unwrap(); for changed in [ ProcessIdentity { - boot: "boot-b".into(), + boot: "0f1e2d3c-4b5a-4968-8776-655443322111".into(), ..identity.clone() }, ProcessIdentity { @@ -234,4 +337,120 @@ mod tests { assert!(verify(&identity, &changed).is_err()); } } + + /// An owned, unreaped group leader: its PID cannot be reused while the test runs. + struct Leader(std::process::Child); + impl Leader { + fn spawn() -> Self { + Self( + std::process::Command::new("/bin/sh") + .args(["-c", "read line"]) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .process_group(0) + .spawn() + .unwrap(), + ) + } + fn pid(&self) -> i32 { + self.0.id() as i32 + } + } + impl Drop for Leader { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + + #[cfg(target_os = "macos")] + #[test] + fn a_wall_clock_step_keeps_the_owned_identity_and_a_new_boot_session_refuses() { + let child = Leader::spawn(); + let host = host::test::Host::new(); + // Darwin's timeval: 64-bit seconds, 32-bit microseconds, 4 padding bytes. + let calendar_boot = |micros: i64| { + let mut bytes = (micros / 1_000_000).to_ne_bytes().to_vec(); + bytes.extend(((micros % 1_000_000) as i32).to_ne_bytes()); + bytes.extend([0; 4]); + bytes + }; + let booted = 1_700_000_000_250_000; + host.set("kern.boottime", Some(&calendar_boot(booted)[..])); + host.set( + "kern.bootsessionuuid", + Some(&b"4F1C2B3A-5D6E-4F70-8192-A3B4C5D6E7F3\0"[..]), + ); + let spawned = observe(child.pid()).unwrap(); + assert_eq!(spawned.boot, "4f1c2b3a-5d6e-4f70-8192-a3b4c5d6e7f3"); + verify(&spawned, &spawned).unwrap(); + // `timed` steps the wall clock with settimeofday (HACK-1429: -73 ms, 15 ms + // after spawn), and XNU moves kern.boottime by the same step either way. + for step in [-73_000, 234_785, -5_000_000] { + host.set("kern.boottime", Some(&calendar_boot(booted + step)[..])); + let stepped = observe(child.pid()).unwrap(); + assert_eq!(stepped, spawned); + verify(&spawned, &stepped).unwrap(); + } + // A different boot session is a different incarnation: refused. + host.set( + "kern.bootsessionuuid", + Some(&b"4F1C2B3A-5D6E-4F70-8192-A3B4C5D6E7F4\0"[..]), + ); + assert_eq!( + verify(&spawned, &observe(child.pid()).unwrap()) + .unwrap_err() + .message, + "Owned child boot/start/UID/group identity mismatch; no signal sent." + ); + // An unavailable or malformed session fails closed; calendar time is never a + // fallback, even though kern.boottime still reads. + for session in [ + None, + Some(&b""[..]), + Some(&b"4F1C2B3A-5D6E-4F70-8192-A3B4C5D6E7F3"[..]), + Some(&b"00000000-0000-0000-0000-000000000000\0"[..]), + Some(&b"4F1C2B3A_5D6E-4F70-8192-A3B4C5D6E7F3\0"[..]), + Some(&b"4F1C2B3A-5D6E-4F70-8192-A3B4C5D6E7FG\0"[..]), + ] { + host.set("kern.bootsessionuuid", session); + assert_eq!( + inspect(child.pid()).unwrap_err().message, + "Host boot identity unavailable." + ); + } + drop(host); + // The real host's session is canonical and stable, and start is the kernel's + // own fork-time snapshot, independent of every simulated host value. + let real = observe(child.pid()).unwrap(); + assert!(canonical_boot(&real.boot), "{real:?}"); + assert_eq!(real.start, spawned.start); + assert_eq!(real, observe(child.pid()).unwrap()); + } + + #[test] + fn an_identity_recorded_in_the_retired_calendar_boot_format_never_authorizes_a_signal() { + let child = Leader::spawn(); + let live = observe(child.pid()).unwrap(); + verify(&live, &live).unwrap(); + // A pre-upgrade Darwin receipt (boot = kern.boottime "seconds:micros") still + // deserializes, so journals holding in-flight or historical jobs stay readable. + let legacy: ProcessIdentity = serde_json::from_value(serde_json::json!({ + "pid": live.pid, + "uid": live.uid, + "process_group": live.process_group, + "start": live.start, + "boot": "1700000000:250000", + })) + .unwrap(); + // It names no stable boot, so it refuses against this same live child and + // against an observer that still reports the retired format. + for observed in [&live, &legacy] { + assert_eq!( + verify(&legacy, observed).unwrap_err().message, + "Owned child identity uses a retired boot format; no signal sent." + ); + } + } } diff --git a/packages/runtime-core/src/node/supervisor.rs b/packages/runtime-core/src/node/supervisor.rs index cbcfa1284..d5298929c 100644 --- a/packages/runtime-core/src/node/supervisor.rs +++ b/packages/runtime-core/src/node/supervisor.rs @@ -7,6 +7,10 @@ use std::path::Path; use std::process::{Child, Command, Stdio}; use std::time::{Duration, Instant}; +/// How long a child gets to settle after this owner's first SIGTERM or SIGKILL +/// before a passed termination deadline quarantines the job. +const TERMINATION_GRACE: Duration = Duration::from_millis(2000); + /// Only this live parent may signal its unreaped child's process group. Recovery never /// signals a recorded PID. A crashed supervisor leaves an explicitly uncertain receipt. pub fn supervise(root: &Path, executable: &Path, checkout: &Path, id: &str) -> Result<()> { @@ -114,6 +118,7 @@ pub fn supervise(root: &Path, executable: &Path, checkout: &Path, id: &str) -> R &mut child, receipt.execution_timeout_ms, identity.as_ref(), + &mut super::ownership::inspect, &mut reaped, ); if result.is_err() && !reaped { @@ -139,12 +144,16 @@ pub fn supervise(root: &Path, executable: &Path, checkout: &Path, id: &str) -> R } result } +/// Owns `child` until it is reaped. `inspect` is the native identity read taken +/// before every signal (`ownership::inspect` in production; tests inject kernel +/// answers such as Darwin's `Gone` window to drive each `Authority`). fn run_child( store: &mut Store, id: &str, child: &mut Child, timeout_ms: u64, identity: std::result::Result<&super::ProcessIdentity, &crate::CandidateError>, + inspect: &mut dyn FnMut(i32) -> Result, reaped: &mut bool, ) -> Result<()> { let identity = identity.map_err(|e| error(&e.message))?; @@ -171,6 +180,7 @@ fn run_child( let started = Instant::now(); let mut persisted = Instant::now(); let mut stopping: Option<(Instant, &'static str)> = None; + let mut killed: Option = None; loop { drain(&mut stdout, &mut output, &mut truncated)?; drain(&mut stderr, &mut errors, &mut truncated)?; @@ -198,11 +208,7 @@ fn run_child( } else { "timed_out" }; - match verify_before_signal( - child, - identity, - super::ownership::inspect(child.id() as i32), - )? { + match verify_before_signal(child, identity, inspect(child.id() as i32))? { Authority::Signal => { signal_group(child, libc::SIGTERM)?; stopping = Some((Instant::now(), reason)); @@ -214,12 +220,11 @@ fn run_child( } if let Some((since, _)) = stopping { if since.elapsed() >= Duration::from_millis(150) { - match verify_before_signal( - child, - identity, - super::ownership::inspect(child.id() as i32), - )? { - Authority::Signal => signal_group(child, libc::SIGKILL)?, + match verify_before_signal(child, identity, inspect(child.id() as i32))? { + Authority::Signal => { + signal_group(child, libc::SIGKILL)?; + killed.get_or_insert_with(Instant::now); + } Authority::Exited => break, // Our SIGKILL is settling: Darwin drops the leader's metadata // before waitid can report the exit. Keep the stop reason. @@ -227,7 +232,14 @@ fn run_child( } } } - if started.elapsed() > Duration::from_millis(timeout_ms + 2000) { + // The deadline bounds an unresponsive child, not a late owner. A pass that + // reaches its first SIGTERM or SIGKILL after the deadline (CPU starvation, a + // journal busy wait while a restarted node opens it) still gives that signal + // its own settling window. Each anchor is set once, so this stays bounded. + let settling = killed.or(stopping.map(|(since, _)| since)); + if started.elapsed() > Duration::from_millis(timeout_ms + 2000) + && settling.is_none_or(|since| since.elapsed() > TERMINATION_GRACE) + { return Err(error("Child termination deadline exceeded.")); } std::thread::sleep(Duration::from_millis(10)); @@ -549,4 +561,243 @@ mod tests { assert_eq!(inspect(child.0.id() as i32).unwrap(), Observed::Gone); assert!(verify_before_signal(&child.0, &identity, Ok(Observed::Gone)).is_err()); } + + /// One owned job for `run_child`: a private journal holding a running receipt, and + /// a TERM-ignoring group leader that exits 3 on its own when its stdin closes. + /// + /// `new` returns only after the leader has printed `ready`, which its shell does + /// after `trap '' TERM` is installed. Without that handshake a starved shell can + /// still have the default TERM action when a short timeout signals it, and the + /// tests that rely on SIGTERM being ignored would then flake. + struct Job { + root: std::path::PathBuf, + store: Store, + id: String, + child: Child, + stdin: Option, + identity: super::super::ProcessIdentity, + } + impl Job { + fn new(timeout_ms: u64) -> Self { + let root = crate::private_test_root( + &std::env::temp_dir().canonicalize().unwrap(), + "hkl-node-run", + ); + let store = Store::open(&root, true).unwrap(); + let id = "a".repeat(64); + super::super::test_support::job(&store, &id, "running", timeout_ms); + let mut child = Command::new("/bin/sh") + .args(["-c", "trap '' TERM; echo ready; read line; exit 3"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .process_group(0) + .spawn() + .unwrap(); + let stdin = child.stdin.take(); + // Exactly the handshake line: run_child later drains the rest of stdout. + let mut ready = [0_u8; 6]; + child + .stdout + .as_mut() + .unwrap() + .read_exact(&mut ready) + .unwrap(); + assert_eq!(&ready, b"ready\n"); + let identity = super::super::ownership::observe(child.id() as i32).unwrap(); + Self { + root, + store, + id, + child, + stdin, + identity, + } + } + fn pid(&self) -> i32 { + self.child.id() as i32 + } + fn run( + &mut self, + inspect: &mut dyn FnMut(i32) -> Result, + ) -> Result { + let mut reaped = false; + let timeout = self.store.get(&self.id)?.execution_timeout_ms; + run_child( + &mut self.store, + &self.id, + &mut self.child, + timeout, + Ok(&self.identity), + inspect, + &mut reaped, + )?; + assert!(reaped); + self.store.get(&self.id) + } + } + impl Drop for Job { + fn drop(&mut self) { + // Child caches a reaped status, so this never signals a reused PID. + let _ = self.child.kill(); + let _ = self.child.wait(); + let _ = std::fs::remove_dir_all(&self.root); + } + } + /// Whether the kernel has published this child's exit, observed without reaping. + fn exit_published(pid: i32) -> bool { + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: WNOWAIT observes this test's own child without reaping it. + assert_eq!( + unsafe { + libc::waitid( + libc::P_PID, + pid as libc::id_t, + &mut info, + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + }, + 0 + ); + #[cfg(target_os = "macos")] + return info.si_pid != 0; + #[cfg(target_os = "linux")] + return unsafe { info.si_pid() } != 0; + } + fn until_exit_published(pid: i32) { + let deadline = Instant::now() + Duration::from_secs(5); + while !exit_published(pid) { + assert!(Instant::now() < deadline, "child exit was not published"); + std::thread::sleep(Duration::from_millis(5)); + } + } + + #[test] + fn run_child_defers_a_gone_live_child_at_timeout_and_keeps_its_own_exit() { + use super::super::ownership::Observed; + let mut job = Job::new(50); + let pid = job.pid(); + let mut stdin = job.stdin.take(); + let mut calls = 0; + let receipt = job + .run(&mut |observed| { + assert_eq!(observed, pid); + calls += 1; + if calls == 1 { + // Darwin's ESRCH window: no live entry, exit not yet published. + // Exiting sends nothing and records no stop reason. + assert!(!exit_published(pid)); + } else { + // The child then exits on its own before anything signalled it. + drop(stdin.take()); + until_exit_published(pid); + } + Ok(Observed::Gone) + }) + .unwrap(); + assert_eq!(calls, 2); + assert_eq!(receipt.state, "failed", "{receipt:?}"); + assert_eq!(receipt.exit_code, Some(3), "{receipt:?}"); + assert_eq!(receipt.signal, None, "{receipt:?}"); + assert_eq!(receipt.detail, None, "{receipt:?}"); + } + + #[test] + fn run_child_never_kills_during_a_gone_window_and_keeps_its_stop_reason() { + use super::super::ownership::{Observed, inspect}; + let mut job = Job::new(50); + let pid = job.pid(); + let mut calls = 0; + let mut alive_after_gone = None; + let receipt = job + .run(&mut |observed| { + calls += 1; + match calls { + // After SIGTERM (ignored), the kill check reads Gone while the + // child is still alive: no SIGKILL yet, and the reason stays. + 2 => { + assert!(!exit_published(pid)); + Ok(Observed::Gone) + } + 3 => { + alive_after_gone = + Some(!exit_published(pid) && unsafe { libc::kill(pid, 0) } == 0); + inspect(observed) + } + _ => inspect(observed), + } + }) + .unwrap(); + assert!(calls >= 3, "observed again after Gone: {calls}"); + assert_eq!(alive_after_gone, Some(true)); + assert_eq!(receipt.state, "timed_out", "{receipt:?}"); + assert_eq!(receipt.signal, Some(libc::SIGKILL), "{receipt:?}"); + assert_eq!(receipt.detail, None, "{receipt:?}"); + } + + #[test] + fn run_child_bounds_an_endless_gone_window_by_its_deadline_without_signalling() { + use super::super::ownership::Observed; + let mut job = Job::new(20); + let pid = job.pid(); + let failure = job.run(&mut |_| Ok(Observed::Gone)).unwrap_err(); + assert_eq!(failure.message, "Child termination deadline exceeded."); + // Gone never authorized a signal: the TERM-ignoring child still waits for + // input, and exits with its own status once that input closes. + assert!(!exit_published(pid)); + drop(job.stdin.take()); + assert_eq!(job.child.wait().unwrap().code(), Some(3)); + } + + #[test] + fn a_kill_check_delayed_past_the_deadline_still_settles_as_the_owners_timeout() { + use super::super::ownership::inspect; + let mut job = Job::new(50); + let mut first = None; + let receipt = job + .run(&mut |observed| { + match first { + // Timeout: the live TERM-ignoring child is signalled with SIGTERM. + None => first = Some(Instant::now()), + // The kill check runs late, as when the owner is starved or waits + // on the journal: its SIGKILL lands after the termination deadline + // (timeout + 2 s from a start at least 50 ms before `first`). + Some(first) => { + let late = first + Duration::from_millis(2100); + std::thread::sleep(late.saturating_duration_since(Instant::now())); + } + } + inspect(observed) + }) + .unwrap(); + assert_eq!(receipt.state, "timed_out", "{receipt:?}"); + assert_eq!(receipt.signal, Some(libc::SIGKILL), "{receipt:?}"); + assert_eq!(receipt.detail, None, "{receipt:?}"); + } + + #[test] + fn a_stop_that_never_settles_is_still_bounded_by_its_grace() { + use super::super::ownership::{Observed, inspect}; + let mut job = Job::new(50); + let pid = job.pid(); + let begun = Instant::now(); + let mut calls = 0; + let failure = job + .run(&mut |observed| { + calls += 1; + // A regression that never quarantines fails here instead of hanging. + assert!(begun.elapsed() < Duration::from_secs(10), "unbounded stop"); + if calls == 1 { + inspect(observed) + } else { + Ok(Observed::Gone) + } + }) + .unwrap_err(); + assert_eq!(failure.message, "Child termination deadline exceeded."); + // SIGTERM was ignored and Gone never authorized SIGKILL: still alive. + assert!(!exit_published(pid)); + drop(job.stdin.take()); + assert_eq!(job.child.wait().unwrap().code(), Some(3)); + } } diff --git a/packages/runtime-core/src/node/transport.rs b/packages/runtime-core/src/node/transport.rs index fac9b797f..0e9474c3e 100644 --- a/packages/runtime-core/src/node/transport.rs +++ b/packages/runtime-core/src/node/transport.rs @@ -319,4 +319,74 @@ mod tests { assert!(read_response(&mut client).is_err()); assert!(start.elapsed() < Duration::from_secs(2)); } + + #[test] + fn an_upgraded_node_never_signals_a_pre_upgrade_in_flight_job() { + use std::os::unix::process::CommandExt; + let root = crate::private_test_root( + &std::env::temp_dir().canonicalize().unwrap(), + "hkl-node-upgrade", + ); + let mut store = Store::open(&root, true).unwrap(); + // The pre-upgrade job's live child, recorded with the retired Darwin boot format. + let mut child = Command::new("/bin/sh") + .args(["-c", "read line; exit 3"]) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .process_group(0) + .spawn() + .unwrap(); + let pid = child.id() as i32; + let legacy = crate::node::ProcessIdentity { + pid, + uid: unsafe { libc::geteuid() }, + process_group: pid, + start: 1, + boot: "1700000000:250000".into(), + }; + let id = "b".repeat(64); + super::super::test_support::job(&store, &id, "running", 60_000); + store + .update(&id, |r| { + r.supervisor_pid = Some(pid); + r.supervisor_identity = Some(legacy.clone()); + r.child_pid = Some(pid); + r.child_identity = Some(legacy.clone()); + r.starts = 1; + Ok(()) + }) + .unwrap(); + let before = serde_json::to_value(store.get(&id).unwrap()).unwrap(); + let unused = Path::new("/nonexistent"); + let mut workers = HashMap::new(); + // The old supervisor still holds the job lock: the new node leaves it alone. + let held = try_lock(&store.lock_path(&id).unwrap()).unwrap().unwrap(); + schedule(&mut store, unused, unused, &mut workers).unwrap(); + assert_eq!( + serde_json::to_value(store.get(&id).unwrap()).unwrap(), + before + ); + // Once that supervisor is gone the job is quarantined, still without a signal. + // A sibling test thread's fork can briefly share the lock's open file until its + // exec closes the CLOEXEC descriptor, so allow a bounded retry. + drop(held); + let deadline = Instant::now() + Duration::from_secs(10); + loop { + schedule(&mut store, unused, unused, &mut workers).unwrap(); + if store.get(&id).unwrap().state != "running" || Instant::now() >= deadline { + break; + } + std::thread::sleep(Duration::from_millis(20)); + } + let recovered = store.get(&id).unwrap(); + assert_eq!(recovered.state, "quarantined", "{recovered:?}"); + assert_eq!(recovered.child_identity, Some(legacy)); + assert!(workers.is_empty()); + // The recorded child was never signalled: it exits with its own status. + drop(child.stdin.take()); + assert_eq!(child.wait().unwrap().code(), Some(3)); + drop(store); + std::fs::remove_dir_all(root).unwrap(); + } } diff --git a/packages/runtime-core/src/provider/graph/dependency_slots.rs b/packages/runtime-core/src/provider/graph/dependency_slots.rs index 7d31c5b22..6fa96d81f 100644 --- a/packages/runtime-core/src/provider/graph/dependency_slots.rs +++ b/packages/runtime-core/src/provider/graph/dependency_slots.rs @@ -777,43 +777,85 @@ mod tests { other ); } + /// Hang guard for the claim-writer test, not a correctness bound. A writer can + /// wait behind seven fsync-backed critical sections whose latency is unbounded on + /// a loaded host (HACK-1430: a 3 s per-writer budget expired at load average + /// ~190), so only a lease that is never released should reach this limit. + const CLAIM_LEASE_HANG_GUARD: Duration = Duration::from_secs(120); #[test] fn concurrent_claim_writers_serialize_without_overlapping_slots() { + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering::SeqCst}; + const WRITERS: u8 = 8; let fixture = super::super::tests::Fixture::new(); let provider = fixture.0.join("provider"); let directory = fixture.0.join("claims"); + // Hold the lease until every writer has been refused once, so all of them + // contend before any may claim a slot, whatever the scheduler does. + let held = state::Lock::acquire(&provider).unwrap(); + let refused = Arc::new(AtomicUsize::new(0)); + let inside = Arc::new(AtomicUsize::new(0)); let mut threads = Vec::new(); - for index in 0..8 { + for index in 0..WRITERS { let provider = provider.clone(); let directory = directory.clone(); + let refused = Arc::clone(&refused); + let inside = Arc::clone(&inside); threads.push(std::thread::spawn(move || { - let deadline = Instant::now() + Duration::from_secs(3); - let _lock = loop { + let deadline = Instant::now() + CLAIM_LEASE_HANG_GUARD; + let mut was_refused = false; + let lock = loop { match state::Lock::acquire(&provider) { Ok(lock) => break lock, - Err(e) if e.code == "provider_busy" && Instant::now() < deadline => { - std::thread::yield_now() + Err(e) if e.code == "provider_busy" => { + if !was_refused { + was_refused = true; + refused.fetch_add(1, SeqCst); + } + assert!(Instant::now() < deadline, "E_TEST_CLAIM_LEASE_HANG"); + std::thread::sleep(Duration::from_millis(1)); } Err(e) => panic!("unexpected reservation lock refusal: {}", e.code), } }; + assert!(was_refused, "E_TEST_CLAIM_WRITER_UNCONTENDED"); + // The lease, not the slot selection, must exclude other writers. + assert_eq!(inside.fetch_add(1, SeqCst), 0, "E_TEST_CLAIM_LEASE_OVERLAP"); let mut record = fixture_record(char::from(b'1' + index), 0); - let existing = records(&directory, &record.owner, 8).unwrap(); + let existing = records(&directory, &record.owner, WRITERS).unwrap(); let used = existing .iter() .flat_map(|r| r.slots.values().copied()) .collect(); - record.slots = select(&BTreeSet::from([0]), &used, 8, true).unwrap(); + record.slots = select(&BTreeSet::from([0]), &used, WRITERS, true).unwrap(); store(&directory, &record); + assert_eq!(inside.fetch_sub(1, SeqCst), 1, "E_TEST_CLAIM_LEASE_OVERLAP"); + drop(lock); record.slots[&0] })); } + // No writer can finish while the lease is held; one that did already failed, + // so stop waiting and let join report it. + let deadline = Instant::now() + CLAIM_LEASE_HANG_GUARD; + let mut all_refused = false; + while Instant::now() < deadline && !threads.iter().any(|t| t.is_finished()) { + if refused.load(SeqCst) == usize::from(WRITERS) { + all_refused = true; + break; + } + std::thread::sleep(Duration::from_millis(1)); + } + drop(held); let slots = threads .into_iter() .map(|t| t.join().unwrap()) .collect::>(); - assert_eq!(slots, (0..8).collect()); - assert_eq!(records(&directory, &"a".repeat(32), 8).unwrap().len(), 8); + assert!(all_refused, "E_TEST_CLAIM_WRITERS_NOT_REFUSED"); + assert_eq!(slots, (0..WRITERS).collect()); + assert_eq!( + records(&directory, &"a".repeat(32), WRITERS).unwrap().len(), + usize::from(WRITERS) + ); } #[test] fn legacy_exact_selection_cannot_steal_unbound_durable_reservation() { diff --git a/packages/runtime-core/src/provider/relay_owner/publication/tests.rs b/packages/runtime-core/src/provider/relay_owner/publication/tests.rs index 0377b6359..119f17676 100644 --- a/packages/runtime-core/src/provider/relay_owner/publication/tests.rs +++ b/packages/runtime-core/src/provider/relay_owner/publication/tests.rs @@ -462,6 +462,11 @@ fn dead_witness_rejects_replacement_listener_and_changed_lock() { break pin; } assert!(Instant::now() < deadline); + // Fail fast if the owner died before publishing, not at the deadline. + assert!( + child.0.try_wait().unwrap().is_none(), + "E_TEST_PUBLICATION_CHILD_EXITED" + ); std::thread::sleep(Duration::from_millis(1)); }; kill_and_reap(&mut child, &pin); diff --git a/packages/runtime-core/tests/node_contract.rs b/packages/runtime-core/tests/node_contract.rs index eacd2fb53..7daf26a4d 100644 --- a/packages/runtime-core/tests/node_contract.rs +++ b/packages/runtime-core/tests/node_contract.rs @@ -272,8 +272,16 @@ fn restart_does_not_own_or_duplicate_supervisor() { let request = n.submit("restart", "tree", 10_000, 1200); let accepted = n.ask(&request); let id = accepted["job_id"].as_str().unwrap(); - n.until(|| n.receipt(id).child_pid.is_some()); - let original = n.receipt(id); + // Judge the first receipt that names the child: a second read could land after + // the 1.2 s timeout under load and see a later, legitimately terminal state. + let mut first = None; + n.until(|| { + let receipt = n.receipt(id); + let named = receipt.child_pid.is_some(); + first = named.then_some(receipt); + named + }); + let original = first.unwrap(); assert_eq!(original.state, "running", "{original:?}"); assert!(original.child_identity.is_some(), "{original:?}"); // The restart may land before or after the timeout fires, and the supervisor may diff --git a/tests/e2e/native-config-networks-acceptance.ts b/tests/e2e/native-config-networks-acceptance.ts index b01e572d4..ec8358b56 100644 --- a/tests/e2e/native-config-networks-acceptance.ts +++ b/tests/e2e/native-config-networks-acceptance.ts @@ -11,6 +11,7 @@ import { realpath, symlink, } from "node:fs/promises"; +import { constants as osConstants } from "node:os"; import { dirname, join, resolve } from "node:path"; import type { Readable } from "node:stream"; import type { @@ -1048,10 +1049,14 @@ export async function runNativeNetworkFixtureCommand(opts: { }, timeout); const exited = new Promise((resolveExit, reject) => { launched.once("error", reject); - launched.once("exit", (code) => { + launched.once("exit", (code, signal) => { // Reap descendants of this just-launched detached group; no stale PID reclamation. kill(); - resolveExit(code ?? 128); + // A signaled command reports 128 + its signal number, as a shell + // does, so the signal stays visible (143 is SIGTERM) (HACK-1428). + resolveExit( + code ?? 128 + (signal === null ? 0 : osConstants.signals[signal]) + ); }); }); const stdoutCapture = capture(launched.stdout, output); diff --git a/tests/native-networks-fixture.test.ts b/tests/native-networks-fixture.test.ts index 433324a80..146b5c0f9 100644 --- a/tests/native-networks-fixture.test.ts +++ b/tests/native-networks-fixture.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test"; +import { expect, setDefaultTimeout, test } from "bun:test"; import { createHash } from "node:crypto"; import { chmod, @@ -50,6 +50,16 @@ import { runNativeNetworkFixtureCommand, } from "./e2e/native-config-networks-acceptance.ts"; +/** + * Many tests here start Bun child processes in sequence, and each start can + * take a second on a loaded host. Bun's 5 s default then fires mid-command: + * bun test SIGTERMs the child and the test fails on that child's exit (143, + * or 128 before HACK-1428) instead of on its own assertion. Every fixture + * command and probe keeps its own, tighter timeout. + */ +const SUBPROCESS_TEST_TIMEOUT_MS = 60_000; +setDefaultTimeout(SUBPROCESS_TEST_TIMEOUT_MS); + test("whole-run and cleanup budgets gate command starts and never clip an in-flight command", () => { const started = 1_000_000; const budget = NATIVE_NETWORK_FIXTURE_SCENARIO_TIMEOUT; @@ -512,6 +522,139 @@ test("peer probe attributes a failure to one URL and phase in closed stderr evid } }); +/** + * Runs the peer probe with `Bun.dns.lookup` replaced by a stub that answers + * every name with 127.0.0.1, after `delayMs` for names starting with `slow`. + * The probe names use RFC 6761 `.invalid`, so only the stubbed answer can + * reach the local listener: a second, real resolution fails (HACK-1428). + */ +async function runStubbedPeerProbe(opts: { + readonly root: string; + readonly origin: string; + readonly alias: string; + readonly timeoutMs: number; + readonly delayMs: number; +}): Promise<{ readonly code: number; readonly stderr: string }> { + const stub = `Bun.dns.lookup=async(h)=>{if(h.startsWith("slow"))await Bun.sleep(${opts.delayMs});return [{address:"127.0.0.1",family:4}]};`; + const child = Bun.spawn( + [ + process.execPath, + "-e", + stub + + nativeNetworkFixturePeerProbe({ + origin: opts.origin, + alias: opts.alias, + expected: "primary:peer", + timeoutMs: opts.timeoutMs, + }), + ], + { + cwd: opts.root, + env: { + PATH: process.env.PATH ?? "/usr/bin:/bin", + PEER_ORIGIN: opts.origin, + }, + stdout: "ignore", + stderr: "pipe", + } + ); + const [code, stderr] = await Promise.all([ + child.exited, + new Response(child.stderr).text(), + ]); + return { code, stderr }; +} + +// HACK-1428: an HTTP-phase failure must not hide a second, uncounted resolution. +test("peer probe fetches the address it just resolved", async () => { + const root = await mkdtemp(join(tmpdir(), "hack-network-probe-")); + const hosts: (string | null)[] = []; + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + hosts.push(request.headers.get("host")); + return new Response("primary:peer"); + }, + }); + try { + const origin = `http://probe-1428.invalid:${server.port}`; + expect( + await runStubbedPeerProbe({ + root, + origin, + alias: `http://probe-1428-alias.invalid:${server.port}/a`, + timeoutMs: 4000, + delayMs: 0, + }) + ).toEqual({ code: 0, stderr: "" }); + // Both requests went to the stub's address, not to the unresolvable names. + expect(hosts).toEqual([ + `127.0.0.1:${server.port}`, + `127.0.0.1:${server.port}`, + ]); + } finally { + server.stop(true); + await rm(root, { recursive: true, force: true }); + } +}); + +// HACK-1428: resolution time is charged against the URL's one shared budget. +test("peer probe charges resolution time against the shared per-URL budget", async () => { + const root = await mkdtemp(join(tmpdir(), "hack-network-probe-")); + const server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + fetch(request) { + return new URL(request.url).pathname === "/hang" + ? new Promise(() => undefined) + : new Response("primary:peer"); + }, + }); + try { + const origin = `http://probe-1428.invalid:${server.port}`; + const alias = `http://slow-1428.invalid:${server.port}/hang`; + const budget = 2000; + const delay = 1200; + const hung = await runStubbedPeerProbe({ + root, + origin, + alias, + timeoutMs: budget, + delayMs: delay, + }); + expect(hung.code).toBe(NATIVE_NETWORK_PROBE_EXIT.timeout); + const evidence = nativeNetworkFixtureProbeEvidence(hung.stderr); + expect(evidence).toMatchObject({ url: "alias", phase: "http" }); + expect(evidence?.dnsMs).toBeGreaterThanOrEqual(delay - 50); + expect(evidence?.elapsedMs).toBeGreaterThanOrEqual(budget - 50); + // The HTTP phase gets only what resolution left: a fresh full budget + // would end near budget + delay (3200 ms). + expect(evidence?.elapsedMs).toBeLessThan(budget + delay / 2); + + // Resolution alone exceeding the budget is a resolution timeout. + const stalled = await runStubbedPeerProbe({ + root, + origin, + alias, + timeoutMs: 500, + delayMs: delay, + }); + expect(stalled.code).toBe(NATIVE_NETWORK_PROBE_EXIT.dnsTimeout); + const stalledEvidence = nativeNetworkFixtureProbeEvidence(stalled.stderr); + expect(stalledEvidence).toMatchObject({ + url: "alias", + phase: "dns", + addresses: 0, + }); + expect(stalledEvidence?.elapsedMs).toBeGreaterThanOrEqual(450); + expect(stalledEvidence?.elapsedMs).toBeLessThan(delay); + } finally { + server.stop(true); + await rm(root, { recursive: true, force: true }); + } +}); + test("probe evidence admits only the closed stderr line", () => { const secret = "/private/fixture/secret-path"; expect( @@ -1574,6 +1717,27 @@ test("bounded private capture preserves literal argv, rejects inherited caller c } }); +// HACK-1428: a bare 128 hid which signal ended a fixture command. +test("capture reports a signaled command as 128 plus its signal number", async () => { + const root = await mkdtemp(join(tmpdir(), "native-network-signal-")); + try { + for (const [signal, expected] of [ + ["TERM", 143], + ["KILL", 137], + ] as const) { + const result = await runNativeNetworkFixtureCommand({ + argv: ["/bin/sh", "-c", `kill -${signal} $$`], + cwd: root, + env: { PATH: "/usr/bin:/bin" }, + captures: join(root, "captures"), + }); + expect(result.exitCode).toBe(expected); + } + } finally { + await rm(root, { recursive: true, force: true }); + } +}); + test("capture output overflow fails rather than returning a truncated green command", async () => { const root = await mkdtemp(join(tmpdir(), "native-network-output-")); try {