From 02c808040ae6e2fb6d4e6b92898e0ce56aae89a0 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 10:52:10 -0400 Subject: [PATCH 1/4] fix(runtime-core): name a node job's macOS boot by its session UUID, not calendar boot time On macOS the node job identity recorded `kern.boottime` as its boot. XNU moves that value whenever the wall clock is stepped, and `timed` steps it routinely, so a step between spawn and the timeout signal turned a normal timeout into "Owned child boot/start/UID/group identity mismatch; no signal sent." and a quarantine. The boot is now `kern.bootsessionuuid` (canonical lowercase UUID, the Darwin peer of Linux `boot_id`), read through a sysctl seam with no calendar fallback. The process start stays the kernel's fork-time `p_start` snapshot, which a clock step does not rewrite. `verify` refuses any recorded boot that is not a canonical UUID, so a receipt written before this change in the `seconds:micros` format never authorizes a signal. Tests cover a simulated clock step and boot-session change through the seam, the retired format, a pre-upgrade in-flight job under recovery, the rollback-journal race between check and open, and run_child's Gone-before-exit paths through an injected inspect. The restart contract test now judges the first receipt that names the child. Refs HACK-1429 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/node/journal.rs | 35 +++ packages/runtime-core/src/node/mod.rs | 81 ++++++ packages/runtime-core/src/node/ownership.rs | 257 +++++++++++++++++-- packages/runtime-core/src/node/supervisor.rs | 190 +++++++++++++- packages/runtime-core/src/node/transport.rs | 61 +++++ packages/runtime-core/tests/node_contract.rs | 12 +- 6 files changed, 605 insertions(+), 31 deletions(-) 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..376a55a0d 100644 --- a/packages/runtime-core/src/node/supervisor.rs +++ b/packages/runtime-core/src/node/supervisor.rs @@ -114,6 +114,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 +140,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))?; @@ -198,11 +203,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,11 +215,7 @@ 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), - )? { + match verify_before_signal(child, identity, inspect(child.id() as i32))? { Authority::Signal => signal_group(child, libc::SIGKILL)?, Authority::Exited => break, // Our SIGKILL is settling: Darwin drops the leader's metadata @@ -549,4 +546,177 @@ 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. + 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; read line; exit 3"]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .process_group(0) + .spawn() + .unwrap(); + let stdin = child.stdin.take(); + 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)); + } } diff --git a/packages/runtime-core/src/node/transport.rs b/packages/runtime-core/src/node/transport.rs index fac9b797f..5437cf668 100644 --- a/packages/runtime-core/src/node/transport.rs +++ b/packages/runtime-core/src/node/transport.rs @@ -319,4 +319,65 @@ 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. + drop(held); + schedule(&mut store, unused, unused, &mut workers).unwrap(); + 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/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 From 99ab7626713b4c1bab3ff18d726fa82bf96c97da Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 11:29:01 -0400 Subject: [PATCH 2/4] fix(runtime-core): give a node job's late stop signal its settling window Under load, `restart_does_not_own_or_duplicate_supervisor` failed once in 20 runs as quarantined with "Child termination deadline exceeded." Instrumentation showed the SIGTERM 1.5 s into the job and the first SIGKILL 4.4 s in: one supervisor pass stalled for about 2.8 s (CPU starvation, or a journal busy wait while the restarted node opened the database). The deadline check ran in the same pass as that SIGKILL, so a normal timeout was quarantined before the kill could settle. The termination deadline (timeout + 2 s from start) now also requires 2 s since the owner's first SIGTERM or SIGKILL. Each anchor is set once, so termination stays bounded. A child that is never signalled, for example one that reads as Gone forever, keeps the original deadline. Refs HACK-1429 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/node/supervisor.rs | 71 +++++++++++++++++++- 1 file changed, 69 insertions(+), 2 deletions(-) diff --git a/packages/runtime-core/src/node/supervisor.rs b/packages/runtime-core/src/node/supervisor.rs index 376a55a0d..2f8847f2c 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<()> { @@ -176,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)?; @@ -216,7 +221,10 @@ fn run_child( if let Some((since, _)) = stopping { if since.elapsed() >= Duration::from_millis(150) { match verify_before_signal(child, identity, inspect(child.id() as i32))? { - Authority::Signal => signal_group(child, libc::SIGKILL)?, + 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. @@ -224,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)); @@ -719,4 +734,56 @@ mod tests { 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)); + } } From c98e300496d8032d42506d17d38d63500c7f6149 Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 11:46:17 -0400 Subject: [PATCH 3/4] test(runtime-core): tolerate a sibling fork briefly sharing the recovery test's job lock Under load, a concurrent test thread's fork can share the dropped job-lock file until its exec closes the CLOEXEC descriptor, so the next schedule pass still saw the lock as held (2 failures in 30 loaded runs). The test now retries schedule for a bounded time. It still requires quarantine and an unsignalled child. Refs HACK-1429 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/node/transport.rs | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/packages/runtime-core/src/node/transport.rs b/packages/runtime-core/src/node/transport.rs index 5437cf668..0e9474c3e 100644 --- a/packages/runtime-core/src/node/transport.rs +++ b/packages/runtime-core/src/node/transport.rs @@ -368,8 +368,17 @@ mod tests { 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); - schedule(&mut store, unused, unused, &mut workers).unwrap(); + 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)); From 4cbde7fa1a6717a8c035430e519d278828a8c09c Mon Sep 17 00:00:00 2001 From: Dimitri Kennedy Date: Sat, 24 Oct 2026 13:36:48 -0400 Subject: [PATCH 4/4] test(runtime-core): wait for the run_child fixture's TERM trap before timing out Job::new now returns only after the TERM-ignoring shell prints 'ready', which it does after 'trap '' TERM' is installed. Under CPU load the 50 ms timeout could otherwise SIGTERM the shell before the trap ran, killing it and failing the Gone-window and kill-branch tests that cover Authority::Exiting. Refs HACK-1429 Co-Authored-By: Claude Opus 5.5 --- packages/runtime-core/src/node/supervisor.rs | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/packages/runtime-core/src/node/supervisor.rs b/packages/runtime-core/src/node/supervisor.rs index 2f8847f2c..d5298929c 100644 --- a/packages/runtime-core/src/node/supervisor.rs +++ b/packages/runtime-core/src/node/supervisor.rs @@ -564,6 +564,11 @@ mod tests { /// 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, @@ -582,7 +587,7 @@ mod tests { 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; read line; exit 3"]) + .args(["-c", "trap '' TERM; echo ready; read line; exit 3"]) .stdin(Stdio::piped()) .stdout(Stdio::piped()) .stderr(Stdio::piped()) @@ -590,6 +595,15 @@ mod tests { .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,