From 3bc9ac827951be9304f7ad75d1adbc07e708c5ed Mon Sep 17 00:00:00 2001 From: Malte Ubl Date: Fri, 30 Oct 2026 03:01:06 -0700 Subject: [PATCH] Start POSIX checker servers outside this process's children Node reaps children only from its event loop, which a synchronous compile never returns to, so every replaced or closed tsc-ts process stayed a zombie until the compile ended: 797 on a pi build, most of them syntax parsers replaced every 256 K of source. On Linux and macOS a shell that spawnSync runs and reaps now creates two FIFOs, starts the server in the background on them and exits; init reaps the server. The compiler opens the FIFOs with openSync, as it already opens the Windows named pipe, and closing them ends the session: the server reads end of file and exits, and is never signaled. This also removes the dependency on Node's private stream handles. Windows is unchanged. Co-Authored-By: Claude Opus 5.5 --- .../src/frontend/ts7/rpc-process.test.ts | 76 +++-- .../compiler/src/frontend/ts7/rpc-process.ts | 278 ++++++++++++------ 2 files changed, 235 insertions(+), 119 deletions(-) diff --git a/packages/compiler/src/frontend/ts7/rpc-process.test.ts b/packages/compiler/src/frontend/ts7/rpc-process.test.ts index faad0cc028..e91b0b8200 100644 --- a/packages/compiler/src/frontend/ts7/rpc-process.test.ts +++ b/packages/compiler/src/frontend/ts7/rpc-process.test.ts @@ -1,37 +1,69 @@ -import { execFile } from "node:child_process"; +import { execFile, execFileSync } from "node:child_process"; import { promisify } from "node:util"; import { expect, test } from "vitest"; +import { Ts7FifoIo } from "./rpc-process.js"; const execFileAsync = promisify(execFile); +const posix = process.platform !== "win32"; -test.skipIf(process.platform === "win32")( - "closing the owned server is quiet even when its cancellation handler writes stderr", +function alive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +test.skipIf(!posix)("a FIFO server echoes, and exits when its transport closes", () => { + const io = new Ts7FifoIo("/bin/cat", []); + const bytes = Buffer.from("ping\n"); + expect(io.write(bytes, 0, bytes.length)).toBe(bytes.length); + const reply = Buffer.alloc(16); + expect(reply.subarray(0, io.read(reply, 0, reply.length)).toString()).toBe("ping\n"); + io.close(); + io.close(); + expect(() => io.read(reply, 0, 1)).toThrow("closed"); + // The server is not this process's child: init reaps it once it exits. + const deadline = Date.now() + 5000; + while (alive(io.pid) && Date.now() < deadline) execFileSync("sleep", ["0.01"]); + expect(alive(io.pid)).toBe(false); +}); + +test.skipIf(!posix)("a FIFO server that cannot start fails its first read", () => { + const io = new Ts7FifoIo("/nonexistent/scriptc-ts7-server", []); + try { + expect(io.read(Buffer.alloc(1), 0, 1)).toBe(0); + } finally { + io.close(); + } +}); + +test.skipIf(!posix)( + "replaced and closed checker servers leave no zombie children behind", async () => { - // A ready message ensures the signal handler is installed before close. - // Keep stdin idle, matching an API server waiting for its next request. - const server = ` - process.on('SIGTERM', () => { - process.stderr.write('context canceled\\n'); - process.exit(0); - }); - process.stdout.write(Buffer.from([0x93, 4, 0xc4, 0, 0xc4, 0])); - setInterval(() => {}, 1000); - `; - const moduleUrl = new URL("./rpc-process.ts", import.meta.url).href; + // Source budget 1: every parse after the first replaces the server. + const parserUrl = new URL("./source-parser.ts", import.meta.url).href; + const apiUrl = new URL("./rpc-api.ts", import.meta.url).href; const host = ` - import { spawnTs7Wire } from ${JSON.stringify(moduleUrl)}; - const wire = spawnTs7Wire(process.execPath, ['--eval', ${JSON.stringify(server)}]); - const ready = wire.read(); - if (ready.kind !== 4) throw new Error('server was not ready'); - wire.close(); - wire.close(); + import { execFileSync } from 'node:child_process'; + import { Ts7SourceParser } from ${JSON.stringify(parserUrl)}; + import { Ts7Api } from ${JSON.stringify(apiUrl)}; + const parser = new Ts7SourceParser((options) => new Ts7Api(options), process.cwd(), 1); + for (let n = 0; n < 12; n++) parser.parse('f' + n + '.ts', 'export const v' + n + ' = ' + n + ';', 'ts'); + parser.close(); + const zombies = execFileSync('ps', ['-A', '-o', 'ppid=,stat='], { encoding: 'utf8' }) + .split('\\n') + .map((line) => line.trim().split(/\\s+/)) + .filter(([ppid, stat]) => Number(ppid) === process.pid && (stat ?? '').startsWith('Z')); + console.log(JSON.stringify({ zombies: zombies.length })); `; const result = await execFileAsync( process.execPath, ["--import", "tsx", "--input-type=module", "--eval", host], - { timeout: 10_000 }, + { timeout: 60_000 }, ); - expect(result.stdout).toBe(""); + expect(JSON.parse(result.stdout)).toEqual({ zombies: 0 }); expect(result.stderr).toBe(""); }, ); diff --git a/packages/compiler/src/frontend/ts7/rpc-process.ts b/packages/compiler/src/frontend/ts7/rpc-process.ts index 416f5e4f75..24d8f5a99a 100644 --- a/packages/compiler/src/frontend/ts7/rpc-process.ts +++ b/packages/compiler/src/frontend/ts7/rpc-process.ts @@ -1,98 +1,96 @@ -import { spawn, type ChildProcess } from "node:child_process"; +import { spawn, spawnSync, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { closeSync, openSync, readSync, writeSync } from "node:fs"; -import type { Readable, Writable } from "node:stream"; +import { closeSync, constants, mkdtempSync, openSync, readSync, rmSync, writeSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { Ts7Wire, type Ts7WireIo } from "./rpc-wire.js"; -// Node's public spawn API does not expose synchronous pipe descriptors. -// Keep this host-specific boundary out of the statically compiled protocol. -// Windows uses a named pipe because libuv's Windows handles are not CRT fds. -interface PipeHandle { - fd: number; - setBlocking: (blocking: boolean) => void; -} -type PipeStream = (Readable | Writable) & { - _handle?: PipeHandle; - unref: () => void; -}; +// Node's public spawn API does not expose synchronous pipe descriptors, so +// the server's end of the channel is a named pipe on Windows and a FIFO pair +// elsewhere, both opened with openSync. Keep this host-specific boundary out +// of the statically compiled protocol. -const liveTransports = new Set(); +const liveTransports = new Set(); const sleepBuffer = new Int32Array(new SharedArrayBuffer(4)); function closeLiveTransports(): void { for (const transport of liveTransports) transport.close(); } -/** Own the child and its descriptors together: every failure path closes - * both, and the exit hook exists only while at least one host is alive. */ -class Ts7ProcessIo implements Ts7WireIo { +function track(transport: Ts7WireIo): void { + if (liveTransports.size === 0) process.on("exit", closeLiveTransports); + liveTransports.add(transport); +} + +function untrack(transport: Ts7WireIo): void { + liveTransports.delete(transport); + if (liveTransports.size === 0) process.removeListener("exit", closeLiveTransports); +} + +/** A synchronous read or write that retries an interrupted call and waits + * out a full pipe. */ +function transfer( + operation: ( + fd: number, + buffer: Uint8Array, + offset: number, + length: number, + position: null, + ) => number, + fd: number, + buffer: Uint8Array, + offset: number, + length: number, +): number { + for (;;) { + try { + return operation(fd, buffer, offset, length, null); + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code === "EINTR") continue; + if (code !== "EAGAIN") throw error; + Atomics.wait(sleepBuffer, 0, 0, 1); + } + } +} + +/** Windows: the server is a child process that creates a named pipe + * (`--pipe`), which this process opens. Windows has no zombie processes; the + * child's handle closes with the process. */ +class Ts7NamedPipeIo implements Ts7WireIo { private readonly child: ChildProcess; - private readFd = -1; - private writeFd = -1; private pipeFd = -1; private closed = false; private spawnError: Error | undefined; constructor(executable: string, args: string[]) { - const windows = process.platform === "win32"; const pipe = `\\\\.\\pipe\\scriptc-ts7-${process.pid}-${randomUUID()}`; - this.child = spawn(executable, windows ? [...args, "--pipe", pipe] : args, { - stdio: windows ? ["ignore", "ignore", "inherit"] : ["pipe", "pipe", "inherit"], + this.child = spawn(executable, [...args, "--pipe", pipe], { + stdio: ["ignore", "ignore", "inherit"], windowsHide: true, }); - // A failed spawn emits asynchronously even when descriptor validation - // below has already thrown. Always consume that event. + // A failed spawn emits asynchronously even when the connection below has + // already thrown. Always consume that event. this.child.on("error", (error: Error) => { this.spawnError = error; }); try { if (this.child.pid === undefined) throw new Error(`Unable to start TypeScript server: ${executable}`); - if (windows) { - for (let attempt = 0; attempt < 500; attempt++) { - try { - this.pipeFd = openSync(pipe, "r+"); - break; - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code !== "ENOENT" && code !== "EBUSY") throw error; - Atomics.wait(sleepBuffer, 0, 0, 10); - } + for (let attempt = 0; attempt < 500; attempt++) { + try { + this.pipeFd = openSync(pipe, "r+"); + break; + } catch (error) { + const code = (error as NodeJS.ErrnoException).code; + if (code !== "ENOENT" && code !== "EBUSY") throw error; + Atomics.wait(sleepBuffer, 0, 0, 10); } - if (this.pipeFd === -1) - throw new Error("TypeScript server: timed out connecting to named pipe"); - this.readFd = this.pipeFd; - this.writeFd = this.pipeFd; - } else { - const stdout = this.child.stdout as PipeStream | null; - const stdin = this.child.stdin as PipeStream | null; - const readHandle = stdout?._handle; - const writeHandle = stdin?._handle; - if ( - !stdout || - !stdin || - !readHandle || - !writeHandle || - !Number.isInteger(readHandle.fd) || - readHandle.fd < 0 || - !Number.isInteger(writeHandle.fd) || - writeHandle.fd < 0 || - typeof readHandle.setBlocking !== "function" || - typeof writeHandle.setBlocking !== "function" - ) { - throw new Error("TypeScript server: Node synchronous pipe handles are unavailable"); - } - readHandle.setBlocking(true); - writeHandle.setBlocking(true); - this.child.stdout!.pause(); - stdout.unref(); - stdin.unref(); - this.readFd = readHandle.fd; - this.writeFd = writeHandle.fd; } + if (this.pipeFd === -1) + throw new Error("TypeScript server: timed out connecting to named pipe"); this.child.unref(); - if (liveTransports.size === 0) process.on("exit", closeLiveTransports); - liveTransports.add(this); + track(this); } catch (error) { this.close(); throw error; @@ -102,52 +100,27 @@ class Ts7ProcessIo implements Ts7WireIo { read = (buffer: Uint8Array, offset: number, length: number): number => { if (this.spawnError) throw this.spawnError; if (this.closed) throw new Error("TypeScript server transport is closed"); - for (;;) { - try { - return readSync(this.readFd, buffer, offset, length, null); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === "EINTR") continue; - if (code !== "EAGAIN") throw error; - Atomics.wait(sleepBuffer, 0, 0, 1); - } - } + return transfer(readSync, this.pipeFd, buffer, offset, length); }; write = (buffer: Uint8Array, offset: number, length: number): number => { if (this.spawnError) throw this.spawnError; if (this.closed) throw new Error("TypeScript server transport is closed"); - for (;;) { - try { - return writeSync(this.writeFd, buffer, offset, length, null); - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === "EINTR") continue; - if (code !== "EAGAIN") throw error; - Atomics.wait(sleepBuffer, 0, 0, 1); - } - } + return transfer(writeSync, this.pipeFd, buffer, offset, length); }; close = (): void => { if (this.closed) return; this.closed = true; - liveTransports.delete(this); - if (liveTransports.size === 0) process.removeListener("exit", closeLiveTransports); + untrack(this); if (this.pipeFd !== -1) { try { closeSync(this.pipeFd); } catch { - /* still close the child */ + /* still end the child */ } this.pipeFd = -1; } - this.readFd = -1; - this.writeFd = -1; - this.child.stdout?.destroy(); - this.child.stdin?.destroy(); - // A failed spawn can leave libuv's process handle without a PID. - // Never signal it: PID zero has process-group semantics on POSIX. // The channel is closed and all its work is discarded. SIGTERM invokes // tsgo's cancellation handler, which can race EOF and print "context // canceled" into an otherwise successful compile's stderr. @@ -155,6 +128,117 @@ class Ts7ProcessIo implements Ts7WireIo { }; } +/* POSIX: the server is not this process's child. A shell that spawnSync runs + * and reaps creates two FIFOs, starts the server in the background with its + * standard input and output on them, prints its PID and exits; the server's + * parent becomes init (or the nearest subreaper), which reaps it. A child of + * this process would stay a zombie after it exits: Node reaps children only + * from its event loop, which a synchronous compile never returns to. The + * subshell points its standard streams at /dev/null before it opens the + * FIFOs, so spawnSync's output pipe closes as soon as the shell exits. */ +const FIFO_LAUNCH = + 'i=$1 o=$2; shift 2; mkfifo -m 600 "$i" "$o" || exit 1; ' + + '( exec /dev/null; exec "$@" <"$i" >"$o" ) & echo $!'; + +/** How long a starting server may take to open its input, in milliseconds. */ +const FIFO_OPEN_TIMEOUT = 10_000; + +function processAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +export class Ts7FifoIo implements Ts7WireIo { + private readFd = -1; + private writeFd = -1; + private closed = false; + /** The server's process ID, for tests: it is never signaled. */ + readonly pid: number; + + constructor(executable: string, args: string[]) { + const directory = mkdtempSync(join(tmpdir(), "scriptc-ts7-")); + const input = join(directory, "in"); + const output = join(directory, "out"); + try { + const shell = spawnSync( + "/bin/sh", + ["-c", FIFO_LAUNCH, "sh", input, output, executable, ...args], + { stdio: ["ignore", "pipe", "inherit"], encoding: "utf8" }, + ); + const pid = Number(shell.stdout?.trim()); + if (shell.status !== 0 || !Number.isSafeInteger(pid) || pid <= 0) + throw new Error(`Unable to start TypeScript server: ${executable}`); + this.pid = pid; + // A non-blocking open fails with ENXIO until the server opens its end, + // so a server that exits as it starts cannot block this process. + let probe = -1; + for (let waited = 0; probe < 0; waited++) { + try { + probe = openSync(input, constants.O_WRONLY | constants.O_NONBLOCK); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENXIO") throw error; + if (waited >= FIFO_OPEN_TIMEOUT || !processAlive(pid)) + throw new Error(`Unable to start TypeScript server: ${executable}`); + Atomics.wait(sleepBuffer, 0, 0, 1); + } + } + try { + // The server reads now, so a blocking open returns at once. + this.writeFd = openSync(input, "w"); + } finally { + closeSync(probe); + } + // The server's shell opens its output right after its input. + this.readFd = openSync(output, "r"); + track(this); + } catch (error) { + this.close(); + throw error; + } finally { + // Both ends are open, or the start failed: the names are not needed. + rmSync(directory, { recursive: true, force: true }); + } + } + + read = (buffer: Uint8Array, offset: number, length: number): number => { + if (this.closed) throw new Error("TypeScript server transport is closed"); + return transfer(readSync, this.readFd, buffer, offset, length); + }; + + write = (buffer: Uint8Array, offset: number, length: number): number => { + if (this.closed) throw new Error("TypeScript server transport is closed"); + return transfer(writeSync, this.writeFd, buffer, offset, length); + }; + + /** Closing both FIFOs ends the session: the server reads end of file on + * its input and exits, and a write to its closed output ends it with + * SIGPIPE. It is never signaled, because once init has reaped it its PID + * can belong to another process. */ + close = (): void => { + if (this.closed) return; + this.closed = true; + untrack(this); + for (const fd of [this.writeFd, this.readFd]) { + if (fd === -1) continue; + try { + closeSync(fd); + } catch { + /* close the other end too */ + } + } + this.writeFd = -1; + this.readFd = -1; + }; +} + export function spawnTs7Wire(executable: string, args: string[]): Ts7Wire { - return new Ts7Wire(new Ts7ProcessIo(executable, args)); + return new Ts7Wire( + process.platform === "win32" + ? new Ts7NamedPipeIo(executable, args) + : new Ts7FifoIo(executable, args), + ); }