From 0c9962a9bceafa3924d87ee643311dbd6cfe4433 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 5 Nov 2026 08:37:16 -0600 Subject: [PATCH] fix(plugins): replace a retired worker with a fresh process (#19245) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Sandbox providers such as Daytona run as plugin workers that the host manages in `server/src/services/plugin-worker-manager.ts` > - The host retires a worker when it cannot prove that a login pseudo-terminal or a duplex channel route is gone (fail closed) > - Retirement used `killProcess()`, which marks a deliberate stop, so the handle stayed `stopped` and no restart followed > - Every later call then failed with `Cannot call "" — worker for "" is stopped` until the server process restarted > - This pull request makes a retirement replace the worker with a fresh process through the existing backoff timer, with a bound on repeated retirements > - The benefit is that one unconfirmed close no longer makes a sandbox provider unavailable for the whole server ## Linked Issues or Issue Description Refs #19150 **What happened?** A Daytona plugin worker was retired after an unacknowledged duplex close. After that, every run that used the Daytona environment failed setup with `Cannot call "environmentExecute" — worker for is stopped`. 15 runs failed in about 12 minutes. The plugin stayed unavailable until the server restarted. #19150 removed the specific trigger. It keeps fail-closed retirement for close replies that the host cannot confirm, so the same outage can still occur. **Expected behavior** The retired process, its routes, and its tombstones are never reused. The plugin comes back on a fresh process after a short, bounded backoff. A deliberate stop (plugin disable, upgrade, shutdown) stays stopped. **Steps to reproduce** 1. Start a plugin worker and open a duplex channel whose close acknowledgement is wrong or missing. 2. Close the channel. The host retires the worker. 3. Call any worker method. Before this change, the call fails with `is stopped` forever. ## What Changed - Add `retireWorkerProcess()`. It kills the process but does not mark a deliberate stop. The three fail-closed retirement paths use it instead of `killProcess()`. - In the exit handler, a retired exit settles every route as before, then counts the retirement and schedules a fresh process through the existing backoff timer (`crashed` → `backoff` → `running`). It emits the existing `crash` event with `willRestart`. - A deliberate stop wins. If `stop()` runs before the exit, the exit ends `stopped`. If `stop()` runs during backoff, it cancels the restart. - Bound repeated retirements: at most 5 retirements in the 10-minute crash window. A successful restart does not reset this count (it does reset the crash count), so a worker that starts cleanly and is retired again cannot loop forever. After the bound, the worker stays `crashed`. - `scheduleRestart()` / `computeBackoffMs()` take the attempt number, so the crash path and the retirement path each use their own counter. - Add an optional `crashRecovery` start option (`minBackoffMs`, `maxConsecutiveRetirements`) so tests can use short waits. - Update the Daytona worker close test: a stalled close now shows `backoff`, then a fresh `running` process with a new pid. ## Verification - Red before the fix: the three new tests in `server/src/__tests__/plugin-worker-manager-duplex.test.ts` (`plugin worker replacement after a fail-closed retirement`) failed with `expected 'stopped' to be 'running'` / `'backoff'`. - Green after the fix. The tests prove: - the retired route and a sibling route on the same worker both settle with `{ exitCode: null }`; - the fresh process has a new pid, a write to a retired route does not revive it, and a new route on the fresh process echoes data; - `stop()` during backoff stays `stopped` with no process; - with `maxConsecutiveRetirements: 2`, the third retirement leaves the worker `crashed` and `call()` rejects. - The real Daytona plugin worker test (`plugin-worker-manager-daytona-close.test.ts`, `stalled-close`) now shows the replacement with the actual provider bundle. The closed lease still rejects on the fresh process. - `cd server && npx vitest run src/__tests__/plugin src/__tests__/managed-plugin src/services/plugin` → 48 files, 649 tests pass. The new tests passed 5 runs in a row. - `npx tsc --noEmit -p server` shows no errors in the changed files. - Not run: a live staging run. This PR does not deploy. ## Risks - A plugin that keeps causing retirements now restarts up to 5 times in 10 minutes (1s, 2s, 4s, 8s, 16s backoff) instead of staying dead. After that, it stays `crashed`, as a crash loop does now. - A retirement now emits `plugin.worker.crashed`, so host-service cleanup runs for the plugin, the same as for a crash. A restart emits `plugin.worker.restarted`. A retirement also counts in `totalCrashes`, so idle sleep treats the worker as it treats a crashed worker. - A call made during backoff fails with `is backoff` rather than `is stopped`. The outage is bounded to the backoff, not to a server restart. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use (shell, file edit, test runs). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- ...lugin-worker-manager-daytona-close.test.ts | 18 ++- .../plugin-worker-manager-duplex.test.ts | 107 ++++++++++++++++ server/src/services/plugin-worker-manager.ts | 114 ++++++++++++++++-- 3 files changed, 222 insertions(+), 17 deletions(-) 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;