diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index e16c40beaf..436c7bc85c 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -136,7 +136,12 @@ at least one identity source. Supported-platform process probes fail explicitly instead of silently treating a live PID as either the original owner or a recycled process when identity cannot be established. -Use `--drain-required` only when the deploy intentionally requires the old terminate-and-retry behavior. Without that flag, the old server verifies that the marker targets its own PID, snapshots currently running heartbeat run IDs and child PIDs, and skips the shutdown drain so eligible detached local-agent processes can keep running. On startup the new server writes `$PAPERCLIP_HOME/instances/${PAPERCLIP_INSTANCE_ID:-default}/hot-restart-report.json` with `previousServerPid`, `newServerPid`, `previousServerVersion`, `newServerVersion`, `adoptedRunIds`, `finalizedWhileDownRunIds`, `lostRunIds`, and per-run classifications before the normal orphan reaper runs. +Use `--drain-required` only when the deploy intentionally requires the old terminate-and-retry behavior. Without that flag, the old server verifies that the marker targets its own PID, stops new scheduler work, waits for any queue-claim callback already in flight, snapshots currently running heartbeat run IDs and child PIDs, and skips the shutdown drain so eligible detached local-agent processes can keep running. ACP-backed local runs use server-owned stdio and cannot survive their parent server, so the old server instead persists their complete snapshot, changes the marker to `drainRequired` with `drainReason: "active_acp_run"`, and drains only those runs to queued retries. Detached CLI runs remain eligible for adoption during the same mixed restart. If an ACP process terminates but its terminal run update does not persist, startup classifies it as lost with reason `selective_drain_not_finalized` rather than treating the drain as successful. On startup the new server writes `$PAPERCLIP_HOME/instances/${PAPERCLIP_INSTANCE_ID:-default}/hot-restart-report.json` with `previousServerPid`, `newServerPid`, `previousServerVersion`, `newServerVersion`, `drainReason`, `adoptedRunIds`, `finalizedWhileDownRunIds`, `lostRunIds`, and per-run classifications before the normal orphan reaper runs. + +When Paperclip manages embedded PostgreSQL, it suppresses that dependency's eager +`SIGINT`/`SIGTERM` cleanup hooks. Paperclip owns signal ordering so the heartbeat +snapshot and any required drain complete while the database is still available; +the coordinated shutdown path stops embedded PostgreSQL afterward. The request command records the preflight set of running heartbeat IDs and writes an instance-scoped marker plus a PID-targeted legacy home-root handoff marker. @@ -187,6 +192,13 @@ An alive child appears in `adoptedRunIds`; a child that completed during the restart window appears in `finalizedWhileDownRunIds`. Either is continuous. A `lostRunIds` entry remains a failed deploy and must not be waived. +For a recovery from a version that can stop embedded PostgreSQL before writing +its shutdown snapshot, use `--drain-required` once to cross the broken boundary. +After the fixed server is live, perform another ordinary hot restart. Require +`lostRunIds` to be empty and every preflight run to appear in either +`adoptedRunIds` or `finalizedWhileDownRunIds`; an ACP-backed original should be +finalized and have a queued retry rather than be adopted. + Tailscale/private-auth dev mode: ```sh diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 413d2d29d2..7eb04e6807 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1438,6 +1438,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentStatus: "running", processPid: child.pid ?? null, processGroupId: null, + contextSnapshot: { + executionEngine: "cli", + processTopology: "detached", + }, }); await withTempPaperclipHome(async () => { @@ -1490,6 +1494,246 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("snapshots and drains a server-stdio ACP run before embedded database shutdown", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeGreaterThan(0); + const { agentId, runId } = await seedRunFixture({ + agentStatus: "running", + processPid: child.pid ?? null, + processGroupId: null, + contextSnapshot: { + executionEngine: "acp", + processTopology: "server_stdio", + }, + }); + + await withTempPaperclipHome(async (home) => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-acp-version", + requestedAt: new Date("2026-08-04T00:05:00.000Z"), + preflightActiveRunIds: [runId], + }); + const heartbeat = heartbeatService(db); + + await expect(heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-08-04T00:06:00.000Z"), + )).resolves.toEqual({ + mode: "acp_drain_required", + skipDrain: false, + activeRunIds: [runId], + activeAcpRunIds: [runId], + drainRunIds: [runId], + drainReason: "active_acp_run", + }); + await expect(readHotRestartIntent()).resolves.toMatchObject({ + drainRequired: true, + drainReason: "active_acp_run", + drainRunIds: [runId], + shutdownSnapshot: { + activeRuns: [expect.objectContaining({ runId, processPid: child.pid })], + }, + }); + + const drain = await heartbeat.drainRunningRunsForShutdown( + "SIGTERM", + new Date("2026-08-04T00:06:01.000Z"), + [runId], + ); + expect(drain.interruptedRunIds).toEqual([runId]); + expect(drain.retryRunIds).toHaveLength(1); + await waitForPidExit(child.pid!); + + const reconciliation = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-08-04T00:07:00.000Z"), + ); + expect(reconciliation).toMatchObject({ + mode: "reported", + adoptedRunIds: [], + finalizedWhileDownRunIds: [runId], + lostRunIds: [], + skippedRunIds: [], + }); + + const runs = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.agentId, agentId)); + expect(runs.find((run) => run.id === runId)).toMatchObject({ + status: "interrupted", + errorCode: "server_shutdown_interrupted", + }); + expect(runs.find((run) => run.retryOfRunId === runId)).toMatchObject({ + status: "queued", + }); + + const report = JSON.parse( + await fs.readFile(resolveHotRestartReportPath(home), "utf8"), + ) as Record; + expect(report).toMatchObject({ + drainRequired: true, + drainReason: "active_acp_run", + adoptedRunIds: [], + finalizedWhileDownRunIds: [runId], + lostRunIds: [], + }); + }); + }); + + it("reports a selectively drained ACP run as lost when terminal persistence fails", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeGreaterThan(0); + const { runId } = await seedRunFixture({ + agentStatus: "running", + processPid: child.pid ?? null, + processGroupId: null, + contextSnapshot: { + executionEngine: "acp", + processTopology: "server_stdio", + }, + }); + + await withTempPaperclipHome(async (home) => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-acp-persistence-failure-version", + requestedAt: new Date("2026-08-04T00:15:00.000Z"), + preflightActiveRunIds: [runId], + }); + const heartbeat = heartbeatService(db); + + await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-08-04T00:16:00.000Z"), + ); + + // Model the failure boundary precisely: termination succeeded, but the + // interrupted status write never landed, so the durable row is running. + process.kill(child.pid!, "SIGKILL"); + await waitForPidExit(child.pid!); + + const reconciliation = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-08-04T00:17:00.000Z"), + ); + expect(reconciliation).toMatchObject({ + mode: "reported", + adoptedRunIds: [], + finalizedWhileDownRunIds: [], + lostRunIds: [runId], + skippedRunIds: [], + }); + + const report = JSON.parse( + await fs.readFile(resolveHotRestartReportPath(home), "utf8"), + ) as { runs: Array<{ runId: string; classification: string; reason: string }> }; + expect(report.runs).toContainEqual(expect.objectContaining({ + runId, + classification: "lost", + reason: "selective_drain_not_finalized", + })); + }); + }); + + it("drains only server-stdio runs and preserves detached CLI adoption in a mixed restart", async () => { + const acpChild = spawnAliveProcess(); + const cliChild = spawnAliveProcess(); + childProcesses.add(acpChild); + childProcesses.add(cliChild); + expect(acpChild.pid).toBeGreaterThan(0); + expect(cliChild.pid).toBeGreaterThan(0); + + const acp = await seedRunFixture({ + agentStatus: "running", + processPid: acpChild.pid ?? null, + processGroupId: null, + contextSnapshot: { + executionEngine: "acp", + processTopology: "server_stdio", + }, + }); + const cli = await seedRunFixture({ + agentStatus: "running", + processPid: cliChild.pid ?? null, + processGroupId: null, + contextSnapshot: { + executionEngine: "cli", + processTopology: "detached", + }, + }); + + await withTempPaperclipHome(async (home) => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-mixed-version", + requestedAt: new Date("2026-08-04T01:05:00.000Z"), + preflightActiveRunIds: [acp.runId, cli.runId], + }); + const heartbeat = heartbeatService(db); + + const preparation = await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-08-04T01:06:00.000Z"), + ); + expect(preparation).toMatchObject({ + mode: "acp_drain_required", + skipDrain: false, + activeAcpRunIds: [acp.runId], + drainRunIds: [acp.runId], + drainReason: "active_acp_run", + }); + if (preparation.mode !== "acp_drain_required") { + throw new Error(`Expected selective ACP drain, received ${preparation.mode}`); + } + expect(new Set(preparation.activeRunIds)).toEqual(new Set([acp.runId, cli.runId])); + + const drain = await heartbeat.drainRunningRunsForShutdown( + "SIGTERM", + new Date("2026-08-04T01:06:01.000Z"), + preparation.drainRunIds, + ); + expect(drain.interruptedRunIds).toEqual([acp.runId]); + await waitForPidExit(acpChild.pid!); + expect(isPidAlive(cliChild.pid)).toBe(true); + + const reconciliation = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-08-04T01:07:00.000Z"), + ); + expect(reconciliation).toMatchObject({ + mode: "reported", + adoptedRunIds: [cli.runId], + finalizedWhileDownRunIds: [acp.runId], + lostRunIds: [], + skippedRunIds: [], + }); + + const originalRuns = await db + .select() + .from(heartbeatRuns) + .where(inArray(heartbeatRuns.id, [acp.runId, cli.runId])); + expect(originalRuns.find((run) => run.id === acp.runId)).toMatchObject({ + status: "interrupted", + errorCode: "server_shutdown_interrupted", + }); + expect(originalRuns.find((run) => run.id === cli.runId)).toMatchObject({ + status: "running", + }); + + const report = JSON.parse( + await fs.readFile(resolveHotRestartReportPath(home), "utf8"), + ) as Record; + expect(report).toMatchObject({ + drainRequired: true, + drainReason: "active_acp_run", + adoptedRunIds: [cli.runId], + finalizedWhileDownRunIds: [acp.runId], + lostRunIds: [], + }); + }); + }); + it("adopts an old-server legacy snapshot written for a new instance-scoped marker", async () => { const child = spawnAliveProcess(); childProcesses.add(child); @@ -1675,6 +1919,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { runStatus: "queued", processPid: null, processGroupId: null, + contextSnapshot: { + executionEngine: "cli", + processTopology: "detached", + }, includeIssue: false, }); const heartbeat = heartbeatService(db); @@ -1757,6 +2005,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentStatus: "running", processPid: child.pid ?? null, processGroupId: null, + contextSnapshot: { + executionEngine: "cli", + processTopology: "detached", + }, }); await withTempPaperclipHome(async (home) => { @@ -1825,6 +2077,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { agentStatus: "running", processPid: orphan.processPid, processGroupId: orphan.processGroupId, + contextSnapshot: { + executionEngine: "cli", + processTopology: "detached", + }, }); await withTempPaperclipHome(async () => { diff --git a/server/src/index.ts b/server/src/index.ts index 26ccfe9c39..6f7f24c658 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -78,7 +78,10 @@ import { initTelemetry, getTelemetryClient } from "./telemetry.js"; import { conflict } from "./errors.js"; import { ensureDecisionSigningSecret } from "./services/decision-signing.js"; import { createDecisionRetentionNotifyOriginAgent, createDecisionWakeOriginAgent } from "./services/decision-wakeup.js"; -import { coordinateHeartbeatSchedulerShutdown } from "./shutdown.js"; +import { + coordinateHeartbeatSchedulerShutdown, + loadWithoutCoordinatedShutdownSignalHooks, +} from "./shutdown.js"; import { systemdNotify } from "./services/systemd-notify.js"; import { flushInFlightRunLogMirrors } from "./services/run-log-store.js"; import type { @@ -350,7 +353,13 @@ export async function startServer(): Promise { const moduleName = "embedded-postgres"; let EmbeddedPostgres: EmbeddedPostgresCtor; try { - const mod = await import(moduleName); + // embedded-postgres registers async-exit-hook handlers as an import side + // effect. Those handlers stop PostgreSQL immediately on SIGINT/SIGTERM, + // racing Paperclip's later heartbeat snapshot query. Paperclip explicitly + // stops the managed cluster in its own ordered shutdown path instead. + const mod = await loadWithoutCoordinatedShutdownSignalHooks( + () => import(moduleName), + ); EmbeddedPostgres = mod.default as EmbeddedPostgresCtor; } catch { throw new Error( @@ -900,8 +909,14 @@ export async function startServer(): Promise { throw err; } - let drainHeartbeatRunsForShutdown: ((signal: "SIGINT" | "SIGTERM") => Promise) | null = null; - let prepareHotRestartShutdown: ((signal: "SIGINT" | "SIGTERM") => Promise<{ skipDrain: boolean }>) | null = null; + let drainHeartbeatRunsForShutdown: (( + signal: "SIGINT" | "SIGTERM", + runIds?: readonly string[] | null, + ) => Promise) | null = null; + let prepareHotRestartShutdown: ((signal: "SIGINT" | "SIGTERM") => Promise<{ + skipDrain: boolean; + drainRunIds?: string[]; + }>) | null = null; let heartbeatSchedulerStopped = false; let heartbeatSchedulerInterval: ReturnType | null = null; const heartbeatSchedulerInFlight = new Set>(); @@ -946,7 +961,9 @@ export async function startServer(): Promise { const retentionExecutor = decisionRetentionService(db as any, { notifyOriginAgent: createDecisionRetentionNotifyOriginAgent(heartbeat.wakeup), }); - drainHeartbeatRunsForShutdown = heartbeat.drainRunningRunsForShutdown; + drainHeartbeatRunsForShutdown = (signal, runIds) => ( + heartbeat.drainRunningRunsForShutdown(signal, new Date(), runIds) + ); prepareHotRestartShutdown = heartbeat.prepareHotRestartShutdown; const environmentCustomImages = environmentCustomImageService(db as any, { pluginWorkerManager }); const routines = routineService(db as any, { pluginWorkerManager }); @@ -1102,10 +1119,10 @@ export async function startServer(): Promise { await runRetentionSweep(); startHeartbeatSchedulerInterval(() => { - // Async so the suppression checks below can honor the override-aware - // resolver (e.g. worktree run-execution opt-in). The gated work is still - // wrapped in trackHeartbeatSchedulerWork with its own error handling. - void (async () => { + // Track the outer async callback as well as the work it starts. Shutdown + // can then wait through an already-running suppression check before it + // captures the authoritative set of running heartbeat rows. + trackHeartbeatSchedulerWork((async () => { if (heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => { logger.error({ err }, "decision expiry sweep failed"); @@ -1260,7 +1277,9 @@ export async function startServer(): Promise { logger.error({ err }, "periodic heartbeat recovery failed"); })); } - })(); + })().catch((err) => { + logger.error({ err }, "heartbeat scheduler tick failed"); + })); }); } else { startHeartbeatSchedulerInterval(() => { @@ -1383,10 +1402,11 @@ export async function startServer(): Promise { waitForHeartbeatSchedulerIdle, }); const skipHeartbeatDrain = heartbeatShutdown.hotRestart?.skipDrain === true; + const selectiveDrainRunIds = heartbeatShutdown.hotRestart?.drainRunIds ?? null; if (skipHeartbeatDrain) { logger.info( { signal, hotRestart: heartbeatShutdown.hotRestart }, - "hot-restart shutdown prepared; skipping heartbeat scheduler idle wait and graceful run drain", + "hot-restart shutdown prepared after scheduler quiescence; skipping graceful run drain", ); } else if (heartbeatShutdown.preparationError) { logger.error( @@ -1403,7 +1423,7 @@ export async function startServer(): Promise { if (!skipHeartbeatDrain && drainHeartbeatRunsForShutdown) { try { - const drain = await drainHeartbeatRunsForShutdown(signal); + const drain = await drainHeartbeatRunsForShutdown(signal, selectiveDrainRunIds); logger.info({ signal, drain }, "graceful heartbeat run drain complete"); } catch (err) { logger.error({ err, signal }, "graceful heartbeat run drain failed"); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 3f31597604..1adf891798 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -9990,6 +9990,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; } + function isServerStdioBoundHotRestartRun(input: { + run: typeof heartbeatRuns.$inferSelect; + adapterType: string; + adapterConfig: unknown; + }) { + const context = parseObject(input.run.contextSnapshot); + if (context.processTopology === "server_stdio" || context.executionEngine === "acp") { + return true; + } + if (context.processTopology === "detached" || context.executionEngine === "cli") { + return false; + } + if (!["claude_local", "codex_local", "gemini_local"].includes(input.adapterType)) { + return false; + } + return readNonEmptyString(parseObject(input.adapterConfig).engine) !== "cli"; + } + async function prepareHotRestartShutdown(signal: "SIGINT" | "SIGTERM", now = new Date()) { let intent: Awaited>; try { @@ -10013,6 +10031,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .select({ run: heartbeatRuns, adapterType: agents.adapterType, + adapterConfig: agents.adapterConfig, }) .from(heartbeatRuns) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) @@ -10023,6 +10042,39 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) previousServerVersion: intent.previousServerVersion ?? serverVersion, }; + const serverStdioRuns = activeRuns.filter(isServerStdioBoundHotRestartRun); + if (serverStdioRuns.length > 0) { + const activeServerStdioRunIds = serverStdioRuns.map(({ run }) => run.id); + await writeHotRestartShutdownSnapshot({ + intent: intentWithVersion, + signal, + activeRuns: snapshotRuns, + drainReason: "active_acp_run", + drainRunIds: activeServerStdioRunIds, + capturedAt: now, + }); + + logger.warn( + { + signal, + previousServerPid: intent.previousServerPid, + activeRunIds: snapshotRuns.map((run) => run.runId), + activeServerStdioRunIds, + drainReason: "active_acp_run", + }, + "server-stdio agent run prevents hot-restart adoption; using graceful drain and retry", + ); + + return { + mode: "acp_drain_required" as const, + skipDrain: false as const, + activeRunIds: snapshotRuns.map((run) => run.runId), + activeAcpRunIds: activeServerStdioRunIds, + drainRunIds: activeServerStdioRunIds, + drainReason: "active_acp_run" as const, + }; + } + await writeHotRestartShutdownSnapshot({ intent: intentWithVersion, signal, @@ -10083,12 +10135,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } if (!intent.shutdownSnapshot) { - logger.warn( + const log = intent.drainRequired ? logger.info.bind(logger) : logger.warn.bind(logger); + log( { previousServerPid: intent.previousServerPid, preflightActiveRunIds: intent.preflightActiveRunIds, + drainReason: intent.drainReason ?? null, }, - "hot-restart intent present but shutdown snapshot is missing; no runs can be adopted", + intent.drainRequired + ? "drain-required restart intent has no adoption snapshot" + : "hot-restart intent present but shutdown snapshot is missing; no runs can be adopted", ); } const candidates = intent.shutdownSnapshot?.activeRuns ?? []; @@ -10170,7 +10226,20 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) continue; } - if (intent.drainRequired) { + const hasSelectiveAcpDrain = intent.drainReason === "active_acp_run" + && (intent.drainRunIds?.length ?? 0) > 0; + if (hasSelectiveAcpDrain && intent.drainRunIds?.includes(candidate.runId)) { + // A selective ACP drain is expected to persist a terminal row before + // the new server starts. If the process was terminated but that write + // failed, surface the run as lost instead of hiding it as an expected + // drain skip. + classify(candidate, "lost", "selective_drain_not_finalized", patch); + continue; + } + if ( + intent.drainRequired + && !hasSelectiveAcpDrain + ) { classify(candidate, "skipped", "drain_required", patch); continue; } @@ -10250,6 +10319,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) requestedAt: intent.requestedAt, completedAt: now.toISOString(), drainRequired: intent.drainRequired, + drainReason: intent.drainReason ?? (intent.drainRequired ? "requested" : null), previousServerPid: intent.previousServerPid, newServerPid: process.pid, previousServerVersion: intent.previousServerVersion, @@ -10284,7 +10354,15 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }; } - async function drainRunningRunsForShutdown(signal: "SIGINT" | "SIGTERM", now = new Date()) { + async function drainRunningRunsForShutdown( + signal: "SIGINT" | "SIGTERM", + now = new Date(), + runIds: readonly string[] | null = null, + ) { + const selectedRunIds = runIds ? [...new Set(runIds)] : null; + if (selectedRunIds?.length === 0) { + return { interrupted: 0, interruptedRunIds: [], retryRunIds: [] }; + } const activeRuns = await db .select({ run: heartbeatRuns, @@ -10292,7 +10370,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }) .from(heartbeatRuns) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) - .where(eq(heartbeatRuns.status, "running")); + .where( + selectedRunIds + ? and( + eq(heartbeatRuns.status, "running"), + inArray(heartbeatRuns.id, selectedRunIds), + ) + : eq(heartbeatRuns.status, "running"), + ); const interruptedRunIds: string[] = []; const retryRunIds: string[] = []; diff --git a/server/src/services/hot-restart.ts b/server/src/services/hot-restart.ts index 02700ff096..0e823ea24b 100644 --- a/server/src/services/hot-restart.ts +++ b/server/src/services/hot-restart.ts @@ -35,6 +35,8 @@ export type HotRestartIntent = { previousServerStartedAt?: string | null; previousServerVersion: string | null; drainRequired: boolean; + drainReason?: "requested" | "active_acp_run" | null; + drainRunIds?: string[]; requestedByRunId: string | null; preflightActiveRunIds: string[]; shutdownSnapshot?: { @@ -58,6 +60,7 @@ export type HotRestartReport = { requestedAt: string; completedAt: string; drainRequired: boolean; + drainReason: "requested" | "active_acp_run" | null; previousServerPid: number; newServerPid: number; previousServerVersion: string | null; @@ -123,6 +126,10 @@ function asBoolean(value: unknown): boolean { return value === true; } +function asDrainReason(value: unknown) { + return value === "requested" || value === "active_acp_run" ? value : null; +} + function asDateString(value: unknown): string | null { const candidate = asString(value); if (!candidate) return null; @@ -412,6 +419,8 @@ export function parseHotRestartIntent(value: unknown): HotRestartIntent | null { previousServerStartedAt: asDateString(value.previousServerStartedAt), previousServerVersion: asString(value.previousServerVersion), drainRequired: asBoolean(value.drainRequired), + drainReason: asDrainReason(value.drainReason), + drainRunIds: asStringArray(value.drainRunIds), requestedByRunId: asString(value.requestedByRunId), preflightActiveRunIds: asStringArray(value.preflightActiveRunIds), }; @@ -479,6 +488,7 @@ export async function writeHotRestartIntent(input: { previousServerStartedAt?: string | null; previousServerVersion?: string | null; drainRequired?: boolean; + drainReason?: "requested" | "active_acp_run" | null; requestedByRunId?: string | null; preflightActiveRunIds?: string[]; requestedAt?: Date; @@ -502,6 +512,7 @@ export async function writeHotRestartIntent(input: { previousServerStartedAt, previousServerVersion: input.previousServerVersion ?? null, drainRequired: input.drainRequired ?? false, + drainReason: input.drainReason ?? (input.drainRequired ? "requested" : null), requestedByRunId: input.requestedByRunId ?? null, preflightActiveRunIds: asStringArray(input.preflightActiveRunIds), }; @@ -527,11 +538,20 @@ export async function writeHotRestartShutdownSnapshot(input: { intent: HotRestartIntent; signal: "SIGINT" | "SIGTERM"; activeRuns: HotRestartIntentRun[]; + drainReason?: "active_acp_run"; + drainRunIds?: string[]; capturedAt?: Date; homeDir?: string; }) { const updated: HotRestartIntent = { ...input.intent, + ...(input.drainReason + ? { + drainRequired: true, + drainReason: input.drainReason, + drainRunIds: asStringArray(input.drainRunIds), + } + : {}), shutdownSnapshot: { capturedAt: (input.capturedAt ?? new Date()).toISOString(), signal: input.signal, diff --git a/server/src/shutdown.test.ts b/server/src/shutdown.test.ts index 524cc2418c..74bf4b0fc7 100644 --- a/server/src/shutdown.test.ts +++ b/server/src/shutdown.test.ts @@ -1,12 +1,77 @@ +import { EventEmitter } from "node:events"; import { describe, expect, it, vi } from "vitest"; -import { coordinateHeartbeatSchedulerShutdown } from "./shutdown.js"; +import { + coordinateHeartbeatSchedulerShutdown, + loadWithoutCoordinatedShutdownSignalHooks, +} from "./shutdown.js"; + +describe("loadWithoutCoordinatedShutdownSignalHooks", () => { + it("removes the eager signal handlers from the real embedded-postgres import", async () => { + const before = { + SIGINT: process.rawListeners("SIGINT"), + SIGTERM: process.rawListeners("SIGTERM"), + }; + const moduleName = "embedded-postgres"; + + await loadWithoutCoordinatedShutdownSignalHooks(() => import(moduleName)); + + expect(process.rawListeners("SIGINT")).toEqual(before.SIGINT); + expect(process.rawListeners("SIGTERM")).toEqual(before.SIGTERM); + }); + + it("keeps the database available for a marker-backed SIGTERM snapshot", async () => { + const signalTarget = new EventEmitter(); + const preexistingSignalListener = vi.fn(); + signalTarget.on("SIGTERM", preexistingSignalListener); + + let databaseAvailable = true; + const embeddedPostgresExitHook = vi.fn(() => { + databaseAvailable = false; + }); + await loadWithoutCoordinatedShutdownSignalHooks( + async () => { + signalTarget.on("SIGINT", embeddedPostgresExitHook); + signalTarget.on("SIGTERM", embeddedPostgresExitHook); + return { default: class EmbeddedPostgres {} }; + }, + signalTarget, + ); + + let shutdown: Promise | null = null; + let snapshotCaptured = false; + signalTarget.once("SIGTERM", () => { + shutdown = coordinateHeartbeatSchedulerShutdown({ + signal: "SIGTERM", + prepareHotRestartShutdown: async () => { + // This models the real failure path: a valid intent exists, and the + // snapshot must query embedded PostgreSQL after SIGTERM is delivered. + expect(databaseAvailable).toBe(true); + snapshotCaptured = true; + return { mode: "hot_restart" as const, skipDrain: true }; + }, + waitForHeartbeatSchedulerIdle: vi.fn(async () => undefined), + }); + }); + + signalTarget.emit("SIGTERM"); + await shutdown; + + expect(preexistingSignalListener).toHaveBeenCalledOnce(); + expect(embeddedPostgresExitHook).not.toHaveBeenCalled(); + expect(snapshotCaptured).toBe(true); + }); +}); describe("coordinateHeartbeatSchedulerShutdown", () => { - it("captures a hot-restart snapshot without waiting for active scheduler work", async () => { + it("quiesces active scheduler work before capturing a hot-restart snapshot", async () => { let snapshotCaptured = false; - const waitForHeartbeatSchedulerIdle = vi.fn(() => new Promise(() => undefined)); + let releaseScheduler!: () => void; + const schedulerIdle = new Promise((resolve) => { + releaseScheduler = resolve; + }); + const waitForHeartbeatSchedulerIdle = vi.fn(() => schedulerIdle); - const result = await coordinateHeartbeatSchedulerShutdown({ + const shutdown = coordinateHeartbeatSchedulerShutdown({ signal: "SIGTERM", prepareHotRestartShutdown: vi.fn(async () => { snapshotCaptured = true; @@ -15,12 +80,41 @@ describe("coordinateHeartbeatSchedulerShutdown", () => { waitForHeartbeatSchedulerIdle, }); + await vi.waitFor(() => expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce()); + expect(snapshotCaptured).toBe(false); + releaseScheduler(); + + const result = await shutdown; expect(snapshotCaptured).toBe(true); - expect(waitForHeartbeatSchedulerIdle).not.toHaveBeenCalled(); expect(result).toEqual({ hotRestart: { mode: "prepared", skipDrain: true }, preparationError: null, - waitedForSchedulerIdle: false, + waitedForSchedulerIdle: true, + }); + }); + + it("quiesces scheduler work before selecting server-stdio runs to drain", async () => { + const waitForHeartbeatSchedulerIdle = vi.fn(async () => undefined); + + const result = await coordinateHeartbeatSchedulerShutdown({ + signal: "SIGTERM", + prepareHotRestartShutdown: vi.fn(async () => ({ + mode: "acp_drain_required" as const, + skipDrain: false, + drainRunIds: ["acp-run"], + })), + waitForHeartbeatSchedulerIdle, + }); + + expect(waitForHeartbeatSchedulerIdle).toHaveBeenCalledOnce(); + expect(result).toEqual({ + hotRestart: { + mode: "acp_drain_required", + skipDrain: false, + drainRunIds: ["acp-run"], + }, + preparationError: null, + waitedForSchedulerIdle: true, }); }); diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index a0e657eae7..3100e5312b 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -2,6 +2,51 @@ type HotRestartShutdownPreparation = { skipDrain: boolean; }; +const COORDINATED_SHUTDOWN_SIGNALS = ["SIGINT", "SIGTERM"] as const; + +type ShutdownSignalTarget = { + rawListeners(eventName: string): Function[]; + removeListener(eventName: string, listener: (...args: any[]) => void): unknown; +}; + +/** + * Some dependencies eagerly install process signal handlers as an import side + * effect. Paperclip must remain the sole owner of SIGINT/SIGTERM ordering: its + * handler first snapshots live heartbeat runs and only then stops embedded + * infrastructure. Remove only listeners added by the supplied import, while + * preserving every listener that was already registered. + */ +export async function loadWithoutCoordinatedShutdownSignalHooks( + load: () => Promise, + signalTarget: ShutdownSignalTarget = process, +) { + const listenersBeforeLoad = new Map( + COORDINATED_SHUTDOWN_SIGNALS.map((signal) => [ + signal, + signalTarget.rawListeners(signal), + ]), + ); + + let loaded: T; + try { + loaded = await load(); + } finally { + for (const signal of COORDINATED_SHUTDOWN_SIGNALS) { + const remainingBeforeLoad = [...(listenersBeforeLoad.get(signal) ?? [])]; + for (const listener of signalTarget.rawListeners(signal)) { + const existingIndex = remainingBeforeLoad.indexOf(listener); + if (existingIndex >= 0) { + remainingBeforeLoad.splice(existingIndex, 1); + continue; + } + signalTarget.removeListener(signal, listener as (...args: any[]) => void); + } + } + } + + return loaded; +} + export async function coordinateHeartbeatSchedulerShutdown< TPreparation extends HotRestartShutdownPreparation, >(input: { @@ -16,6 +61,12 @@ export async function coordinateHeartbeatSchedulerShutdown< let hotRestart: TPreparation | null = null; let preparationError: unknown = null; + // The signal handler stops the scheduler before entering this coordinator. + // Quiesce any callback that was already in flight before querying running + // rows for the shutdown snapshot, otherwise a late queue claim can create a + // run that is absent from both the snapshot and the selective drain set. + await input.waitForHeartbeatSchedulerIdle(); + if (input.prepareHotRestartShutdown) { try { hotRestart = await input.prepareHotRestartShutdown(input.signal); @@ -24,15 +75,6 @@ export async function coordinateHeartbeatSchedulerShutdown< } } - if (hotRestart?.skipDrain) { - return { - hotRestart, - preparationError, - waitedForSchedulerIdle: false, - }; - } - - await input.waitForHeartbeatSchedulerIdle(); return { hotRestart, preparationError,