diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 84141d4f0d5..5bfe2a89fd4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3292,6 +3292,11 @@ A graceful shutdown can also suspend a restart-recovery claim whose execution has not yet reached provider dispatch. That execution has lost its authority: it leaves the run running, its budget hold held and its workspace unmarked, and the next controller adopts the same run. It does not settle the run as failed. +Shutdown closes the runtime API listener before it drains runs. A native +execution that then needs the listener (for example, to give a new runner its +callback origin) detaches the same way instead of failing. It keeps its claim, +the drain suspends it, and the next controller adopts the run. A listener that +never started is still a startup failure. Remote Runner processes use their authenticated connection lease as the bound on disconnected lifetime. The controller's shorter connection-recovery budget limits its own attempt; it must not terminate the remote owner while another diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index cdf63886b62..c340db6c085 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { authUsers, budgetReservations, companyMemberships, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db"; import { expect, it, vi } from "vitest"; +import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; @@ -74,6 +75,56 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut 90_000, ); + it("hands a recovery to the next controller when shutdown closed the listener before backend setup", async () => { + const f = await seedRecoverableRun("LISTENER-CLOSED"); + const listener = fixture.current.server!; + const address = listener.address(); + if (!address || typeof address === "string") throw new Error("Expected a TCP runtime API listener"); + let listenerClosed: Promise | undefined; + // Graceful shutdown closes HTTP admission before it drains heartbeat runs. + // Backend setup then asks for the callback origin, as runnerd setup does. + const dying = vi.fn(() => ({ + descriptor: async () => { resolveLocalRuntimeApiUrl(); throw new Error("unreachable"); }, + openSession: async () => { resolveLocalRuntimeApiUrl(); throw new Error("unreachable"); }, + })); + const heartbeat = heartbeatService(f.db, { + nativeSessionBackendFactory: dying as never, + beforeNativeRuntimeSelection: async () => { + listener.closeIdleConnections(); + listenerClosed = new Promise(resolve => listener.close(() => resolve())); + }, + }); + let drained: unknown; + try { + try { + const { claims } = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId })]); + } finally { await heartbeat.drainActiveRunExecutions(); } + expect(dying).toHaveBeenCalled(); + drained = await heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + await heartbeat.drainActiveRunExecutions(); + } finally { + await listenerClosed; + await new Promise(resolve => listener.listen(address.port, "127.0.0.1", resolve)); + } + expect(drained).toMatchObject({ restartSuspendedRunIds: [f.runId] }); + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect({ status: run.status, errorCode: run.errorCode, finishedAt: run.finishedAt }).toEqual({ + status: "running", errorCode: null, finishedAt: null }); + expect(owner).toMatchObject({ phase: "observed", resultId: null, recoveryState: "awaiting_runner_reattach" }); + + await f.db.update(nativeRunFinalizations).set({ controllerPid: 2_000_000_001 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const resumed = vi.fn(() => { throw new Error("next boot reached provider dispatch"); }); + const next = heartbeatService(f.db, { nativeSessionBackendFactory: resumed }); + try { + const { claims } = await next.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId })]); + } finally { await next.drainActiveRunExecutions(); } + expect(resumed).toHaveBeenCalledOnce(); + }, 90_000); + it("still settles a recovery whose claim merely expired without a shutdown suspension", async () => { const f = await seedRecoverableRun("EXPIRED-NOT-SUSPENDED"); const dying = vi.fn(() => { throw new Error("An expired claim must not open a provider session"); }); diff --git a/server/src/__tests__/helpers/native-restart.ts b/server/src/__tests__/helpers/native-restart.ts index deccd592af1..8d642f4ca12 100644 --- a/server/src/__tests__/helpers/native-restart.ts +++ b/server/src/__tests__/helpers/native-restart.ts @@ -244,7 +244,7 @@ export function nativeRestartFixture() { }; } - return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor }; } }; + return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor, server }; } }; } export function remotePreparedWorkspace() { diff --git a/server/src/http/server.test.ts b/server/src/http/server.test.ts index b548858d23f..c4616ec6751 100644 --- a/server/src/http/server.test.ts +++ b/server/src/http/server.test.ts @@ -8,7 +8,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import express from "express"; import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js"; import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals } from "../realtime/runner-prp-ws.js"; -import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; +import { LocalRuntimeApiListenerClosedError, resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { createPaperclipHttpServer } from "./server.js"; import { privateHostnameGuard } from "../middleware/private-hostname-guard.js"; @@ -206,6 +206,30 @@ describe("Paperclip HTTP server runner transport", () => { } }); + it("distinguishes a listener closed for shutdown from one that never served", async () => { + expect(() => resolveLocalRuntimeApiUrl()).toThrow("local_runtime_api_listener_unavailable"); + expect(() => resolveLocalRuntimeApiUrl()).not.toThrow(LocalRuntimeApiListenerClosedError); + const server = createPaperclipHttpServer((_req, res) => res.end("closing"), { apiUrl: "http://ignored.example.test" }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const client = await fetch(resolveLocalRuntimeApiUrl(), { keepalive: true }); + // `close()` stops admission at once; its close event waits for open requests. + const closed = new Promise(resolve => server.close(() => resolve())); + expect(() => resolveLocalRuntimeApiUrl()).toThrow(LocalRuntimeApiListenerClosedError); + await client.text(); + server.closeAllConnections(); + await closed; + expect(() => resolveLocalRuntimeApiUrl()).toThrow(LocalRuntimeApiListenerClosedError); + expect(() => resolveLocalRuntimeApiUrl()).toThrow("local_runtime_api_listener_unavailable"); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + try { + expect(resolveLocalRuntimeApiUrl()).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); + } finally { + await new Promise(resolve => server.close(() => resolve())); + } + }); + it.each(["127.0.0.1", "::1", "::ffff:127.0.0.1"])("routes a guarded native read through the real %s listener", async host => { await guardedNativeRead(host); }); diff --git a/server/src/runtime-api.ts b/server/src/runtime-api.ts index 4c4f87868a6..b7d1a3d64f4 100644 --- a/server/src/runtime-api.ts +++ b/server/src/runtime-api.ts @@ -4,15 +4,37 @@ import { isIP } from "node:net"; const runtimeApiListeners = new Set(); const registeredRuntimeApiListeners = new WeakSet(); +let runtimeApiListenerClosed = false; + +/** + * The process had a runtime API listener and has since closed it, which only + * happens while it shuts down. Work that needs a callback origin must hand off + * to the next controller instead of failing as a provider or startup error. + */ +export class LocalRuntimeApiListenerClosedError extends Error { + constructor() { + // Keep the established failure code for classifiers and persisted rows. + super("local_runtime_api_listener_unavailable"); + this.name = "LocalRuntimeApiListenerClosedError"; + } +} /** Track listener ownership without treating configured or advertised URLs as authority. */ export function registerLocalRuntimeApiServer(server: Server): void { if (!registeredRuntimeApiListeners.has(server)) { registeredRuntimeApiListeners.add(server); - server.on("listening", () => runtimeApiListeners.add(server)); - server.on("close", () => runtimeApiListeners.delete(server)); + server.on("listening", () => { + runtimeApiListeners.add(server); + runtimeApiListenerClosed = false; + }); + server.on("close", () => { + if (runtimeApiListeners.delete(server)) runtimeApiListenerClosed = true; + }); + } + if (server.listening) { + runtimeApiListeners.add(server); + runtimeApiListenerClosed = false; } - if (server.listening) runtimeApiListeners.add(server); } function normalizeHost(value: string | null | undefined): string { @@ -94,6 +116,11 @@ export function chooseLocalRuntimeApiUrl(input: { export function resolveLocalRuntimeApiUrl(): string { const listeners = [...runtimeApiListeners].filter(server => server.listening); + // `server.close()` clears `listening` at once; its close event waits for + // admitted requests. Either state means this process stopped serving. + if (listeners.length === 0 && (runtimeApiListenerClosed || runtimeApiListeners.size > 0)) { + throw new LocalRuntimeApiListenerClosedError(); + } if (listeners.length !== 1) { throw new Error(listeners.length ? "local_runtime_api_listener_ambiguous" : "local_runtime_api_listener_unavailable"); } @@ -103,7 +130,10 @@ export function resolveLocalRuntimeApiUrl(): string { } export const runtimeApiInternals = { - resetForTests(): void { runtimeApiListeners.clear(); }, + resetForTests(): void { + runtimeApiListeners.clear(); + runtimeApiListenerClosed = false; + }, }; export function choosePrimaryRuntimeApiUrl(input: { diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index f12aad1871f..8185661619d 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -25,7 +25,7 @@ import { import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; const toolDefinitions = vi.hoisted(() => vi.fn()); -vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" })); +vi.mock("../../runtime-api.js", async importOriginal => ({ ...await importOriginal(), resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" })); vi.mock("./paperclip-runner-tool-authority.js", () => ({ PaperclipRunnerToolAuthority: class { readonly binding = {}; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 47c09f6bde5..0a22e680602 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -274,7 +274,7 @@ const state = vi.hoisted(() => ({ release: null as null | (() => void), })); -vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl })); +vi.mock("../../runtime-api.js", async importOriginal => ({ ...await importOriginal(), resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl })); const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise), duplicate: undefined as undefined | ((event: PrpEvent) => Promise) })); vi.mock("./paperclip-control-plane-port.js", async importOriginal => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6e1833712bb..2ac1d64ab47 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5,7 +5,7 @@ import { logger } from "../../middleware/logger.js"; import { canRestoreMigratedNativeSession } from "./native-workspace-migration.js"; import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES, nativeHarnessBackupManifestSha256, retiredSourceAuthorizesNativeBackupRestore } from "./native-retired-backup-source.js"; import { prepareCursorCloudRuntime } from "./cursor-cloud-runtime.js"; -import { resolveLocalRuntimeApiUrl } from "../../runtime-api.js"; +import { LocalRuntimeApiListenerClosedError, resolveLocalRuntimeApiUrl } from "../../runtime-api.js"; import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js"; import { readProviderFailure } from "@paperclipai/shared"; import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; @@ -10706,7 +10706,12 @@ async function executePaperclipNativeSessionWithinScope( } else { error = identityRedactor.redact(error); } - if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { + // A closed runtime API listener means this process is shutting down; it + // closes HTTP admission before it drains runs. Keep the claim for the + // drain to suspend, so the next controller adopts the run without spending + // a failure or a same-run retry. + if (nativeRunsDetachingForRestart.has(input.execution.binding.runId) + || error instanceof LocalRuntimeApiListenerClosedError) { await leaseRenewal.stop().catch(() => undefined); liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId);