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/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