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 index 5fa083ef278..4c2fa314d75 100644 --- a/doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md +++ b/doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md @@ -51,6 +51,10 @@ explicit token. 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. + Closed without the epoch (2026-11-10): the sweep now writes its marker under the + coordinator row lock and only while no executor lease on the run is live + (`setRunStatusFromLive` option `onlyWhileNativeLeaseLapsed`). Test: + `server/src/__tests__/heartbeat-native-orphan-sweep-lease-race.test.ts`. 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. diff --git a/server/src/__tests__/heartbeat-native-orphan-sweep-lease-race.test.ts b/server/src/__tests__/heartbeat-native-orphan-sweep-lease-race.test.ts new file mode 100644 index 00000000000..e8b2b26a9bc --- /dev/null +++ b/server/src/__tests__/heartbeat-native-orphan-sweep-lease-race.test.ts @@ -0,0 +1,129 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, closeLivenessDb, companies, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { heartbeatService } from "../services/heartbeat.ts"; +import { + lockNativeExecutionLeaseForReacquire, nativeExecutionLeaseHeldCondition, renewNativeSessionExecutionLease, +} from "../services/native-runtime/native-execution-lease.js"; +import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../services/native-runtime/native-runner-ownership.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; +if (!support.supported) console.warn(`Skipping orphan-sweep lease race tests: ${support.reason ?? "unsupported environment"}`); + +/** The owning server: another process on the same database. */ +const controller = { bootId: `boot-${randomUUID()}`, pid: 4243, processStartedAt: new Date("2026-11-10T05:15:54.000Z") }; +const GENERATION = 1; + +/** + * Host sleep with two servers on one database (plan + * doc/plans/2026-11-09-explicit-native-lease-claim-tokens.md, gap 2). The + * owner's executor lease lapsed only because its host slept. Another server's + * orphan sweep reads the lapsed lease; the owner wakes and takes the lease back + * before the sweep writes. The sweep must not then mark the healthy run + * ownership-unverified: that marker tells the board automatic recovery is + * blocked, and it fences the owner out of its own lease after the next sleep, + * so the turn is interrupted and the user must click Retry. + */ +describeEmbeddedPostgres("another server's orphan sweep racing the owner's lease take-back", () => { + let db!: ReturnType; + let owner!: ReturnType; + let stopDatabase: (() => Promise) | undefined; + let companyId!: string; + let agentId!: string; + + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("paperclip-orphan-sweep-lease-race-"); + stopDatabase = started.cleanup; + db = createDb(started.connectionString); + owner = createDb(started.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Sweep race", issuePrefix: "SWR", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", role: "engineer", status: "running", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex" }, runtimeConfig: {}, permissions: {} }); + }, 120_000); + + 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() { + const issueId = randomUUID(); + const runId = randomUUID(); + const leaseOwner = `runner-slept:${randomUUID()}`; + await db.insert(issues).values({ id: issueId, companyId, identifier: `SWR-${runId.slice(0, 8)}`, + title: "slept", status: "in_progress", priority: "medium", workMode: "standard", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", + nativeIssueId: issueId, nativePhase: "observed", invocationSource: "assignment", triggerDetail: "system", + contextSnapshot: { issueId }, startedAt: sql`clock_timestamp() - interval '41 minutes'` as unknown as Date }); + 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 }); + return { runId, companyId, issueId, leaseOwner }; + } + + async function waitForLockWaiter() { + for (let attempt = 0; attempt < 200; attempt++) { + const rows = await owner.execute<{ waiting: number }>(sql`select count(*)::int as waiting from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock'`); + if (Number((rows as unknown as Array<{ waiting: number }>)[0]?.waiting ?? 0) > 0) return true; + await new Promise(resolve => setTimeout(resolve, 25)); + } + return false; + } + + it("leaves the run unmarked when the owner takes its lapsed lease back first", async () => { + const run = await seedSleptRun(); + const sweep = heartbeatService(db); + let sweepDone: Promise | undefined; + // The owner's take-back, held open: coordinator lock, then the run (shared). + await owner.transaction(async tx => { + const live = await lockNativeExecutionLeaseForReacquire(tx, run); + expect(live).toBe(false); + // The sweep reads the lapsed lease now and blocks at its write. + sweepDone = sweep.reapOrphanedRuns(); + expect(await waitForLockWaiter()).toBe(true); + const renewed = await tx.update(nativeRunFinalizations) + .set({ leaseExpiresAt: sql`clock_timestamp() + interval '20 minutes'`, updatedAt: sql`clock_timestamp()` }) + .where(and(eq(nativeRunFinalizations.runId, run.runId), eq(nativeRunFinalizations.leaseOwner, run.leaseOwner), + eq(nativeRunFinalizations.attempt, 1), eq(nativeRunFinalizations.controllerBootId, controller.bootId), + eq(nativeRunFinalizations.controllerPid, controller.pid), + eq(nativeRunFinalizations.controllerProcessStartedAt, controller.processStartedAt), + nativeExecutionLeaseHeldCondition({ remote: false, controllerGeneration: GENERATION }))) + .returning({ runId: nativeRunFinalizations.runId }); + expect(renewed).toHaveLength(1); + }); + await sweepDone; + + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.runId)); + expect(saved).toMatchObject({ status: "running", errorCode: null, error: null }); + const warnings = await db.select({ message: heartbeatRunEvents.message }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.runId, run.runId), eq(heartbeatRunEvents.level, "warn"))); + expect(warnings).toEqual([]); + + // The host sleeps again before the turn ends: the owner must still take its own lease back. + await db.update(nativeRunFinalizations).set({ leaseExpiresAt: sql`clock_timestamp() - interval '40 minutes'` }) + .where(eq(nativeRunFinalizations.runId, run.runId)); + await expect(renewNativeSessionExecutionLease({ db: owner, ...run, attempt: 1, controller, + reacquire: { controllerGeneration: GENERATION } })).resolves.toBeUndefined(); + }); + + it("still marks a lapsed run whose owner never came back", async () => { + const run = await seedSleptRun(); + await heartbeatService(db).reapOrphanedRuns(); + const [saved] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.runId)); + expect(saved).toMatchObject({ status: "running", errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE }); + // ... and the late owner stays fenced out. + await expect(renewNativeSessionExecutionLease({ db: owner, ...run, attempt: 1, controller, + reacquire: { controllerGeneration: GENERATION } })).rejects.toThrow("native_session_lease_lost"); + }); +}); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 71a86bd8c7a..4eaef8c8842 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -164,7 +164,8 @@ export interface HeartbeatRecoveryDependencies extends Pick Promise; setRunStatus: (runId: string, status: string, patch?: Partial) => Promise; - setRunStatusFromLive: (runId: string, status: string, fromStatuses: string[], patch?: Partial) => Promise<{ + setRunStatusFromLive: (runId: string, status: string, fromStatuses: string[], patch?: Partial, + failureReport?: undefined, cancellationCondition?: undefined, options?: { onlyWhileNativeLeaseLapsed?: true }) => Promise<{ updated: boolean; run: HeartbeatRun | null; }>; @@ -1980,6 +1981,12 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD | "native_chat_workspace_scope_mismatch"; processPidAlive?: boolean; processGroupAlive?: boolean; + /** + * The orphan sweep acted on a lapsed executor lease it read earlier. The + * owner may have taken that lease back since (host sleep), so mark only + * while it is still lapsed, under the coordinator lock. + */ + leaseLapsed?: boolean; }, ) { const durableOwnershipHold = @@ -1992,21 +1999,20 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ) return run; const blockedStatus = run.status === "failed" ? "failed" : "running"; - const blockedWrite = await setRunStatusFromLive( - run.id, - blockedStatus, - [blockedStatus], - { - error: NATIVE_OWNERSHIP_UNVERIFIED_MESSAGE, - errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, - ...(durableOwnershipHold - ? { - nativePhase: "terminal_failure", - nativePhaseUpdatedAt: new Date(), - } - : {}), - }, - ); + const blockedPatch = { + error: NATIVE_OWNERSHIP_UNVERIFIED_MESSAGE, + errorCode: NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE, + ...(durableOwnershipHold + ? { + nativePhase: "terminal_failure", + nativePhaseUpdatedAt: new Date(), + } + : {}), + }; + const blockedWrite = evidence.leaseLapsed + ? await setRunStatusFromLive(run.id, blockedStatus, [blockedStatus], blockedPatch, + undefined, undefined, { onlyWhileNativeLeaseLapsed: true }) + : await setRunStatusFromLive(run.id, blockedStatus, [blockedStatus], blockedPatch); if (!blockedWrite.updated || !blockedWrite.run) { return blockedWrite.run ?? run; } @@ -2506,6 +2512,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD : "observed_owner_unverified", processPidAlive: nativeProcessPidAlive, processGroupAlive: nativeProcessGroupAlive, + // Read as lapsed (or unset): its owner may take it back before this write. + leaseLapsed: nativeRun && !(nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now), }); if (nativeControllerLeaseExpiresAt && nativeControllerLeaseExpiresAt > now) scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 650f7a504a4..7c6189b3863 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -1220,7 +1220,15 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl patch?: Partial, failureReport?: RunFailureReportOptions, cancellationCondition?: ReturnType, - options?: { setupFailure: true }, + options?: { + setupFailure?: true; + /** + * Write only while no native executor lease on the run is live. Locks the + * coordinator row first, the order the owner takes back a lapsed lease in + * (`lockNativeExecutionLeaseForReacquire`), so exactly one side wins. + */ + onlyWhileNativeLeaseLapsed?: true; + }, ) { // fromStatuses can name a terminal status as its own source (for example, // an idempotent "still failed" patch), so the write below is not always a @@ -1272,6 +1280,11 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl eq(nativeRunFinalizations.companyId, previousStatus.companyId), eq(nativeRunFinalizations.issueId, previousStatus.nativeIssueId))) .for("update"); + } else if (options?.onlyWhileNativeLeaseLapsed && previousStatus) { + await writer.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, previousStatus.companyId))) + .for("update"); } return previousStatus && previousStatus.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(status) ? terminalizeLegacyExecution({ @@ -1314,6 +1327,11 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl ...(isHeartbeatRunTerminalStatus(status) ? [nativeRunnerOwnershipNotHeldCondition()] : []), + ...(options?.onlyWhileNativeLeaseLapsed + ? [sql`not exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.leaseExpiresAt} > clock_timestamp())`] + : []), ), ) .returning()