mirror of
https://github.com/morluto/rea.git
synced 2026-10-26 17:35:11 +00:00
refactor(tests): consolidate compiled CLI session ownership
This commit is contained in:
morluto committed
1 parent
6021efd371
commit
2bd981b017
7 files changed
+598
-919
No files matched your search
+10
-7
@@ -233,13 +233,16 @@ the next command starts.
|
||||
IL2CPP metadata, and dylib-resolution commands use that session. File teardown
|
||||
stops the shared CLI process, including startup still in flight, and does not
|
||||
open another after close.
|
||||
`exportedFunctionBindingsCli.test.ts` keeps its own compiled CLI parity
|
||||
process for the file and still compares that process's JSON command results
|
||||
with the MCP session. A command that does not respond is rejected. That
|
||||
process stays owned until it exits, and another process does not start while
|
||||
it remains. A process that stays running after SIGKILL remains owned so a
|
||||
later close can retry. Closing the session reports a failure when the process
|
||||
does not exit after stdin closes, once the child has been stopped.
|
||||
`exportedFunctionBindingsCli.test.ts` uses the shared compiled CLI session for
|
||||
its file and compares its JSON command results with the MCP session. It selects
|
||||
20-second startup and command deadlines; other callers retain their existing
|
||||
deadline behavior. Retired processes stay owned until exit, block replacement,
|
||||
and remain available for a later close to retry after failed termination.
|
||||
Shared session cleanup closes stdin so the worker can release JavaScript
|
||||
resources, waits the bounded grace, then escalates to SIGTERM and SIGKILL only
|
||||
if needed. Confirmed exit succeeds; an unconfirmed exit reports cleanup failure.
|
||||
When startup or a command fails and cleanup also fails, the reported aggregate
|
||||
retains both failures and the original failure as its cause.
|
||||
`applicationWorkflowCli.test.ts` keeps one compiled CLI process for its JSON
|
||||
commands, including a non-zero status whose stdout is still a JSON document,
|
||||
and closes that process after the file.
|
||||
|
||||
@@ -15,9 +15,9 @@ import { javascriptApplicationAnalysisResultSchema } from "../../../src/domain/j
|
||||
import { javaScriptExportShapeComparisonResultSchema } from "../../../src/domain/javascript/javascriptExportShapeComparisonSchemas.js";
|
||||
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
import {
|
||||
startCompiledCliSession,
|
||||
type CompiledCliSession,
|
||||
} from "../../support/cli/compiledCliSession.js";
|
||||
createCliCommandSession,
|
||||
type CliCommandSession,
|
||||
} from "../../support/cli/cliCommandSession.js";
|
||||
|
||||
const execute = promisify(execFile);
|
||||
const requireFixture = createRequire(import.meta.url);
|
||||
@@ -27,11 +27,10 @@ const comparisonEnvelope = z.object({
|
||||
const applicationEnvelope = z.object({
|
||||
normalized_result: javascriptApplicationAnalysisResultSchema,
|
||||
});
|
||||
let cliSession: CompiledCliSession | undefined;
|
||||
let cliSession: CliCommandSession | undefined;
|
||||
|
||||
// Source semantics use one real stdio MCP session. Cases that also check the
|
||||
// compiled CLI adapter share one warm process: it loads the production command
|
||||
// runtime once and serves the same JSON commands as scripts/rea.mjs.
|
||||
// Source semantics use one real stdio MCP session. CLI parity uses the shared
|
||||
// file-scoped session, including its worker rotation and cleanup policy.
|
||||
let mcpConnection:
|
||||
| {
|
||||
readonly client: Client;
|
||||
@@ -1207,8 +1206,22 @@ const expectCountChange = (
|
||||
};
|
||||
|
||||
const runCli = async (args: readonly string[]): Promise<unknown> => {
|
||||
cliSession ??= startCompiledCliSession();
|
||||
return cliSession.run(args);
|
||||
cliSession ??= createCliCommandSession({
|
||||
startTimeoutMs: 20_000,
|
||||
requestTimeoutMs: 20_000,
|
||||
});
|
||||
const environment = Object.fromEntries(
|
||||
Object.entries({ ...process.env, REA_LOG_LEVEL: "silent" }).filter(
|
||||
(entry): entry is [string, string] => entry[1] !== undefined,
|
||||
),
|
||||
);
|
||||
const result = await cliSession.run({
|
||||
arguments: args,
|
||||
cwd: process.cwd(),
|
||||
environment,
|
||||
});
|
||||
expect(result.exitCode, result.stderr || result.stdout).toBe(0);
|
||||
return result.json;
|
||||
};
|
||||
|
||||
const analyzeVersions = async (
|
||||
|
||||
@@ -1,266 +0,0 @@
|
||||
import { EventEmitter, PassThrough } from "node:stream";
|
||||
import type { ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { expect, onTestFinished, test } from "vitest";
|
||||
|
||||
import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js";
|
||||
import { startCompiledCliSession } from "../../support/cli/compiledCliSession.js";
|
||||
|
||||
interface FakeCliProcess extends EventEmitter {
|
||||
stdin: PassThrough;
|
||||
stdout: PassThrough;
|
||||
stderr: PassThrough;
|
||||
pid: number;
|
||||
exitCode: number | null;
|
||||
signalCode: NodeJS.Signals | null;
|
||||
kill(signal?: NodeJS.Signals): boolean;
|
||||
}
|
||||
|
||||
const fakeProcess = (pid: number): FakeCliProcess => {
|
||||
const child = new EventEmitter() as FakeCliProcess;
|
||||
child.stdin = new PassThrough();
|
||||
child.stdout = new PassThrough();
|
||||
child.stderr = new PassThrough();
|
||||
child.pid = pid;
|
||||
child.exitCode = null;
|
||||
child.signalCode = null;
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return true;
|
||||
child.signalCode = signal;
|
||||
child.emit("exit", null, signal);
|
||||
child.emit("close", null, signal);
|
||||
return true;
|
||||
};
|
||||
return child;
|
||||
};
|
||||
|
||||
const asCliProcess = (child: FakeCliProcess): ChildProcessWithoutNullStreams =>
|
||||
child as unknown as ChildProcessWithoutNullStreams;
|
||||
|
||||
const ready = (child: FakeCliProcess): void => {
|
||||
child.stdout.write(`${JSON.stringify({ ready: true })}\n`);
|
||||
};
|
||||
|
||||
const exitWhenStdinCloses = (child: FakeCliProcess): void => {
|
||||
child.stdin.on("finish", () => {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
child.exitCode = 0;
|
||||
child.emit("exit", 0, null);
|
||||
child.emit("close", 0, null);
|
||||
});
|
||||
};
|
||||
|
||||
const replyWhenAsked = (
|
||||
child: FakeCliProcess,
|
||||
stdout: string,
|
||||
exitCode = 0,
|
||||
): void => {
|
||||
let buffer = "";
|
||||
child.stdin.setEncoding("utf8");
|
||||
child.stdin.on("data", (chunk: string) => {
|
||||
buffer += chunk;
|
||||
let newline = buffer.indexOf("\n");
|
||||
while (newline !== -1) {
|
||||
const line = buffer.slice(0, newline);
|
||||
buffer = buffer.slice(newline + 1);
|
||||
if (line.length > 0) {
|
||||
const request = JSON.parse(line) as { id: number };
|
||||
child.stdout.write(
|
||||
`${JSON.stringify({ id: request.id, exitCode, stdout })}\n`,
|
||||
);
|
||||
}
|
||||
newline = buffer.indexOf("\n");
|
||||
}
|
||||
});
|
||||
};
|
||||
|
||||
test("rejects a compiled CLI command when close wins the ready handshake", async () => {
|
||||
let child: FakeCliProcess | undefined;
|
||||
let releaseReady: (() => void) | undefined;
|
||||
const session = startCompiledCliSession({
|
||||
stopTimeoutMs: 50,
|
||||
spawnWorker: () => {
|
||||
const started = fakeProcess(60);
|
||||
child = started;
|
||||
exitWhenStdinCloses(started);
|
||||
releaseReady = () => {
|
||||
ready(started);
|
||||
};
|
||||
return asCliProcess(started);
|
||||
},
|
||||
});
|
||||
|
||||
const pending = session.run(["early"]);
|
||||
for (let attempt = 0; child === undefined && attempt < 30; attempt += 1)
|
||||
await new Promise((resolve) => {
|
||||
setImmediate(resolve);
|
||||
});
|
||||
if (child === undefined) throw new Error("compiled CLI worker did not start");
|
||||
const closing = session.close();
|
||||
releaseReady?.();
|
||||
await expect(pending).rejects.toThrow("Compiled CLI session is closed");
|
||||
await closing;
|
||||
});
|
||||
|
||||
test("rejects a compiled CLI command that never replies and serves the next one", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const session = startCompiledCliSession({
|
||||
requestTimeoutMs: 30,
|
||||
spawnWorker: () => {
|
||||
const child = fakeProcess(70 + children.length);
|
||||
children.push(child);
|
||||
exitWhenStdinCloses(child);
|
||||
ready(child);
|
||||
if (children.length > 1)
|
||||
replyWhenAsked(child, JSON.stringify({ ok: true }));
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(session.run(["stuck"])).rejects.toThrow(/did not respond/);
|
||||
expect(children[0]?.signalCode).toBe("SIGKILL");
|
||||
await expect(session.run(["next"])).resolves.toEqual({ ok: true });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("keeps a timed-out compiled CLI worker until it exits", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
let releaseExit: (() => void) | undefined;
|
||||
const session = startCompiledCliSession({
|
||||
requestTimeoutMs: 20,
|
||||
stopTimeoutMs: 40,
|
||||
spawnWorker: () => {
|
||||
const child = fakeProcess(90 + children.length);
|
||||
children.push(child);
|
||||
if (children.length === 1) {
|
||||
child.kill = () => {
|
||||
releaseExit = () => {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
child.signalCode = "SIGKILL";
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
child.emit("close", null, "SIGKILL");
|
||||
};
|
||||
return true;
|
||||
};
|
||||
} else {
|
||||
exitWhenStdinCloses(child);
|
||||
replyWhenAsked(child, JSON.stringify({ ok: true }));
|
||||
}
|
||||
ready(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
|
||||
const stalled = session.run(["stuck"]);
|
||||
const next = session.run(["next"]);
|
||||
await expect(stalled).rejects.toThrow(/did not respond/);
|
||||
expect(children).toHaveLength(1);
|
||||
expect(children[0]?.signalCode).toBeNull();
|
||||
releaseExit?.();
|
||||
await expect(next).resolves.toEqual({ ok: true });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("retries a timed-out compiled CLI worker that survives SIGKILL", async () => {
|
||||
let child: FakeCliProcess | undefined;
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = startCompiledCliSession({
|
||||
requestTimeoutMs: 20,
|
||||
stopTimeoutMs: 20,
|
||||
spawnWorker: () => {
|
||||
child = fakeProcess(91);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
return true;
|
||||
};
|
||||
ready(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(session.run(["stuck"])).rejects.toThrow(/did not respond/);
|
||||
const replacement = session.run(["next"]);
|
||||
await expect(replacement).rejects.toThrow(/stayed running after SIGKILL/);
|
||||
await expect(session.close()).rejects.toThrow(/stayed running after SIGKILL/);
|
||||
expect(signals).toEqual(["SIGKILL", "SIGKILL"]);
|
||||
const stalledChild = child;
|
||||
if (stalledChild === undefined) throw new Error("missing stalled worker");
|
||||
stalledChild.kill = (signal: NodeJS.Signals = "SIGKILL") => {
|
||||
signals.push(signal);
|
||||
stalledChild.signalCode = signal;
|
||||
stalledChild.emit("exit", null, signal);
|
||||
stalledChild.emit("close", null, signal);
|
||||
return true;
|
||||
};
|
||||
await session.close();
|
||||
expect(stalledChild.signalCode).toBe("SIGKILL");
|
||||
});
|
||||
|
||||
test("reports a compiled CLI session that needed a signal after stdin closed", async () => {
|
||||
let child: FakeCliProcess | undefined;
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = startCompiledCliSession({
|
||||
stopTimeoutMs: 20,
|
||||
spawnWorker: () => {
|
||||
child = fakeProcess(81);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
if (signal !== "SIGKILL") return true;
|
||||
if (child?.exitCode !== null || child?.signalCode !== null) return true;
|
||||
child.signalCode = signal;
|
||||
child.emit("exit", null, signal);
|
||||
child.emit("close", null, signal);
|
||||
return true;
|
||||
};
|
||||
ready(child);
|
||||
replyWhenAsked(child, JSON.stringify({ ok: true }));
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(session.run(["once"])).resolves.toEqual({ ok: true });
|
||||
await expect(session.close()).rejects.toThrow(
|
||||
"Compiled CLI session did not exit after stdin closed",
|
||||
);
|
||||
expect(signals).toEqual(["SIGTERM", "SIGKILL"]);
|
||||
expect(child?.signalCode).toBe("SIGKILL");
|
||||
});
|
||||
|
||||
test("closes a compiled CLI session that exits after stdin closes", async () => {
|
||||
const session = startCompiledCliSession({
|
||||
spawnWorker: () => {
|
||||
const child = fakeProcess(82);
|
||||
exitWhenStdinCloses(child);
|
||||
ready(child);
|
||||
replyWhenAsked(child, JSON.stringify({ ok: true }));
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
|
||||
await expect(session.run(["once"])).resolves.toEqual({ ok: true });
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("keeps a compiled CLI failure status and serves the next command", async () => {
|
||||
const session = startCompiledCliSession();
|
||||
onTestFinished(() => session.close());
|
||||
const root = await createTestTempDirectory("rea-warm-cli-status-");
|
||||
await writeFile(
|
||||
join(root, "app.mjs"),
|
||||
"export function current() { return 1; }\n",
|
||||
);
|
||||
|
||||
await expect(
|
||||
session.run(["compare-javascript-export-shapes", "{", "--json"]),
|
||||
).rejects.toThrow(/status 1:.*invalid_request/);
|
||||
await expect(
|
||||
session.run(["analyze-javascript-application", root, "--json"]),
|
||||
).resolves.toMatchObject({
|
||||
normalized_result: expect.any(Object),
|
||||
});
|
||||
await session.close();
|
||||
});
|
||||
@@ -1,7 +1,8 @@
|
||||
import { EventEmitter, PassThrough } from "node:stream";
|
||||
import type { ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
import { expect, test } from "vitest";
|
||||
import { expect, onTestFinished, test } from "vitest";
|
||||
import { z } from "zod";
|
||||
|
||||
import { javascriptApplicationAnalysisResultSchema } from "../../../src/domain/javascript/javascriptApplicationAnalysis.js";
|
||||
@@ -126,8 +127,13 @@ test("stops an unready shared CLI process when the session closes", async () =>
|
||||
command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }),
|
||||
);
|
||||
await waitFor(() => spawned !== undefined);
|
||||
await session.close();
|
||||
await expect(pending).rejects.toThrow("Shared CLI session is closed");
|
||||
const rejected = expect(pending).rejects.toThrow(
|
||||
"Shared CLI session is closed",
|
||||
);
|
||||
const closing = session.close();
|
||||
spawned?.stdout.write(encodeCliSessionFrame({ ready: true }));
|
||||
await closing;
|
||||
await rejected;
|
||||
expect(spawned?.signalCode).toBe("SIGTERM");
|
||||
await expect(
|
||||
session.run(
|
||||
@@ -155,8 +161,11 @@ test("stops an in-flight shared CLI command when the session closes", async () =
|
||||
command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }),
|
||||
);
|
||||
await waitFor(() => received);
|
||||
const rejected = expect(pending).rejects.toThrow(
|
||||
"Shared CLI session is closed",
|
||||
);
|
||||
await session.close();
|
||||
await expect(pending).rejects.toThrow("Shared CLI session is closed");
|
||||
await rejected;
|
||||
expect(spawned?.signalCode).toBe("SIGTERM");
|
||||
});
|
||||
|
||||
@@ -249,6 +258,369 @@ const respondingProcess = (
|
||||
return child;
|
||||
};
|
||||
|
||||
test("recovers after a shared CLI command exceeds its selected deadline", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const session = createCliCommandSession({
|
||||
requestTimeoutMs: 20,
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(
|
||||
130 + children.length,
|
||||
(request, current) => {
|
||||
if (children.length > 1)
|
||||
reply(current, request.id, { stdout: "recovered" });
|
||||
},
|
||||
);
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
await expect(
|
||||
session.run(command(["stuck"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not respond/);
|
||||
expect(children[0]?.signalCode).not.toBeNull();
|
||||
await expect(
|
||||
session.run(command(["next"], { PATH: "/usr/bin" })),
|
||||
).resolves.toMatchObject({ stdout: "recovered" });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("times out a shared CLI command whose stdin write callback stays pending", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
let releaseWrite: (() => void) | undefined;
|
||||
let observeWrite: (() => void) | undefined;
|
||||
const writing = new Promise<void>((resolve) => {
|
||||
observeWrite = resolve;
|
||||
});
|
||||
const session = createCliCommandSession({
|
||||
requestTimeoutMs: 20,
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(
|
||||
135 + children.length,
|
||||
(request, current) =>
|
||||
reply(current, request.id, { stdout: "recovered" }),
|
||||
);
|
||||
if (children.length === 0) {
|
||||
child.stdin._write = (_chunk, _encoding, callback) => {
|
||||
releaseWrite = () => callback();
|
||||
observeWrite?.();
|
||||
};
|
||||
}
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
const failed = expect(
|
||||
session.run(command(["stuck"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not respond/);
|
||||
await writing;
|
||||
expect(releaseWrite).toEqual(expect.any(Function));
|
||||
await failed;
|
||||
expect(children[0]?.signalCode).not.toBeNull();
|
||||
releaseWrite?.();
|
||||
await expect(
|
||||
session.run(command(["next"], { PATH: "/usr/bin" })),
|
||||
).resolves.toMatchObject({ stdout: "recovered" });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("recovers after a shared CLI worker misses its selected readiness deadline", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const session = createCliCommandSession({
|
||||
startTimeoutMs: 20,
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const child =
|
||||
children.length === 0
|
||||
? fakeProcess(140)
|
||||
: respondingProcess(141, (request, current) =>
|
||||
reply(current, request.id, { stdout: "ready" }),
|
||||
);
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
await expect(
|
||||
session.run(command(["stuck"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not become ready/);
|
||||
expect(children[0]?.signalCode).not.toBeNull();
|
||||
await expect(
|
||||
session.run(command(["next"], { PATH: "/usr/bin" })),
|
||||
).resolves.toMatchObject({ stdout: "ready" });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("waits for the timed-out shared CLI worker to exit before starting its replacement", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
let releaseExit: (() => void) | undefined;
|
||||
let observeStop: (() => void) | undefined;
|
||||
const stopping = new Promise<void>((resolve) => {
|
||||
observeStop = resolve;
|
||||
});
|
||||
const session = createCliCommandSession({
|
||||
requestTimeoutMs: 20,
|
||||
shutdownGraceMs: 100,
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(
|
||||
150 + children.length,
|
||||
(request, current) => {
|
||||
if (children.length > 1)
|
||||
reply(current, request.id, { stdout: "next" });
|
||||
},
|
||||
);
|
||||
if (children.length === 0) {
|
||||
const stop = child.kill.bind(child);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
releaseExit = () => stop(signal);
|
||||
observeStop?.();
|
||||
return true;
|
||||
};
|
||||
}
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
const stalled = expect(
|
||||
session.run(command(["stuck"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not respond/);
|
||||
const next = session.run(command(["next"], { PATH: "/usr/bin" }));
|
||||
await stopping;
|
||||
expect(children).toHaveLength(1);
|
||||
releaseExit?.();
|
||||
await stalled;
|
||||
await expect(next).resolves.toMatchObject({ stdout: "next" });
|
||||
expect(children).toHaveLength(2);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("retains a shared CLI worker that survives SIGKILL and retries its exact owner on close", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = createCliCommandSession({
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(
|
||||
160 + children.length,
|
||||
(request, current) => {
|
||||
reply(current, request.id, {
|
||||
exitCode: 1,
|
||||
stderr: "fatal",
|
||||
fatal: true,
|
||||
});
|
||||
},
|
||||
);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
return true;
|
||||
};
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
await expect(
|
||||
session.run(command(["fatal"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not exit after SIGKILL/);
|
||||
await expect(
|
||||
session.run(command(["blocked"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not exit after SIGKILL/);
|
||||
expect(children).toHaveLength(1);
|
||||
await expect(session.close()).rejects.toThrow(/did not exit after SIGKILL/);
|
||||
const retained = children[0];
|
||||
if (retained === undefined) throw new Error("missing retained worker");
|
||||
const stopsBeforeRecovery = signals.length;
|
||||
retained.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
retained.signalCode = signal;
|
||||
retained.emit("exit", null, signal);
|
||||
retained.emit("close", null, signal);
|
||||
return true;
|
||||
};
|
||||
await session.close();
|
||||
expect(signals.length).toBeGreaterThan(stopsBeforeRecovery);
|
||||
expect(retained.signalCode).not.toBeNull();
|
||||
expect(children).toHaveLength(1);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test.each([
|
||||
{
|
||||
phase: "command",
|
||||
reason: /did not respond/,
|
||||
},
|
||||
{
|
||||
phase: "startup",
|
||||
reason: /did not become ready/,
|
||||
},
|
||||
])(
|
||||
"preserves the $phase timeout when shared CLI cleanup also fails",
|
||||
async ({ phase, reason }) => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = createCliCommandSession({
|
||||
...(phase === "command"
|
||||
? { requestTimeoutMs: 20 }
|
||||
: { startTimeoutMs: 20 }),
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const child =
|
||||
phase === "command"
|
||||
? respondingProcess(190, () => undefined)
|
||||
: fakeProcess(190);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
return true;
|
||||
};
|
||||
children.push(child);
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
const failure: unknown = await session
|
||||
.run(command(["stuck"], { PATH: "/usr/bin" }))
|
||||
.catch((error: unknown) => error);
|
||||
expect(failure).toBeInstanceOf(AggregateError);
|
||||
if (!(failure instanceof AggregateError))
|
||||
throw new Error("missing combined command and cleanup failure", {
|
||||
cause: failure,
|
||||
});
|
||||
expect(failure.message).toMatch(reason);
|
||||
expect(failure.message).toMatch(/did not exit after SIGKILL/);
|
||||
expect(failure.errors).toHaveLength(2);
|
||||
expect(failure.errors[0]).toBeInstanceOf(Error);
|
||||
expect(failure.errors[0]).toHaveProperty(
|
||||
"message",
|
||||
expect.stringMatching(reason),
|
||||
);
|
||||
expect(failure.errors[1]).toHaveProperty(
|
||||
"message",
|
||||
expect.stringMatching(/did not exit after SIGKILL/),
|
||||
);
|
||||
expect(failure.cause).toBe(failure.errors[0]);
|
||||
expect(signals).toEqual(["SIGTERM", "SIGKILL"]);
|
||||
await expect(
|
||||
session.run(command(["blocked"], { PATH: "/usr/bin" })),
|
||||
).rejects.toThrow(/did not exit after SIGKILL/);
|
||||
expect(children).toHaveLength(1);
|
||||
await expect(session.close()).rejects.toThrow(/did not exit after SIGKILL/);
|
||||
const retained = children[0];
|
||||
if (retained === undefined) throw new Error("missing retained worker");
|
||||
retained.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
retained.signalCode = signal;
|
||||
retained.emit("exit", null, signal);
|
||||
retained.emit("close", null, signal);
|
||||
return true;
|
||||
};
|
||||
await session.close();
|
||||
expect(retained.signalCode).not.toBeNull();
|
||||
expect(children).toHaveLength(1);
|
||||
},
|
||||
);
|
||||
|
||||
test("confirms shared CLI shutdown after escalating an ignored SIGTERM", async () => {
|
||||
let child: FakeCliProcess | undefined;
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = createCliCommandSession({
|
||||
shutdownGraceMs: 20,
|
||||
spawnWorker: () => {
|
||||
const started = respondingProcess(170, (request, current) =>
|
||||
reply(current, request.id, { stdout: "done" }),
|
||||
);
|
||||
const stop = started.kill.bind(started);
|
||||
started.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
return signal === "SIGKILL" ? stop(signal) : true;
|
||||
};
|
||||
child = started;
|
||||
return asCliProcess(started);
|
||||
},
|
||||
});
|
||||
await expect(
|
||||
session.run(command(["once"], { PATH: "/usr/bin" })),
|
||||
).resolves.toMatchObject({ stdout: "done" });
|
||||
await session.close();
|
||||
expect(signals).toEqual(["SIGTERM", "SIGKILL"]);
|
||||
expect(child?.signalCode).toBe("SIGKILL");
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("lets the shared CLI worker finish its resource cleanup before signaling", async () => {
|
||||
const events: string[] = [];
|
||||
const session = createCliCommandSession({
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(180, (request, current) =>
|
||||
reply(current, request.id, { stdout: "done" }),
|
||||
);
|
||||
child.stdin.once("finish", () => {
|
||||
events.push("stdin closed", "resources released");
|
||||
child.exitCode = 0;
|
||||
child.emit("exit", 0, null);
|
||||
child.emit("close", 0, null);
|
||||
});
|
||||
const stop = child.kill.bind(child);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
events.push(signal);
|
||||
return stop(signal);
|
||||
};
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
await expect(
|
||||
session.run(command(["once"], { PATH: "/usr/bin" })),
|
||||
).resolves.toMatchObject({ stdout: "done" });
|
||||
await session.close();
|
||||
expect(events).toEqual(["stdin closed", "resources released"]);
|
||||
await session.close();
|
||||
expect(events).toHaveLength(2);
|
||||
});
|
||||
|
||||
test("shares one shared CLI cleanup attempt across concurrent close calls", async () => {
|
||||
let cleanupStarts = 0;
|
||||
let releaseExit: (() => void) | undefined;
|
||||
let observeCleanup: (() => void) | undefined;
|
||||
const cleaning = new Promise<void>((resolve) => {
|
||||
observeCleanup = resolve;
|
||||
});
|
||||
const signals: NodeJS.Signals[] = [];
|
||||
const session = createCliCommandSession({
|
||||
spawnWorker: () => {
|
||||
const child = respondingProcess(185, (request, current) =>
|
||||
reply(current, request.id, { stdout: "done" }),
|
||||
);
|
||||
child.stdin.once("finish", () => {
|
||||
cleanupStarts += 1;
|
||||
releaseExit = () => {
|
||||
child.exitCode = 0;
|
||||
child.emit("exit", 0, null);
|
||||
child.emit("close", 0, null);
|
||||
};
|
||||
observeCleanup?.();
|
||||
});
|
||||
const stop = child.kill.bind(child);
|
||||
child.kill = (signal: NodeJS.Signals = "SIGTERM") => {
|
||||
signals.push(signal);
|
||||
return stop(signal);
|
||||
};
|
||||
return asCliProcess(child);
|
||||
},
|
||||
});
|
||||
await session.run(command(["once"], { PATH: "/usr/bin" }));
|
||||
const first = session.close();
|
||||
const second = session.close();
|
||||
expect(second).toBe(first);
|
||||
await cleaning;
|
||||
expect(cleanupStarts).toBe(1);
|
||||
expect(signals).toEqual([]);
|
||||
releaseExit?.();
|
||||
await Promise.all([first, second]);
|
||||
await session.close();
|
||||
expect(cleanupStarts).toBe(1);
|
||||
expect(signals).toEqual([]);
|
||||
});
|
||||
|
||||
test("starts another process after the shared CLI process exits", async () => {
|
||||
const children: FakeCliProcess[] = [];
|
||||
const session = createCliCommandSession({
|
||||
@@ -457,6 +829,81 @@ test("starts a new process after the real worker reports a fatal failure", async
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("keeps a malformed JSON command's failure status and serves the next command", async () => {
|
||||
const session = createCliCommandSession();
|
||||
onTestFinished(() => session.close());
|
||||
const environment = {
|
||||
PATH: process.env.PATH ?? "",
|
||||
REA_LOG_LEVEL: "silent",
|
||||
};
|
||||
const failed = await session.run(
|
||||
command(["compare-javascript-export-shapes", "{", "--json"], environment),
|
||||
);
|
||||
expect(failed.exitCode).toBe(1);
|
||||
expect(failed.stdout).toContain("invalid_request");
|
||||
const worker = session.processId();
|
||||
expect(worker).toEqual(expect.any(Number));
|
||||
const root = await createTestTempDirectory("rea-warm-cli-status-");
|
||||
await writeFixtureFiles(root, {
|
||||
"app.mjs": "export function current() { return 1; }\n",
|
||||
});
|
||||
const recovered = await session.run(
|
||||
command(["analyze-javascript-application", root, "--json"], environment),
|
||||
);
|
||||
expect(recovered.exitCode, recovered.stderr).toBe(0);
|
||||
expect(recovered.json).toMatchObject({
|
||||
normalized_result: expect.any(Object),
|
||||
});
|
||||
expect(session.processId()).toBe(worker);
|
||||
await session.close();
|
||||
});
|
||||
|
||||
test("closes the real JavaScript CLI worker through its resource cleanup", async () => {
|
||||
let worker: ChildProcessWithoutNullStreams | undefined;
|
||||
const session = createCliCommandSession({
|
||||
spawnWorker: (environment) => {
|
||||
worker = spawn(
|
||||
process.execPath,
|
||||
[
|
||||
fileURLToPath(
|
||||
new URL(
|
||||
"../../support/cli/cliCommandSessionWorker.mjs",
|
||||
import.meta.url,
|
||||
),
|
||||
),
|
||||
],
|
||||
{
|
||||
cwd: process.cwd(),
|
||||
env: environment,
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
},
|
||||
);
|
||||
return worker;
|
||||
},
|
||||
});
|
||||
onTestFinished(() => session.close());
|
||||
const root = await createTestTempDirectory("rea-warm-cli-graceful-");
|
||||
await writeFixtureFiles(root, {
|
||||
"app.mjs": "export function current() { return 1; }\n",
|
||||
});
|
||||
const analyzed = await session.run(
|
||||
command(["analyze-javascript-application", root, "--json"], {
|
||||
PATH: process.env.PATH ?? "",
|
||||
REA_LOG_LEVEL: "silent",
|
||||
}),
|
||||
);
|
||||
expect(analyzed.exitCode, analyzed.stderr).toBe(0);
|
||||
expect(analyzed.json).toMatchObject({
|
||||
normalized_result: expect.any(Object),
|
||||
});
|
||||
expect(worker?.exitCode).toBeNull();
|
||||
expect(worker?.signalCode).toBeNull();
|
||||
await session.close();
|
||||
expect(worker?.exitCode).toBe(0);
|
||||
expect(worker?.signalCode).toBeNull();
|
||||
expect(session.processId()).toBeUndefined();
|
||||
});
|
||||
|
||||
const requireSharedCli = (cli: {
|
||||
sharedCliRun?: (invocation: CliInvocation) => Promise<CliResult>;
|
||||
}): ((invocation: CliInvocation) => Promise<CliResult>) => {
|
||||
|
||||
@@ -61,12 +61,23 @@ interface CliCommandSessionDependencies {
|
||||
readonly commandLimit?: number;
|
||||
/** Resident-set bytes, reported by the worker, that start a new process. */
|
||||
readonly residentLimitBytes?: number;
|
||||
/** Optional caller-selected startup and command deadlines; unset by default. */
|
||||
readonly startTimeoutMs?: number;
|
||||
readonly requestTimeoutMs?: number;
|
||||
/** Grace for EOF cleanup and each signal before escalation. */
|
||||
readonly shutdownGraceMs?: number;
|
||||
}
|
||||
|
||||
interface PendingCommand {
|
||||
readonly id: number;
|
||||
readonly resolve: (result: CliCommandResult) => void;
|
||||
readonly reject: (error: unknown) => void;
|
||||
readonly timer: ReturnType<typeof setTimeout> | undefined;
|
||||
}
|
||||
|
||||
interface RetiredWorker {
|
||||
readonly child: ChildProcessWithoutNullStreams;
|
||||
attempt: Promise<void> | undefined;
|
||||
}
|
||||
|
||||
type SessionResponse = z.infer<typeof responseSchema>;
|
||||
@@ -78,6 +89,11 @@ const positiveLimit = (value: number | undefined, fallback: number): number =>
|
||||
? value
|
||||
: fallback;
|
||||
|
||||
const selectedTimeout = (value: number | undefined): number | undefined =>
|
||||
value !== undefined && Number.isSafeInteger(value) && value > 0
|
||||
? value
|
||||
: undefined;
|
||||
|
||||
const envKey = (environment: Readonly<Record<string, string>>): string =>
|
||||
JSON.stringify(
|
||||
Object.keys(environment)
|
||||
@@ -110,13 +126,17 @@ const signalChild = (
|
||||
|
||||
const stopChild = async (
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
graceMs: number,
|
||||
): Promise<void> => {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
// EOF lets the worker await closeCliJavaScriptResources before exiting.
|
||||
if (!child.stdin.writableEnded) child.stdin.end();
|
||||
if (await waitForExit(child, graceMs)) return;
|
||||
signalChild(child, "SIGTERM");
|
||||
if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return;
|
||||
if (await waitForExit(child, graceMs)) return;
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
signalChild(child, "SIGKILL");
|
||||
if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return;
|
||||
if (await waitForExit(child, graceMs)) return;
|
||||
throw new Error(
|
||||
`shared CLI process ${String(child.pid)} did not exit after SIGKILL`,
|
||||
);
|
||||
@@ -157,11 +177,14 @@ class SharedCliProcess implements CliCommandSession {
|
||||
|
||||
readonly #commandLimit: number;
|
||||
readonly #residentLimitBytes: number;
|
||||
readonly #startTimeoutMs: number | undefined;
|
||||
readonly #requestTimeoutMs: number | undefined;
|
||||
readonly #shutdownGraceMs: number;
|
||||
#child: ChildProcessWithoutNullStreams | undefined;
|
||||
#childEnvKey: string | undefined;
|
||||
#commands = 0;
|
||||
#residentBytes: number | undefined;
|
||||
#retiring: Promise<void> | undefined;
|
||||
#retiring: RetiredWorker | undefined;
|
||||
#closed = false;
|
||||
#closing: Promise<void> | undefined;
|
||||
#starting: Promise<void> | undefined;
|
||||
@@ -181,6 +204,12 @@ class SharedCliProcess implements CliCommandSession {
|
||||
dependencies.residentLimitBytes,
|
||||
RESIDENT_REUSE_LIMIT_BYTES,
|
||||
);
|
||||
this.#startTimeoutMs = selectedTimeout(dependencies.startTimeoutMs);
|
||||
this.#requestTimeoutMs = selectedTimeout(dependencies.requestTimeoutMs);
|
||||
this.#shutdownGraceMs = positiveLimit(
|
||||
dependencies.shutdownGraceMs,
|
||||
SHUTDOWN_GRACE_MS,
|
||||
);
|
||||
}
|
||||
|
||||
run(request: CliCommandRequest): Promise<CliCommandResult> {
|
||||
@@ -207,45 +236,83 @@ class SharedCliProcess implements CliCommandSession {
|
||||
close(): Promise<void> {
|
||||
this.#closed = true;
|
||||
if (this.#closing !== undefined) return this.#closing;
|
||||
const attempt = this.#shutdown();
|
||||
let attempt: Promise<void>;
|
||||
attempt = this.#shutdown().catch((error: unknown) => {
|
||||
if (this.#closing === attempt) this.#closing = undefined;
|
||||
throw error;
|
||||
});
|
||||
this.#closing = attempt;
|
||||
return attempt;
|
||||
}
|
||||
|
||||
async #shutdown(): Promise<void> {
|
||||
try {
|
||||
const owner = this.#child;
|
||||
if (owner !== undefined) await stopChild(owner);
|
||||
const retiring = this.#retiring;
|
||||
if (retiring !== undefined) await retiring;
|
||||
} catch (error) {
|
||||
this.#closing = undefined;
|
||||
throw error;
|
||||
}
|
||||
const pendingStart = this.#starting;
|
||||
this.#failReady(closedError());
|
||||
this.#failPending(closedError());
|
||||
const owner = this.#child;
|
||||
if (owner !== undefined) this.#retire(owner);
|
||||
await this.#finishRetirement();
|
||||
if (pendingStart !== undefined) await pendingStart.catch(() => undefined);
|
||||
}
|
||||
|
||||
async #invoke(request: CliCommandRequest): Promise<CliCommandResult> {
|
||||
if (this.#closed) throw closedError();
|
||||
try {
|
||||
return await this.#run(request);
|
||||
} finally {
|
||||
await this.#finishRetirement();
|
||||
const current = this.#child;
|
||||
if (
|
||||
!this.#reusable(current, envKey(request.environment)) &&
|
||||
current !== undefined
|
||||
) {
|
||||
this.#retire(current);
|
||||
await this.#finishRetirement();
|
||||
}
|
||||
if (this.#closed) throw closedError();
|
||||
let result: CliCommandResult;
|
||||
try {
|
||||
if (this.#child === undefined) await this.#start(request.environment);
|
||||
result = await this.#run(request);
|
||||
} catch (primary) {
|
||||
try {
|
||||
await this.#finishRetirement();
|
||||
} catch (cleanup) {
|
||||
const primaryReason =
|
||||
primary instanceof Error ? primary.message : String(primary);
|
||||
const cleanupReason =
|
||||
cleanup instanceof Error ? cleanup.message : String(cleanup);
|
||||
throw new AggregateError(
|
||||
[primary, cleanup],
|
||||
`${primaryReason}; shared CLI cleanup failed: ${cleanupReason}`,
|
||||
{ cause: primary },
|
||||
);
|
||||
}
|
||||
throw primary;
|
||||
}
|
||||
await this.#finishRetirement();
|
||||
return result;
|
||||
}
|
||||
|
||||
async #run(request: CliCommandRequest): Promise<CliCommandResult> {
|
||||
await this.#ensure(request.environment);
|
||||
const owner = this.#child;
|
||||
if (this.#closed || owner === undefined) throw closedError();
|
||||
const id = this.#nextId;
|
||||
this.#nextId += 1;
|
||||
const result = new Promise<CliCommandResult>((resolve, reject) => {
|
||||
this.#pending = { id, resolve, reject };
|
||||
const timer =
|
||||
this.#requestTimeoutMs === undefined
|
||||
? undefined
|
||||
: setTimeout(() => {
|
||||
this.#failTransport(
|
||||
owner,
|
||||
new Error(
|
||||
`Shared CLI command did not respond within ${String(this.#requestTimeoutMs)} ms`,
|
||||
),
|
||||
);
|
||||
}, this.#requestTimeoutMs);
|
||||
timer?.unref();
|
||||
this.#pending = { id, resolve, reject, timer };
|
||||
});
|
||||
try {
|
||||
await writeChild(
|
||||
void writeChild(
|
||||
owner,
|
||||
encodeCliSessionFrame({
|
||||
id,
|
||||
@@ -253,7 +320,9 @@ class SharedCliProcess implements CliCommandSession {
|
||||
cwd: request.cwd,
|
||||
env: request.environment,
|
||||
}),
|
||||
);
|
||||
).catch((error: unknown) => {
|
||||
this.#failTransport(owner, error);
|
||||
});
|
||||
} catch (error) {
|
||||
this.#failTransport(owner, error);
|
||||
}
|
||||
@@ -263,11 +332,14 @@ class SharedCliProcess implements CliCommandSession {
|
||||
async #finishRetirement(): Promise<void> {
|
||||
const retiring = this.#retiring;
|
||||
if (retiring === undefined) return;
|
||||
try {
|
||||
await retiring;
|
||||
} finally {
|
||||
if (this.#retiring === retiring) this.#retiring = undefined;
|
||||
}
|
||||
retiring.attempt ??= stopChild(
|
||||
retiring.child,
|
||||
this.#shutdownGraceMs,
|
||||
).finally(() => {
|
||||
retiring.attempt = undefined;
|
||||
});
|
||||
await retiring.attempt;
|
||||
if (this.#retiring === retiring) this.#retiring = undefined;
|
||||
}
|
||||
|
||||
#reusable(
|
||||
@@ -289,16 +361,6 @@ class SharedCliProcess implements CliCommandSession {
|
||||
);
|
||||
}
|
||||
|
||||
async #ensure(environment: Readonly<Record<string, string>>): Promise<void> {
|
||||
if (this.#closed) throw closedError();
|
||||
const key = envKey(environment);
|
||||
const current = this.#child;
|
||||
if (this.#reusable(current, key)) return;
|
||||
if (current !== undefined) await stopChild(current);
|
||||
if (this.#closed) throw closedError();
|
||||
await this.#start(environment);
|
||||
}
|
||||
|
||||
#start(environment: Readonly<Record<string, string>>): Promise<void> {
|
||||
if (this.#closed) return Promise.reject(closedError());
|
||||
const started = this.#spawnWorker(environment);
|
||||
@@ -311,12 +373,25 @@ class SharedCliProcess implements CliCommandSession {
|
||||
this.#rejectReady = reject;
|
||||
});
|
||||
this.#starting = ready;
|
||||
const timer =
|
||||
this.#startTimeoutMs === undefined
|
||||
? undefined
|
||||
: setTimeout(() => {
|
||||
this.#failTransport(
|
||||
started,
|
||||
new Error(
|
||||
`Shared CLI worker did not become ready within ${String(this.#startTimeoutMs)} ms`,
|
||||
),
|
||||
);
|
||||
}, this.#startTimeoutMs);
|
||||
timer?.unref();
|
||||
this.#watch(started);
|
||||
if (this.#closed) {
|
||||
signalChild(started, "SIGTERM");
|
||||
this.#failReady(closedError());
|
||||
this.#retire(started);
|
||||
}
|
||||
return ready.finally(() => {
|
||||
clearTimeout(timer);
|
||||
if (this.#starting === ready) this.#starting = undefined;
|
||||
});
|
||||
}
|
||||
@@ -378,6 +453,7 @@ class SharedCliProcess implements CliCommandSession {
|
||||
return;
|
||||
}
|
||||
this.#pending = undefined;
|
||||
clearTimeout(current.timer);
|
||||
if (message.fatal === true) {
|
||||
this.#retire(owner);
|
||||
} else {
|
||||
@@ -394,7 +470,7 @@ class SharedCliProcess implements CliCommandSession {
|
||||
}
|
||||
this.#commands = 0;
|
||||
this.#residentBytes = undefined;
|
||||
this.#retiring = stopChild(owner);
|
||||
this.#retiring ??= { child: owner, attempt: undefined };
|
||||
}
|
||||
|
||||
#onClose(
|
||||
@@ -427,6 +503,7 @@ class SharedCliProcess implements CliCommandSession {
|
||||
#failPending(error: unknown): void {
|
||||
const current = this.#pending;
|
||||
this.#pending = undefined;
|
||||
clearTimeout(current?.timer);
|
||||
current?.reject(error);
|
||||
}
|
||||
|
||||
|
||||
@@ -1,444 +0,0 @@
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
import { z } from "zod";
|
||||
|
||||
import { waitForExit } from "../process/processFixture.js";
|
||||
|
||||
const START_TIMEOUT_MS = 20_000;
|
||||
const REQUEST_TIMEOUT_MS = 20_000;
|
||||
const STOP_TIMEOUT_MS = 1_000;
|
||||
const STDERR_TAIL_CHARS = 4_000;
|
||||
const workerPath = fileURLToPath(
|
||||
new URL("./compiledCliWorker.mjs", import.meta.url),
|
||||
);
|
||||
const readyMessage = z.object({ ready: z.literal(true) }).strict();
|
||||
const responseMessage = z
|
||||
.object({
|
||||
id: z.number(),
|
||||
exitCode: z.number(),
|
||||
stdout: z.string(),
|
||||
error: z.string().optional(),
|
||||
})
|
||||
.strict();
|
||||
|
||||
type CliResponse = z.infer<typeof responseMessage>;
|
||||
|
||||
interface PendingResponse {
|
||||
readonly resolve: (response: CliResponse) => void;
|
||||
readonly reject: (cause: unknown) => void;
|
||||
}
|
||||
|
||||
interface OwnedStop {
|
||||
readonly child: ChildProcessWithoutNullStreams;
|
||||
attempt: Promise<boolean> | undefined;
|
||||
}
|
||||
|
||||
interface SessionState {
|
||||
child: ChildProcessWithoutNullStreams | undefined;
|
||||
starting: Promise<void> | undefined;
|
||||
buffer: string;
|
||||
nextId: number;
|
||||
readonly pending: Map<number, PendingResponse>;
|
||||
readonly stopping: OwnedStop[];
|
||||
queue: Promise<void>;
|
||||
stderr: string;
|
||||
closed: boolean;
|
||||
closing: Promise<void> | undefined;
|
||||
settleStart: ((cause?: Error) => void) | undefined;
|
||||
ready: boolean;
|
||||
readonly requestTimeoutMs: number;
|
||||
readonly stopTimeoutMs: number;
|
||||
readonly spawnWorker: () => ChildProcessWithoutNullStreams;
|
||||
}
|
||||
|
||||
/** One warm compiled CLI process serving production JSON commands. */
|
||||
export interface CompiledCliSession {
|
||||
run(args: readonly string[]): Promise<unknown>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
/** Limits and process factory for session tests. Production callers omit them. */
|
||||
export interface CompiledCliSessionOptions {
|
||||
readonly spawnWorker?: () => ChildProcessWithoutNullStreams;
|
||||
readonly requestTimeoutMs?: number;
|
||||
readonly stopTimeoutMs?: number;
|
||||
}
|
||||
|
||||
/** Start a session that loads the compiled CLI once and reuses that process. */
|
||||
export const startCompiledCliSession = (
|
||||
options: CompiledCliSessionOptions = {},
|
||||
): CompiledCliSession => {
|
||||
const state = createSessionState(options);
|
||||
return {
|
||||
run: (args) => enqueue(state, () => dispatch(state, args)),
|
||||
close: () => closeSession(state),
|
||||
};
|
||||
};
|
||||
|
||||
const positiveMilliseconds = (
|
||||
value: number | undefined,
|
||||
fallback: number,
|
||||
): number =>
|
||||
value !== undefined && Number.isSafeInteger(value) && value > 0
|
||||
? value
|
||||
: fallback;
|
||||
|
||||
const defaultSpawnWorker = (): ChildProcessWithoutNullStreams =>
|
||||
spawn(process.execPath, [workerPath], {
|
||||
cwd: process.cwd(),
|
||||
env: { ...process.env, REA_LOG_LEVEL: "silent" },
|
||||
stdio: ["pipe", "pipe", "pipe"],
|
||||
});
|
||||
|
||||
const createSessionState = (
|
||||
options: CompiledCliSessionOptions,
|
||||
): SessionState => ({
|
||||
child: undefined,
|
||||
starting: undefined,
|
||||
buffer: "",
|
||||
nextId: 1,
|
||||
pending: new Map(),
|
||||
stopping: [],
|
||||
queue: Promise.resolve(),
|
||||
stderr: "",
|
||||
closed: false,
|
||||
closing: undefined,
|
||||
settleStart: undefined,
|
||||
ready: false,
|
||||
requestTimeoutMs: positiveMilliseconds(
|
||||
options.requestTimeoutMs,
|
||||
REQUEST_TIMEOUT_MS,
|
||||
),
|
||||
stopTimeoutMs: positiveMilliseconds(options.stopTimeoutMs, STOP_TIMEOUT_MS),
|
||||
spawnWorker: options.spawnWorker ?? defaultSpawnWorker,
|
||||
});
|
||||
|
||||
const enqueue = (
|
||||
state: SessionState,
|
||||
task: () => Promise<unknown>,
|
||||
): Promise<unknown> => {
|
||||
const run = state.queue.then(task);
|
||||
state.queue = run.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
return run;
|
||||
};
|
||||
|
||||
const dispatch = async (
|
||||
state: SessionState,
|
||||
args: readonly string[],
|
||||
): Promise<unknown> => {
|
||||
await ensureStarted(state);
|
||||
const child = state.child;
|
||||
if (child === undefined)
|
||||
throw new Error("Compiled CLI session is not running");
|
||||
const id = state.nextId;
|
||||
state.nextId += 1;
|
||||
const response = await request(state, child, id, args);
|
||||
if (response.exitCode !== 0) throw new Error(failureText(state, response));
|
||||
return JSON.parse(response.stdout) as unknown;
|
||||
};
|
||||
|
||||
const request = (
|
||||
state: SessionState,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
id: number,
|
||||
args: readonly string[],
|
||||
): Promise<CliResponse> =>
|
||||
new Promise((resolve, reject) => {
|
||||
let settled = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const settle = (finish: () => void): void => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
state.pending.delete(id);
|
||||
finish();
|
||||
};
|
||||
timer = setTimeout(() => {
|
||||
settle(() => {
|
||||
reject(
|
||||
new Error(
|
||||
`Compiled CLI command did not respond within ${String(state.requestTimeoutMs)} ms`,
|
||||
),
|
||||
);
|
||||
retireChild(state, child);
|
||||
});
|
||||
}, state.requestTimeoutMs);
|
||||
timer.unref();
|
||||
state.pending.set(id, {
|
||||
resolve: (response) => {
|
||||
settle(() => {
|
||||
resolve(response);
|
||||
});
|
||||
},
|
||||
reject: (cause) => {
|
||||
settle(() => {
|
||||
reject(cause);
|
||||
});
|
||||
},
|
||||
});
|
||||
child.stdin.write(`${JSON.stringify({ id, args })}\n`, (cause) => {
|
||||
if (cause === undefined || cause === null) return;
|
||||
settle(() => {
|
||||
reject(cause);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
const failureText = (state: SessionState, response: CliResponse): string => {
|
||||
const detail = response.error ?? response.stdout.slice(0, 500);
|
||||
const stderr = state.stderr.trim();
|
||||
return stderr.length === 0
|
||||
? `Compiled CLI command failed with status ${String(response.exitCode)}: ${detail}`
|
||||
: `Compiled CLI command failed with status ${String(response.exitCode)}: ${detail}\n${stderr}`;
|
||||
};
|
||||
|
||||
const ensureStarted = async (state: SessionState): Promise<void> => {
|
||||
if (state.closed) throw sessionClosed();
|
||||
await releaseRetired(state);
|
||||
if (state.closed) throw sessionClosed();
|
||||
if (state.ready && state.child !== undefined) return;
|
||||
state.starting ??= launch(state);
|
||||
await state.starting;
|
||||
if (state.closed) throw sessionClosed();
|
||||
};
|
||||
|
||||
const launch = (state: SessionState): Promise<void> =>
|
||||
new Promise((resolve, reject) => {
|
||||
const child = state.spawnWorker();
|
||||
state.child = child;
|
||||
let settled = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const finish = (cause?: Error): void => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
if (state.settleStart === finish) state.settleStart = undefined;
|
||||
if (cause === undefined) resolve();
|
||||
else reject(cause);
|
||||
};
|
||||
timer = setTimeout(() => {
|
||||
signalChild(child, "SIGKILL");
|
||||
finish(
|
||||
new Error(
|
||||
`Compiled CLI session did not start within ${String(START_TIMEOUT_MS)} ms\n${state.stderr}`,
|
||||
),
|
||||
);
|
||||
}, START_TIMEOUT_MS);
|
||||
timer.unref();
|
||||
state.settleStart = finish;
|
||||
const fail = (cause: unknown): void => {
|
||||
finish(cause instanceof Error ? cause : new Error(String(cause)));
|
||||
};
|
||||
const currentChild = (): boolean => state.child === child;
|
||||
child.once("error", fail);
|
||||
child.once("exit", (code, signal) => {
|
||||
if (!currentChild()) return;
|
||||
state.child = undefined;
|
||||
state.ready = false;
|
||||
state.starting = undefined;
|
||||
state.buffer = "";
|
||||
const cause = new Error(
|
||||
`Compiled CLI session exited (${String(code)}, ${String(signal)})\n${state.stderr}`,
|
||||
);
|
||||
fail(cause);
|
||||
rejectPending(state, cause);
|
||||
});
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stderr.on("data", (chunk: string) => {
|
||||
if (!currentChild()) return;
|
||||
state.stderr = `${state.stderr}${chunk}`.slice(-STDERR_TAIL_CHARS);
|
||||
});
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk: string) => {
|
||||
if (!currentChild()) return;
|
||||
state.buffer += chunk;
|
||||
drainLines(state, () => {
|
||||
state.ready = true;
|
||||
finish();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
const hasExited = (child: ChildProcessWithoutNullStreams): boolean =>
|
||||
child.exitCode !== null || child.signalCode !== null;
|
||||
|
||||
const signalChild = (
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
signal: NodeJS.Signals,
|
||||
): void => {
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch (error) {
|
||||
const code =
|
||||
error instanceof Error && "code" in error ? error.code : undefined;
|
||||
if (code !== "ESRCH") throw error;
|
||||
}
|
||||
};
|
||||
|
||||
const detachChild = (
|
||||
state: SessionState,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
): void => {
|
||||
if (state.child !== child) return;
|
||||
state.child = undefined;
|
||||
state.ready = false;
|
||||
state.starting = undefined;
|
||||
state.buffer = "";
|
||||
};
|
||||
|
||||
const stopOwned = (state: SessionState, entry: OwnedStop): Promise<boolean> => {
|
||||
entry.attempt ??= (async () => {
|
||||
if (hasExited(entry.child)) return true;
|
||||
signalChild(entry.child, "SIGKILL");
|
||||
return waitForExit(entry.child, state.stopTimeoutMs);
|
||||
})().finally(() => {
|
||||
entry.attempt = undefined;
|
||||
});
|
||||
return entry.attempt;
|
||||
};
|
||||
|
||||
const trackRetired = (
|
||||
state: SessionState,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
): OwnedStop => {
|
||||
const existing = state.stopping.find((entry) => entry.child === child);
|
||||
if (existing !== undefined) return existing;
|
||||
const entry: OwnedStop = { child, attempt: undefined };
|
||||
state.stopping.push(entry);
|
||||
if (!hasExited(child)) void stopOwned(state, entry).catch(() => undefined);
|
||||
return entry;
|
||||
};
|
||||
|
||||
const releaseRetired = async (state: SessionState): Promise<void> => {
|
||||
let failed = false;
|
||||
for (const entry of [...state.stopping]) {
|
||||
if (hasExited(entry.child)) continue;
|
||||
const stopped = await stopOwned(state, entry).catch(() => false);
|
||||
if (!stopped || !hasExited(entry.child)) failed = true;
|
||||
}
|
||||
const retained = state.stopping.filter((entry) => !hasExited(entry.child));
|
||||
state.stopping.length = 0;
|
||||
state.stopping.push(...retained);
|
||||
if (failed)
|
||||
throw new Error("Compiled CLI session stayed running after SIGKILL");
|
||||
};
|
||||
|
||||
const retireChild = (
|
||||
state: SessionState,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
): void => {
|
||||
detachChild(state, child);
|
||||
if (hasExited(child)) return;
|
||||
trackRetired(state, child);
|
||||
};
|
||||
|
||||
const drainLines = (state: SessionState, onReady: () => void): void => {
|
||||
let newline = state.buffer.indexOf("\n");
|
||||
while (newline !== -1) {
|
||||
const line = state.buffer.slice(0, newline);
|
||||
state.buffer = state.buffer.slice(newline + 1);
|
||||
if (line.length > 0) acceptLine(state, line, onReady);
|
||||
newline = state.buffer.indexOf("\n");
|
||||
}
|
||||
};
|
||||
|
||||
const acceptLine = (
|
||||
state: SessionState,
|
||||
line: string,
|
||||
onReady: () => void,
|
||||
): void => {
|
||||
let message: unknown;
|
||||
try {
|
||||
message = JSON.parse(line) as unknown;
|
||||
} catch (cause) {
|
||||
rejectPending(
|
||||
state,
|
||||
cause instanceof Error ? cause : new Error(String(cause)),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (readyMessage.safeParse(message).success) {
|
||||
onReady();
|
||||
return;
|
||||
}
|
||||
const parsed = responseMessage.safeParse(message);
|
||||
if (!parsed.success) {
|
||||
rejectPending(
|
||||
state,
|
||||
new Error("Compiled CLI session returned an unexpected message"),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const pending = state.pending.get(parsed.data.id);
|
||||
state.pending.delete(parsed.data.id);
|
||||
if (pending === undefined) {
|
||||
rejectPending(
|
||||
state,
|
||||
new Error(
|
||||
`Compiled CLI session returned unexpected id ${String(parsed.data.id)}`,
|
||||
),
|
||||
);
|
||||
return;
|
||||
}
|
||||
pending.resolve(parsed.data);
|
||||
};
|
||||
|
||||
const rejectPending = (state: SessionState, cause: Error): void => {
|
||||
for (const pending of state.pending.values()) pending.reject(cause);
|
||||
state.pending.clear();
|
||||
};
|
||||
|
||||
const closeSession = (state: SessionState): Promise<void> => {
|
||||
state.closed = true;
|
||||
if (state.closing !== undefined) return state.closing;
|
||||
let attempt: Promise<void>;
|
||||
attempt = shutdown(state).catch((error: unknown) => {
|
||||
if (state.closing === attempt) state.closing = undefined;
|
||||
throw error;
|
||||
});
|
||||
state.closing = attempt;
|
||||
return attempt;
|
||||
};
|
||||
|
||||
const shutdown = async (state: SessionState): Promise<void> => {
|
||||
const child = state.child;
|
||||
let forced = false;
|
||||
const stillRunning: OwnedStop[] = [];
|
||||
if (child !== undefined) {
|
||||
rejectPending(state, sessionClosed());
|
||||
state.settleStart?.(sessionClosed());
|
||||
detachChild(state, child);
|
||||
const stopped = await stopGracefully(state, child);
|
||||
if (stopped === "forced") forced = true;
|
||||
if (stopped === "stuck" && !hasExited(child))
|
||||
stillRunning.push({ child, attempt: undefined });
|
||||
}
|
||||
try {
|
||||
await releaseRetired(state);
|
||||
} finally {
|
||||
state.stopping.push(...stillRunning);
|
||||
}
|
||||
if (state.stopping.length > 0)
|
||||
throw new Error("Compiled CLI session stayed running after SIGKILL");
|
||||
if (forced)
|
||||
throw new Error("Compiled CLI session did not exit after stdin closed");
|
||||
};
|
||||
|
||||
const stopGracefully = async (
|
||||
state: SessionState,
|
||||
child: ChildProcessWithoutNullStreams,
|
||||
): Promise<"clean" | "forced" | "stuck"> => {
|
||||
child.stdin.end();
|
||||
if (await waitForExit(child, state.stopTimeoutMs)) return "clean";
|
||||
signalChild(child, "SIGTERM");
|
||||
if (await waitForExit(child, state.stopTimeoutMs)) return "forced";
|
||||
signalChild(child, "SIGKILL");
|
||||
if (await waitForExit(child, state.stopTimeoutMs)) return "forced";
|
||||
return "stuck";
|
||||
};
|
||||
|
||||
const sessionClosed = (): Error => new Error("Compiled CLI session is closed");
|
||||
@@ -1,151 +0,0 @@
|
||||
import { Writable } from "node:stream";
|
||||
import { createInterface } from "node:readline";
|
||||
|
||||
const { closeCliJavaScriptResources, createCli } =
|
||||
await import("../../../dist/cli-runtime/cli.js");
|
||||
const { createCliResultOutput } =
|
||||
await import("../../../dist/cli-runtime/cliResultOutput.js");
|
||||
const {
|
||||
cliConfigurationError,
|
||||
normalizeCliArguments,
|
||||
renderCliOutputArgumentError,
|
||||
renderEmptyFilteredCliOutput,
|
||||
parseCliOutputArguments,
|
||||
} = await import("../../../dist/cli-runtime/cliOutput.js");
|
||||
|
||||
const MAX_STDOUT_CHARS = 16 * 1024 * 1024;
|
||||
|
||||
const writeMessage = (message) => {
|
||||
process.stdout.write(`${JSON.stringify(message)}\n`);
|
||||
};
|
||||
|
||||
// One CLI instance serves every command. A fresh instance per command retains
|
||||
// its command registrations after the call returns; production exits instead.
|
||||
const outputParts = [];
|
||||
let outputCharacters = 0;
|
||||
const destination = new Writable({
|
||||
write(chunk, _encoding, callback) {
|
||||
const text = Buffer.isBuffer(chunk)
|
||||
? chunk.toString("utf8")
|
||||
: String(chunk);
|
||||
outputCharacters += text.length;
|
||||
if (outputCharacters > MAX_STDOUT_CHARS) {
|
||||
callback(new Error("Compiled CLI stdout exceeded 16 MiB"));
|
||||
return;
|
||||
}
|
||||
outputParts.push(text);
|
||||
callback();
|
||||
},
|
||||
});
|
||||
Object.defineProperty(destination, "isTTY", { value: false });
|
||||
const resultOutput = createCliResultOutput(
|
||||
parseCliOutputArguments(["--json"]),
|
||||
destination,
|
||||
);
|
||||
const cli = createCli(process.env, resultOutput);
|
||||
|
||||
const resetOutput = () => {
|
||||
outputParts.length = 0;
|
||||
outputCharacters = 0;
|
||||
};
|
||||
|
||||
const runCommand = async (args) => {
|
||||
const previousExitCode = process.exitCode;
|
||||
process.exitCode = undefined;
|
||||
resetOutput();
|
||||
try {
|
||||
const cliArgs = normalizeCliArguments(args);
|
||||
const outputOptions = parseCliOutputArguments(cliArgs);
|
||||
const outputArguments = outputOptions.validation;
|
||||
if (!outputArguments.ok)
|
||||
return {
|
||||
exitCode: 1,
|
||||
stdout: renderCliOutputArgumentError(outputArguments),
|
||||
};
|
||||
const configurationError = cliConfigurationError(process.env);
|
||||
if (configurationError !== undefined)
|
||||
return { exitCode: 1, stdout: "", error: configurationError };
|
||||
|
||||
let wroteOutput = false;
|
||||
let explicitExit;
|
||||
resultOutput?.begin(outputOptions);
|
||||
try {
|
||||
await cli.serve(cliArgs, {
|
||||
stdout: (output) => {
|
||||
if (resultOutput?.handled) return;
|
||||
if (output.length === 0) return;
|
||||
wroteOutput = true;
|
||||
if (resultOutput?.captureSdkOutput(output)) return;
|
||||
destination.write(output);
|
||||
},
|
||||
exit: (code) => {
|
||||
explicitExit = code;
|
||||
},
|
||||
});
|
||||
} catch (cause) {
|
||||
return {
|
||||
exitCode: 1,
|
||||
stdout: outputParts.join(""),
|
||||
error:
|
||||
cause instanceof Error
|
||||
? (cause.stack ?? cause.message)
|
||||
: String(cause),
|
||||
};
|
||||
}
|
||||
if (!wroteOutput && !resultOutput?.handled) {
|
||||
const empty = renderEmptyFilteredCliOutput(outputOptions);
|
||||
if (empty !== undefined) destination.write(empty);
|
||||
}
|
||||
await resultOutput?.finishSdkOutput(
|
||||
explicitExit ??
|
||||
(typeof process.exitCode === "number" ? process.exitCode : 0),
|
||||
);
|
||||
// logCliCommand reports some failures only through process.exitCode.
|
||||
return {
|
||||
exitCode:
|
||||
explicitExit ??
|
||||
(typeof process.exitCode === "number" ? process.exitCode : 0),
|
||||
stdout: outputParts.join(""),
|
||||
};
|
||||
} finally {
|
||||
process.exitCode = previousExitCode;
|
||||
}
|
||||
};
|
||||
|
||||
const acceptRequest = async (line) => {
|
||||
let request;
|
||||
try {
|
||||
request = JSON.parse(line);
|
||||
} catch (cause) {
|
||||
writeMessage({
|
||||
id: -1,
|
||||
exitCode: 1,
|
||||
stdout: "",
|
||||
error: cause instanceof Error ? cause.message : String(cause),
|
||||
});
|
||||
return;
|
||||
}
|
||||
const args = request?.args;
|
||||
if (
|
||||
typeof request?.id !== "number" ||
|
||||
!Array.isArray(args) ||
|
||||
args.some((argument) => typeof argument !== "string")
|
||||
) {
|
||||
writeMessage({
|
||||
id: typeof request?.id === "number" ? request.id : -1,
|
||||
exitCode: 1,
|
||||
stdout: "",
|
||||
error: "Compiled CLI session expected an id and string arguments",
|
||||
});
|
||||
return;
|
||||
}
|
||||
writeMessage({ id: request.id, ...(await runCommand(args)) });
|
||||
};
|
||||
|
||||
const lines = createInterface({ input: process.stdin, crlfDelay: Infinity });
|
||||
writeMessage({ ready: true });
|
||||
for await (const line of lines) {
|
||||
if (line.trim().length === 0) continue;
|
||||
await acceptRequest(line);
|
||||
}
|
||||
await closeCliJavaScriptResources();
|
||||
Reference in new issue
Block a user