diff --git a/packages/env/CHANGELOG.md b/packages/env/CHANGELOG.md index 7f16fe545..046169688 100644 --- a/packages/env/CHANGELOG.md +++ b/packages/env/CHANGELOG.md @@ -4,7 +4,7 @@ ### Added -- The daemon's `process` operation runs a program with plain pipes for stdin, stdout and stderr, for programs that speak a protocol on stdin and stdout such as MCP servers: `processWrite` and `processCloseStdin` feed its stdin in arrival order, output arrives as raw bytes per stream, and cancelling or a lost connection kills its process group. `hello` reports it in `features`. `openProcess()` returns a `RemoteProcess`; daemons released before it fail the open with `not_supported`. Unlike terminals, it works on Windows too. +- The daemon's `process` operation runs a program with plain pipes for stdin, stdout and stderr, for programs that speak a protocol on stdin and stdout such as MCP servers: `processWrite` and `processCloseStdin` feed its stdin in arrival order, output arrives as raw bytes per stream, and cancelling, a lost connection or the program's own exit kills its process group (on Windows, its job object), so a wrapper such as `npx` that exits leaves nothing running. `hello` reports it in `features`. `openProcess()` returns a `RemoteProcess`; daemons released before it fail the open with `not_supported`. Unlike terminals, it works on Windows too. ### Fixed diff --git a/packages/env/README.md b/packages/env/README.md index 0ff5c819f..3a864db1f 100644 --- a/packages/env/README.md +++ b/packages/env/README.md @@ -57,10 +57,11 @@ Windows yet. `openProcess(connection, options)` runs a program with plain pipes, for programs that speak a protocol on stdin and stdout, such as an MCP server: `write()` and `closeStdin()` for its input, raw stdout and stderr bytes through -`onStdout` and `onStderr`, `close()` to kill its process group, and `exited` for its exit code. Nothing is echoed, -translated or merged, unlike a terminal; it works on every supported system. A daemon released before it fails the open -with `not_supported`. `argv` runs as an `exec` argv command does: on Windows, batch files such as `npx.cmd` are refused, -as Node refuses them without a shell, so callers start them through `cmd.exe`. +`onStdout` and `onStderr`, `close()` to kill its process group, and `exited` for its exit code. When the program exits, +what it started in its process group is killed too, so a wrapper such as `npx` leaves nothing running. Nothing is +echoed, translated or merged, unlike a terminal; it works on every supported system. A daemon released before it fails +the open with `not_supported`. `argv` runs as an `exec` argv command does: on Windows, batch files such as `npx.cmd` are +refused, as Node refuses them without a shell, so callers start them through `cmd.exe`. ```ts const server = await openProcess(connection, { diff --git a/packages/env/daemon/src/exec.rs b/packages/env/daemon/src/exec.rs index 072a5ad70..6cb279369 100644 --- a/packages/env/daemon/src/exec.rs +++ b/packages/env/daemon/src/exec.rs @@ -215,16 +215,39 @@ pub fn spawn( Ok(child) } -/// Reap `child` on a thread of its own, which sends its exit to `sender`. A thread that cannot start hands the child -/// back with the error, so the caller kills and reaps it. -pub fn start_waiter(child: Child, sender: Sender) -> Result<(), (Child, std::io::Error)> { +/// A `process` program's group, killed once the program exits, and the daemon's groups, which forget the program +/// before it is reaped. +pub struct Ending { + pub group: sys::Group, + pub groups: Groups, +} + +/// Reap `child` on a thread of its own, which sends its exit to `sender`. With `ending`, what the child started in its +/// group is killed once it exits, before the exit is sent. A thread that cannot start hands the child back with the +/// error, so the caller kills and reaps it. +pub fn start_waiter( + child: Child, + ending: Option, + sender: Sender, +) -> Result<(), (Child, std::io::Error)> { // The waiter receives the child once it runs, so a waiter that cannot start leaves the child here to reap. let (give, take) = mpsc::channel::(); let started = thread::Builder::new().spawn(move || { let Ok(mut child) = take.recv() else { return; }; - let _ = sender.send(match child.wait() { + let status = match &ending { + None => child.wait(), + Some(ending) => { + // After a failed wait, ending the group kills the program, so the reap below still returns. + let exited = ending.group.wait_for_exit(&mut child); + ending.group.end(); + // Forgotten before it is reaped: a kill that checks `groups` under its lock never reaches a reused pid. + ending.groups.lock().unwrap().remove(&child.id()); + exited.and(child.wait()) + } + }; + let _ = sender.send(match status { Ok(status) => Message::Exit(status), Err(error) => Message::WaitFailed(error.to_string()), }); @@ -392,7 +415,7 @@ pub fn run( if let Err(error) = spawn_reader(1, stderr, sender.clone(), stop.clone(), gate) { return Err(abandon(child, error)); } - if let Err((child, error)) = start_waiter(child, sender.clone()) { + if let Err((child, error)) = start_waiter(child, None, sender.clone()) { return Err(abandon(child, error)); } drop(sender); diff --git a/packages/env/daemon/src/process.rs b/packages/env/daemon/src/process.rs index 82f4da32f..5854eafa8 100644 --- a/packages/env/daemon/src/process.rs +++ b/packages/env/daemon/src/process.rs @@ -4,7 +4,8 @@ use crate::errors::Failure; use crate::exec::{ - Control, Groups, Launch, Message, kill_group, running, spawn, spawn_reader, start_waiter, + Control, Ending, Groups, Launch, Message, kill_group, running, spawn, spawn_reader, + start_waiter, }; use crate::frame::{EVENT, Frame}; use crate::output::Output; @@ -118,6 +119,12 @@ pub fn run( groups.lock().unwrap().remove(&pid); Failure::new("spawn_error", error.to_string()) }; + // What the program starts in its group dies when it exits: a wrapper such as `npx` that exits does not leave the + // server it started running, out of reach of `close()` and of a lost connection. + let group = match sys::group(&child) { + Ok(group) => group, + Err(error) => return Err(abandon(child, error)), + }; // Like a full pipe: a program that writes faster than the link waits. let stdout = child.stdout.take().unwrap(); if let Err(error) = spawn_reader( @@ -160,7 +167,14 @@ pub fn run( if let Err(error) = started { return Err(abandon(child, error)); } - if let Err((child, error)) = start_waiter(child, sender.clone()) { + if let Err((child, error)) = start_waiter( + child, + Some(Ending { + group, + groups: groups.clone(), + }), + sender.clone(), + ) { return Err(abandon(child, error)); } drop(sender); @@ -184,7 +198,11 @@ pub fn run( let kill = |killed: &mut bool| { if !*killed { *killed = true; - kill_group(pid); + // Only while the program is in `groups`: the waiter removes it under this lock before reaping it. + let groups = groups.lock().unwrap(); + if groups.contains(&pid) { + kill_group(pid); + } } }; if control.killed.load(Ordering::SeqCst) { @@ -222,6 +240,8 @@ pub fn run( } Ok(Message::Eof(stream)) => ended[stream] = true, Ok(Message::Exit(exit)) => { + // The waiter already ended the group; nothing is left to kill. + killed = true; status = Some(Ok(exit)); idle_until = Some(Instant::now() + EXIT_STDIO_GRACE); } diff --git a/packages/env/daemon/src/pty.rs b/packages/env/daemon/src/pty.rs index 9df9dd9fd..a41606e42 100644 --- a/packages/env/daemon/src/pty.rs +++ b/packages/env/daemon/src/pty.rs @@ -220,7 +220,7 @@ pub fn run( if let Err(error) = started { return Err(abandon(child, error)); } - if let Err((child, error)) = start_waiter(child, sender.clone()) { + if let Err((child, error)) = start_waiter(child, None, sender.clone()) { return Err(abandon(child, error)); } drop(sender); diff --git a/packages/env/daemon/src/sys/unix.rs b/packages/env/daemon/src/sys/unix.rs index a8564857a..4125e3abb 100644 --- a/packages/env/daemon/src/sys/unix.rs +++ b/packages/env/daemon/src/sys/unix.rs @@ -354,6 +354,47 @@ pub fn kill_tree(pid: u32) { } } +/// The process group of a `process` program: what the program started there is killed once it exits. +pub struct Group { + pid: u32, +} + +/// The group of a program just started; `configure` already made it a process group of its own. +pub fn group(child: &Child) -> io::Result { + Ok(Group { pid: child.id() }) +} + +impl Group { + /// Wait until the program has exited, leaving it unreaped: its pid, which is the group's id, names no other + /// process until it is reaped. + pub fn wait_for_exit(&self, child: &mut Child) -> io::Result<()> { + loop { + // SAFETY: a zeroed siginfo_t for waitid to fill; WNOWAIT leaves the program unreaped. + let waited = unsafe { + let mut info: libc::siginfo_t = std::mem::zeroed(); + libc::waitid( + libc::P_PID, + child.id() as libc::id_t, + &mut info, + libc::WEXITED | libc::WNOWAIT, + ) + }; + if waited == 0 { + return Ok(()); + } + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::Interrupted { + return Err(error); + } + } + } + + /// Kill what is left in the group, the program too if it still runs. + pub fn end(&self) { + kill_tree(self.pid); + } +} + /// A process killed by a signal has no exit code; report 128 + the signal, as the shell does. pub fn exit_code(status: ExitStatus) -> i64 { i64::from( @@ -381,3 +422,45 @@ pub fn read_dir(path: &str) -> io::Result { pub fn path_identity(_path: &str, metadata: &Metadata, _follow: bool) -> (u64, u64) { identity(metadata) } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + use std::time::{Duration, Instant}; + + #[test] + fn ending_a_group_kills_what_the_program_left_running() { + // A wrapper that starts a background process in its group, prints its pid and exits, as `npx` can. + let mut command = Command::new("/bin/sh"); + command.stdout(Stdio::piped()); + let script = "sleep 30 /dev/null 2>&1 & echo $!"; + configure( + &mut command, + &["-c".to_string(), script.to_string()], + &Map::new(), + true, + ); + let mut child = command.spawn().unwrap(); + let group = group(&child).unwrap(); + let mut stdout = String::new(); + child + .stdout + .take() + .unwrap() + .read_to_string(&mut stdout) + .unwrap(); + let sleeper: i32 = stdout.trim().parse().unwrap(); + group.wait_for_exit(&mut child).unwrap(); + group.end(); + assert!(child.wait().unwrap().success()); + // The killed sleeper is an orphan; it is gone once init has reaped it. + let deadline = Instant::now() + Duration::from_secs(5); + // SAFETY: signal 0 only checks that the process exists. + while unsafe { libc::kill(sleeper, 0) } == 0 && Instant::now() < deadline { + std::thread::sleep(Duration::from_millis(20)); + } + // SAFETY: as above. + assert_ne!(unsafe { libc::kill(sleeper, 0) }, 0); + } +} diff --git a/packages/env/daemon/src/sys/windows.rs b/packages/env/daemon/src/sys/windows.rs index f2a7516d0..0dfa9c07f 100644 --- a/packages/env/daemon/src/sys/windows.rs +++ b/packages/env/daemon/src/sys/windows.rs @@ -33,7 +33,7 @@ use windows_sys::Win32::System::JobObjects::{ AssignProcessToJobObject, CreateJobObjectW, JOB_OBJECT_LIMIT_BREAKAWAY_OK, JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION, JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE, JOB_OBJECT_LIMIT_SILENT_BREAKAWAY_OK, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, - JobObjectExtendedLimitInformation, SetInformationJobObject, + JobObjectExtendedLimitInformation, SetInformationJobObject, TerminateJobObject, }; use windows_sys::Win32::System::Threading::{CREATE_NO_WINDOW, CreateEventW, WaitForSingleObject}; use windows_sys::core::BOOL; @@ -685,6 +685,57 @@ pub fn kill_tree(pid: u32) { .spawn(); } +/// The job object of a `process` program: what is left in it is killed once the program exits, and when the daemon +/// ends (the job is closed with it). It is nested in the daemon's job. +pub struct Group { + job: OwnedHandle, +} + +/// Put a program just started into a job object of its own. The program already runs: a child it starts before this +/// assignment stays outside the job, as one started before `after_spawn` stays outside the daemon's job. Starting it +/// suspended would close that gap. +pub fn group(child: &Child) -> io::Result { + // SAFETY: plain Win32 calls with a zeroed, correctly sized information struct; the job handle is owned from here. + unsafe { + let job = CreateJobObjectW(std::ptr::null(), std::ptr::null()); + if job.is_null() { + return Err(io::Error::last_os_error()); + } + let job = OwnedHandle::from_raw_handle(job as _); + let mut information: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = std::mem::zeroed(); + // No breakaway: a child started after the assignment stays in this job, even if the daemon's job allows + // breakaway. + information.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE; + if SetInformationJobObject( + job.as_raw_handle() as _, + JobObjectExtendedLimitInformation, + (&information as *const JOBOBJECT_EXTENDED_LIMIT_INFORMATION).cast(), + std::mem::size_of::() as u32, + ) == FALSE + || AssignProcessToJobObject(job.as_raw_handle() as _, child.as_raw_handle() as _) + == FALSE + { + return Err(io::Error::last_os_error()); + } + Ok(Group { job }) + } +} + +impl Group { + /// Wait until the program has exited; the child's open handle keeps its pid from naming another process. + pub fn wait_for_exit(&self, child: &mut Child) -> io::Result<()> { + child.wait().map(drop) + } + + /// Kill what is left in the job, the program too if it still runs. + pub fn end(&self) { + // SAFETY: the job handle is valid while `self` lives. + unsafe { + TerminateJobObject(self.job.as_raw_handle() as _, 1); + } + } +} + /// Exit codes are unsigned 32-bit values on Windows (an access violation is 3221225477). pub fn exit_code(status: ExitStatus) -> i64 { i64::from(status.code().unwrap_or(1) as u32) diff --git a/packages/env/docs/protocol.md b/packages/env/docs/protocol.md index b83d876ef..64396fb1c 100644 --- a/packages/env/docs/protocol.md +++ b/packages/env/docs/protocol.md @@ -64,9 +64,10 @@ command as a full pipe would. Each side sends `ping` every 5 seconds. The daemon kills every process group it started and exits after 30 seconds without receiving any bytes, when stdin ends, or on `SIGTERM`, `SIGINT` or `SIGHUP`. A dropped connection or a stopped daemon therefore stops remote commands. On Windows, a terminated daemon cannot clean up: a job object kills -each command's own process, but not processes that one started. The client fails requests of a lost connection with -`{ code: "unknown", lost: true }` and starts a new daemon on the next request; handles belong to the daemon that opened -them, so requests on them fail instead of reaching the new one. +each command's own process, but not processes that one started, except for `process` programs, whose own job object +holds what they start (see `process` for the gap right after the start). The client fails requests of a lost +connection with `{ code: "unknown", lost: true }` and starts a new daemon on the next request; handles belong to the +daemon that opened them, so requests on them fail instead of reaching the new one. ## Operations @@ -155,11 +156,15 @@ build) from the prompt; background jobs are not reported. It fails with `EBADF` `process { argv, cwd, env, inheritEnv }` runs a program with plain pipes for stdin, stdout and stderr (feature `process`), for programs that speak a protocol on stdin and stdout, such as MCP servers. `argv` is required and runs directly, without a shell, resolved and started as an `exec` argv command is (its own process group, the same -environment rules, a job object on Windows). Once the program runs, an event `{ kind: "started", pid }` comes first; -then `{ kind: "output", stream }` events carry raw bytes of `stdout` or `stderr` as payload (not decoded, not -translated). Output waits while more than 4 MiB of output is unsent, like a full pipe. After the program exits, output -is collected until both pipes close or 100 ms pass without output. The result is `{ exitCode }`; a program that cannot -start fails with `spawn_error`. +environment rules), and on Windows also in a job object of its own. When the program exits, whatever it started in its +process group (on Windows, its job object) is killed: a wrapper such as `npx` that exits does not leave the server it +started running. On Unix, a process that moved to a process group or session of its own (`setpgid`, `setsid`) has left +the group and keeps running. On Windows, the program runs before the daemon assigns it to its job objects: a child it +starts in that moment stays outside them and keeps running; children started later cannot leave the job. Once the +program runs, an event `{ kind: "started", pid }` comes first; then `{ kind: "output", stream }` events carry raw bytes +of `stdout` or `stderr` as payload (not decoded, not translated). Output waits while more than 4 MiB of output is +unsent, like a full pipe. After the program exits, output is collected until both pipes close or 100 ms pass without +output. The result is `{ exitCode }`; a program that cannot start fails with `spawn_error`. While a `process` request runs, `processWrite { target }` writes its payload to the program's stdin and `processCloseStdin { target }` closes stdin after the input queued before it; `target` is the id of the `process` diff --git a/packages/env/src/process.ts b/packages/env/src/process.ts index 473e7d4a6..9b461d4c8 100644 --- a/packages/env/src/process.ts +++ b/packages/env/src/process.ts @@ -19,7 +19,9 @@ export interface ProcessOptions { /** * A program running with plain pipes in a pi-env daemon, for programs that speak a protocol on stdin and stdout. It - * ends when the program exits, `close()` kills it, or the connection is lost (the daemon then kills it). + * ends when the program exits, `close()` kills it, or the connection is lost (the daemon then kills it). When the + * program exits, the daemon kills whatever it started in its process group, so a wrapper such as `npx` that exits + * leaves nothing running. */ export class RemoteProcess { readonly pid: number; diff --git a/packages/env/test/process.test.ts b/packages/env/test/process.test.ts index db1bbb505..0397b7f0b 100644 --- a/packages/env/test/process.test.ts +++ b/packages/env/test/process.test.ts @@ -246,13 +246,36 @@ describe("ending", () => { expect(await waitUntilGone(child)).toBe(true); }); - // On Windows the daemon's pipe ends are overlapped; a writer blocked in the standard library's write could not be - // cancelled, so it kept stdin open, still writing, for as long as a descendant held it. - it("stops writing to stdin a descendant still holds once the program has exited", async () => { + it("kills what the program started when it exits", async () => { const { connection, cwd } = connect(); - const count = join(cwd, "count.txt"); - // The descendant inherits stdin and reads it only after the program exited, until its end. - const descendant = ` + const stdout = collector(); + const program = await openProcess(connection, { + cwd, + // A wrapper that exits and leaves the server it started running, as npx can once stdin closes. On Windows the + // server is detached, so libuv's own job object does not take it down with the wrapper: only the daemon's can. + argv: node(` + const child = require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 30000)"], { stdio: "ignore", detached: ${process.platform === "win32"} }); + process.stdout.write(child.pid + "\\n", () => process.exit(0)); + `), + onStdout: stdout.onData, + }); + expect(await program.exited).toBe(0); + const child = Number.parseInt(stdout.bytes().toString("utf8"), 10); + expect(Number.isInteger(child)).toBe(true); + expect(await waitUntilGone(child)).toBe(true); + }); + + // On Windows the daemon's pipe ends are overlapped; a writer blocked in the standard library's write could not be + // cancelled, so it kept stdin open, still writing, for as long as a descendant held it. There a descendant now dies + // with the program's job object; only one started before the program was assigned to the job (see the protocol) + // can still hold stdin, which a test cannot arrange reliably. On Unix the descendant leaves the process group. + it.skipIf(process.platform === "win32")( + "stops writing to stdin a descendant still holds once the program has exited", + async () => { + const { connection, cwd } = connect(); + const count = join(cwd, "count.txt"); + // The descendant inherits stdin and reads it only after the program exited, until its end. + const descendant = ` setTimeout(() => { let length = 0; process.stdin.on("data", (chunk) => { length += chunk.length; }); @@ -260,22 +283,24 @@ describe("ending", () => { }, 1500); setTimeout(() => process.exit(1), 15000); `; - const program = await openProcess(connection, { - cwd, - argv: node(` + const program = await openProcess(connection, { + cwd, + argv: node(` require("node:child_process").spawn(process.execPath, ["-e", ${JSON.stringify(descendant)}], { stdio: ["inherit", "ignore", "ignore"], detached: true }).unref(); setTimeout(() => process.exit(0), 300); `), - onStdout: () => {}, - }); - // Far more than the pipe holds: the writer blocks until it is stopped. - program.write(new Uint8Array(4 * 1024 * 1024)); - expect(await program.exited).toBe(0); - const deadline = Date.now() + 8000; - while (!existsSync(count) && Date.now() < deadline) await new Promise((done) => setTimeout(done, 50)); - const read = Number(existsSync(count) ? readFileSync(count, "utf8") : Number.NaN); - expect(read).toBeLessThan(4 * 1024 * 1024); - }, 20_000); + onStdout: () => {}, + }); + // Far more than the pipe holds: the writer blocks until it is stopped. + program.write(new Uint8Array(4 * 1024 * 1024)); + expect(await program.exited).toBe(0); + const deadline = Date.now() + 8000; + while (!existsSync(count) && Date.now() < deadline) await new Promise((done) => setTimeout(done, 50)); + const read = Number(existsSync(count) ? readFileSync(count, "utf8") : Number.NaN); + expect(read).toBeLessThan(4 * 1024 * 1024); + }, + 20_000, + ); it("kills the program when the connection closes", async () => { const { connection, cwd } = connect();