diff --git a/crates/freshell-freshagent/src/claude/teardown_tests.rs b/crates/freshell-freshagent/src/claude/teardown_tests.rs index 19ab7cdd2..4d4051c9a 100644 --- a/crates/freshell-freshagent/src/claude/teardown_tests.rs +++ b/crates/freshell-freshagent/src/claude/teardown_tests.rs @@ -304,31 +304,30 @@ async fn server_shutdown_stops_every_session_concurrently_within_its_share_of_th .all(|(_, fields)| fields.get("graceful").map(String::as_str) == Some("false"))); } -#[cfg(unix)] -fn process_exists(pid: i32) -> bool { - // SAFETY: signal 0 only probes for existence. - let result = unsafe { libc::kill(pid, 0) }; - result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) -} - /// G31: the untagged grandchild a sidecar leaves behind is outside the Linux /// `/proc` ownership reap; only the process-tree guard reaches it. #[cfg(unix)] #[tokio::test] async fn teardown_kills_a_grandchild_the_sidecar_left_behind() { + use crate::process_probe::ProcessProbe; + let dir = tempfile::tempdir().unwrap(); let record = dir.path().join("record.txt"); let (session, ready) = fake_session(FAKE_ORPHANS_GRANDCHILD, &record, "orphans").await; - let grandchild: i32 = ready - .strip_prefix("ready ") - .and_then(|pid| pid.trim().parse().ok()) - .expect("the readiness line carries the grandchild pid"); - assert!(process_exists(grandchild)); + let grandchild = ProcessProbe::open( + ready + .strip_prefix("ready ") + .and_then(|pid| pid.trim().parse().ok()) + .expect("the readiness line carries the grandchild pid"), + ); + assert!(grandchild.running()); let confirmation = teardown_removed_session(session, TREE_DEATH_CONFIRM_ROUNDS).await; + // The orphaned grandchild may still be a zombie its adopter has not + // reaped yet; `kill(pid, 0)` would find that, the probe does not. assert!( - !process_exists(grandchild), + grandchild.exited(), "the grandchild did not outlive the teardown" ); assert_eq!(confirmation.label(), "confirmed"); diff --git a/crates/freshell-freshagent/src/claude_spawn.rs b/crates/freshell-freshagent/src/claude_spawn.rs index 5718e479d..ca5e6c699 100644 --- a/crates/freshell-freshagent/src/claude_spawn.rs +++ b/crates/freshell-freshagent/src/claude_spawn.rs @@ -17,14 +17,23 @@ //! - **Unix** (Linux and macOS): the child leads its own process group //! (`process_group(0)`); the SDK spawns the CLI non-detached, so the CLI //! shares that group. Escalation signals the group, and "the tree is gone" -//! means no live process is left in it (on Linux zombies are not counted; -//! they hold no pipes and write nothing). +//! means no live process is left in it. On Linux a zombie is not counted +//! once its last thread has ended (it holds no pipes and writes nothing), +//! but one whose main thread alone ended still runs and is counted. //! - **Windows**: the child is created suspended, assigned to a //! kill-on-close Job Object (`freshell_platform::process_job`), and only //! then resumed, so every process it starts is born in the job. //! Escalation terminates the job, and "the tree is gone" means the job has -//! no active process. If the job cannot be created or assigned, the child -//! runs unguarded and `freshagent.claude.tree_guard_unavailable` is logged. +//! no active process AND every process the guard saw in it has exited (its +//! handle is signaled, as node's is once reaped). The job stops counting a +//! process as soon as it starts to exit, a few milliseconds (more for a +//! large process) before Windows has finished that exit, and may stop +//! listing it then too. So when a stop begins the guard records the job's members +//! and has the job report every process that joins it from then on, and +//! at every check it watches the members reported since, even one already +//! exiting (`freshell_platform::process_job::JobMembers`). If the job +//! cannot be created or assigned, the child runs unguarded and +//! `freshagent.claude.tree_guard_unavailable` is logged. //! - Dropping the guard (an early-return path, a panic, a dropped session) //! kills whatever is left of the tree (Windows: closing the job handle does //! it too), so no path leaks the CLI. @@ -344,6 +353,9 @@ pub async fn graceful_then_kill( ) -> SidecarExit { let started = Instant::now(); let deadline = started + grace; + // Before anything can end a member (node's exit can: on Windows node + // keeps the children it spawns in a kill-on-close job of its own). + guard.watch_members(); request_shutdown(stdin, deadline).await; if let Ok(Ok(status)) = tokio::time::timeout_at(deadline, child.wait()).await { let (stop, drain) = if status.success() { @@ -445,7 +457,7 @@ pub fn spawn_guarded(command: &mut Command) -> std::io::Result<(Child, SidecarTr #[cfg(windows)] fn spawn_in_job(command: &mut Command) -> std::io::Result<(Child, SidecarTreeGuard)> { use freshell_platform::process_job::{ - resume_suspended_process, KillOnCloseJob, CREATE_SUSPENDED, + resume_suspended_process, JobMembers, KillOnCloseJob, CREATE_SUSPENDED, }; let job = match KillOnCloseJob::new() { @@ -466,7 +478,11 @@ fn spawn_in_job(command: &mut Command) -> std::io::Result<(Child, SidecarTreeGua }; let guard = match job.assign_process(process_id) { Ok(()) => SidecarTreeGuard { - tree: Tree::Job { job, gone: false }, + tree: Tree::Job { + job, + members: JobMembers::default(), + gone: false, + }, }, Err(error) => { tree_guard_unavailable(&format!("job assignment failed: {error}")); @@ -504,10 +520,12 @@ enum Tree { gone: bool, }, /// Closing the job's last handle (dropping it) kills every process - /// still in it. + /// still in it. `members` holds the processes seen in it, which must + /// have exited too before the tree is gone (see the module docs). #[cfg(windows)] Job { job: freshell_platform::process_job::KillOnCloseJob, + members: freshell_platform::process_job::JobMembers, gone: bool, }, } @@ -605,6 +623,28 @@ impl SidecarTreeGuard { (status, tree) } + /// Windows: start watching every process in the job not watched yet, + /// and (the first time) every process that joins it from now on (see + /// the module docs). Nothing to do elsewhere: the process group is + /// checked as a whole. + fn watch_members(&mut self) { + #[cfg(windows)] + if let Tree::Job { + job, + members, + gone: false, + } = &mut self.tree + { + if let Err(error) = members.record(job) { + tracing::warn!(target: "freshell_freshagent::claude", + event = "freshagent.claude.tree_members_unwatched", + cause = %error, + "the Claude child's job members could not be watched; the tree stays unconfirmed until they can" + ); + } + } + } + /// Poll until no tree member is alive (marking the guard spent) or /// `budget` elapses. async fn wait_gone(&mut self, budget: Duration) -> bool { @@ -621,8 +661,8 @@ impl SidecarTreeGuard { } } - fn tree_alive(&self) -> bool { - match &self.tree { + fn tree_alive(&mut self) -> bool { + match &mut self.tree { Tree::None => false, #[cfg(unix)] Tree::ProcessGroup { gone: true, .. } => false, @@ -630,10 +670,19 @@ impl SidecarTreeGuard { Tree::ProcessGroup { pgid, gone: false } => process_group_alive(*pgid), #[cfg(windows)] Tree::Job { gone: true, .. } => false, - // A failed query counts as alive: the bounded wait then ends - // as Lingering rather than claiming a confirmation. + // Alive while the job counts an active process or a member it + // held has not finished exiting (see the module docs). A failed + // query counts as alive: the bounded wait then ends as Lingering + // rather than claiming a confirmation. #[cfg(windows)] - Tree::Job { job, gone: false } => job.active_processes().map_or(true, |n| n > 0), + Tree::Job { + job, + members, + gone: false, + } => members + .record(job) + .and_then(|()| Ok(job.active_processes()? > 0 || members.any_running()?)) + .unwrap_or(true), } } @@ -648,11 +697,16 @@ impl SidecarTreeGuard { } /// Hard-kill every process still in the tree. - fn kill_now(&self) { + fn kill_now(&mut self) { + // While the job still lists them (Windows). + self.watch_members(); #[cfg(unix)] self.signal_tree(libc::SIGKILL); #[cfg(windows)] - if let Tree::Job { job, gone: false } = &self.tree { + if let Tree::Job { + job, gone: false, .. + } = &self.tree + { if let Err(error) = job.terminate() { tracing::warn!(target: "freshell_freshagent::claude", event = "freshagent.claude.tree_terminate_failed", @@ -682,7 +736,8 @@ impl Drop for SidecarTreeGuard { } } -/// Whether any live (non-zombie) process remains in process group `pgid`. +/// Whether any live process remains in process group `pgid` (on Linux a +/// zombie is not live once its last thread has ended). #[cfg(unix)] fn process_group_alive(pgid: i32) -> bool { // SAFETY: signal 0 only probes for existence. @@ -694,7 +749,7 @@ fn process_group_alive(pgid: i32) -> bool { { // The group exists; it is alive only if a member is not a zombie // (an orphan waiting for a reaper that may never come in a - // container without one). + // container without one) or is one that still has threads. linux_group_has_live_member(pgid) } #[cfg(not(target_os = "linux"))] @@ -720,15 +775,28 @@ fn linux_group_has_live_member(pgid: i32) -> bool { let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { continue; }; - // `comm` may hold spaces or parentheses: fields resume after the - // last ')' with field 3 (state), then ppid, then pgrp. + // `comm` may hold spaces or parentheses: the fields resume after + // the last ')' at field 3 (state). `proc(5)` numbers them from 1. let Some(rest) = stat.rsplit(')').next() else { continue; }; - let mut fields = rest.split_whitespace(); - let state = fields.next(); - let group = fields.nth(1).and_then(|group| group.parse::().ok()); - if group == Some(pgid) && !matches!(state, Some("Z") | Some("X")) { + let fields: Vec<&str> = rest.split_whitespace().collect(); + let field = |number: usize| fields.get(number - 3).copied(); + if field(5).and_then(|group| group.parse::().ok()) != Some(pgid) { + continue; + } + let live = match field(3) { + Some("X" | "x") => false, + // `Z` is the main thread's state. The process has exited only + // once no other thread (field 20 counts them) is left: one + // whose main thread alone ended runs on, and a killed one's + // other threads may still be finishing their exit. + Some("Z") => field(20) + .and_then(|threads| threads.parse::().ok()) + .is_none_or(|threads| threads > 1), + _ => true, + }; + if live { return true; } } diff --git a/crates/freshell-freshagent/src/lib.rs b/crates/freshell-freshagent/src/lib.rs index f92c8df0e..f1e680085 100644 --- a/crates/freshell-freshagent/src/lib.rs +++ b/crates/freshell-freshagent/src/lib.rs @@ -54,6 +54,11 @@ pub mod native_history; pub mod opencode_ws; pub mod pane_ops; mod pane_resize; +/// Zombie-aware process and process-group probes, shared with the +/// `claude_sidecar_teardown` integration test. +#[cfg(all(test, unix))] +#[path = "../tests/support/process_probe.rs"] +mod process_probe; pub mod rollback_record; pub mod session_handoff; pub mod session_lease; diff --git a/crates/freshell-freshagent/src/model_capabilities_tests.rs b/crates/freshell-freshagent/src/model_capabilities_tests.rs index 71919bba0..7c87dc443 100644 --- a/crates/freshell-freshagent/src/model_capabilities_tests.rs +++ b/crates/freshell-freshagent/src/model_capabilities_tests.rs @@ -667,14 +667,19 @@ export function query() { let error = result.expect_err("a hung catalog probe times out"); assert!(error.message.contains("timed out"), "{error:?}"); - let grandchild: i32 = std::fs::read_to_string(&pid_file) - .expect("the fake SDK started its grandchild") - .trim() - .parse() - .expect("a pid"); - // SAFETY: signal 0 only probes for existence. - let alive = unsafe { libc::kill(grandchild, 0) } == 0; - assert!(!alive, "the probe's grandchild did not outlive the timeout"); + let grandchild = crate::process_probe::ProcessProbe::open( + std::fs::read_to_string(&pid_file) + .expect("the fake SDK started its grandchild") + .trim() + .parse() + .expect("a pid"), + ); + // The killed grandchild may still be a zombie its adopter has not reaped + // yet; `kill(pid, 0)` would find that, the probe does not. + assert!( + grandchild.exited(), + "the probe's grandchild did not outlive the timeout" + ); } // ── route level (model-capabilities-router.ts ports) ───────────────────────── diff --git a/crates/freshell-freshagent/tests/claude_sidecar_teardown.rs b/crates/freshell-freshagent/tests/claude_sidecar_teardown.rs index 7b0a0bd4e..d229d00d6 100644 --- a/crates/freshell-freshagent/tests/claude_sidecar_teardown.rs +++ b/crates/freshell-freshagent/tests/claude_sidecar_teardown.rs @@ -6,10 +6,14 @@ //! macOS in the `electron-build` workflow (the crate's unit tests do not //! compile on Windows, so the cross-platform behavior lives in this //! integration target). They need `node` on PATH (or `FRESHELL_CLAUDE_NODE`), -//! like the crate's other sidecar tests. Only +//! like the crate's other sidecar tests, and on Linux `python3`. Only //! `a_session_stop_confirms_the_whole_sidecar_tree_gone` sets process //! environment (`FRESHELL_CLAUDE_SIDECAR`, `CLAUDE_CONFIG_DIR`), which no //! other test here reads. +//! +//! Whether a process is still running is asked of `support/process_probe.rs`, +//! never of `kill(pid, 0)`, which also finds a killed orphan its adopter has +//! not reaped yet (see that module). use std::path::Path; use std::process::Stdio; @@ -22,6 +26,12 @@ use freshell_freshagent::claude_spawn::{ use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::{Child, ChildStdin, ChildStdout}; +#[path = "support/process_probe.rs"] +mod process_probe; +#[cfg(unix)] +use process_probe::ProcessGroupProbe; +use process_probe::ProcessProbe; + /// Honors the protocol like `index.mjs`: the shutdown line ends the process /// with exit 0. Its SIGTERM handler records that a signal was needed. const FAKE_HONORS_SHUTDOWN: &str = r#" @@ -228,117 +238,214 @@ fn grandchild_pid(ready: &str) -> u32 { .unwrap_or_else(|| panic!("readiness line carries the grandchild pid: {ready}")) } -/// Whether signal 0 reaches `target` (a pid, or `-pgid` for a group). -#[cfg(unix)] -fn signal_zero_reaches(target: i32) -> bool { - // SAFETY: signal 0 only probes for existence. - let result = unsafe { libc::kill(target, 0) }; - result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +/// A node that runs until it is killed, as a test's own (unreaped) child. +/// On Unix it leads its own process group. +fn spawn_idle_node() -> std::process::Child { + let mut command = std::process::Command::new(node()); + command + .args(["-e", "setInterval(() => {}, 1000)"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + #[cfg(unix)] + std::os::unix::process::CommandExt::process_group(&mut command, 0); + command.spawn().expect("spawn an idle node") } -/// `(state, process group)` of `pid` from `/proc//stat`, or `None` once -/// the process is gone. `comm` may hold spaces or parentheses, so the fields -/// resume after the last ')': state, then ppid, then pgrp. -/// -/// The probes below need it because Linux keeps a killed process in the -/// table as a zombie until a reaper collects it: an orphaned grandchild waits -/// for init (asynchronously), or forever in a container without a reaper. -/// `kill(pid, 0)` still reaches a zombie, but a zombie runs nothing and the -/// guard's own tree check does not count one, so these probes do not either. -#[cfg(target_os = "linux")] -fn linux_proc_state(pid: &str) -> Option<(String, i32)> { - let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; - let mut fields = stat.rsplit(')').next()?.split_whitespace(); - let state = fields.next()?.to_string(); - let group = fields.nth(1)?.parse().ok()?; - Some((state, group)) -} +/// How long the probe tests below watch a process that keeps running, to +/// see that it is never reported as exited. +const KEEPS_RUNNING_WATCH: Duration = Duration::from_millis(300); -#[cfg(target_os = "linux")] -fn is_zombie(state: &str) -> bool { - matches!(state, "Z" | "X") +/// The probe the tree assertions below rely on reports a running process +/// (and, on Unix, its group) as running, and as exited once it was killed +/// and reaped. +#[tokio::test] +async fn the_probe_sees_a_running_process_as_running_until_it_is_killed() { + let mut child = spawn_idle_node(); + let process = ProcessProbe::open(child.id()); + #[cfg(unix)] + let group = ProcessGroupProbe::new(child.id() as i32); + + assert!(process.running(), "the process runs"); + #[cfg(unix)] + { + assert!(group.running(), "its group has a running member"); + let (process_exited, group_exited) = tokio::join!( + process.exits_within(KEEPS_RUNNING_WATCH), + group.exits_within(KEEPS_RUNNING_WATCH) + ); + assert!(!process_exited, "a running process does not exit"); + assert!(!group_exited, "nor does its group"); + } + #[cfg(windows)] + assert!( + !process.exits_within(KEEPS_RUNNING_WATCH).await, + "a running process does not exit" + ); + + child.kill().expect("kill the idle node"); + child.wait().expect("reap the idle node"); + + assert!(process.exited(), "a killed, reaped process has exited"); + #[cfg(unix)] + assert!(group.exited(), "its group has no member left"); } -/// Whether a live (non-zombie, on Linux) process remains in group `pgid`. +/// Block until `pid`, a child of this process, has exited, WITHOUT reaping +/// it: it stays a zombie until the caller waits on it. #[cfg(unix)] -fn process_group_has_live_member(pgid: i32) -> bool { - if !signal_zero_reaches(-pgid) { - return false; +fn wait_until_zombie(pid: u32) { + loop { + // SAFETY: `siginfo_t` is plain data that waitid fills in. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: waits on this process's own child; WNOWAIT leaves it + // unreaped. + let result = unsafe { + libc::waitid( + libc::P_PID, + pid as libc::id_t, + &mut info, + libc::WEXITED | libc::WNOWAIT, + ) + }; + if result == 0 { + return; + } + let error = std::io::Error::last_os_error(); + assert_eq!( + error.kind(), + std::io::ErrorKind::Interrupted, + "wait for the child to exit: {error}" + ); } +} + +/// A process that was killed but not yet reaped (a zombie) has exited: it +/// runs nothing, holds no pipes and writes nothing. An orphaned grandchild +/// the guard kills is exactly that until whoever adopted it (init, or the +/// nearest subreaper such as `systemd --user`) reaps it, and on a loaded +/// machine that can come after the guard has confirmed the tree gone. +#[cfg(unix)] +#[tokio::test] +async fn the_probe_counts_a_killed_but_unreaped_process_as_exited() { + let mut child = spawn_idle_node(); + let pid = child.id(); + let process = ProcessProbe::open(pid); + let group = ProcessGroupProbe::new(pid as i32); + assert!(process.running()); + + child.kill().expect("kill the idle node"); + wait_until_zombie(pid); + #[cfg(target_os = "linux")] { - let Ok(entries) = std::fs::read_dir("/proc") else { - // Cannot look: the signal probe above is the answer. - return true; - }; - entries.flatten().any(|entry| { - entry - .file_name() - .to_str() - .filter(|name| name.bytes().all(|b| b.is_ascii_digit())) - .and_then(linux_proc_state) - .is_some_and(|(state, group)| group == pgid && !is_zombie(&state)) - }) + // SAFETY: signal 0 only probes for existence. + assert_eq!( + unsafe { libc::kill(pid as i32, 0) }, + 0, + "the zombie is still in the process table, where kill(pid, 0) finds it" + ); + // SAFETY: as above. + assert_eq!( + unsafe { libc::kill(-(pid as i32), 0) }, + 0, + "and kill(-pgid, 0) finds the group it is still a member of" + ); + // Linux tells the zombie apart at once: no reaper is needed. + let process_exited = process.exited(); + let group_exited = group.exited(); + child.wait().expect("reap the zombie"); + assert!(process_exited, "an unreaped zombie has exited"); + assert!(group_exited, "so has a group holding only a zombie"); } + // Without /proc a zombie cannot be told from a running process, so the + // probes see it exit once it is reaped; reap it shortly, as its adopter + // would. #[cfg(not(target_os = "linux"))] { - true + const REAPED_WITHIN: Duration = Duration::from_secs(5); + let reaper = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(300)); + child.wait().expect("reap the zombie"); + }); + let (process_exited, group_exited) = tokio::join!( + process.exits_within(REAPED_WITHIN), + group.exits_within(REAPED_WITHIN) + ); + reaper.join().expect("the reaper thread"); + assert!(process_exited, "a zombie reaped shortly after has exited"); + assert!(group_exited, "so has the group that held only it"); } } -/// Observes one process's exit. On Windows the handle is opened while the -/// process is known to run, so a reused id can never be mistaken for it. -struct ProcessProbe { - #[cfg(unix)] - pid: u32, - #[cfg(windows)] - watch: freshell_platform::process_job::ProcessExitWatch, -} - -impl ProcessProbe { - fn open(pid: u32) -> Self { - Self { - #[cfg(unix)] - pid, - #[cfg(windows)] - watch: freshell_platform::process_job::ProcessExitWatch::open(pid) - .expect("open the grandchild while it runs"), - } - } - - fn alive(&self) -> bool { - // A zombie is not running (see `linux_proc_state`). - #[cfg(target_os = "linux")] - { - signal_zero_reaches(self.pid as i32) - && linux_proc_state(&self.pid.to_string()) - .is_some_and(|(state, _)| !is_zombie(&state)) - } - #[cfg(all(unix, not(target_os = "linux")))] - { - signal_zero_reaches(self.pid as i32) - } - #[cfg(windows)] +/// Python code whose main thread ends alone (`pthread_exit`) while another +/// thread runs on. Linux then reports the process's state as `Z` (zombie) +/// although it still runs. +#[cfg(target_os = "linux")] +const MAIN_THREAD_ENDS_FIRST: &str = " +import ctypes, threading, time +threading.Thread(target=lambda: time.sleep(120)).start() +ctypes.CDLL(None).pthread_exit(None) +"; + +/// Wait until `pid`'s main thread has ended (its state reads `Z`) and +/// return its `/proc//stat`. +#[cfg(target_os = "linux")] +async fn wait_until_main_thread_ended(pid: u32) -> String { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + loop { + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).expect("read its stat"); + // The state follows the last ')' (the command name may hold one). + if stat + .rsplit(')') + .next() + .map(str::trim_start) + .is_some_and(|rest| rest.starts_with('Z')) { - !self - .watch - .wait(Duration::ZERO) - .expect("wait on the grandchild") + return stat; } + assert!( + std::time::Instant::now() < deadline, + "the main thread never ended: {stat}" + ); + tokio::time::sleep(Duration::from_millis(20)).await; } +} - /// Whether the process exits within `budget`. - #[cfg(windows)] - async fn exits_within(&self, budget: Duration) -> bool { - let deadline = std::time::Instant::now() + budget; - while self.alive() { - if std::time::Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - true - } +/// A process whose main thread ended while another thread still runs is +/// still running, although Linux then reports its state as `Z` (zombie): +/// the probe stays as strict as `kill(pid, 0)` for it. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn the_probe_sees_a_process_whose_main_thread_ended_as_running() { + let mut command = std::process::Command::new("python3"); + command + .args(["-c", MAIN_THREAD_ENDS_FIRST]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + std::os::unix::process::CommandExt::process_group(&mut command, 0); + let mut child = command + .spawn() + .expect("python3 is required by the main-thread-ended probe test"); + let pid = child.id(); + let process = ProcessProbe::open(pid); + let group = ProcessGroupProbe::new(pid as i32); + let stat = wait_until_main_thread_ended(pid).await; + + let running = process.running(); + let group_running = group.running(); + let (process_exited, group_exited) = tokio::join!( + process.exits_within(KEEPS_RUNNING_WATCH), + group.exits_within(KEEPS_RUNNING_WATCH) + ); + child.kill().expect("kill python"); + child.wait().expect("reap python"); + + assert!(running, "its other thread still runs: {stat}"); + assert!(!process_exited, "it does not exit: {stat}"); + assert!(group_running, "its group has a running member: {stat}"); + assert!(!group_exited, "its group does not exit: {stat}"); } #[tokio::test] @@ -355,7 +462,7 @@ async fn a_graceful_exit_does_not_leave_the_grandchild_running() { "node and its grandchild both run inside the guard's job: {:?}", fake.guard.active_processes() ); - assert!(grandchild.alive()); + assert!(grandchild.running()); let grace = Duration::from_millis(500); let exit = graceful_then_kill(fake.child, fake.stdin, fake.guard, grace).await; @@ -365,7 +472,7 @@ async fn a_graceful_exit_does_not_leave_the_grandchild_running() { "node itself honored the shutdown line: {exit:?}" ); assert!( - !grandchild.alive(), + grandchild.exited(), "the grandchild node left behind was killed: {exit:?}" ); assert_eq!( @@ -381,12 +488,63 @@ async fn a_graceful_exit_does_not_leave_the_grandchild_running() { ); let pgid = pgid.expect("the sidecar ran in its own process group"); assert!( - !process_group_has_live_member(pgid), + ProcessGroupProbe::new(pgid).exited(), "no group member survives" ); } } +/// [`FAKE_ORPHANS_GRANDCHILD`] with a grandchild whose main thread ends +/// while another of its threads runs on (`PYTHON` stands for +/// [`MAIN_THREAD_ENDS_FIRST`]): Linux reports it as a zombie although it +/// still runs. +#[cfg(target_os = "linux")] +const FAKE_ORPHANS_A_GRANDCHILD_WHOSE_MAIN_THREAD_ENDED: &str = r#" +const { spawn } = require('child_process') +const grandchild = spawn('python3', ['-c', PYTHON], { stdio: 'ignore' }) +require('readline').createInterface({ input: process.stdin }).on('line', (line) => { + if (JSON.parse(line).type === 'shutdown') process.exit(0) +}) +setInterval(() => {}, 1000) +grandchild.on('spawn', () => process.stdout.write(`ready ${grandchild.pid}\n`)) +"#; + +/// A tree member whose main thread has ended while another of its threads +/// still runs is not gone, although Linux reports its state as `Z`: the +/// guard waits for it and kills it like any other member left behind. +#[cfg(target_os = "linux")] +#[tokio::test] +async fn a_graceful_exit_does_not_leave_a_member_whose_main_thread_ended_running() { + let dir = tempfile::tempdir().unwrap(); + let record = dir.path().join("record.txt"); + let script = FAKE_ORPHANS_A_GRANDCHILD_WHOSE_MAIN_THREAD_ENDED.replace( + "PYTHON", + &serde_json::to_string(MAIN_THREAD_ENDS_FIRST).expect("encode the Python code"), + ); + let fake = spawn_fake(&script, &record).await; + let pid = grandchild_pid(&fake.ready); + let grandchild = ProcessProbe::open(pid); + let stat = wait_until_main_thread_ended(pid).await; + assert!(grandchild.running(), "its other thread runs: {stat}"); + let grace = Duration::from_millis(500); + + let exit = graceful_then_kill(fake.child, fake.stdin, fake.guard, grace).await; + + assert!( + exit.graceful(), + "node itself honored the shutdown line: {exit:?}" + ); + assert_eq!(exit.tree, TreeConfirmation::Confirmed, "{exit:?}"); + assert!( + exit.elapsed >= grace, + "the member counted as running, so it got the rest of the grace period: {exit:?}" + ); + assert!( + grandchild.exited(), + "the member whose main thread ended was killed: {exit:?}" + ); +} + /// Exits 1 on the shutdown line WITHOUT stopping its grandchild: a sidecar /// that crashed on its way down (for example on a write to a stdout nobody /// reads any more) before its shutdown handler could abort the SDK query, @@ -412,9 +570,13 @@ async fn an_abnormal_exit_is_not_graceful_and_the_tree_is_killed_at_once() { let record = dir.path().join("record.txt"); let fake = spawn_fake(FAKE_CRASHES_LEAVING_GRANDCHILD, &record).await; let grandchild = ProcessProbe::open(grandchild_pid(&fake.ready)); - assert!(grandchild.alive()); + assert!(grandchild.running()); + // Far longer than node needs to crash even on a loaded machine (it can + // take over a second there), so granting the tree the rest of it is + // never mistaken for a slow crash. + let grace = Duration::from_secs(20); - let exit = graceful_then_kill(fake.child, fake.stdin, fake.guard, SIDECAR_GRACEFUL_EXIT).await; + let exit = graceful_then_kill(fake.child, fake.stdin, fake.guard, grace).await; assert_eq!( exit.stop, @@ -427,11 +589,144 @@ async fn an_abnormal_exit_is_not_graceful_and_the_tree_is_killed_at_once() { "node's own exit status is reported: {exit:?}" ); assert!( - exit.elapsed < SIDECAR_GRACEFUL_EXIT / 2, + exit.elapsed < grace / 2, "the orphaned tree did not get the rest of the grace period: {exit:?}" ); - assert!(!grandchild.alive(), "the grandchild was killed: {exit:?}"); + assert!(grandchild.exited(), "the grandchild was killed: {exit:?}"); + assert_eq!(exit.tree, TreeConfirmation::Confirmed, "{exit:?}"); +} + +/// [`FAKE_CRASHES_LEAVING_GRANDCHILD`] with a grandchild that holds 512 MB it +/// has written to, so once it is killed, finishing its exit (releasing that +/// memory) takes tens of milliseconds. The readiness line comes once the +/// grandchild holds the memory. +/// +/// That gap is what these tests are about on Windows: a job stops counting a +/// process as active the moment it starts to exit, but Windows signals the +/// process (it has exited) only once the exit is done. On windows-2022 that +/// came about 4 ms after the count for an idle node and 25-30 ms after it +/// for one holding 768 MB, every time. +const FAKE_CRASHES_LEAVING_A_SLOW_EXITING_GRANDCHILD: &str = r#" +const { spawn } = require('child_process') +const grandchild = spawn(process.execPath, ['-e', "const held = Buffer.alloc(512 * 1024 * 1024, 1); setInterval(() => held[0]++, 1000); console.log('holding')"], { stdio: ['ignore', 'pipe', 'ignore'] }) +require('readline').createInterface({ input: process.stdin }).on('line', (line) => { + if (JSON.parse(line).type === 'shutdown') process.exit(1) +}) +setInterval(() => {}, 1000) +grandchild.stdout.once('data', () => process.stdout.write(`ready ${grandchild.pid}\n`)) +"#; + +/// The guard confirms the tree gone only once every process in it has +/// exited, not as soon as the last one has started to: a grandchild still +/// finishing its exit at that moment is not gone yet. On Windows a crashed +/// node's own exit is what kills this grandchild (node keeps the children it +/// spawns in a kill-on-close job of its own), so the guard must already be +/// watching it when node goes. +#[tokio::test] +async fn an_abnormal_exit_is_confirmed_only_once_a_slow_exiting_grandchild_has_exited() { + let dir = tempfile::tempdir().unwrap(); + let record = dir.path().join("record.txt"); + let fake = spawn_fake(FAKE_CRASHES_LEAVING_A_SLOW_EXITING_GRANDCHILD, &record).await; + let grandchild = ProcessProbe::open(grandchild_pid(&fake.ready)); + assert!(grandchild.running()); + + let exit = graceful_then_kill(fake.child, fake.stdin, fake.guard, SIDECAR_GRACEFUL_EXIT).await; + + assert_eq!(exit.stop, SidecarStop::AbnormalExit, "{exit:?}"); + assert_eq!(exit.tree, TreeConfirmation::Confirmed, "{exit:?}"); + assert!( + grandchild.exited(), + "the grandchild had exited when the tree was confirmed gone: {exit:?}" + ); +} + +/// Starts nothing until the shutdown line. Then it starts a grandchild like +/// [`FAKE_CRASHES_LEAVING_A_SLOW_EXITING_GRANDCHILD`]'s, writes its pid to +/// the record file once it holds its memory, and exits 1 as soon as the +/// test creates `.go`. +const FAKE_STARTS_A_SLOW_EXITING_GRANDCHILD_ON_SHUTDOWN_THEN_CRASHES: &str = r#" +const fs = require('fs') +const { spawn } = require('child_process') +const out = process.argv[1] +require('readline').createInterface({ input: process.stdin }).on('line', (line) => { + if (JSON.parse(line).type !== 'shutdown') return + const grandchild = spawn(process.execPath, ['-e', "const held = Buffer.alloc(512 * 1024 * 1024, 1); setInterval(() => held[0]++, 1000); console.log('holding')"], { stdio: ['ignore', 'pipe', 'ignore'] }) + grandchild.stdout.once('data', () => { + fs.writeFileSync(`${out}.tmp`, String(grandchild.pid)) + fs.renameSync(`${out}.tmp`, out) + setInterval(() => { if (fs.existsSync(`${out}.go`)) process.exit(1) }, 5) + }) +}) +setInterval(() => {}, 1000) +process.stdout.write('ready\n') +"#; + +/// The guard also watches a process that joins the tree after the stop has +/// begun, not only those running when it began: here node starts the +/// slow-exiting grandchild after the shutdown line and then crashes. On +/// Windows that crash is what kills the grandchild (node's own kill-on-close +/// job), so it may already be exiting when the guard first looks. +#[tokio::test] +async fn an_abnormal_exit_is_confirmed_only_once_a_grandchild_started_during_the_stop_has_exited() { + let dir = tempfile::tempdir().unwrap(); + let record = dir.path().join("record.txt"); + let fake = spawn_fake( + FAKE_STARTS_A_SLOW_EXITING_GRANDCHILD_ON_SHUTDOWN_THEN_CRASHES, + &record, + ) + .await; + // Ample, so a slow machine's grandchild holds its memory well inside it; + // the crash ends the wait as soon as the test lets it. + let grace = Duration::from_secs(20); + let stop = tokio::spawn(graceful_then_kill( + fake.child, fake.stdin, fake.guard, grace, + )); + let deadline = std::time::Instant::now() + grace; + let pid: u32 = loop { + if let Some(pid) = std::fs::read_to_string(&record) + .ok() + .and_then(|pid| pid.trim().parse().ok()) + { + break pid; + } + assert!( + std::time::Instant::now() < deadline, + "the fake started its grandchild" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + }; + let grandchild = ProcessProbe::open(pid); + assert!(grandchild.running()); + + let mut go = record.into_os_string(); + go.push(".go"); + std::fs::write(go, "").expect("let the fake crash"); + let exit = stop.await.expect("the stop task"); + + assert_eq!(exit.stop, SidecarStop::AbnormalExit, "{exit:?}"); assert_eq!(exit.tree, TreeConfirmation::Confirmed, "{exit:?}"); + assert!( + grandchild.exited(), + "the grandchild had exited when the tree was confirmed gone: {exit:?}" + ); +} + +/// As above, for a tree the guard itself kills while all of it runs. +#[tokio::test] +async fn terminating_the_guard_is_confirmed_only_once_a_slow_exiting_grandchild_has_exited() { + let dir = tempfile::tempdir().unwrap(); + let record = dir.path().join("record.txt"); + let mut fake = spawn_fake(FAKE_CRASHES_LEAVING_A_SLOW_EXITING_GRANDCHILD, &record).await; + let grandchild = ProcessProbe::open(grandchild_pid(&fake.ready)); + assert!(grandchild.running()); + + let confirmation = fake.guard.terminate(&mut fake.child).await; + + assert_eq!(confirmation, TreeConfirmation::Confirmed); + assert!( + grandchild.exited(), + "the grandchild had exited when the tree was confirmed gone" + ); } /// Windows: the guard's job holds node and every descendant; terminating it @@ -450,7 +745,7 @@ async fn terminating_the_guard_empties_the_job() { assert_eq!(confirmation, TreeConfirmation::Confirmed); assert_eq!(fake.guard.active_processes(), Some(0), "the job is empty"); assert!( - !grandchild.alive(), + grandchild.exited(), "the grandchild was in the job and died" ); } @@ -484,25 +779,6 @@ fn kernel_process_group(pid: u32) -> i32 { unsafe { libc::getpgid(pid as i32) } } -/// Whether `pid` is gone, waiting up to five seconds for whoever adopted it to -/// reap it (a killed orphan is briefly a zombie, which `kill(pid, 0)` still -/// finds). -#[cfg(unix)] -async fn process_gone_within_five_seconds(pid: u32) -> bool { - let deadline = std::time::Instant::now() + Duration::from_secs(5); - loop { - // SAFETY: signal 0 only probes for existence. - let result = unsafe { libc::kill(pid as i32, 0) }; - if result != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) { - return true; - } - if std::time::Instant::now() >= deadline { - return false; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } -} - #[cfg(unix)] #[tokio::test] async fn terminating_the_guard_kills_every_member_of_the_process_group() { @@ -535,16 +811,18 @@ async fn terminating_the_guard_kills_every_member_of_the_process_group() { "the background member is in the child's group" ); assert_eq!(guard.process_group(), Some(leader as i32)); + let member = ProcessProbe::open(member); + assert!(member.running()); let confirmation = guard.terminate(&mut child).await; assert_eq!(confirmation, TreeConfirmation::Confirmed); assert!( - process_gone_within_five_seconds(member).await, + member.exited(), "the background member of the group was killed" ); assert!( - !process_group_has_live_member(leader as i32), + ProcessGroupProbe::new(leader as i32).exited(), "no group member survives" ); } @@ -576,14 +854,14 @@ async fn dropping_the_guard_kills_the_process_group() { Some(9), "the drop sent SIGKILL: {status:?}" ); - let deadline = std::time::Instant::now() + Duration::from_secs(5); - while process_group_has_live_member(pgid) { - assert!( - std::time::Instant::now() < deadline, - "the background member of the group survived the drop" - ); - tokio::time::sleep(Duration::from_millis(20)).await; - } + // Nothing confirms a drop's kill, so the member gets time to finish + // exiting. + assert!( + ProcessGroupProbe::new(pgid) + .exits_within(Duration::from_secs(5)) + .await, + "the background member of the group survived the drop" + ); } /// The session sidecar for the `FreshClaudeState` stop test: it answers @@ -712,7 +990,7 @@ async fn a_session_stop_confirms_the_whole_sidecar_tree_gone() { tokio::time::sleep(Duration::from_millis(20)).await; }; let grandchild = ProcessProbe::open(grandchild_pid); - assert!(grandchild.alive()); + assert!(grandchild.running()); state.shutdown().await; std::env::remove_var("CLAUDE_CONFIG_DIR"); @@ -725,7 +1003,7 @@ async fn a_session_stop_confirms_the_whole_sidecar_tree_gone() { "{logged:?}" ); assert!( - !grandchild.alive(), + grandchild.exited(), "the grandchild the sidecar left behind was killed" ); } diff --git a/crates/freshell-freshagent/tests/support/process_probe.rs b/crates/freshell-freshagent/tests/support/process_probe.rs new file mode 100644 index 000000000..0da793de5 --- /dev/null +++ b/crates/freshell-freshagent/tests/support/process_probe.rs @@ -0,0 +1,222 @@ +//! Whether a process a test started, or its whole process group, is still +//! running. A zombie (a process that was killed but not yet reaped) is not: +//! it runs nothing, holds no pipes and writes nothing. +//! +//! Tests must not ask `kill(pid, 0)` instead, because it still finds a +//! zombie. When the sidecar's tree guard kills a grandchild whose parent has +//! already exited, the grandchild is adopted by init or the nearest +//! subreaper (`systemd --user` on a desktop, a container's init) and stays a +//! zombie until that process gets round to reaping it. On a loaded machine +//! that can be well after the guard rightly confirmed the tree gone, which +//! is how `kill(pid, 0)` made `claude_sidecar_teardown` flaky in CI. +//! +//! - **Linux**: [`ProcessProbe::running`] is exact, from `/proc//stat`. +//! A process runs until it is gone, dead (`X`), or a zombie (`Z`) with no +//! other thread left. A `Z` process that still has threads is either one +//! whose main thread ended while another thread runs on, or one a signal +//! killed whose other threads are still finishing their exit. +//! - **Other Unix** (macOS): without `/proc` a zombie cannot be told from a +//! running process, so a zombie counts as running until it is reaped. +//! - **Windows**: exact, from a process handle opened while the process ran. +//! Windows has no zombies. +//! +//! These are the tree guard's own tests of a member being gone +//! (`claude_spawn`: on Linux the same `/proc` rule; on macOS nothing left in +//! the process group, zombies included; on Windows every member's handle +//! signaled). So once the guard has confirmed a tree gone, every member it +//! held has exited by then, and [`ProcessProbe::exited`] asks at once: a +//! guard that confirmed too early fails it even when the member would have +//! died a moment later. Only a kill nothing confirms (dropping the guard, a +//! test reaping its own zombie later) is given time, with `exits_within`. +//! +//! Shared by the `claude_sidecar_teardown` integration test and the crate's +//! unit tests (`lib.rs` includes this file under `cfg(all(test, unix))`). +#![allow(dead_code)] // each including test crate uses a different subset + +use std::time::{Duration, Instant}; + +const POLL: Duration = Duration::from_millis(20); + +/// Observes one process. Open it while the process is known to run: on +/// Linux and Windows a process that later reuses its id is then never +/// mistaken for it. +pub struct ProcessProbe { + #[cfg(unix)] + pid: u32, + /// The kernel's start time for `pid` when the probe was opened. + #[cfg(target_os = "linux")] + start_time: Option, + #[cfg(windows)] + watch: freshell_platform::process_job::ProcessExitWatch, +} + +impl ProcessProbe { + pub fn open(pid: u32) -> Self { + Self { + #[cfg(unix)] + pid, + #[cfg(target_os = "linux")] + start_time: linux::stat(pid).map(|stat| stat.start_time), + #[cfg(windows)] + watch: freshell_platform::process_job::ProcessExitWatch::open(pid) + .expect("open the process while it runs"), + } + } + + /// Whether the process is running now (see the module docs for what + /// that means on each platform). + pub fn running(&self) -> bool { + #[cfg(target_os = "linux")] + { + match linux::stat(self.pid) { + None => false, + Some(stat) + if self + .start_time + .is_some_and(|start| start != stat.start_time) => + { + // The id now names a newer process: ours has exited. + false + } + Some(stat) => stat.running, + } + } + #[cfg(all(unix, not(target_os = "linux")))] + { + signal_zero_finds(self.pid as i32) + } + #[cfg(windows)] + { + !self + .watch + .wait(Duration::ZERO) + .expect("wait on the process") + } + } + + /// Whether the process has exited by now (a zombie has; see the module + /// docs for why a check after a confirmed stop asks at once). + pub fn exited(&self) -> bool { + !self.running() + } + + /// Whether the process exits within `budget`. + pub async fn exits_within(&self, budget: Duration) -> bool { + poll_until_stopped(budget, || self.running()).await + } +} + +/// Observes a process group: it has exited once none of its members runs. +#[cfg(unix)] +pub struct ProcessGroupProbe { + pgid: i32, +} + +#[cfg(unix)] +impl ProcessGroupProbe { + pub fn new(pgid: i32) -> Self { + Self { pgid } + } + + /// Whether any member of the group is running now, in the sense of + /// [`ProcessProbe::running`]. + pub fn running(&self) -> bool { + // ESRCH: nothing at all, not even a zombie, is left in the group. + if !signal_zero_finds(-self.pgid) { + return false; + } + #[cfg(target_os = "linux")] + { + linux::group_has_running_member(self.pgid) + } + #[cfg(not(target_os = "linux"))] + { + true + } + } + + /// Whether every member of the group has exited by now (zombies have). + pub fn exited(&self) -> bool { + !self.running() + } + + /// Whether every member of the group exits within `budget`. + pub async fn exits_within(&self, budget: Duration) -> bool { + poll_until_stopped(budget, || self.running()).await + } +} + +/// `kill(target, 0)` finds a process (or, for a negative target, a group +/// member), zombies included. +#[cfg(unix)] +fn signal_zero_finds(target: i32) -> bool { + // SAFETY: signal 0 only probes for existence. + let result = unsafe { libc::kill(target, 0) }; + result == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) +} + +/// Poll `running` until it answers `false` (`true`) or `budget` ends (`false`). +async fn poll_until_stopped(budget: Duration, running: impl Fn() -> bool) -> bool { + let deadline = Instant::now() + budget; + while running() { + if Instant::now() >= deadline { + return false; + } + tokio::time::sleep(POLL).await; + } + true +} + +#[cfg(target_os = "linux")] +mod linux { + /// What `/proc//stat` says about a process. + pub struct Stat { + pub process_group: i32, + pub start_time: u64, + pub running: bool, + } + + /// The process's stat, or `None` once there is no such process (or, for + /// another user's process, when `/proc` hides it). + pub fn stat(pid: u32) -> Option { + let text = std::fs::read_to_string(format!("/proc/{pid}/stat")).ok()?; + // The command name (field 2) may hold spaces and parentheses, so the + // fields resume after the last ')', at field 3 (state). + let (_, rest) = text + .rsplit_once(')') + .unwrap_or_else(|| panic!("malformed /proc/{pid}/stat: {text}")); + let fields: Vec<&str> = rest.split_whitespace().collect(); + // `proc(5)` numbers the fields from 1. + let field = |index: usize| { + *fields + .get(index - 3) + .unwrap_or_else(|| panic!("/proc/{pid}/stat has no field {index}: {text}")) + }; + let number = |index: usize| -> i64 { + field(index) + .parse() + .unwrap_or_else(|_| panic!("/proc/{pid}/stat field {index} is no number: {text}")) + }; + let threads = number(20); + Some(Stat { + process_group: number(5) as i32, + start_time: number(22) as u64, + running: match field(3) { + // Threads other than the ended main thread are left. + "Z" => threads > 1, + "X" | "x" => false, + _ => true, + }, + }) + } + + /// Whether a running process is in group `pgid`. + pub fn group_has_running_member(pgid: i32) -> bool { + std::fs::read_dir("/proc") + .expect("list /proc") + .flatten() + .filter_map(|entry| entry.file_name().to_str()?.parse::().ok()) + .filter_map(stat) + .any(|stat| stat.process_group == pgid && stat.running) + } +} diff --git a/crates/freshell-platform/Cargo.toml b/crates/freshell-platform/Cargo.toml index 52cc8b24c..61799d6d3 100644 --- a/crates/freshell-platform/Cargo.toml +++ b/crates/freshell-platform/Cargo.toml @@ -26,7 +26,9 @@ windows-sys = { version = "=0.61.2", features = [ "Win32_Foundation", "Win32_Security", "Win32_System_Diagnostics_ToolHelp", + "Win32_System_IO", "Win32_System_JobObjects", + "Win32_System_SystemServices", "Win32_System_Threading", ] } diff --git a/crates/freshell-platform/src/process_job.rs b/crates/freshell-platform/src/process_job.rs index aeaeafe0b..243b3ae9f 100644 --- a/crates/freshell-platform/src/process_job.rs +++ b/crates/freshell-platform/src/process_job.rs @@ -7,6 +7,16 @@ //! job handle (dropping [`KillOnCloseJob`]) terminates every process still in //! the job (`JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`). //! +//! A job stops counting a process as active the moment the process starts +//! to exit (and stops listing it around then), but Windows signals the +//! process (it has exited) only once that exit is done: on windows-2022 about 4 ms later +//! for an idle node and 25-30 ms later for one holding 768 MB, every time. +//! An owner that must know the whole tree has exited therefore watches the +//! members' own handles too ([`JobMembers`]), learning of each member from +//! the job's list and from the job's completion port, which reports every +//! process that joins the job, even one that has started to exit by the +//! time the report is read. +//! //! This crate is tokio-free and does no logging: the blocking wait //! ([`KillOnCloseJob::wait_signaled`]) is meant to run on a blocking thread, //! and errors are returned for the caller to log under its own event names. @@ -111,6 +121,51 @@ impl KillOnCloseJob { Ok(accounting.ActiveProcesses) } + /// The ids of the processes the job counts as active, those in jobs + /// nested in it included. + pub fn process_ids(&self) -> io::Result> { + use windows_sys::Win32::Foundation::ERROR_MORE_DATA; + use windows_sys::Win32::System::JobObjects::{ + JobObjectBasicProcessIdList, QueryInformationJobObject, JOBOBJECT_BASIC_PROCESS_ID_LIST, + }; + + // A header, then the ids; `usize` storage keeps both aligned. + let header = std::mem::offset_of!(JOBOBJECT_BASIC_PROCESS_ID_LIST, ProcessIdList) + / std::mem::size_of::(); + let mut capacity = 16; + loop { + let mut buffer = vec![0usize; header + capacity]; + let queried = unsafe { + QueryInformationJobObject( + self.handle, + JobObjectBasicProcessIdList, + buffer.as_mut_ptr().cast(), + u32::try_from(std::mem::size_of_val(buffer.as_slice())).unwrap_or(u32::MAX), + std::ptr::null_mut(), + ) + }; + let error = (queried == 0).then(io::Error::last_os_error); + // SAFETY: the buffer is `usize`-aligned and longer than the + // header, which the query filled in. + let list = unsafe { &*buffer.as_ptr().cast::() }; + let assigned = list.NumberOfAssignedProcesses as usize; + let listed = (list.NumberOfProcessIdsInList as usize).min(capacity); + let more = match error { + Some(error) if error.raw_os_error() == Some(ERROR_MORE_DATA as i32) => true, + Some(error) => return Err(error), + None => listed == capacity && assigned > listed, + }; + if !more { + return Ok(buffer[header..header + listed] + .iter() + .map(|&id| id as u32) + .collect()); + } + // More processes than room: ask again with room for them. + capacity = assigned.max(capacity) * 2; + } + } + /// Terminate every process in the job (exit code 1). pub fn terminate(&self) -> io::Result<()> { let terminated = @@ -217,6 +272,40 @@ impl ProcessExitWatch { Ok(Self { handle }) } + /// Watch `process_id` only if it is a member of `job` (directly or + /// through a job nested in it), so an id reused by a process outside the + /// job is never watched in its place. `Ok(None)` when no member has that + /// id any more. + pub fn open_member(process_id: u32, job: &KillOnCloseJob) -> io::Result> { + use windows_sys::Win32::Foundation::ERROR_INVALID_PARAMETER; + use windows_sys::Win32::System::JobObjects::IsProcessInJob; + use windows_sys::Win32::System::Threading::{ + OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION, PROCESS_SYNCHRONIZE, + }; + + let handle = unsafe { + OpenProcess( + PROCESS_SYNCHRONIZE | PROCESS_QUERY_LIMITED_INFORMATION, + 0, + process_id, + ) + }; + if handle.is_null() { + let error = io::Error::last_os_error(); + // The id names no process any more: that member is long gone. + if error.raw_os_error() == Some(ERROR_INVALID_PARAMETER as i32) { + return Ok(None); + } + return Err(error); + } + let watch = Self { handle }; + let mut in_job = 0; + if unsafe { IsProcessInJob(watch.handle, job.handle, &mut in_job) } == 0 { + return Err(io::Error::last_os_error()); + } + Ok((in_job != 0).then_some(watch)) + } + /// Whether the process has exited, waiting at most `timeout` for it. pub fn wait(&self, timeout: Duration) -> io::Result { use windows_sys::Win32::Foundation::{WAIT_FAILED, WAIT_OBJECT_0, WAIT_TIMEOUT}; @@ -240,6 +329,154 @@ impl Drop for ProcessExitWatch { } } +/// Handles on the processes in a job, for an owner that must know when +/// every one of them has exited, not just stopped being counted (see the +/// module docs). +/// +/// The first [`Self::record`] watches the processes the job lists and +/// subscribes to the job's reports of every process that joins it from +/// then on. A process that started to exit before that first call may not +/// be listed any more, so the first call belongs before anything that ends +/// the members (a shutdown request, [`KillOnCloseJob::terminate`]); each +/// later call, at every check, also watches every process that joined +/// since, including one already exiting. +#[derive(Default)] +pub struct JobMembers { + watched: Vec<(u32, ProcessExitWatch)>, + joined: Option, +} + +impl JobMembers { + /// Start watching every process in `job` that is not watched yet: those + /// it lists and those it reported joining. + pub fn record(&mut self, job: &KillOnCloseJob) -> io::Result<()> { + // A watched member keeps its id taken until its watch is closed, so + // once the exited ones are forgotten an id seen again is new. + self.forget_exited()?; + let joined = match &self.joined { + Some(joined) => joined, + None => self.joined.insert(JoinedProcesses::subscribe(job)?), + }; + let mut process_ids = joined.take()?; + process_ids.extend(job.process_ids()?); + for process_id in process_ids { + if self + .watched + .iter() + .any(|(watched, _)| *watched == process_id) + { + continue; + } + if let Some(watch) = ProcessExitWatch::open_member(process_id, job)? { + self.watched.push((process_id, watch)); + } + } + Ok(()) + } + + /// Whether a recorded member has not exited yet. + pub fn any_running(&mut self) -> io::Result { + self.forget_exited()?; + Ok(!self.watched.is_empty()) + } + + fn forget_exited(&mut self) -> io::Result<()> { + let mut failure = None; + self.watched + .retain(|(_, watch)| match watch.wait(Duration::ZERO) { + Ok(exited) => !exited, + Err(error) => { + failure.get_or_insert(error); + true + } + }); + failure.map_or(Ok(()), Err) + } +} + +/// A job's completion port, which the job posts a report to for every +/// process that joins it (and, when the port is attached, for every process +/// already in it). Unlike the job's process list, a report stays readable +/// after its process has started to exit. +struct JoinedProcesses { + port: HANDLE, +} + +// SAFETY: a completion port handle is a kernel object handle; dequeuing from +// it is thread-safe, and it is closed exactly once (in `Drop`). +unsafe impl Send for JoinedProcesses {} +unsafe impl Sync for JoinedProcesses {} + +impl JoinedProcesses { + /// Attach a new completion port to `job` (a job takes only one). + fn subscribe(job: &KillOnCloseJob) -> io::Result { + use windows_sys::Win32::Foundation::INVALID_HANDLE_VALUE; + use windows_sys::Win32::System::JobObjects::{ + JobObjectAssociateCompletionPortInformation, SetInformationJobObject, + JOBOBJECT_ASSOCIATE_COMPLETION_PORT, + }; + use windows_sys::Win32::System::IO::CreateIoCompletionPort; + + let port = + unsafe { CreateIoCompletionPort(INVALID_HANDLE_VALUE, std::ptr::null_mut(), 0, 1) }; + if port.is_null() { + return Err(io::Error::last_os_error()); + } + let joined = Self { port }; + let association = JOBOBJECT_ASSOCIATE_COMPLETION_PORT { + CompletionKey: std::ptr::null_mut(), + CompletionPort: joined.port, + }; + let associated = unsafe { + SetInformationJobObject( + job.handle, + JobObjectAssociateCompletionPortInformation, + std::ptr::addr_of!(association).cast(), + std::mem::size_of::() as u32, + ) + }; + if associated == 0 { + return Err(io::Error::last_os_error()); + } + Ok(joined) + } + + /// The ids of the processes reported joining since the last call. Other + /// reports (a process exiting, the job emptying) are dropped. + fn take(&self) -> io::Result> { + use windows_sys::Win32::Foundation::WAIT_TIMEOUT; + use windows_sys::Win32::System::SystemServices::JOB_OBJECT_MSG_NEW_PROCESS; + use windows_sys::Win32::System::IO::{GetQueuedCompletionStatus, OVERLAPPED}; + + let mut joined = Vec::new(); + loop { + let mut message = 0u32; + let mut key = 0usize; + // A job report carries its process id in place of an OVERLAPPED. + let mut process_id: *mut OVERLAPPED = std::ptr::null_mut(); + let dequeued = unsafe { + GetQueuedCompletionStatus(self.port, &mut message, &mut key, &mut process_id, 0) + }; + if dequeued == 0 { + let error = io::Error::last_os_error(); + if process_id.is_null() && error.raw_os_error() == Some(WAIT_TIMEOUT as i32) { + return Ok(joined); + } + return Err(error); + } + if message == JOB_OBJECT_MSG_NEW_PROCESS { + joined.push(process_id as usize as u32); + } + } + } +} + +impl Drop for JoinedProcesses { + fn drop(&mut self) { + unsafe { windows_sys::Win32::Foundation::CloseHandle(self.port) }; + } +} + #[cfg(test)] mod tests { use super::*; @@ -287,6 +524,175 @@ mod tests { child.wait().expect("reap the child"); } + #[test] + fn the_job_lists_the_child_and_its_descendants() { + let job = KillOnCloseJob::new().expect("create the job"); + let mut child = spawn_tree_in(&job); + + let ids = job.process_ids().expect("list the job's processes"); + + assert!(ids.contains(&child.id()), "the child is listed: {ids:?}"); + assert!(ids.len() >= 3, "so are its descendants: {ids:?}"); + job.terminate().expect("terminate the job"); + child.wait().expect("reap the child"); + } + + #[test] + fn a_member_watch_opens_only_on_a_process_in_the_job() { + let job = KillOnCloseJob::new().expect("create the job"); + let mut member = spawn_tree_in(&job); + let mut outsider = Command::new("ping") + .args(["-n", "120", "127.0.0.1"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn a process outside the job"); + + let watched_member = ProcessExitWatch::open_member(member.id(), &job); + let watched_outsider = ProcessExitWatch::open_member(outsider.id(), &job); + + outsider.kill().expect("kill the outsider"); + outsider.wait().expect("reap the outsider"); + job.terminate().expect("terminate the job"); + member.wait().expect("reap the member"); + assert!( + watched_member.expect("open the member").is_some(), + "the job's member is watched" + ); + assert!( + watched_outsider.expect("open the outsider").is_none(), + "a process outside the job is not" + ); + } + + #[test] + fn recorded_members_run_until_every_one_has_exited() { + let job = KillOnCloseJob::new().expect("create the job"); + let mut child = spawn_tree_in(&job); + let separately: Vec = job + .process_ids() + .expect("list the job's processes") + .into_iter() + .map(|id| ProcessExitWatch::open(id).expect("watch a member")) + .collect(); + let mut members = JobMembers::default(); + members.record(&job).expect("record the members"); + assert!(members.any_running().unwrap(), "the members run"); + + job.terminate().expect("terminate the job"); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while members.any_running().expect("check the members") { + assert!( + std::time::Instant::now() < deadline, + "the terminated members exit" + ); + std::thread::sleep(Duration::from_millis(1)); + } + + for watch in &separately { + assert!( + watch.wait(Duration::ZERO).unwrap(), + "every member had exited when none was reported running" + ); + } + child.wait().expect("reap the child"); + } + + /// A long `ping`, spawned suspended into `job` and then resumed. + fn spawn_ping_in(job: &KillOnCloseJob) -> std::process::Child { + use std::os::windows::process::CommandExt; + let child = Command::new("ping") + .args(["-n", "120", "127.0.0.1"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .creation_flags(CREATE_SUSPENDED) + .spawn() + .expect("spawn ping suspended"); + job.assign_suspended_process(child.id()) + .expect("assign the suspended ping and resume it"); + child + } + + #[test] + fn the_job_reports_the_processes_in_it_and_every_one_that_joins() { + let job = KillOnCloseJob::new().expect("create the job"); + let mut tree = spawn_tree_in(&job); + let listed = job.process_ids().expect("list the job's processes"); + + let joined = JoinedProcesses::subscribe(&job).expect("subscribe to the job"); + let present = joined.take().expect("read the reports"); + let mut later = spawn_ping_in(&job); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + let reported_later = loop { + let reported = joined.take().expect("read the reports"); + if !reported.is_empty() || std::time::Instant::now() >= deadline { + break reported; + } + std::thread::sleep(Duration::from_millis(10)); + }; + + job.terminate().expect("terminate the job"); + tree.wait().expect("reap the tree's root"); + later.wait().expect("reap the later process"); + for id in &listed { + assert!( + present.contains(id), + "a process already in the job is reported: {id} of {listed:?}, reported {present:?}" + ); + } + assert_eq!( + reported_later, + [later.id()], + "a process that joins later is reported once" + ); + } + + #[test] + fn a_member_the_job_no_longer_lists_is_still_watched_once_it_was_reported() { + let job = KillOnCloseJob::new().expect("create the job"); + let mut members = JobMembers::default(); + members.record(&job).expect("record the (empty) job"); + let mut member = spawn_ping_in(&job); + // Kept open, this keeps the exited process (and its id) around, as + // a process still finishing its exit is. + let watch = ProcessExitWatch::open(member.id()).expect("watch the member"); + member.kill().expect("kill the member"); + assert!( + watch + .wait(Duration::from_secs(5)) + .expect("wait on the member"), + "the member exits" + ); + // The job drops an exiting process from its list, sometimes only + // just after the process is signaled. + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while job + .process_ids() + .expect("list the job's processes") + .contains(&member.id()) + { + assert!( + std::time::Instant::now() < deadline, + "the job stops listing a member that has exited" + ); + std::thread::sleep(Duration::from_millis(1)); + } + + members.record(&job).expect("record the members"); + + assert!( + members.watched.iter().any(|(id, _)| *id == member.id()), + "the member the job reported joining is watched" + ); + assert!( + !members.any_running().expect("check the members"), + "and it has exited" + ); + member.wait().expect("reap the member"); + } + #[test] fn closing_the_job_kills_every_process_in_it() { let job = KillOnCloseJob::new().expect("create the job");