diff --git a/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts b/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts index c9e5c649bad..3074cb2778a 100644 --- a/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts +++ b/server/src/__tests__/plugin-worker-manager-daytona-close.test.ts @@ -3,7 +3,7 @@ import os from "node:os"; import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; import { build } from "esbuild"; -import { expect, it } from "vitest"; +import { expect, it, vi } from "vitest"; import { createPluginWorkerManager } from "../services/plugin-worker-manager.js"; const repoRoot = fileURLToPath(new URL("../../..", import.meta.url)); @@ -71,11 +71,19 @@ it.each(["release", "cancel", "stop", "destroy", "resume", "stalled-close"] as c const execution = manager.call(pluginId, "environmentExecute", { ...base, lease: next, command: "printf", args: ["monitor-after-release"], bypassSession: true, }); - if (mode === "stalled-close") await expect(execution).rejects.toThrow("is stopped"); - else await expect(execution).resolves.toMatchObject({ exitCode: 0, stdout: "saved work remains intact" }); - expect(worker.status).toBe(mode === "stalled-close" ? "stopped" : "running"); + if (mode === "stalled-close") { + // The unconfirmed close retires the process. The plugin is unavailable only + // while the bounded restart backoff runs, then a fresh process replaces it. + await expect(execution).rejects.toThrow("is backoff"); + await vi.waitFor(() => expect(worker.status).toBe("running"), { timeout: 5_000 }); + expect(worker.diagnostics().pid).not.toBe(pid); + expect(worker.diagnostics().pid).not.toBeNull(); + } else { + await expect(execution).resolves.toMatchObject({ exitCode: 0, stdout: "saved work remains intact" }); + expect(worker.diagnostics().pid).toBe(pid); + } + expect(worker.status).toBe("running"); expect(manager.getWorker(pluginId)).toBe(worker); - expect(worker.diagnostics().pid).toBe(mode === "stalled-close" ? null : pid); channel.write(new TextEncoder().encode("closed channel must remain closed")); channel.kill(); if (mode !== "resume") await expect(manager.call(pluginId, "environmentExecute", { diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts index ac53ee57691..f01aa8524f4 100644 --- a/server/src/__tests__/plugin-worker-manager-duplex.test.ts +++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts @@ -962,3 +962,110 @@ describe("plugin worker manager duplex channel route", () => { } }); }); + +// A fail-closed retirement kills the worker process because the host cannot +// prove a route is gone. The retired process, its routes, and its tombstones are +// never reused, but the plugin itself must come back on a fresh process instead +// of staying stopped until the server restarts. +describe("plugin worker replacement after a fail-closed retirement", () => { + const fastRecovery = { crashRecovery: { minBackoffMs: 20 } }; + + async function retireOnce(handle: ReturnType) { + const exited = new Promise((resolve) => { + const onExit = () => { + handle.off("exit", onExit); + resolve(); + }; + handle.on("exit", onExit); + }); + const session = await handle.openDuplexChannel( + duplexOpenInput({ mode: "normal", closeMode: "bad-ack" }), + ); + const waited = session.wait(); + await session.close(); + await exited; + return { session, waited }; + } + + it("replaces the retired worker with a fresh process", async () => { + const handle = makeDuplexHandle({ + ...fastRecovery, + duplexChannelLimits: { closeTimeoutMs: 200 }, + }); + try { + await handle.start(); + const retiredPid = handle.diagnostics().pid; + const sibling = await handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })); + const siblingChunks: string[] = []; + sibling.onData((chunk) => siblingChunks.push(new TextDecoder().decode(chunk))); + const { session, waited } = await retireOnce(handle); + // Every route on the retired worker settles as failed. + await expect(waited).resolves.toEqual({ exitCode: null }); + await expect(sibling.wait()).resolves.toEqual({ exitCode: null }); + + await vi.waitFor(() => expect(handle.status).toBe("running"), { timeout: 5_000 }); + const freshPid = handle.diagnostics().pid; + expect(freshPid).not.toBeNull(); + expect(freshPid).not.toBe(retiredPid); + + // No retired route is revived on the fresh process. + session.write(new TextEncoder().encode("retired route stays closed")); + sibling.write(new TextEncoder().encode("retired route stays closed")); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(siblingChunks).toEqual([]); + + // The fresh process serves new routes. + const next = await handle.openDuplexChannel( + duplexOpenInput({ workerSessionId: "ws-next", echoInput: true }), + ); + const chunks: string[] = []; + next.onData((chunk) => chunks.push(new TextDecoder().decode(chunk))); + next.write(new TextEncoder().encode("fresh")); + await vi.waitFor(() => expect(chunks).toContain("echo:fresh")); + await next.close(); + expect(handle.status).toBe("running"); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("keeps a deliberate stop stopped while a replacement is pending", async () => { + const handle = makeDuplexHandle({ + crashRecovery: { minBackoffMs: 300 }, + duplexChannelLimits: { closeTimeoutMs: 200 }, + }); + try { + await handle.start(); + await retireOnce(handle); + await vi.waitFor(() => expect(handle.status).toBe("backoff")); + await handle.stop(); + expect(handle.status).toBe("stopped"); + await new Promise((resolve) => setTimeout(resolve, 700)); + expect(handle.status).toBe("stopped"); + expect(handle.diagnostics().pid).toBeNull(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("bounds repeated retirements instead of restarting forever", async () => { + const handle = makeDuplexHandle({ + crashRecovery: { minBackoffMs: 20, maxConsecutiveRetirements: 2 }, + duplexChannelLimits: { closeTimeoutMs: 200 }, + }); + try { + await handle.start(); + for (let attempt = 0; attempt < 2; attempt += 1) { + await retireOnce(handle); + await vi.waitFor(() => expect(handle.status).toBe("running"), { timeout: 5_000 }); + } + await retireOnce(handle); + await new Promise((resolve) => setTimeout(resolve, 500)); + expect(handle.status).toBe("crashed"); + expect(handle.diagnostics().pid).toBeNull(); + await expect(handle.call("health", {} as never)).rejects.toThrow("is crashed"); + } finally { + await handle.stop().catch(() => undefined); + } + }); +}); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 121f731460e..221be9cad40 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -119,6 +119,14 @@ const MAX_CONSECUTIVE_CRASHES = 10; /** Time window in which crashes are considered consecutive (10 minutes). */ const CRASH_WINDOW_MS = 10 * 60 * 1_000; +/** + * Maximum fail-closed retirements within {@link CRASH_WINDOW_MS} before the host + * stops replacing the worker. Unlike the crash counter, a successful restart does + * not reset this count, so a worker that initializes cleanly and is then retired + * again cannot restart in an unbounded loop. + */ +const MAX_CONSECUTIVE_RETIREMENTS = 5; + /** Maximum number of stderr characters retained for worker failure context. */ const MAX_STDERR_EXCERPT_CHARS = 8_000; @@ -428,6 +436,16 @@ export interface WorkerStartOptions { rpcTimeoutMs?: number; /** Whether to auto-restart on crash. Defaults to true. */ autoRestart?: boolean; + /** + * Crash-recovery tuning. A test shortens the backoff and the retirement bound + * to exercise replacement without long waits. + */ + crashRecovery?: { + /** The first restart delay, in milliseconds. Later attempts double it. */ + minBackoffMs?: number; + /** Max fail-closed retirements in the crash window before restarts stop. */ + maxConsecutiveRetirements?: number; + }; /** Node.js execArgv passed to the child process. */ execArgv?: string[]; /** Environment variables passed to the child process. */ @@ -1083,6 +1101,14 @@ export function createPluginWorkerHandle( let backoffTimer: ReturnType | null = null; let nextRestartAt: number | null = null; + // Fail-closed retirement tracking. A retirement kills the current process + // because the host cannot prove one of its routes is gone. The exit that + // follows replaces the process with a fresh one, bounded by + // `maxConsecutiveRetirements` within the crash window. + let retirementPending = false; + let consecutiveRetirements = 0; + let lastRetirementAt: number | null = null; + // Track open stream channels so we can emit synthetic close on crash. // Maps channel → companyId. const openStreamChannels = new Map(); @@ -1092,6 +1118,9 @@ export function createPluginWorkerHandle( const rpcTimeoutMs = options.rpcTimeoutMs ?? DEFAULT_RPC_TIMEOUT_MS; const autoRestart = options.autoRestart ?? true; + const minBackoffMs = options.crashRecovery?.minBackoffMs ?? MIN_BACKOFF_MS; + const maxConsecutiveRetirements = + options.crashRecovery?.maxConsecutiveRetirements ?? MAX_CONSECUTIVE_RETIREMENTS; // ----------------------------------------------------------------------- // Status management @@ -1636,7 +1665,7 @@ export function createPluginWorkerHandle( { pluginId }, "login pseudo-terminal close not acknowledged; retiring worker", ); - void killProcess(); + retireWorkerProcess(); } } @@ -2127,7 +2156,7 @@ export function createPluginWorkerHandle( // overflow. The host never logs the raw frame content on a violation. function retireDuplexWorkerOnViolation(reason: string): void { log.error({ pluginId }, `duplex channel ownership violation (${reason}); retiring worker`); - void killProcess(); + retireWorkerProcess(); } // Install a tombstone for a bound pair atomically with the removal of its live @@ -2234,7 +2263,7 @@ export function createPluginWorkerHandle( { pluginId }, "duplex channel close not acknowledged; retiring worker", ); - void killProcess(); + retireWorkerProcess(); } } @@ -3102,6 +3131,8 @@ export function createPluginWorkerHandle( signal: NodeJS.Signals | null, ): void { const wasIntentional = intentionalStop; + const wasRetired = retirementPending; + retirementPending = false; // Clean up readline interfaces if (readline) { @@ -3157,12 +3188,18 @@ export function createPluginWorkerHandle( emitter.emit("exit", { pluginId, code, signal }); if (wasIntentional) { - // Graceful stop — status is already "stopping" or will be set to "stopped" + // Graceful stop — status is already "stopping" or will be set to "stopped". + // A deliberate stop that races a retirement stays stopped. setStatus("stopped"); log.info({ code, signal }, "worker process stopped"); return; } + if (wasRetired) { + handleRetiredProcessExit(code, signal); + return; + } + // Unexpected exit — crash recovery totalCrashes++; const now = Date.now(); @@ -3186,7 +3223,7 @@ export function createPluginWorkerHandle( emitter.emit("crash", { pluginId, code, signal, willRestart }); if (willRestart) { - scheduleRestart(); + scheduleRestart(consecutiveCrashes); } else { log.error( { consecutiveCrashes, maxCrashes: MAX_CONSECUTIVE_CRASHES }, @@ -3195,6 +3232,44 @@ export function createPluginWorkerHandle( } } + // The retired process is gone, together with every route, reservation, and + // tombstone it held (the exit path above already settled them as failed). + // Replace it with a fresh process through the same backoff timer a crash uses. + // The fresh process starts with empty route state, and host route ids are + // monotonic, so no retired route can resolve against it. The retirement count + // survives a successful restart, so repeated retirements stop restarting after + // `maxConsecutiveRetirements` within the crash window. + function handleRetiredProcessExit( + code: number | null, + signal: NodeJS.Signals | null, + ): void { + totalCrashes++; + const now = Date.now(); + if (lastRetirementAt !== null && now - lastRetirementAt > CRASH_WINDOW_MS) { + consecutiveRetirements = 0; + } + consecutiveRetirements++; + lastRetirementAt = now; + + const willRestart = autoRestart && consecutiveRetirements <= maxConsecutiveRetirements; + log.error( + { code, signal, consecutiveRetirements, totalCrashes, willRestart }, + "retired worker process exited", + ); + + setStatus("crashed"); + emitter.emit("crash", { pluginId, code, signal, willRestart }); + + if (willRestart) { + scheduleRestart(consecutiveRetirements); + } else { + log.error( + { consecutiveRetirements, maxRetirements: maxConsecutiveRetirements }, + "max consecutive worker retirements reached, not restarting", + ); + } + } + function rejectAllPending(error: Error): void { for (const [id, pending] of pendingRequests) { clearTimeout(pending.timer); @@ -3217,23 +3292,23 @@ export function createPluginWorkerHandle( // Crash recovery with exponential backoff // ----------------------------------------------------------------------- - function computeBackoffMs(): number { - // Exponential backoff: MIN_BACKOFF * MULTIPLIER^(consecutiveCrashes - 1) + function computeBackoffMs(attempt: number): number { + // Exponential backoff: MIN_BACKOFF * MULTIPLIER^(attempt - 1) const delay = - MIN_BACKOFF_MS * Math.pow(BACKOFF_MULTIPLIER, consecutiveCrashes - 1); + minBackoffMs * Math.pow(BACKOFF_MULTIPLIER, Math.max(attempt, 1) - 1); // Add jitter: ±25% const jitter = delay * 0.25 * (Math.random() * 2 - 1); return Math.min(Math.round(delay + jitter), MAX_BACKOFF_MS); } - function scheduleRestart(): void { - const delay = computeBackoffMs(); + function scheduleRestart(attempt: number): void { + const delay = computeBackoffMs(attempt); nextRestartAt = Date.now() + delay; setStatus("backoff"); log.info( - { delayMs: delay, consecutiveCrashes }, + { delayMs: delay, attempt }, "scheduling restart with backoff", ); @@ -3243,7 +3318,7 @@ export function createPluginWorkerHandle( // A temporary idle hold must not consume the crash-recovery attempt. // Keep the bounded backoff until admission reopens. if (readTaskDrain(new Date())?.ownerId) { - scheduleRestart(); + scheduleRestart(attempt); return; } try { @@ -3275,6 +3350,7 @@ export function createPluginWorkerHandle( } intentionalStop = false; + retirementPending = false; setStatus("starting"); stderrExcerpt = ""; @@ -3432,6 +3508,20 @@ export function createPluginWorkerHandle( }); } + // Retire the current process on a fail-closed route violation. Unlike + // `killProcess`, this is not a deliberate stop: the exit replaces the process + // with a fresh one. A deliberate stop already in progress wins, so a plugin + // disable, upgrade, or shutdown still ends stopped. + function retireWorkerProcess(): void { + if (!childProcess) return; + if (!intentionalStop) retirementPending = true; + try { + childProcess.kill("SIGKILL"); + } catch { + // Process may already be dead; its exit event still runs the exit path. + } + } + async function killProcess(): Promise { if (!childProcess) return; intentionalStop = true;