mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
fix(native): hand a shutdown-interrupted startup to the next controller
Graceful shutdown closes the runtime API listener before it drains heartbeat runs. A native execution between provider dispatch and runnerd backend setup then failed with local_runtime_api_listener_unavailable, so the drain no longer saw it as running and never suspended it for the next boot. resolveLocalRuntimeApiUrl now raises LocalRuntimeApiListenerClosedError (same failure code) once this process stopped serving, and the executor treats it as a restart detach: the run stays running with its claim, the drain suspends it, and the next controller adopts it. A listener that never started is still a startup failure. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
65aee998a3
commit
a35ec8b845
8 files changed
+125
-10
No files matched your search
@@ -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:
|
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
|
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.
|
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
|
Remote Runner processes use their authenticated connection lease as the bound
|
||||||
on disconnected lifetime. The controller's shorter connection-recovery budget
|
on disconnected lifetime. The controller's shorter connection-recovery budget
|
||||||
limits its own attempt; it must not terminate the remote owner while another
|
limits its own attempt; it must not terminate the remote owner while another
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto";
|
|||||||
import { and, eq } from "drizzle-orm";
|
import { and, eq } from "drizzle-orm";
|
||||||
import { authUsers, budgetReservations, companyMemberships, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db";
|
import { authUsers, budgetReservations, companyMemberships, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db";
|
||||||
import { expect, it, vi } from "vitest";
|
import { expect, it, vi } from "vitest";
|
||||||
|
import { resolveLocalRuntimeApiUrl } from "../runtime-api.js";
|
||||||
import { heartbeatService } from "../services/heartbeat.js";
|
import { heartbeatService } from "../services/heartbeat.js";
|
||||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||||
import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js";
|
import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js";
|
||||||
@@ -74,6 +75,56 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut
|
|||||||
90_000,
|
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<void> | 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<void>(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<void>(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 () => {
|
it("still settles a recovery whose claim merely expired without a shutdown suspension", async () => {
|
||||||
const f = await seedRecoverableRun("EXPIRED-NOT-SUSPENDED");
|
const f = await seedRecoverableRun("EXPIRED-NOT-SUSPENDED");
|
||||||
const dying = vi.fn(() => { throw new Error("An expired claim must not open a provider session"); });
|
const dying = vi.fn(() => { throw new Error("An expired claim must not open a provider session"); });
|
||||||
|
|||||||
@@ -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() {
|
export function remotePreparedWorkspace() {
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
|
|||||||
import express from "express";
|
import express from "express";
|
||||||
import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js";
|
import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js";
|
||||||
import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals } from "../realtime/runner-prp-ws.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 { createPaperclipHttpServer } from "./server.js";
|
||||||
import { privateHostnameGuard } from "../middleware/private-hostname-guard.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<void>(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<void>(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 => {
|
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);
|
await guardedNativeRead(host);
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -4,15 +4,37 @@ import { isIP } from "node:net";
|
|||||||
|
|
||||||
const runtimeApiListeners = new Set<Server>();
|
const runtimeApiListeners = new Set<Server>();
|
||||||
const registeredRuntimeApiListeners = new WeakSet<Server>();
|
const registeredRuntimeApiListeners = new WeakSet<Server>();
|
||||||
|
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. */
|
/** Track listener ownership without treating configured or advertised URLs as authority. */
|
||||||
export function registerLocalRuntimeApiServer(server: Server): void {
|
export function registerLocalRuntimeApiServer(server: Server): void {
|
||||||
if (!registeredRuntimeApiListeners.has(server)) {
|
if (!registeredRuntimeApiListeners.has(server)) {
|
||||||
registeredRuntimeApiListeners.add(server);
|
registeredRuntimeApiListeners.add(server);
|
||||||
server.on("listening", () => runtimeApiListeners.add(server));
|
server.on("listening", () => {
|
||||||
server.on("close", () => runtimeApiListeners.delete(server));
|
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 {
|
function normalizeHost(value: string | null | undefined): string {
|
||||||
@@ -94,6 +116,11 @@ export function chooseLocalRuntimeApiUrl(input: {
|
|||||||
|
|
||||||
export function resolveLocalRuntimeApiUrl(): string {
|
export function resolveLocalRuntimeApiUrl(): string {
|
||||||
const listeners = [...runtimeApiListeners].filter(server => server.listening);
|
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) {
|
if (listeners.length !== 1) {
|
||||||
throw new Error(listeners.length ? "local_runtime_api_listener_ambiguous" : "local_runtime_api_listener_unavailable");
|
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 = {
|
export const runtimeApiInternals = {
|
||||||
resetForTests(): void { runtimeApiListeners.clear(); },
|
resetForTests(): void {
|
||||||
|
runtimeApiListeners.clear();
|
||||||
|
runtimeApiListenerClosed = false;
|
||||||
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
export function choosePrimaryRuntimeApiUrl(input: {
|
export function choosePrimaryRuntimeApiUrl(input: {
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ import {
|
|||||||
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js";
|
||||||
|
|
||||||
const toolDefinitions = vi.hoisted(() => vi.fn());
|
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<typeof import("../../runtime-api.js")>(), resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" }));
|
||||||
vi.mock("./paperclip-runner-tool-authority.js", () => ({
|
vi.mock("./paperclip-runner-tool-authority.js", () => ({
|
||||||
PaperclipRunnerToolAuthority: class {
|
PaperclipRunnerToolAuthority: class {
|
||||||
readonly binding = {};
|
readonly binding = {};
|
||||||
|
|||||||
@@ -274,7 +274,7 @@ const state = vi.hoisted(() => ({
|
|||||||
release: null as null | (() => void),
|
release: null as null | (() => void),
|
||||||
}));
|
}));
|
||||||
|
|
||||||
vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl }));
|
vi.mock("../../runtime-api.js", async importOriginal => ({ ...await importOriginal<typeof import("../../runtime-api.js")>(), resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl }));
|
||||||
|
|
||||||
const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise<void>), duplicate: undefined as undefined | ((event: PrpEvent) => Promise<void>) }));
|
const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise<void>), duplicate: undefined as undefined | ((event: PrpEvent) => Promise<void>) }));
|
||||||
vi.mock("./paperclip-control-plane-port.js", async importOriginal => {
|
vi.mock("./paperclip-control-plane-port.js", async importOriginal => {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import { logger } from "../../middleware/logger.js";
|
|||||||
import { canRestoreMigratedNativeSession } from "./native-workspace-migration.js";
|
import { canRestoreMigratedNativeSession } from "./native-workspace-migration.js";
|
||||||
import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES, nativeHarnessBackupManifestSha256, retiredSourceAuthorizesNativeBackupRestore } from "./native-retired-backup-source.js";
|
import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES, nativeHarnessBackupManifestSha256, retiredSourceAuthorizesNativeBackupRestore } from "./native-retired-backup-source.js";
|
||||||
import { prepareCursorCloudRuntime } from "./cursor-cloud-runtime.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 { providerFailureFromError } from "../../vendor/paperclip-runner/index.js";
|
||||||
import { readProviderFailure } from "@paperclipai/shared";
|
import { readProviderFailure } from "@paperclipai/shared";
|
||||||
import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js";
|
import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js";
|
||||||
@@ -10706,7 +10706,12 @@ async function executePaperclipNativeSessionWithinScope(
|
|||||||
} else {
|
} else {
|
||||||
error = identityRedactor.redact(error);
|
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);
|
await leaseRenewal.stop().catch(() => undefined);
|
||||||
liveQuestions.close();
|
liveQuestions.close();
|
||||||
activeNativeSessions.delete(input.execution.binding.runId);
|
activeNativeSessions.delete(input.execution.binding.runId);
|
||||||
|
|||||||
Reference in new issue
Block a user