From ee6c101854738bc2a36ebffc45776238fca18178 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 6 Nov 2026 13:55:43 -0800 Subject: [PATCH] Reject failover backups behind the accepted event cursor (#19613) A verified failover backup can still trail events already accepted by the controller. Compare the backup produced cursor with the registered controller state, or the supported local journal before registration, and reject stale files before remote sync. Keep the permanent replay-conflict classification ahead of generic Runner errors. Validation: 1,192 native session and execution-driver tests and 10 workflow tests pass locally, together with the maintained low-memory Runner test project typecheck and fresh server declarations. All applicable checks on cd8c9cd07c192832fa180a845577d36fb0bc41c8 are green (56 success, three skipped); exact-head Apex review is 5/5 with no unresolved findings. One e2e shard was retried after GitHub reported lost self-hosted runner communication, and the same-head rerun passed. No acknowledgement was lowered, provider replay authorized, or recovery files discarded. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: Paperclip Co-Authored-By: Paperclip --- doc/recoverable-disk-cleanup.md | 12 ++ .../native-session-executor.test.ts | 118 ++++++++++++++++++ .../native-runtime/native-session-executor.ts | 38 +++++- 3 files changed, 165 insertions(+), 3 deletions(-) diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md index d268de28f1..0a26bfc799 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -63,6 +63,18 @@ directories, and data Codex downloads again on startup: remote plugin bundles un install staging. After a restore, remote plugins can be unavailable to a turn until Codex finishes downloading them. +A turn that ends without a proven Runner suspension writes no new copy, but the +controller can still acknowledge the events Runner produced after the last copy. +Before it restores a copy into a replacement sandbox, the server compares the +copy's produced event cursor with the controller's acknowledged cursor. If the +copy is behind, the run fails with `native_event_replay_conflict` and automatic +recovery stops. No files are copied to the sandbox. Resuming that copy would make +Runner reject the controller's acknowledgement, and it could repeat work the +controller already accepted. +When a controller is registered, the comparison uses its current live cursor. +The local marker written for external checkpoint storage does not replace that +authority. + An hourly background sweep expires the copy once the session can only resume by reopening closed work. It checks at most 200 sessions and expires at most 50 per batch, starting two minutes after readiness. A copy expires when all of these hold: 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 7f91a73ea8..5ce7ab0c1d 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -18216,6 +18216,124 @@ describe("runnerd provider runtime wiring", () => { } }); + it.each([ + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v1", registered: false }, + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: false }, + { backup: "current", nextSourceSeq: 1518, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: false }, + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: true }, + { backup: "current", nextSourceSeq: 1518, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: true }, + ])("checks the controller cursor before restoring a $backup failover backup ($controlSchema, registered=$registered)", async ({ backup, nextSourceSeq, controlSchema, registered }) => { + const remoteCwd = join(isolatedStateDirectory, "remote"); + await mkdir(join(remoteCwd, ".paperclip-runtime", "paperclip-runner"), { recursive: true }); + const syncIn = vi.fn(async () => undefined); + const remoteExecute = vi.fn(async (command: { command: string; args?: string[] }) => { + if (command.args?.[0] === "--build-metadata") return { + exitCode: 0, timedOut: false, stdout: JSON.stringify({ + schema: "paperclip-runner/runnerd-build-metadata/v1", binaryName: "paperclip-runnerd", + packageName: "@paperclipai/paperclip-runner", binaryContractVersion: 2, + durableSessionCapabilities: ["unlimited_runtime", "connection_lease_renewal"], + prpTransportModes: ["listen_ws"], + }), stderr: "", + }; + if (command.args?.[0] === "--version") return { + exitCode: 0, timedOut: false, stdout: "codex-cli 0.160.0", stderr: "", + }; + if (command.args?.[1]?.includes("base64")) return { + exitCode: 1, timedOut: false, stdout: "", stderr: "", + }; + return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; + }); + const checkpointSink = vi.fn(async () => undefined); + const checkpointRegistration = registered ? vi.spyOn(runnerCheckpoint, "nativeRunnerCheckpointRegistration") + .mockReturnValue(async authority => { + authority.setCheckpointPersistence(checkpointSink); + await authority.flushCheckpoint(); + }) : undefined; + let controller: DurablePrpControlPlane | undefined; + let registration: Awaited>> | undefined; + onTestFinished(async () => { await registration?.release(); await controller?.stop(); checkpointRegistration?.mockRestore(); }); + const controllerArtifact = join(isolatedStateDirectory, "failover-cursor-controller"); + if (registered) await writeFile(controllerArtifact, "fixture bootstrap artifact"); + const runnerInstanceId = "runner-failover-cursor"; + await createRunnerdBackend({ + db: leaseDb(execution), execution, runnerInstanceId, + ...(registered ? { controllerPersistence: { bootId: "failover-cursor-boot" }, + runnerRemoteBinaryPath: controllerArtifact, runnerRemoteCodexPath: controllerArtifact } : {}), + runnerIngressAuthorized: true, + runnerExecutionTarget: { + kind: "remote", transport: "sandbox", remoteCwd, environmentId: "environment", + leaseId: "lease-replacement", providerKey: "daytona", + effectiveCapabilities: { runnerWebSocketIngress: true }, + getRunnerIngressEndpoint: async () => ({ kind: "authenticated_websocket", websocketUrl: "wss://sandbox.invalid/runner", + secretHeaders: [{ name: "Authorization", value: "fixture-ingress" }], generation: "1", refresh: vi.fn(), close: vi.fn(async () => undefined) }), + sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "sandbox-replacement", reason: "not_found" }, + runner: { execute: remoteExecute, syncIn }, + } as never, + }); + state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); + const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { + prepareExternalRunnerState: () => Promise; + controlPlaneRegistration: NonNullable; + }; + const root = options.stateDirectory!; + await mkdir(join(root, "control-plane"), { recursive: true }); + if (!registered) await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ + schema: controlSchema, + identity: options.prpIdentity, ackedSourceSeq: 1517, connectionCount: 1, committedEvents: [], commands: [], + })); + const current = join(root, "failover-backups", "current"); + await mkdir(join(current, "runner"), { recursive: true }); + await mkdir(join(current, "codex-home", "sessions"), { recursive: true }); + await writeFile(join(current, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", ...options.prpIdentity, runnerInstanceId, + lifecycle: "suspended", nextSourceSeq, ackedSourceSeq: nextSourceSeq - 1, outbox: [], + })); + await writeFile(join(current, "codex-home", "sessions", "thread.jsonl"), "thread-state"); + await writeFile(join(current, "manifest.json"), JSON.stringify(buildNativeHarnessBackupManifest({ + backupRoot: current, execution, runnerInstanceId, + providerSessionIdentity: { providerSessionId: "thread-1", providerBackendSessionId: "session-1", providerSessionIdentity: null }, + sourceProviderLeaseId: "sandbox-previous", + }))); + + if (registered) { + controller = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), + identity: options.prpIdentity! as ConstructorParameters[0]["identity"], + expectedRunnerVersion: "failover-cursor-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + controller.store.state.ackedSourceSeq = 1517; + await controller.flushCheckpoint(); + registration = await options.controlPlaneRegistration(controller); + expect(checkpointRegistration).toHaveBeenCalled(); + expect(checkpointSink).toHaveBeenCalled(); + const marker = JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")); + expect(marker.schema).toBe("paperclip.runner.external-control-plane-state.v1"); + expect(marker.ackedSourceSeq).toBeUndefined(); + syncIn.mockClear(); + } + const prepared = options.prepareExternalRunnerState(); + if (backup === "stale") { + await expect(prepared).rejects.toThrow( + "native_event_replay_conflict: failover backup is behind the controller cursor (backup produced 119, controller acknowledged 1517)", + ); + expect(syncIn).not.toHaveBeenCalled(); + const error = await prepared.catch((caught: unknown) => caught); + expect(nativeSessionFailureSourceCode(new Error(`native_session_recovery_failed: ${String(error)}`))) + .toBe("native_event_replay_conflict"); + expect(nativeSessionFailureDisposition(1, new Date(), "native_event_replay_conflict").phase).toBe("terminal_failure"); + } else { + await expect(prepared).rejects.toThrow("runner_harness_state_mismatch: restored_provider_identity_changed"); + expect(syncIn).toHaveBeenCalled(); + } + }); + + it("keeps an integrity conflict permanent when a runner exit wraps it", () => { + expect(nativeSessionFailureSourceCode(new Error( + "native_session_recovery_failed: runnerd exited unexpectedly: native_event_replay_conflict: failover backup is behind the controller cursor", + ))).toBe("native_event_replay_conflict"); + expect(nativeSessionFailureSourceCode(new Error( + "native_session_recovery_failed: Error: runnerd exited while waiting for runner.drain", + ))).toBe("native_runner_process_exited"); + }); + it.each([ ...["current", "preinstalled-exact", "preinstalled-mismatch", "preinstalled-error", "preinstalled-timeout", "stale", "missing", "retained", "retained-mismatch", "retained-error", "retained-timeout", "retained-explicit"] .map((image) => ({ image, version: "0.160.0" })), diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 3d47f6b6d2..5f8201f8ab 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -7057,6 +7057,35 @@ export function verifyNativeHarnessBackup(input: { return null; } +export function nativeHarnessBackupCursorConflict(input: { + backupRoot: string; + stateRoot: string; + controllerState?: { schema: string; identity: { runnerInstanceId: string }; ackedSourceSeq: number }; +}): string | null { + const read = (path: string, maxBytes: number) => { + try { + return existsSync(path) + ? record(JSON.parse(readBoundedNativeFile(path, maxBytes, "runner_durable_identity_too_large").toString("utf8"))) + : null; + } catch { + return null; + } + }; + const control = input.controllerState ?? read(resolve(input.stateRoot, "control-plane", "control-plane-state.json"), + NATIVE_CONTROL_PLANE_STATE_MAX_BYTES); + if (!control || !RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema))) return null; + const runner = read(resolve(input.backupRoot, "runner", "runner-state.json"), NATIVE_RUNNER_STATE_MAX_BYTES); + if (!runner) return null; + const acked = control.ackedSourceSeq; + const next = runner.nextSourceSeq; + if (!Number.isSafeInteger(acked) || !Number.isSafeInteger(next) || (next as number) < 1 || + record(control.identity).runnerInstanceId !== runner.runnerInstanceId) return null; + const produced = (next as number) - 1; + return (acked as number) > produced + ? `native_event_replay_conflict: failover backup is behind the controller cursor (backup produced ${produced}, controller acknowledged ${acked as number})` + : null; +} + function nativeSessionCheckpointDirectory(): string { const directory = resolve( resolvePaperclipInstanceRoot(), @@ -7464,6 +7493,9 @@ export function nativeSessionFailureSourceCode( ) { return "provider_transport_failed"; } + if (/native_event_replay_conflict/i.test(message)) { + return "native_event_replay_conflict"; + } if ( /native_runner_process_exited|runnerd exited|runner process failed/i.test( message, @@ -7474,9 +7506,6 @@ export function nativeSessionFailureSourceCode( if (/planning_mode_unsupported/i.test(message)) { return "planning_mode_unsupported"; } - if (/native_event_replay_conflict/i.test(message)) { - return "native_event_replay_conflict"; - } if (message.toLowerCase().includes(NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ.toLowerCase())) { return NATIVE_CURRENT_WAKE_COMMENTS_CHANGED_AFTER_READ; } @@ -14433,6 +14462,9 @@ async function createRunnerdBackendWithinSessionClaim( runnerInstanceId: input.runnerInstanceId, }); if (!backup) throw new Error("runner_harness_backup_unavailable"); + const cursorConflict = nativeHarnessBackupCursorConflict({ backupRoot: backup.root, stateRoot: root, + controllerState: runnerControlPlane?.store.state }); + if (cursorConflict) throw new Error(cursorConflict); await measureNativeRunnerSpan( input.trace, "harness_state.failover_restore",