diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 70c07a31455..74422770352 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2665,6 +2665,14 @@ turn finishes during shutdown, its release checkpoints the session before returning instead of leaving a new idle owner behind. If checkpointing fails, the retained state continues to block unverified reuse. +Object-backed provider backups use the same completed-run fence during warm +retirement. A committed result can preserve its provider files after its live +controller lease is released. The boot ID and controller generation must still +match. A successor must pass the existing same-controller preparation or native +ownership checks. The backup transaction checks this authority again after +upload, and a late predecessor cannot replace a newer conversation backup. +This grants no new lease or permission to start provider work. + Local durable control-plane state is bounded at 256 MiB in the server, runnerd recovery, and durable control-plane readers. These paths synchronously read and parse the full JSON file, so memory use and parse time grow with file diff --git a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts index c7c2b5a7f0d..4bcda4b168d 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts @@ -2,7 +2,7 @@ import { createHash, randomUUID } from "node:crypto"; import { chmod, mkdir, readFile, rename, stat, symlink, readlink, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { and, eq } from "drizzle-orm"; -import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations } from "@paperclipai/db"; +import { agentTaskSessions, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; import { afterEach, expect, it, vi } from "vitest"; import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; @@ -14,6 +14,7 @@ import { buildNativeHarnessBackupManifest } from "./native-session-executor.js"; import { createNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp, verifyNativeHarnessBackupStampAsync } from "./native-harness-backup-stamp.js"; import { hasLocalNativeHarnessObligations, migrateNativeHarnessBackups } from "./native-harness-backup-migration.js"; import { nativeHarnessBackupExpirationService } from "./native-harness-backup-expiration.js"; +import { legacyControllerBootId } from "../legacy-controller-lease.js"; describeEmbeddedPostgres("durable native provider backups", () => { const environment = nativeRestartFixture(); @@ -70,6 +71,116 @@ describeEmbeddedPostgres("durable native provider backups", () => { expect(rows.filter(row => row.status === "pending")).toHaveLength(1); expect(await readFile(join(f.root, "manifest.json"), "utf8")).toContain("native-harness-backup"); }); + async function commitWarmRun(f: Awaited>) { + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: run.completionContractId!, serverFingerprint: "completed-warm-turn", schemaStatus: "accepted", + resultJson: { summary: "Retained provider history" }, canonicalSha256: "completed-warm-turn" }).returning(); + await f.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, + leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + } + it.each(["idle", "preparing", "native"])("preserves a released committed warm session with a %s successor", async state => { + const f = await fixture(); + const result = await commitWarmRun(f); + let nextRunId: string | null = null; + if (state !== "idle") { + nextRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: environment.current.agentId, + issueId: f.issueId, status: "running", runtimeMode: state === "native" ? "native" : "legacy", executionStage: "preparing", + ...(state === "native" ? { nativeIssueId: f.issueId } : {}), + controllerBootId: legacyControllerBootId, controllerLeaseExpiresAt: new Date(Date.now() + 120_000) }); + if (state === "native") await f.db.insert(nativeRunFinalizations).values({ runId: nextRunId, + companyId: f.companyId, issueId: f.issueId, phase: "observed", controllerBootId: environment.current.successor.bootId, + controllerGeneration: 1, leaseOwner: "preparing-warm-successor", leaseExpiresAt: new Date(Date.now() + 120_000) }); + } + await f.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, f.issueId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + await persistNativeHarnessBackup({ ...f, owner }); + const replacement = join(environment.current.runtimeRoot, randomUUID(), "restored-warm-session"); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex", "sessions", "history.json"), "utf8"))).toEqual(f.identity); + const [coordinator] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator).toMatchObject({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null }); + const [task] = await f.db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task.executionRunId).toBe(nextRunId); + }); + it.each(["unaccepted", "uncommitted", "wrong_boot"])("rejects a released warm owner with %s completion evidence", async state => { + const f = await fixture(); await commitWarmRun(f); + if (state === "unaccepted") await f.db.update(nativeRunFinalizations).set({ resultId: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (state === "uncommitted") await f.db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await expect(nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, runId: f.runId, + bootId: state === "wrong_boot" ? randomUUID() : environment.current.successor.bootId })).rejects.toThrow("native_harness_storage_owner_missing"); + }); + it("rejects an active predecessor after the task moves to a successor", async () => { + const f = await fixture(); + const nextRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: environment.current.agentId, + nativeIssueId: f.issueId, issueId: f.issueId, status: "running", runtimeMode: "native" }); + await f.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, f.issueId)); + await expect(persistNativeHarnessBackup(f)).rejects.toThrow("native_harness_storage_owner_superseded"); + expect(await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest))).toEqual([]); + }); + it.each(["foreign_preparation", "expired_preparation", "dispatching_successor", "foreign_native"])("keeps the selected backup when a released owner has a %s", async state => { + const f = await fixture(); await persistNativeHarnessBackup(f); await commitWarmRun(f); + const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + const nextRunId = randomUUID(); + await f.db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: environment.current.agentId, + issueId: f.issueId, status: "running", runtimeMode: state === "foreign_native" ? "native" : "legacy", + ...(state === "foreign_native" ? { nativeIssueId: f.issueId } : {}), + executionStage: state === "dispatching_successor" ? "dispatching" : "preparing", + controllerBootId: state === "foreign_preparation" ? randomUUID() : legacyControllerBootId, + controllerLeaseExpiresAt: state === "expired_preparation" ? new Date(0) : new Date(Date.now() + 120_000) }); + if (state === "foreign_native") await f.db.insert(nativeRunFinalizations).values({ runId: nextRunId, + companyId: f.companyId, issueId: f.issueId, phase: "observed", controllerBootId: randomUUID(), + controllerGeneration: 1, leaseOwner: "foreign-warm-successor", leaseExpiresAt: new Date(Date.now() + 120_000) }); + await f.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, f.issueId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + await expect(persistNativeHarnessBackup({ ...f, owner })).rejects.toThrow("native_runner_checkpoint_owner_lost"); + expect((await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest), + eq(nativeHarnessBackups.status, "current")))).map(row => row.id)).toEqual([before.id]); + }); + it.each(["generation", "result", "phase"])("rechecks a released owner's %s after object upload", async state => { + const f = await fixture(); await persistNativeHarnessBackup(f); await commitWarmRun(f); + await f.db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId)); + const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + const put = f.storage.putFile.bind(f.storage); + f.storage.putFile = async input => { + const result = await put(input); + await f.db.update(nativeRunFinalizations).set(state === "generation" ? { controllerGeneration: 2 } + : state === "result" ? { resultId: null } : { phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + }; + await expect(persistNativeHarnessBackup({ ...f, owner })).rejects.toThrow(state === "generation" + ? "native_runner_checkpoint_owner_lost" : "native_harness_storage_owner_missing"); + expect((await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest), + eq(nativeHarnessBackups.status, "current")))).map(row => row.id)).toEqual([before.id]); + }); + it("cannot replace a newer conversation backup with a released predecessor's late upload", async () => { + const f = await fixture(); await commitWarmRun(f); + await f.db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId)); + const owner = await nativeHarnessBackupOwner(f.db, { companyId: f.companyId, issueId: f.issueId, + runId: f.runId, bootId: environment.current.successor.bootId }); + const next = await fixture(); + const put = f.storage.putFile.bind(f.storage); + f.storage.putFile = async input => { + const result = await put(input); + await persistNativeHarnessBackup({ ...next, scopeDigest: f.scopeDigest }); + return result; + }; + await expect(persistNativeHarnessBackup({ ...f, owner })).rejects.toThrow("native_harness_storage_owner_superseded"); + const [current] = await f.db.select().from(nativeHarnessBackups).where(and(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest), + eq(nativeHarnessBackups.status, "current"))); + expect(current.runId).toBe(next.runId); + const replacement = join(environment.current.runtimeRoot, randomUUID(), "latest-conversation"); ownedHomes.push(replacement); + expect(await restoreDurableNativeHarnessBackup({ ...f, backupRoot: replacement })).toBe(true); + expect(JSON.parse(await readFile(join(replacement, "current", "codex", "sessions", "history.json"), "utf8"))).toEqual(next.identity); + }); it("fences an old controller whose upload finishes after ownership changes", async () => { const f = await fixture(); await persistNativeHarnessBackup(f); const [before] = await f.db.select().from(nativeHarnessBackups).where(eq(nativeHarnessBackups.scopeDigest, f.scopeDigest)); diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index 5bf06d58037..47ab7e74dc9 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -10,7 +10,7 @@ import { and, eq, inArray, sql } from "drizzle-orm"; import { heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import type { StorageService } from "../../storage/types.js"; -import { assertNativeRunnerCheckpointOwner, type NativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js"; +import { lockNativeRunnerCheckpointOwner, validateNativeRunnerCheckpointOwner, type NativeRunnerCheckpointFenceOwner } from "./native-runner-checkpoint.js"; import { loadConfig } from "../../config.js"; import { getStorageService } from "../../storage/index.js"; import { readNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js"; @@ -85,14 +85,15 @@ export async function nativeHarnessBackupOwner(db: Db, input: { companyId: strin const [row] = await db.select().from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId), eq(nativeRunFinalizations.controllerBootId, input.bootId))); - if (!row?.leaseOwner) fail("owner_missing"); - return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner!, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointOwner; + if (!row || !row.leaseOwner && (row.phase !== "committed" || !row.resultId)) fail("owner_missing"); + return { ...input, authorityRunId: input.runId, leaseOwner: row!.leaseOwner, generation: row!.controllerGeneration } satisfies NativeRunnerCheckpointFenceOwner; } -async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointOwner | OfflineOwner) { +async function fence(tx: Parameters[0]>[0], owner: NativeRunnerCheckpointFenceOwner | OfflineOwner) { + const checkpointOwner = "kind" in owner ? null : await lockNativeRunnerCheckpointOwner(tx, owner); const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) .where(and(eq(issues.companyId, owner.companyId), eq(issues.id, owner.issueId))).for("share"); const authorityRunId = "kind" in owner ? owner.runId : owner.authorityRunId; - if (!task || !("kind" in owner) && task.executionRunId && task.executionRunId !== authorityRunId) fail("owner_superseded"); + if (!task) fail("owner_superseded"); if ("kind" in owner) { if (!owner.writersStopped) fail("offline_writers_required"); const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1); @@ -111,11 +112,16 @@ async function fence(tx: Parameters[0]>[0], owner: } return; } - await assertNativeRunnerCheckpointOwner(tx, owner, false); + const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`); + const owned = { ...checkpointOwner!, now: clock.now }; + const committed = owned.phase === "committed" && owned.resultId !== null && owned.leaseOwner === null; + if (owner.leaseOwner === null && !committed) fail("owner_missing"); + if (task.executionRunId && task.executionRunId !== authorityRunId && !committed) fail("owner_superseded"); + await validateNativeRunnerCheckpointOwner(tx, owner, owned, false); } export async function persistNativeHarnessBackup(input: { - db: Db; storage: StorageService; owner: NativeRunnerCheckpointOwner | OfflineOwner; + db: Db; storage: StorageService; owner: NativeRunnerCheckpointFenceOwner | OfflineOwner; scopeDigest: string; root: string; signal?: AbortSignal; }): Promise { if (!/^[a-f0-9]{64}$/.test(input.scopeDigest)) fail("scope_invalid"); diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 4920ede5437..8c3a4a9c3cd 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -9,6 +9,7 @@ type Binding = { companyId: string; issueId: string; runId: string }; export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024; export const NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES = 8 * 1024 * 1024; export type NativeRunnerCheckpointOwner = Binding & { bootId: string; leaseOwner: string; generation: number; authorityRunId: string }; +export type NativeRunnerCheckpointFenceOwner = Omit & { leaseOwner: string | null }; function encryptParts(binding: Binding, records: ReadonlyMap) { if (!records.size) return []; const cipher = createLocalEncryptedCipher("write"); @@ -49,11 +50,12 @@ type CheckpointTransaction = Parameters[0]>[0]; /** A dependent run-lock query can consume this CTE to preserve coordinator-first * lock order without adding a second database round trip. */ -export function nativeRunnerCheckpointOwnerLock(tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner) { +export function nativeRunnerCheckpointOwnerLock(tx: CheckpointTransaction, owner: NativeRunnerCheckpointFenceOwner) { return tx.$with("checkpoint_owner").as(tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, resultId: nativeRunFinalizations.resultId, + phase: nativeRunFinalizations.phase, }).from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, owner.companyId), eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.runId, owner.authorityRunId), eq(nativeRunFinalizations.controllerBootId, owner.bootId), @@ -63,24 +65,24 @@ export function nativeRunnerCheckpointOwnerLock(tx: CheckpointTransaction, owner /** Locks the coordinator before callers lock a run. A caller that subsequently * waits for another lock must recheck the fence before writing. */ export async function lockNativeRunnerCheckpointOwner( - tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner, + tx: CheckpointTransaction, owner: NativeRunnerCheckpointFenceOwner, ) { // A locking CTE acquires ownership before the outer projection samples // wall time. Sampling in the locking SELECT can precede its lock wait. // Keep both steps in one statement without another database round trip. const locked = nativeRunnerCheckpointOwnerLock(tx, owner); const [owned] = await tx.with(locked).select({ leaseOwner: locked.leaseOwner, - leaseExpiresAt: locked.leaseExpiresAt, resultId: locked.resultId, now: sql`clock_timestamp()` }).from(locked); + leaseExpiresAt: locked.leaseExpiresAt, resultId: locked.resultId, phase: locked.phase, now: sql`clock_timestamp()` }).from(locked); if (!owned) throw new Error("native_runner_checkpoint_owner_lost"); return owned; } export async function validateNativeRunnerCheckpointOwner( - tx: CheckpointTransaction, owner: NativeRunnerCheckpointOwner, - owned: Awaited>, newWork: boolean, + tx: CheckpointTransaction, owner: NativeRunnerCheckpointFenceOwner, + owned: Omit>, "phase">, newWork: boolean, ): Promise { const current = owned; - const activeOwner = current.leaseOwner === owner.leaseOwner + const activeOwner = owner.leaseOwner !== null && current.leaseOwner === owner.leaseOwner && current.leaseExpiresAt !== null && current.leaseExpiresAt > new Date(owned.now); if (!activeOwner) { const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues)