mirror of
https://github.com/vercel-labs/scriptc.git
synced 2026-11-09 17:11:46 +08:00
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 <noreply@anthropic.com>
This commit is contained in:
Malte Ubl
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
2c69918cfe
commit
3bc9ac8279
@@ -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("");
|
||||
},
|
||||
);
|
||||
|
||||
@@ -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<Ts7ProcessIo>();
|
||||
const liveTransports = new Set<Ts7WireIo>();
|
||||
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 >/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),
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user