diff --git a/doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md b/doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md new file mode 100644 index 00000000000..5fa083ef278 --- /dev/null +++ b/doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md @@ -0,0 +1,82 @@ +# Explicit claim tokens for native execution leases + +Status: proposed follow-up (step 2). Step 1 shipped with the host-sleep lease fix. + +## Problem + +`native_run_finalizations.lease_expires_at` means two things today. It means +"nobody renewed this lease", and many actors also read it as "the owner is gone". +A host sleep makes the first true without the second: the wall clock and the +database clock jump past the 20-minute executor lease, while the server's +monotonic timers barely move. When the next renewal tick ran, it found its own +lease expired and interrupted a healthy turn (`native_session_lease_lost`, and the +user had to click Retry). The restart-chaos fault TM2 reproduces this. + +## Step 1 (shipped) + +The owner can take back its own lapsed lease in one conditional UPDATE. The UPDATE +runs under the coordinator row lock, followed by a share lock on the run. It +succeeds only when all of these hold: + +- the owner token, attempt, controller generation, and controller boot, pid and + process start are unchanged; +- the phase is `observed` and no result was accepted; +- `recovery_state`, `recovery_request_id`, `failure_code` and `next_attempt_at` + are null; +- `controller_host_hash` is not the board's attested-stopped stamp; +- the run is still `running`, without the ownership-unverified error code and + without a `nativeStartupCleanup` marker; +- the run is local. + +After a result is accepted, the owner may only relinquish a lapsed lease, and +never renews it. The rule lives in `nativeExecutionLeaseHeldCondition` and +`lockNativeExecutionLeaseForReacquire` (`native-execution-lease.ts`). Renewal, +failure settlement and the final release use it. + +This works because every consumer that acts on a lapsed executor lease of a +running, pre-result run already writes one of those fields under the coordinator +lock first. Step 1 depends on that convention, so step 2 replaces it with an +explicit token. + +## Gaps step 1 leaves open + +1. **Workspace finalization** (`native-workspace-finalization-work.ts` → + `attemptNativeWorkspaceFinalization`). It copies back on a lapsed post-result + lease. It is fenced only by the copyback advisory lock and the + `nativeWorkspaceFinalizationOwner` CAS, never by a coordinator write. For this + reason a post-result lease is never renewed. +2. **Another server's orphan sweep** (`heartbeat/recovery.ts` + `markNativeOwnershipUnverified`). It writes the run's error code without the + coordinator lock. If the marker lands first, the owner is fenced. If the sweep + reads the row before the owner takes the lease back and writes afterwards, + a live run carries an ownership-unverified error. This needs two server + processes on one database, and the timing that strands a run is as rare as before. +3. **Shared-workspace holder detection** (`remote-workspace-writer-hold.ts` + `runOwnerFenceHeldSql`). This is a read. A run silent past the stale threshold + with both leases lapsed stops counting as the holder, and nothing is claimed. +4. **Board copyback-owner confirmation on another host** + (`native-workspace-owner-recovery.ts`). It is accepted on the board's word. +5. **Remote leases.** Unchanged. The remote runner is fenced by its own + controller lease, and step 1 never takes a remote lease back. + +## Step 2 + +Add `lease_claim_token text` (nullable) and `lease_epoch integer not null default 0` +to `native_run_finalizations` (AGENTS.md §6 migration). + +- An owner claim sets `lease_epoch = lease_epoch + 1` and clears `lease_claim_token`. +- A consumer that acts on a lapsed lease first runs + `update … set lease_claim_token = $consumer, lease_epoch = lease_epoch + 1 where run_id = $1 and lease_epoch = $seen and lease_expires_at <= clock_timestamp() and lease_claim_token is null`. + Only the winner acts. +- The owner takes a lease back with + `… where lease_epoch = $mine and lease_claim_token is null`. Expiry is no + longer part of that decision. +- Move gaps 1, 2 and 4 to that claim. Workspace finalization claims before its + copyback, and the orphan sweep claims before it writes its marker. Then + post-result renewal can take the lease back as well. +- Upgrade: rows saved by step 1 have epoch 0 and a null token. A step-1 server + that is still draining never reads the new columns. Ship the consumers' claims + before the owner path relies only on the token, so that both orders stay fenced + during a rolling upgrade. +- Tests: one race test per consumer that runs its real claim against the owner + taking the lease back, and an old-saved/new-resumed row. diff --git a/server/src/services/native-runtime/native-execution-lease-reacquire.test.ts b/server/src/services/native-runtime/native-execution-lease-reacquire.test.ts new file mode 100644 index 00000000000..0a7477eb7ca --- /dev/null +++ b/server/src/services/native-runtime/native-execution-lease-reacquire.test.ts @@ -0,0 +1,288 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, closeLivenessDb, companies, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, +} from "@paperclipai/db"; +import { + lockNativeExecutionLeaseForReacquire, nativeExecutionLeaseHeldCondition, renewNativeSessionExecutionLease, + startNativeSessionExecutionLeaseRenewal, +} from "./native-execution-lease.js"; +import { OPERATOR_ATTESTED_CONTROLLER_HOST } from "./native-controller-host-recovery.js"; +import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "./native-runner-ownership.js"; +import { + getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; +if (!support.supported) console.warn(`Skipping native lease re-acquire tests: ${support.reason ?? "unsupported environment"}`); + +const controller = { bootId: `boot-${randomUUID()}`, pid: 4242, processStartedAt: new Date("2026-11-10T05:15:54.000Z") }; +const GENERATION = 1; + +/** + * Host sleep (restart-chaos TM2): the owner process is frozen while the wall + * clock and database clock move on, so its stored lease lapsed although nobody + * else took it. Each seeded run's lease lapsed 40 minutes ago, untouched. + */ +describeEmbeddedPostgres("native execution lease taken back after a host sleep", () => { + let db!: ReturnType; + let owner!: ReturnType; + let stopDatabase: (() => Promise) | undefined; + let companyId!: string; + let agentId!: string; + + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("paperclip-native-lease-reacquire-"); + stopDatabase = started.cleanup; + db = createDb(started.connectionString); + owner = createDb(started.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Lease", issuePrefix: "LRA", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", role: "engineer", status: "active", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex" }, runtimeConfig: {}, permissions: {} }); + }, 120_000); + + afterEach(() => vi.useRealTimers()); + + afterAll(async () => { + if (owner) await closeLivenessDb(owner, 1).catch(() => undefined); + await owner?.$client.end({ timeout: 1 }); + await db?.$client.end({ timeout: 1 }); + await stopDatabase?.(); + }); + + async function seedSleptRun(label: string) { + const issueId = randomUUID(); + const runId = randomUUID(); + const leaseOwner = `runner-${label}:${randomUUID()}`; + await db.insert(issues).values({ id: issueId, companyId, identifier: `LRA-${runId.slice(0, 8)}`, + title: label, status: "in_progress", priority: "medium", workMode: "standard", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId } }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "observed", attempt: 1, + leaseOwner, leaseExpiresAt: sql`clock_timestamp() - interval '40 minutes'`, + controllerBootId: controller.bootId, controllerPid: controller.pid, + controllerProcessStartedAt: controller.processStartedAt, controllerHostHash: "host-a", + controllerGeneration: GENERATION }); + const binding = { runId, companyId, issueId }; + const renewal = { db: owner, ...binding, leaseOwner, attempt: 1, controller, + reacquire: { controllerGeneration: GENERATION } }; + return { ...binding, leaseOwner, renewal }; + } + type SleptRun = Awaited>; + + async function row(runId: string) { + const [saved] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + return saved!; + } + + it("lets the same owner take back a lease that lapsed untouched", async () => { + const run = await seedSleptRun("untouched"); + await expect(renewNativeSessionExecutionLease(run.renewal)).resolves.toBeUndefined(); + const saved = await row(run.runId); + expect(saved.leaseOwner).toBe(run.leaseOwner); + expect(saved.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now() + 15 * 60_000); + }); + + it("renews from the shared timer after the sleep instead of interrupting the turn", async () => { + const run = await seedSleptRun("timer"); + vi.useFakeTimers({ toFake: ["setInterval"] }); + const onLost = vi.fn(async () => undefined); + const { reacquire: _reacquire, ...input } = run.renewal; + const guard = startNativeSessionExecutionLeaseRenewal({ ...input, remote: false, + controllerGeneration: GENERATION, onLost }); + try { + await vi.advanceTimersByTimeAsync(5 * 60_000); + await vi.waitFor(async () => expect((await row(run.runId)).leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now())); + expect(onLost).not.toHaveBeenCalled(); + } finally { + await guard.stop(); + } + }); + + it("handles rows saved by the previous release: same shape, and never another boot's lease", async () => { + // Rows saved before the host hash existed carry null there; the rule adds no column. + const legacyRow = await seedSleptRun("legacy-row"); + await db.update(nativeRunFinalizations).set({ controllerHostHash: null }) + .where(eq(nativeRunFinalizations.runId, legacyRow.runId)); + await expect(renewNativeSessionExecutionLease(legacyRow.renewal)).resolves.toBeUndefined(); + // A deploy that lands mid-run: the lapsed lease belongs to the previous + // server's boot, so the new process can never take it back as its own. + const previousServer = await seedSleptRun("previous-boot"); + await expect(renewNativeSessionExecutionLease({ ...previousServer.renewal, + controller: { ...controller, bootId: `boot-${randomUUID()}`, pid: controller.pid + 1 } })) + .rejects.toThrow("native_session_lease_lost"); + expect((await row(previousServer.runId)).leaseExpiresAt!.getTime()).toBeLessThan(Date.now()); + }); + + it("still refuses a lapsed lease without the claimed generation, or on a remote target", async () => { + const run = await seedSleptRun("no-generation"); + const { reacquire: _reacquire, ...legacy } = run.renewal; + await expect(renewNativeSessionExecutionLease(legacy)).rejects.toThrow("native_session_lease_lost"); + await expect(renewNativeSessionExecutionLease({ ...legacy, reacquire: { controllerGeneration: GENERATION + 1 } })) + .rejects.toThrow("native_session_lease_lost"); + vi.useFakeTimers({ toFake: ["setInterval"] }); + const onLost = vi.fn(async () => undefined); + const guard = startNativeSessionExecutionLeaseRenewal({ ...legacy, remote: true, + controllerGeneration: GENERATION, onLost }); + await vi.advanceTimersByTimeAsync(60_000); + await vi.waitFor(() => expect(onLost).toHaveBeenCalled()); + await expect(guard.stop()).rejects.toThrow("native_session_lease_lost"); + expect((await row(run.runId)).leaseExpiresAt!.getTime()).toBeLessThan(Date.now()); + }); + + // Each actor that may act on a lapsed executor lease records that first. The + // late owner must stay fenced out and must not undo the record. + const claimants: Array<[string, (run: SleptRun) => Promise]> = [ + ["restart recovery holds the run for evidence", run => db.update(nativeRunFinalizations) + .set({ recoveryState: "awaiting_evidence" }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["a recovery request was recorded", run => db.update(nativeRunFinalizations) + .set({ recoveryRequestId: randomUUID() }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["the finalizer claimed the row", run => db.update(nativeRunFinalizations) + .set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: sql`clock_timestamp() + interval '5 minutes'` }) + .where(eq(nativeRunFinalizations.runId, run.runId))], + ["cleanup claimed the row", run => db.update(nativeRunFinalizations) + .set({ leaseOwner: `native-cleanup:${randomUUID()}` }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["a restarted controller took over", run => db.update(nativeRunFinalizations) + .set({ controllerBootId: `boot-${randomUUID()}`, controllerGeneration: GENERATION + 1 }) + .where(eq(nativeRunFinalizations.runId, run.runId))], + ["the same controller's generation moved on", run => db.update(nativeRunFinalizations) + .set({ controllerGeneration: GENERATION + 1 }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["a new provider attempt was claimed", run => db.update(nativeRunFinalizations) + .set({ attempt: 2 }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["failure settlement recorded a retry", run => db.update(nativeRunFinalizations) + .set({ phase: "retryable_failure", failureCode: "native_session_lease_lost", nextAttemptAt: new Date() }) + .where(eq(nativeRunFinalizations.runId, run.runId))], + ["a retry is scheduled", run => db.update(nativeRunFinalizations) + .set({ nextAttemptAt: new Date() }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["the run was already failed", run => db.update(heartbeatRuns) + .set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.runId))], + ["the run was cancelled", run => db.update(heartbeatRuns) + .set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.runId))], + ["the board confirmed the controller stopped", run => db.update(nativeRunFinalizations) + .set({ controllerHostHash: OPERATOR_ATTESTED_CONTROLLER_HOST }).where(eq(nativeRunFinalizations.runId, run.runId))], + ["another server's orphan sweep marked ownership unverified", run => db.update(heartbeatRuns) + .set({ errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, error: "unverified" }) + .where(eq(heartbeatRuns.id, run.runId))], + ["startup cleanup was promoted", run => db.update(heartbeatRuns) + .set({ resultJson: { nativeStartupCleanup: { state: "pending" } } }).where(eq(heartbeatRuns.id, run.runId))], + ]; + + it.each(claimants)("fences the late owner out when %s first", async (_name, claim) => { + const run = await seedSleptRun("claimed"); + await claim(run); + const before = await row(run.runId); + await expect(renewNativeSessionExecutionLease(run.renewal)).rejects.toThrow("native_session_lease_lost"); + expect(await row(run.runId)).toEqual(before); + }); + + /** The executor's release and failure settlement use the same lock and held condition. */ + async function releaseAsOwner(run: SleptRun) { + return owner.transaction(async tx => { + await lockNativeExecutionLeaseForReacquire(tx, run); + return tx.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(and(eq(nativeRunFinalizations.runId, run.runId), eq(nativeRunFinalizations.leaseOwner, run.leaseOwner), + eq(nativeRunFinalizations.attempt, 1), eq(nativeRunFinalizations.controllerBootId, controller.bootId), + nativeExecutionLeaseHeldCondition({ remote: false, controllerGeneration: GENERATION, relinquishAccepted: true }))) + .returning({ runId: nativeRunFinalizations.runId }); + }); + } + + it("releases its own lapsed lease after the result was accepted, but never renews it", async () => { + const accepted = await seedSleptRun("accepted"); + // Only the coordinator's accepted-result shape matters here; skip the + // completion-contract foreign keys rather than seed a whole contract. + const [result] = await db.transaction(async tx => { + await tx.execute(sql`set local session_replication_role = replica`); + return tx.insert(nativeRunResults).values({ companyId, issueId: accepted.issueId, runId: accepted.runId, + completionContractId: randomUUID(), serverFingerprint: randomUUID(), schemaStatus: "accepted", resultJson: {}, + canonicalSha256: "a".repeat(64) }).returning({ id: nativeRunResults.id }); + }); + await db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing", resultId: result!.id }) + .where(eq(nativeRunFinalizations.runId, accepted.runId)); + // Workspace finalization may act on a lapsed post-result lease, so it is never renewed ... + await expect(renewNativeSessionExecutionLease(accepted.renewal)).rejects.toThrow("native_session_lease_lost"); + // ... only relinquished, which grants nothing. + expect(await releaseAsOwner(accepted)).toHaveLength(1); + expect((await row(accepted.runId)).leaseOwner).toBeNull(); + }); + + it("does not release a lapsed lease over a recorded recovery claim", async () => { + const claimed = await seedSleptRun("release-claimed"); + await db.update(nativeRunFinalizations).set({ recoveryState: "awaiting_runner_reattach" }) + .where(eq(nativeRunFinalizations.runId, claimed.runId)); + expect(await releaseAsOwner(claimed)).toHaveLength(0); + const saved = await row(claimed.runId); + expect(saved.recoveryState).toBe("awaiting_runner_reattach"); + expect(saved.leaseOwner).toBe(claimed.leaseOwner); + }); + + /** A claimant that acts on expiry: lock the coordinator, re-check expiry, record the claim, in one transaction. */ + async function claimIfLapsed(runId: string, holdMs: number) { + return db.transaction(async tx => { + const [locked] = await tx.select({ expired: sql`${nativeRunFinalizations.leaseExpiresAt} <= clock_timestamp()` }) + .from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)).for("update"); + if (holdMs) await tx.execute(sql`select pg_sleep(${holdMs / 1000})`); + if (!locked?.expired) return false; + await tx.update(nativeRunFinalizations).set({ recoveryState: "awaiting_evidence" }) + .where(eq(nativeRunFinalizations.runId, runId)); + return true; + }); + } + + /** A run writer that does not take the coordinator lock. */ + async function failRun(runId: string, holdMs: number) { + return db.transaction(async tx => { + const [run] = await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)).for("update"); + await tx.execute(sql`select pg_sleep(${holdMs / 1000})`); + if (run?.status !== "running") return false; + await tx.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + return true; + }); + } + + const renewed = (run: SleptRun) => renewNativeSessionExecutionLease(run.renewal) + .then(() => true, (error: Error) => { expect(error.message).toBe("native_session_lease_lost"); return false; }); + + it.each([ + ["the claimant locks first", 0, 80, false], + ["the owner locks first", 80, 0, true], + ])("lets exactly one side win a race on the coordinator (%s)", async (_name, claimDelayMs, ownerDelayMs, ownerWins) => { + const run = await seedSleptRun("race"); + const [claimed, ownerWon] = await Promise.all([ + new Promise(resolve => setTimeout(resolve, claimDelayMs)).then(() => claimIfLapsed(run.runId, 300)), + new Promise(resolve => setTimeout(resolve, ownerDelayMs)).then(() => renewed(run)), + ]); + expect(ownerWon).toBe(ownerWins); + expect(claimed).toBe(!ownerWins); + const saved = await row(run.runId); + expect(saved.recoveryState).toBe(ownerWins ? null : "awaiting_evidence"); + expect(saved.leaseExpiresAt!.getTime() > Date.now()).toBe(ownerWins); + }); + + it("waits for an in-flight run status write instead of reading past it", async () => { + const run = await seedSleptRun("status-race"); + const [failed, ownerWon] = await Promise.all([ + failRun(run.runId, 400), + new Promise(resolve => setTimeout(resolve, 100)).then(() => renewed(run)), + ]); + expect(failed).toBe(true); + expect(ownerWon).toBe(false); + expect((await row(run.runId)).leaseExpiresAt!.getTime()).toBeLessThan(Date.now()); + }); + + it("has exactly one winner across simultaneous starts", async () => { + for (let index = 0; index < 8; index++) { + const run = await seedSleptRun(`burst-${index}`); + const [claimed, ownerWon] = await Promise.all([claimIfLapsed(run.runId, 0), renewed(run)]); + expect(Number(claimed) + Number(ownerWon)).toBe(1); + const saved = await row(run.runId); + expect(saved.recoveryState === null).toBe(ownerWon); + } + }); +}); diff --git a/server/src/services/native-runtime/native-execution-lease.ts b/server/src/services/native-runtime/native-execution-lease.ts index 98c95766ef5..1460efa42e4 100644 --- a/server/src/services/native-runtime/native-execution-lease.ts +++ b/server/src/services/native-runtime/native-execution-lease.ts @@ -1,14 +1,112 @@ import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import type { Db } from "@paperclipai/db"; import { heartbeatRuns, livenessDb, nativeRunFinalizations } from "@paperclipai/db"; -import { and, eq, gt, isNull, sql } from "drizzle-orm"; +import { and, eq, gt, isNull, or, sql, type SQL } from "drizzle-orm"; import { currentNativeControllerIdentity, type NativeControllerIdentity } from "./native-restart-recovery.js"; +import { OPERATOR_ATTESTED_CONTROLLER_HOST } from "./native-controller-host-recovery.js"; +import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "./native-runner-ownership.js"; import { REMOTE_CONTROLLER_LEASE_MS } from "./native-runner-recovery-contract.js"; export const NATIVE_SESSION_EXECUTION_LEASE_TTL_MS = 20 * 60_000; const NATIVE_SESSION_EXECUTION_LEASE_RENEW_INTERVAL_MS = 5 * 60_000; const RENEWAL_STATEMENT_TIMEOUT_MS = 10_000; +/** + * Takeover of a native execution lease is explicit, not inferred from expiry. + * + * A lease can lapse with its owner alive: a host sleep jumps the wall clock + * (and the database clock) past `lease_expires_at` while this process's + * monotonic timers barely move. Expiry then means only "nobody renewed", not + * "the owner is gone". The same owner may take its own lapsed lease back when + * no other actor has recorded anything on the row or run since. + * + * Every actor that acts on a lapsed executor lease while the run is still + * running first writes one of the fields checked here, under the coordinator + * row lock: restart recovery and takeover (`recovery_state`, `lease_owner`, + * `controller_*`), the finalizer and cleanup (`lease_owner`), failure + * settlement (`phase`, `failure_code`, `next_attempt_at`, run status), the + * board's controller-stopped confirmation (`controller_host_hash`), another + * server's orphan sweep (the run's ownership-unverified error code) and startup + * cleanup (the run's `nativeStartupCleanup` marker). Exactly one side wins: + * either that write commits first and the owner stays fenced out, or the + * owner's lease is live again and those actors defer to it as before. + * + * Pre-result only: phase `observed` without an accepted result. After a result + * is accepted, workspace finalization may act on a lapsed lease without such a + * write, so a lapsed post-result lease is only ever relinquished, never renewed. + * Local runs only: a remote runner is fenced by its own controller lease. + * + * Callers lock with `lockNativeExecutionLeaseForReacquire` first, so the run + * row this reads is the latest committed version and stays so until commit. + */ +function nativeExecutionLeaseUntouchedCondition(stage: "observed" | "accepted"): SQL { + const coordinatorStage = stage === "observed" + ? sql`${nativeRunFinalizations.phase} = 'observed' and ${nativeRunFinalizations.resultId} is null` + : sql`${nativeRunFinalizations.phase} = 'workspace_finalizing' and ${nativeRunFinalizations.resultId} is not null`; + return sql`(${coordinatorStage} + and ${nativeRunFinalizations.recoveryState} is null + and ${nativeRunFinalizations.recoveryRequestId} is null + and ${nativeRunFinalizations.failureCode} is null + and ${nativeRunFinalizations.nextAttemptAt} is null + and ${nativeRunFinalizations.controllerHostHash} is distinct from ${OPERATOR_ATTESTED_CONTROLLER_HOST} + and exists (select 1 from ${heartbeatRuns} + where ${heartbeatRuns.id} = ${nativeRunFinalizations.runId} + and ${heartbeatRuns.companyId} = ${nativeRunFinalizations.companyId} + and ${heartbeatRuns.status} = 'running' + and ${heartbeatRuns.errorCode} is distinct from ${NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE} + and (${heartbeatRuns.resultJson} is null or ${heartbeatRuns.resultJson} -> 'nativeStartupCleanup' is null)))`; +} + +/** + * The executor's own lease is held: live, or (local runs only) lapsed and + * untouched since, so the same owner takes it back in this statement. The + * caller's WHERE must also match lease owner, attempt and controller identity; + * `controllerGeneration` is required to take back a lapsed lease. + */ +export function nativeExecutionLeaseHeldCondition(input: { + remote: boolean; controllerGeneration?: number; relinquishAccepted?: boolean; +}): SQL { + const live = gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`); + if (input.remote || input.controllerGeneration === undefined) return live; + const untouched = input.relinquishAccepted + ? or(nativeExecutionLeaseUntouchedCondition("observed"), nativeExecutionLeaseUntouchedCondition("accepted"))! + : nativeExecutionLeaseUntouchedCondition("observed"); + return or(live, and(eq(nativeRunFinalizations.controllerGeneration, input.controllerGeneration), untouched))!; +} + +/** + * Lock order for `nativeExecutionLeaseHeldCondition`: the coordinator row, then, + * only when its lease has lapsed, the run row (shared). The share lock waits for + * any in-flight run write and holds the run until commit, so the next statement + * reads the run's latest committed state. A live lease takes no extra lock. + * Returns whether the lease was still live. + */ +export async function lockNativeExecutionLeaseForReacquire(tx: Pick, input: { + runId: string; companyId: string; issueId: string; + /** + * Lock only a row the caller's UPDATE could match: this owner's, and with + * `preResultOnly` one without an accepted result. A row it cannot match + * stays unlocked, as the UPDATE alone would leave it. + */ + leaseOwner?: string; preResultOnly?: boolean; +}): Promise { + const rows = await tx.execute<{ live: boolean }>(sql`select coalesce(${nativeRunFinalizations.leaseExpiresAt} > clock_timestamp(), false) as live + from ${nativeRunFinalizations} + where ${nativeRunFinalizations.runId} = ${input.runId} + and ${nativeRunFinalizations.companyId} = ${input.companyId} + and ${nativeRunFinalizations.issueId} = ${input.issueId} + ${input.leaseOwner === undefined ? sql`` : sql`and ${nativeRunFinalizations.leaseOwner} = ${input.leaseOwner}`} + ${input.preResultOnly ? sql`and ${nativeRunFinalizations.resultId} is null` : sql``} + for update`); + const row = (rows as unknown as Array<{ live: boolean }> | undefined)?.[0]; + if (!row) return false; + if (row.live) return true; + await tx.execute(sql`select 1 from ${heartbeatRuns} + where ${heartbeatRuns.id} = ${input.runId} and ${heartbeatRuns.companyId} = ${input.companyId} + for share`); + return false; +} + /** Preparation can perform provider I/O only under its current recovery claim. */ export async function assertNativeSessionPreparationAuthority(input: { db: Db; companyId: string; runId: string; leaseOwner: string | undefined; @@ -74,6 +172,8 @@ export async function renewNativeSessionExecutionLease(input: { attempt: number; controller?: NativeControllerIdentity; leaseTtlMs?: number; + /** Local runs only: lets this exact owner take back a lease that lapsed untouched (host sleep). */ + reacquire?: { controllerGeneration: number }; }, { statementTimeoutMs = RENEWAL_STATEMENT_TIMEOUT_MS } = {}): Promise { await renewNativeSessionExecutionLeaseDeadline(input, { statementTimeoutMs }); } @@ -102,10 +202,7 @@ async function renewNativeSessionExecutionLeaseDeadline( await tx.execute(sql`select set_config('statement_timeout', ${`${statementTimeoutMs}ms`}, true)`); // Check expiry after acquiring the row lock. An UPDATE alone can evaluate // its predicate before waiting and revive an owner that expired in the wait. - await tx.execute(sql`select 1 from ${nativeRunFinalizations} - where ${nativeRunFinalizations.runId} = ${input.runId} - and ${nativeRunFinalizations.companyId} = ${input.companyId} - and ${nativeRunFinalizations.issueId} = ${input.issueId} for update`); + await lockNativeExecutionLeaseForReacquire(tx, input); // The database extends from UPDATE time. Exclude connection/lock waits, // but conservatively include UPDATE/commit response time without assuming // that this server's wall clock agrees with the database's clock. @@ -129,7 +226,8 @@ async function renewNativeSessionExecutionLeaseDeadline( nativeRunFinalizations.controllerProcessStartedAt, controller.processStartedAt, ), - gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`), + nativeExecutionLeaseHeldCondition({ remote: !input.reacquire, + controllerGeneration: input.reacquire?.controllerGeneration }), ), ) .returning({ runId: nativeRunFinalizations.runId }); @@ -142,6 +240,8 @@ type RenewalInput = { db: Db; runId: string; companyId: string; issueId: string; leaseOwner: string; attempt: number; controller: NativeControllerIdentity; remote?: boolean; + /** The claimed generation. Local runs then take back a lease that lapsed untouched (host sleep). */ + controllerGeneration?: number; confirmedDeadline?: number; onLost: () => Promise; }; @@ -185,7 +285,9 @@ export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): Na const renew = () => { if (state!.lost || state!.busy) return; state!.busy = true; - state!.pending = renewNativeSessionExecutionLeaseDeadline({ ...input, leaseTtlMs: ttlMs }) + state!.pending = renewNativeSessionExecutionLeaseDeadline({ ...input, leaseTtlMs: ttlMs, + reacquire: !input.remote && input.controllerGeneration !== undefined + ? { controllerGeneration: input.controllerGeneration } : undefined }) .then(confirmedDeadline => { // A late database response cannot restore local authority after loss // or recreate a timer after the final caller has stopped. @@ -198,7 +300,10 @@ export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): Na // A cancelled statement can retry at the next tick, but does not // extend the ownership deadline. Never queue overlapping attempts. for (let cause = error, depth = 0; cause && depth < 4; cause = (cause as { cause?: unknown }).cause, depth++) { - if ((cause as { code?: unknown }).code === "57014") return; + // 40P01: taking back a lapsed lease locks the run after the + // coordinator; a writer holding them in the other order can deadlock. + const code = (cause as { code?: unknown }).code; + if (code === "57014" || code === "40P01") return; } lose(error); }).finally(() => { state!.busy = false; }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 25eba55d801..78f65266595 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -45,7 +45,7 @@ import { providerPackWorkspaceId, recordAdmittedRemoteProviderPack, resolveRemot import { nativeProviderPackAdmissionPending, rememberAdmittedNativeProviderPack, rememberRecoveredNativeProviderPack, recordNativeProviderPackAdmission, type NativeProviderPackAdmissionSource } from "./native-provider-pack-admission.js"; import { type ProviderPackInstallation, verifyRemoteProviderPack as verifyRemoteProviderPackIdentity } from "./remote-provider-pack-verification.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; -import { NATIVE_SESSION_EXECUTION_LEASE_TTL_MS, releaseUnstartedNativeExecutionClaim, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; +import { NATIVE_SESSION_EXECUTION_LEASE_TTL_MS, lockNativeExecutionLeaseForReacquire, nativeExecutionLeaseHeldCondition, releaseUnstartedNativeExecutionClaim, startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "./native-execution-lease.js"; export { renewNativeSessionExecutionLease } from "./native-execution-lease.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { writeExecutionControl } from "../execution-control-notifications.js"; @@ -10157,6 +10157,7 @@ async function executePaperclipNativeSessionAttempt( companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, leaseOwner, attempt: claimedAttempt, controller, remote: input.runnerExecutionTarget?.kind === "remote", + controllerGeneration, confirmedDeadline: confirmedLeaseDeadline, onLost: () => { startup.providerPackSourceAbort.abort(new Error("native_session_lease_lost")); @@ -11860,6 +11861,9 @@ async function executePaperclipNativeSessionAttempt( ), ) .for("update"); + // Issue → coordinator → run, as below. A lease that lapsed untouched + // (host sleep) is still this owner's to settle; see the held condition. + await lockNativeExecutionLeaseForReacquire(tx, { ...input.execution.binding, leaseOwner, preResultOnly: true }); await notifyExecutionRunWork(tx, { companyId: input.execution.binding.companyId }); await notifyNativeFinalizationWork(tx, input.execution.binding); const updated = await tx @@ -11992,7 +11996,8 @@ async function executePaperclipNativeSessionAttempt( nativeRunFinalizations.controllerProcessStartedAt, controller.processStartedAt, ), - gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), + nativeExecutionLeaseHeldCondition({ remote: input.runnerExecutionTarget?.kind === "remote", + controllerGeneration }), ), ) .returning({ runId: nativeRunFinalizations.runId, resultId: nativeRunFinalizations.resultId }) @@ -12457,6 +12462,8 @@ async function executePaperclipNativeSessionAttempt( } const releaseNow = new Date(); const released = await timeController("coordinator_release", () => input.db.transaction(async tx => { + // A lease that lapsed untouched (host sleep) is still this owner's to release. + await lockNativeExecutionLeaseForReacquire(tx, { ...input.execution.binding, leaseOwner }); await notifyNativeFinalizationWork(tx, input.execution.binding); return tx.update(nativeRunFinalizations) .set({ @@ -12479,7 +12486,8 @@ async function executePaperclipNativeSessionAttempt( nativeRunFinalizations.controllerProcessStartedAt, controller.processStartedAt, ), - gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), + nativeExecutionLeaseHeldCondition({ remote: input.runnerExecutionTarget?.kind === "remote", + controllerGeneration, relinquishAccepted: true }), ), ) .returning({ runId: nativeRunFinalizations.runId })