diff --git a/server/src/__tests__/fixtures/plugin-worker-persistent.cjs b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs new file mode 100644 index 0000000000..65eff19127 --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-persistent.cjs @@ -0,0 +1,90 @@ +// Long-lived worker fixture for the stdin command-channel supervision tests. +// +// Unlike the other fixtures, this worker deliberately survives losing its +// stdin: readline "close" does not terminate it, and a ref'd keep-alive timer +// holds the event loop open. That reproduces the field shape the supervision +// fix exists for — the host->worker command pipe dies while the worker process +// itself stays alive and uncommandable. +// +// `shutdown` is acknowledged immediately but the exit is deliberately deferred, +// which leaves a window in which a test can kill the command channel *during* +// an intentional stop (the negative control). + +const readline = require("node:readline"); + +// How long the worker waits after acking `shutdown` before exiting. +// +// This has to sit inside the host's post-ack grace period: stopInternal() +// races the shutdown RPC (which resolves as soon as this ack lands) and then +// waits only 500ms more before escalating to SIGTERM. A delay close to that +// ceiling makes the negative-control test a timing race against a real process +// exit on a loaded CI runner, so keep the margin wide. The test does not +// depend on this window being long — it kills the pipe from a write hook the +// moment the shutdown is flushed, not after a poll. +const SHUTDOWN_EXIT_DELAY_MS = 100; + +/** Hard ceiling so a fixture never outlives the test run that spawned it. */ +const MAX_LIFETIME_MS = 30_000; + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +// Ref'd, so the process stays alive after stdin reaches EOF. +const keepAlive = setInterval(() => {}, 1_000); + +const selfDestruct = setTimeout(() => { + process.exit(0); +}, MAX_LIFETIME_MS); + +function exitNow() { + clearInterval(keepAlive); + clearTimeout(selfDestruct); + process.exit(0); +} + +const rl = readline.createInterface({ + input: process.stdin, + crlfDelay: Infinity, +}); + +// Explicitly do NOT exit here: losing the command channel must leave this +// process alive so the host supervision path is the thing under test. +rl.on("close", () => {}); + +rl.on("line", (line) => { + if (!line.trim()) return; + const message = JSON.parse(line); + const method = message && typeof message.method === "string" ? message.method : null; + + if (method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + ok: true, + supportedMethods: [], + }, + }); + return; + } + + if (method === "shutdown") { + send({ + jsonrpc: "2.0", + id: message.id, + result: {}, + }); + setTimeout(exitNow, SHUTDOWN_EXIT_DELAY_MS); + return; + } + + send({ + jsonrpc: "2.0", + id: message.id, + error: { + code: -32601, + message: `Unhandled method: ${method}`, + }, + }); +}); diff --git a/server/src/__tests__/plugin-worker-stdin-supervision.test.ts b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts new file mode 100644 index 0000000000..9ae2fb50e5 --- /dev/null +++ b/server/src/__tests__/plugin-worker-stdin-supervision.test.ts @@ -0,0 +1,259 @@ +/** + * Supervision of the host→worker command channel (child stdin). + * + * A worker can lose its stdin pipe (EPIPE, or the pipe closing) while the + * process itself is still alive. `child.on("exit")` never fires, so the normal + * crash-recovery path is never reached and the worker zombies: alive, holding + * its slot, and rejecting every host→worker RPC with "not writable" forever. + * These tests cover the supervision that turns that into a real exit so the + * existing handleProcessExit() → scheduleRestart() recovery runs. + * + * This lives in its own file rather than in plugin-worker-manager.test.ts + * because it needs to mock `node:child_process` to capture the spawned child, + * and vi.mock is file-scoped — the sibling suite keeps an unmocked fork. + */ + +import path from "node:path"; +import type { ChildProcess } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js"; + +// Hoisted so the vi.mock factory (which is hoisted above the imports) can see +// it. The fork itself is the real one — only the reference is captured. +const { forkedChildren } = vi.hoisted(() => ({ + forkedChildren: [] as ChildProcess[], +})); + +vi.mock("node:child_process", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + fork: (...args: Parameters): ChildProcess => { + const child = actual.fork(...args); + forkedChildren.push(child); + return child; + }, + }; +}); + +const FIXTURES_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures"); +const PERSISTENT_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-persistent.cjs"); + +const TEST_MANIFEST: PaperclipPluginManifestV1 = { + id: "test.plugin", + apiVersion: 1, + version: "1.0.0", + displayName: "Test plugin", + description: "Test plugin", + author: "Paperclip", + categories: ["automation"], + capabilities: [], + entrypoints: { worker: "dist/worker.js" }, +}; + +/** An EPIPE the way Node surfaces one on a dead pipe. */ +function epipe(): NodeJS.ErrnoException { + const err: NodeJS.ErrnoException = new Error("write EPIPE"); + err.code = "EPIPE"; + err.syscall = "write"; + return err; +} + +type Exit = { code: number | null; signal: NodeJS.Signals | null }; + +function nextExit(child: ChildProcess): Promise { + return new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve({ code: child.exitCode, signal: child.signalCode }); + return; + } + child.once("exit", (code, signal) => resolve({ code, signal })); + }); +} + +/** + * Resolve to the child's exit, or to `null` if it is still alive after + * `timeoutMs`. Reporting "still alive" as a value rather than letting the test + * time out is deliberate: an unsupervised worker zombies forever, and the + * assertion below should name that rather than surface as a bare timeout. + */ +function exitWithin(child: ChildProcess, timeoutMs: number): Promise { + return Promise.race([ + nextExit(child), + new Promise((resolve) => setTimeout(() => resolve(null), timeoutMs)), + ]); +} + +/** + * Destroy the command channel at the instant the host's `shutdown` RPC has + * been flushed to the worker. + * + * This is hooked rather than polled on purpose. Waiting for status + * `"stopping"` via vi.waitFor puts the destroy an unbounded number of + * milliseconds after the ack, which then has to beat the fixture's own + * deferred exit — and the fixture in turn has to beat the host's 500ms + * post-ack SIGTERM escalation. Those two deadlines squeeze from opposite + * sides, and on a loaded runner one of them eventually loses. Hooking the + * write removes both races: the destroy lands immediately after the shutdown + * reaches the worker, so the fixture still has its whole exit window left. + * + * It is also a stronger precondition than polling for status. `sendMessage` + * is only reached for `shutdown` from inside stopInternal(), which sets + * `intentionalStop` before it writes — so firing here proves we are inside the + * intentional-stop window rather than inferring it from an observable status. + * + * Resolves with whether the child was still alive when the pipe was killed. + * If it had already exited, the test never exercised the guard at all, and the + * assertion on this value turns that vacuous pass into a failure. + */ +function destroyStdinOnShutdown(child: ChildProcess): Promise<{ aliveAtDestroy: boolean }> { + const stdin = child.stdin; + if (!stdin) throw new Error("expected the forked child to have a stdin pipe"); + + return new Promise((resolve) => { + const originalWrite = stdin.write.bind(stdin) as typeof stdin.write; + let fired = false; + + stdin.write = ((chunk: unknown, ...rest: unknown[]) => { + const accepted = (originalWrite as (...a: unknown[]) => boolean)(chunk, ...rest); + + if (!fired && typeof chunk === "string" && chunk.includes('"shutdown"')) { + fired = true; + // Let the manager's own write callback run first, so the shutdown is + // fully handed off before the pipe dies. + setImmediate(() => { + const aliveAtDestroy = child.exitCode === null && child.signalCode === null; + stdin.destroy(epipe()); + resolve({ aliveAtDestroy }); + }); + } + + return accepted; + }) as typeof stdin.write; + }); +} + +async function startPersistentWorker() { + const before = forkedChildren.length; + const handle = createPluginWorkerHandle("test.plugin", { + entrypointPath: PERSISTENT_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { + instanceId: "instance-1", + hostVersion: "1.0.0", + }, + apiVersion: 1, + hostHandlers: {}, + rpcTimeoutMs: 5_000, + }); + + await handle.start(); + + const child = forkedChildren[before]; + expect(child, "expected the handle to have forked exactly one child").toBeDefined(); + expect(handle.status).toBe("running"); + // Precondition for both tests: the process is alive and the command channel + // is usable. Without this the assertions below could pass vacuously. + expect(child.exitCode).toBeNull(); + expect(child.signalCode).toBeNull(); + expect(child.stdin?.destroyed ?? true).toBe(false); + + return { handle, child }; +} + +afterEach(() => { + for (const child of forkedChildren.splice(0)) { + if (child.exitCode === null && child.signalCode === null) { + try { + child.kill("SIGKILL"); + } catch { + // Already gone. + } + } + } +}); + +describe("plugin worker stdin command-channel supervision", () => { + it("kills the worker and schedules a restart when stdin dies while the process is alive", async () => { + const { handle, child } = await startPersistentWorker(); + + try { + const crashes: Array<{ signal: NodeJS.Signals | null; willRestart: boolean }> = []; + handle.on("crash", (payload) => { + crashes.push({ signal: payload.signal, willRestart: payload.willRestart }); + }); + + const exited = exitWithin(child, 2_000); + + // The failure shape from the field: the command pipe dies, the worker + // process does not. This fixture deliberately survives stdin EOF, so + // nothing but the host supervision can end it. + child.stdin?.destroy(epipe()); + + const exit = await exited; + + expect( + exit, + "worker was left alive and uncommandable after its command channel died", + ).not.toBeNull(); + // SIGKILL is the discriminator: the fixture never exits on its own + // within the test window, and it exits 0 when asked politely. Only the + // supervision path produces a signalled exit here. + expect(exit!.signal).toBe("SIGKILL"); + + await vi.waitFor(() => { + expect(crashes).toHaveLength(1); + }); + expect(crashes[0]?.willRestart).toBe(true); + + // The recovery that matters is the restart, not the kill: a worker that + // is killed and not rescheduled is still gone. + expect(handle.status).toBe("backoff"); + const diagnostics = handle.diagnostics(); + expect(diagnostics.consecutiveCrashes).toBe(1); + expect(diagnostics.nextRestartAt).not.toBeNull(); + expect(diagnostics.nextRestartAt!).toBeGreaterThan(Date.now()); + } finally { + // stop() cancels the pending backoff timer, so no restart escapes. + await handle.stop().catch(() => undefined); + } + }); + + it("does not force-kill or restart when stdin dies during an intentional stop", async () => { + // The risky arm of the change: a graceful stop closes the command channel + // as a matter of course, and must not be mistaken for the failure above. + const { handle, child } = await startPersistentWorker(); + + const crashes: unknown[] = []; + handle.on("crash", (payload) => crashes.push(payload)); + + const exited = nextExit(child); + + // Kill the command channel mid-stop, while the fixture is still inside its + // deferred-exit window. Without the intentionalStop guard this SIGKILLs a + // worker that was already shutting down cleanly. + const destroyed = destroyStdinOnShutdown(child); + const stopping = handle.stop(); + + const { aliveAtDestroy } = await destroyed; + expect( + aliveAtDestroy, + "fixture exited before the command channel was killed — the guard was never exercised", + ).toBe(true); + + await stopping; + const { code, signal } = await exited; + + expect(signal).toBeNull(); + expect(code).toBe(0); + expect(crashes).toHaveLength(0); + expect(handle.status).toBe("stopped"); + + const diagnostics = handle.diagnostics(); + expect(diagnostics.totalCrashes).toBe(0); + expect(diagnostics.nextRestartAt).toBeNull(); + }); +}); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index af6721f3d4..ec066eb877 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -1034,7 +1034,14 @@ export function createPluginWorkerHandle( throw new Error(`Worker process for plugin "${pluginId}" is not writable`); } const serialized = serializeMessage(message as any); - childProcess.stdin.write(serialized); + // Pass a write callback so an async write error (e.g. EPIPE on the command + // pipe) is surfaced rather than swallowed. The stdin "error" handler wired + // in attachStdioHandlers() drives the actual recovery (forced restart). + childProcess.stdin.write(serialized, (err) => { + if (err) { + log.warn({ err: err.message }, "failed to write message to worker stdin"); + } + }); } function errorCodeForWorkerHostError(err: unknown): number { @@ -2936,6 +2943,36 @@ export function createPluginWorkerHandle( ); } }); + + // Supervise the command channel (host -> worker stdin), not just the + // process. If stdin errors (EPIPE) or closes while the process is still + // alive, the worker is uncommandable: every host->worker RPC write will + // throw "not writable" forever, yet child.on("exit") never fires, so the + // crash-recovery path is never reached and the worker silently zombies + // (observed after multi-day uptime). Force a real exit so the standard + // handleProcessExit() -> scheduleRestart() recovery runs. This is the + // missing third failure mode, mirroring the exit/error handlers above — + // event-driven, not a polling watchdog. + if (child.stdin) { + const onCommandChannelLost = (err?: Error): void => { + // Ignore during graceful stop, or if this child was already replaced. + if (intentionalStop || childProcess !== child) return; + // Only act while the process is still alive (a real exit is handled by + // handleProcessExit). exitCode/signalCode are null until the child dies. + if (child.exitCode !== null || child.signalCode !== null) return; + log.error( + { err: err?.message }, + "worker stdin (command channel) lost while process alive — forcing restart", + ); + try { + child.kill("SIGKILL"); + } catch { + // Best effort — handleProcessExit still runs on the eventual exit. + } + }; + child.stdin.on("error", onCommandChannelLost); + child.stdin.on("close", onCommandChannelLost); + } } function handleProcessExit(