From aaea05e158e71845493275b98c37a6e51fb25d5d Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 4 Nov 2026 11:15:58 -0600 Subject: [PATCH] fix(native): preserve backups when completed warm sessions retire (#18986) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native Runner keeps provider history when a warm session ends. > - Object storage must preserve that history before the sandbox is released. > - A completed turn releases its live controller lease. > - The backup owner lookup rejects that released lease and stops the next task preparation. > - This pull request uses the existing completed-run checkpoint fence for backup storage. > - The next turn can prepare after the previous provider state is saved. ## Linked Issues or Issue Description Refs: #18932, #18011, #18984. **What happened?** A warm Codex turn completed and kept its provider session alive. The next turn stopped during preparation with `native_harness_storage_owner_missing`. The backup lookup required a live lease after normal completion had released it. **Expected behavior** Save the completed session under its retained controller generation and committed result. Keep the existing owner and successor checks. Do not start another provider turn during the save. **Steps to reproduce** 1. Run a warm native session with object-backed provider backups. 2. Commit the first turn and release its controller lease. 3. Start the next task preparation and retire the previous session. 4. Observe the backup owner error before provider work starts. **Paperclip version or commit** The regression is present at `86814c6550f52e99ede150a883bdc773cbb540e3`. **Deployment mode** A hosted server with a remote native Runner and S3 storage. ## What Changed - Accept a released backup owner only when its exact controller record has a committed result. - Reuse the checkpoint owner policy for same-controller preparation and native successors. - Recheck the controller generation, committed result, and released state after upload. - Preserve live-owner rejection and the rule that a late upload cannot replace a newer conversation backup. - Add real PostgreSQL storage and restore regressions. Document the completed-run backup rule. ## Verification - On 2026-11-04, two new regressions failed on the unchanged source with `native_harness_storage_owner_missing`. These covered an idle completed session and a same-controller preparing successor. - After the fix, 54 selected tests passed: 30 durable backup tests and 24 existing early-retirement checkpoint tests. After the fixture-only upstream sync, all 58 selected storage, checkpoint, process, and affected fixture cases passed again. - The storage cases use real PostgreSQL, encrypted archives, object transfer, and restoration. They cover invalid completion evidence, foreign or expired successors, active predecessors, changed generations, changes during upload, and newer backup publication. - Command: `pnpm exec vitest run server/src/services/native-runtime/native-harness-backup-storage.integration.test.ts server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts -t 'released|backup|early retirement|durably retires an accepted warm Runner'`. - An independent source review found no outstanding code issue. - Full repository typecheck passed with `node scripts/typecheck-low-memory.mjs` before the upstream fixture-only sync. Production source is unchanged across that sync. - An assembled process probe passed with the Runner binary built from this head and a fake Codex protocol peer. It started real processes, committed a result, released the source lease, claimed the next controller, and completed the source stop, drain, and suspend commands. It recorded one provider turn start and kept the task bound to its successor. - Command: `pnpm exec vitest run server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts -t 'durably retires an accepted warm Runner'`. - CI exposed the old allocation-readiness test setup in three cases. The branch now includes the merged #18984 fixture correction. The same three cases passed locally after that sync. - An earlier CI test shard was cancelled after its runner received a shutdown signal. Its log was retained before one rerun. Another old-head shard reported failure with no retrievable job log or annotations; its cause remains unknown. Current-head CI must pass independently. - Local `pnpm build` passed at this head. All 1,904 chat cases passed across the supported local shards; the earlier shards used byte-identical source before the fixture-only sync. The interrupted default command is retained and is not reported as a pass. - The local bulk server run had separate enrollment acknowledgment, environment-run status, monitor-settle, idle-history teardown, and repository-rehome failures. In isolation, idle-history passed all 75 tests and the unchanged repository cases passed with their original limits (4.9s large files, 6.4s refs). The remaining four local failures reproduced on unchanged master `04a24f4e06` under the same environment and limits. Their causes remain unproven. Logs and the baseline comparison were retained; the local full suite is not claimed as passed. - Current-head CI first hit the unchanged repository-rehome large-file case's 15s timeout. The same isolated case passed in 4.9s. After source/call-path comparison, one bounded failed-shard retry passed all 115 executed suites; no assertion or timeout was changed. The original timeout remains recorded. - Local verification uses macOS 26.5.2 arm64, Node 26.4.0, pnpm 9.15.4, PostgreSQL 18.1, and Cargo 1.97.1. The current tested head is `622f69f76ff68c71eacdcbcbc6ab4d4c6eb83bf9`. - The full unfiltered [PR workflow](https://github.com/paperclipai/paperclip/actions/runs/41868587974) passed at `622f69f76ff68c71eacdcbcbc6ab4d4c6eb83bf9`: all 12 general server, 3 chat, 9 serialized, workspace, native Runner, build/typecheck, E2E, and final verify lanes. The file selector was skipped, so this was the complete fallback matrix. - Greptile reports 5/5 on this exact head with no open review threads. Disposable local test processes were cleaned up and the source tree is clean. - The real hosted Codex and S3 workflow has not been rerun because the local verification environment has no remote provider credentials. The original task remains unchanged. ## Risks The main risk is a stale controller publishing provider history after a new owner starts. Both storage transactions keep the company, run, boot, and generation fence. A released owner must keep its committed result. Existing successor checks and the newer-backup publication guard remain in force. The live checkpoint writer still requires a non-null lease. No migration or API change is required. ## Model Used OpenAI Codex, GPT-6, with reasoning, repository tools, and code execution. The exact serving model ID and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass (focused checks; separate baseline broad-suite failures are disclosed above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/execution-semantics.md | 8 ++ ...harness-backup-storage.integration.test.ts | 113 +++++++++++++++++- .../native-harness-backup-storage.ts | 20 ++-- .../native-runner-checkpoint.ts | 14 ++- 4 files changed, 141 insertions(+), 14 deletions(-) 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)