mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix(plugins): replace a retired worker with a fresh process (#19245)
## 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 "<method>" — worker for "<id>" 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 <plugin> 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
5498db1072
commit
0c9962a9bc
3 files changed
+222
-17
No files matched your search
@@ -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", {
|
||||
|
||||
@@ -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<typeof makeDuplexHandle>) {
|
||||
const exited = new Promise<void>((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);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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<typeof setTimeout> | 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<string, string>();
|
||||
@@ -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<void> {
|
||||
if (!childProcess) return;
|
||||
intentionalStop = true;
|
||||
|
||||
Reference in new issue
Block a user