diff --git a/packages/db/src/migrations/0446_native_lease_holder.sql b/packages/db/src/migrations/0446_native_lease_holder.sql new file mode 100644 index 00000000000..f7592a083ed --- /dev/null +++ b/packages/db/src/migrations/0446_native_lease_holder.sql @@ -0,0 +1,3 @@ +ALTER TABLE "native_run_finalizations" ADD COLUMN "lease_holder" jsonb;--> statement-breakpoint +CREATE INDEX "native_run_finalizations_finalizer_lease_idx" ON "native_run_finalizations" USING btree ("lease_expires_at") WHERE "native_run_finalizations"."lease_owner" like 'native-finalizer:%';--> statement-breakpoint +CREATE INDEX "native_run_finalizations_cleanup_lease_idx" ON "native_run_finalizations" USING btree ("lease_expires_at") WHERE "native_run_finalizations"."lease_owner" like 'native-cleanup:%'; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0440_snapshot.json b/packages/db/src/migrations/meta/0446_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0440_snapshot.json rename to packages/db/src/migrations/meta/0446_snapshot.json index b62a9e53a00..f7c4f4c2756 100644 --- a/packages/db/src/migrations/meta/0440_snapshot.json +++ b/packages/db/src/migrations/meta/0446_snapshot.json @@ -1,6 +1,6 @@ { - "id": "0c990bdc-6df0-44ca-8f21-28124f7b39e7", - "prevId": "d290d707-7d29-43a0-a697-06eecbdb743d", + "id": "782688f1-7843-4a93-8ea3-2cec362d269e", + "prevId": "f0d8430b-7abf-4232-86a9-11a268a0d20e", "version": "7", "dialect": "postgresql", "tables": { @@ -34880,6 +34880,12 @@ "primaryKey": false, "notNull": false }, + "process_host_hash": { + "name": "process_host_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, "legacy_process_settlement": { "name": "legacy_process_settlement", "type": "jsonb", @@ -44760,6 +44766,12 @@ "primaryKey": false, "notNull": false }, + "lease_holder": { + "name": "lease_holder", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "controller_boot_id": { "name": "controller_boot_id", "type": "text", @@ -44772,6 +44784,12 @@ "primaryKey": false, "notNull": false }, + "controller_host_hash": { + "name": "controller_host_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, "controller_process_started_at": { "name": "controller_process_started_at", "type": "timestamp with time zone", @@ -44918,6 +44936,54 @@ "method": "btree", "with": {} }, + "native_run_finalizations_finalizer_lease_idx": { + "name": "native_run_finalizations_finalizer_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-finalizer:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_cleanup_lease_idx": { + "name": "native_run_finalizations_cleanup_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"lease_owner\" like 'native-cleanup:%'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_accepted_result_lease_idx": { + "name": "native_run_finalizations_accepted_result_lease_idx", + "columns": [ + { + "expression": "lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"lease_owner\" is not null AND \"native_run_finalizations\".\"phase\" not in ('committed', 'terminal_failure', 'applied')", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 6c3bfb7b73b..b2011fce23e 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3109,6 +3109,13 @@ "when": 1794104103100, "tag": "0445_native_accepted_result_lease_index", "breakpoints": true + }, + { + "idx": 446, + "version": "7", + "when": 1794105010331, + "tag": "0446_native_lease_holder", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/native_run_finalizations.ts b/packages/db/src/schema/native_run_finalizations.ts index 613a9580b7c..fc4fa11076c 100644 --- a/packages/db/src/schema/native_run_finalizations.ts +++ b/packages/db/src/schema/native_run_finalizations.ts @@ -27,6 +27,11 @@ export const nativeRunFinalizations = pgTable( attempt: integer("attempt").notNull().default(0), leaseOwner: text("lease_owner"), leaseExpiresAt: timestamp("lease_expires_at", { withTimezone: true }), + // Server process that took a token-only lease (finalizer or cleanup): + // {token, bootId, pid, hostHash, processStartedAt}. It names the holder only + // while `token` equals lease_owner, so a later lease never inherits it. The + // controller_* columns stay the executor's. Null on rows saved before it existed. + leaseHolder: jsonb("lease_holder").$type>(), controllerBootId: text("controller_boot_id"), controllerPid: integer("controller_pid"), // PID space of controllerPid (controller-process-identity hostHash). Null @@ -61,6 +66,14 @@ export const nativeRunFinalizations = pgTable( .where(sql`${table.phase} = 'terminal_failure' AND ${table.failureCode} = 'native_workspace_sync_out_unsafe_archive'`), // The `native_accepted_result_lease` ownership kind: outstanding executor // leases on an accepted result that is not final yet (reconciler listing). + // The `native_finalizer_lease` and `native_cleanup_lease` ownership kinds + // (reconciler listing and settle watch). Both are sparse. + finalizerLeaseIdx: index("native_run_finalizations_finalizer_lease_idx") + .on(table.leaseExpiresAt) + .where(sql`${table.leaseOwner} like 'native-finalizer:%'`), + cleanupLeaseIdx: index("native_run_finalizations_cleanup_lease_idx") + .on(table.leaseExpiresAt) + .where(sql`${table.leaseOwner} like 'native-cleanup:%'`), acceptedResultLeaseIdx: index("native_run_finalizations_accepted_result_lease_idx") .on(table.leaseExpiresAt) .where(sql`${table.resultId} is not null AND ${table.leaseOwner} is not null AND ${table.phase} not in ('committed', 'terminal_failure', 'applied')`), diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index bc77912b7ca..9ac9b952ed1 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -2,6 +2,8 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { activityLog, authUsers, budgetReservations, issueRecoveryActions, companyMemberships, completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; import { expect, it, vi } from "vitest"; +import { subscribeDatabaseWork } from "@paperclipai/db/work-signals"; +import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; @@ -10,11 +12,12 @@ import type { NativeSessionBackend } from "@paperclipai/paperclip-runner"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { CONTROL_PLANE_CONFORMANCE_OPEN, CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; -import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; +import { currentControllerBootId, currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { createOwnershipReportState, runOwnershipReconciliationPass, settleOwnableKind, type OwnershipReport } from "../services/recovery/ownership-invariant-report.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; import { registerOwnershipReconciliationWork } from "../services/native-runtime/native-finalization-reconciler.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +import { proveOwnerGone } from "../services/owner-proof.js"; describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shutdown", () => { @@ -572,4 +575,119 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut await coordinator.stop(); } }, 180_000); + + // Phase 2: the token-only finalizer and cleanup leases record their holder + // process (`lease_holder`, bound to the token) and settle on the shared + // ownership path when it is proven gone, never by lease age. + type Holder = { bootId: string; pid: number; hostHash: string; processStartedAt: string | null }; + const deadHolder = (hostHash = currentProcessSpaceHash()): Holder => ({ bootId: randomUUID(), pid: 2_000_000_001, + hostHash, processStartedAt: new Date(Date.now() - 60_000).toISOString() }); + async function heldBy(f: Awaited>, holder: Holder, token?: string) { + const [row] = await f.db.select({ leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(nativeRunFinalizations).set({ leaseHolder: { token: token ?? row.leaseOwner, ...holder } }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + } + const tokenKindId = (prefix: string) => prefix === "native-finalizer" ? "native_finalizer_lease" : "native_cleanup_lease"; + /** This run's record of the kind and its proof (the fixture database is shared across cases). */ + async function proofOf(f: Awaited>, prefix: string) { + const kind = OWNABLE_KINDS.find(candidate => candidate.id === tokenKindId(prefix))!; + const records = await kind.listForRuns!(f.db, [f.runId]); + if (!records.length) return null; + const owner = records[0]!.owner; + if (owner.kind === "process") return { owner: owner.kind, ...await proveOwnerGone(owner.stamp) }; + if (owner.kind === "fence") return { owner: owner.kind, ...await proveOwnerGone(owner.stamp, owner.fence) }; + return { owner: owner.kind }; + } + async function leaseOf(f: Awaited>) { + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + return owner; + } + + it("finalizes promptly when a hard kill left the finalizer lease to its dead holder, and stamps the next holder", async () => { + // Crash point: the finalizer claimed the accepted result (lease written) + // and the server was killed before it committed. The 5-minute lease must + // not hold the run `running`; the next boot proves the holder gone. + const f = await seedAcceptedResult("FINALIZER-THEN-SIGKILL", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder()); + // A claim wakes the ownership worker, which may have gone quiet, so it + // watches the new holder. + const ownershipWakes: string[] = []; + const stop = subscribeDatabaseWork(f.db, DELIVERY_QUEUES.ownershipInvariantReport, (event, hints) => { + if (event === "settled") ownershipWakes.push(...(hints ?? [])); + }); + const { run, owner } = await restart(f).finally(stop); + expect(ownershipWakes.some(hint => hint.includes(f.runId))).toBe(true); + expect({ status: run.status, phase: owner.phase }).toEqual({ status: "succeeded", phase: "committed" }); + // This boot's own finalizer claim recorded itself as the holder of its token, + // without rewriting the executor's controller identity. + expect(owner.leaseHolder).toMatchObject({ token: expect.stringMatching(/^native-finalizer:/), + bootId: currentControllerBootId(), pid: process.pid, hostHash: currentProcessSpaceHash() }); + expect(owner.controllerPid).toBe((await currentNativeControllerIdentity()).pid); + }, 90_000); + + it.each(["native-finalizer", "native-cleanup"])("settles a dead %s holder in the reconciler pass, never in a read-only pass", async (prefix) => { + const f = await seedAcceptedResult(`TOKEN-PASS-${prefix}`, () => `${prefix}:${randomUUID()}`); + await heldBy(f, deadHolder()); + expect(await proofOf(f, prefix)).toEqual({ owner: "process", verdict: "gone", reason: "pid_dead" }); + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState(), record: false }); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + const pass = await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + expect(pass.kinds.find(kind => kind.kind === tokenKindId(prefix))!.settled).toBeGreaterThanOrEqual(1); + const released = await leaseOf(f); + expect(released.leaseExpiresAt!.getTime()).toBeLessThanOrEqual(Date.now()); + // Ownership only: the accepted result and its workspace work are untouched. + expect({ phase: released.phase, resultId: released.resultId }).toEqual({ phase: "workspace_finalizing", + resultId: expect.any(String) }); + // An expired lease is free for the next claim and no longer in progress. + expect(await proofOf(f, prefix)).toBeNull(); + }, 60_000); + + it.each([ + { label: "live earlier-boot holder", proof: { verdict: "alive", reason: "pid_alive" }, holder: async (): Promise => ({ bootId: randomUUID(), + pid: process.pid, hostHash: currentProcessSpaceHash(), + processStartedAt: (await currentNativeControllerIdentity()).processStartedAt.toISOString() }) }, + { label: "current boot", proof: { verdict: "alive", reason: "current_process" }, holder: async (): Promise => ({ bootId: currentControllerBootId(), + pid: process.pid, hostHash: currentProcessSpaceHash(), processStartedAt: null }) }, + { label: "dead PID in another PID space", proof: { verdict: "unprovable", reason: "different_process_space" }, holder: async () => deadHolder("another-pid-space") }, + ])("keeps a finalizer lease held by a $label", async ({ holder, proof }) => { + const f = await seedAcceptedResult(`FINALIZER-KEPT-${randomUUID()}`, () => `native-finalizer:${randomUUID()}`); + await heldBy(f, await holder()); + expect(await proofOf(f, "native-finalizer")).toEqual({ owner: "process", ...proof }); + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + await restart(f); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + }, 60_000); + + it("never lets a later lease inherit an earlier holder, and keeps holderless leases of the previous release", async () => { + // A holder written for another token names nobody; a row saved by the + // previous release has none. Both stay token fences until expiry, also on + // an ephemeral server. + const f = await seedAcceptedResult("FINALIZER-STALE-HOLDER", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder(), `native-finalizer:${randomUUID()}`); + const legacy = await seedAcceptedResult("FINALIZER-PREVIOUS-RELEASE", () => `native-finalizer:${randomUUID()}`); + vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); + try { + for (const row of [f, legacy]) { + expect(await proofOf(row, "native-finalizer")).toEqual({ owner: "fence", verdict: "alive", reason: "fence_held" }); + } + await runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState() }); + } finally { vi.unstubAllEnvs(); } + await restart(f); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + expect((await leaseOf(legacy)).leaseExpiresAt).toEqual(legacy.leaseExpiresAt); + }, 90_000); + + it("settles only the exact lease it proved: a takeover after the proof is kept", async () => { + const f = await seedAcceptedResult("FINALIZER-TAKEN-OVER", () => `native-finalizer:${randomUUID()}`); + await heldBy(f, deadHolder()); + const kind = OWNABLE_KINDS.find(candidate => candidate.id === "native_finalizer_lease")!; + const [record] = await kind.listForRuns!(f.db, [f.runId]); + const successor = `native-finalizer:${randomUUID()}`; + await f.db.update(nativeRunFinalizations).set({ leaseOwner: successor, + leaseHolder: { token: successor, ...deadHolder() } }).where(eq(nativeRunFinalizations.runId, f.runId)); + const outcome = await kind.settle!(f.db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } }); + expect(outcome).toEqual({ settled: false, reason: "owner_changed_since_proof" }); + expect((await leaseOf(f)).leaseExpiresAt).toEqual(f.leaseExpiresAt); + }, 60_000); }); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index ecc246bb638..1d26ce393a7 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -140,7 +140,8 @@ describe("report-only ownership reconciliation", () => { const names = OWNABLE_KINDS.map(kind => kind.id); expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); - expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)).toEqual(["native_accepted_result_lease", "wake_request_claimed"]); + expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "wake_request_claimed"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index cd33ccb1ed3..5858b719e0a 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -954,6 +954,12 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD payload: { restartKind: "hard", runnerDisposition: "awaiting_result_finalization", reason: released.reason }, }).catch(error => logger.warn({ err: error, runId: released.runId }, "failed to record released native result lease")); } + // Finalizer and cleanup leases of a stopped holder, so the finalization + // and cleanup below run now. A failure is retried by the reconciler pass. + for (const kind of ["native_finalizer_lease", "native_cleanup_lease"]) { + await settleOwnableKind(db, kind, { now, runIds }) + .catch(error => logger.warn({ err: error, kind }, "failed to settle native token leases at restart")); + } if (!runIds && !options?.skipDelegatedWork) { await reconcileRetainedWorkspaces(db, environmentRuntime, resumeRemoteStopComments); await reconcileNativeFinalizations(db, undefined, { diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index e584d1fb407..1fae0f2768f 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -3,6 +3,7 @@ import { isContextCompactionRun } from "../context-compaction.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js"; import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js"; +import { nativeLeaseHolder, notifyNativeTokenLeaseClaimed } from "./native-token-lease-settlement.js"; import { notifyNativeRunProjection } from "../native-run-projection-notifications.js"; import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js"; import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; @@ -237,11 +238,16 @@ async function claimCoordinator(input: { throw new NativeFinalizationDeferredError("lease_busy"); if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator); else await notifyNativeResultCommit(tx, coordinator); + await notifyNativeTokenLeaseClaimed(tx, coordinator); const [updated] = await tx .update(nativeRunFinalizations) .set({ leaseOwner, leaseExpiresAt: new Date(now.getTime() + 5 * 60_000), + // This process, bound to the token, so a hard kill mid-finalization is + // proven and settled on the shared ownership path. The controller_* + // columns stay the executor's. + leaseHolder: nativeLeaseHolder(leaseOwner), attempt: input.preserveProviderAttempt || coordinator.controllerBootId !== null ? coordinator.attempt diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0ffe3f696a8..a8f48d8c39b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,4 +1,5 @@ import { remoteRecoveryLeaseMismatch } from "./remote-recovery-lease-mismatch.js"; +import { nativeLeaseHolder, notifyNativeTokenLeaseClaimed } from "./native-token-lease-settlement.js"; import { captureRemoteNativeProviderArtifacts } from "./native-provider-artifact-capture.js"; import { NativeSessionDiagnostics, nativeSessionDiagnostics, type NativeDiagnosticLease } from "./native-session-diagnostics.js"; import { recordAgentProviderRateLimits } from "../agent-provider-rate-limits.js"; @@ -3896,11 +3897,14 @@ async function inspectRetainedNativeSessionCleanup( }, ]), ]; + await notifyNativeTokenLeaseClaimed(tx, { companyId: run.companyId, runId: run.id }); await tx .update(nativeRunFinalizations) .set({ leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), + // This process, bound to the token (shared ownership path). + leaseHolder: nativeLeaseHolder(leaseOwner), recoveryHistory: history, updatedAt: new Date(), }) diff --git a/server/src/services/native-runtime/native-token-lease-settlement.ts b/server/src/services/native-runtime/native-token-lease-settlement.ts new file mode 100644 index 00000000000..61f512c9ba9 --- /dev/null +++ b/server/src/services/native-runtime/native-token-lease-settlement.ts @@ -0,0 +1,112 @@ +import { and, eq, gt, like, sql } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { currentControllerProcessStamp, readControllerProcess, type ControllerProcess } from "../controller-process-identity.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { notifyNativeResultCommit } from "../native-result-commit-notifications.js"; +import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js"; +import type { LocalOwnerStamp, OwnerProof } from "../owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "../recovery/ownership-registry.js"; + +/** + * Holder stamps and settle for the token-only native leases (plan Phase 2): + * the finalizer lease (`native-finalizer:*`, 5 min) and the retained-session + * cleanup lease (`native-cleanup:*`, 60 s). + * + * Both run inside one server process, but their claims used to record only a + * random token: the controller_* columns belong to the executor and must not + * be overwritten (the finalizer's attempt rule and restart recovery read + * them). A holder killed mid-lease therefore kept the run waiting for the + * whole TTL. The claim now writes `lease_holder` = this process's identity, + * bound to the token, in the same UPDATE (no I/O: `currentControllerProcessStamp`). + * The holder names the lease owner only while its token equals `lease_owner`, + * so renewals keep it and any later owner never inherits it. + * + * Previous release: rows it saved have no holder. They stay token fences, and + * the existing claim takes them over at expiry (at most 5 min / 60 s after the + * upgrade, since neither lease outlives that). Settle never touches them. + */ + +export type NativeTokenLeaseKind = "native_finalizer_lease" | "native_cleanup_lease"; + +const PREFIX: Record = { + native_finalizer_lease: "native-finalizer:", + native_cleanup_lease: "native-cleanup:", +}; + +/** Wakes the ownership worker for a newly claimed token lease, so a quiet + * worker (on any server) starts watching the holder. Call it in the claim's + * transaction, before the write it announces. */ +export async function notifyNativeTokenLeaseClaimed(tx: object, scope: { companyId: string; runId: string }) { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.ownershipInvariantReport, + JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); +} + +/** The `lease_holder` value for a token-only lease claimed by this process. */ +export function nativeLeaseHolder(token: string): Record { + return { token, ...currentControllerProcessStamp() }; +} + +/** The process holding `token`, or null when the holder was written for + * another token (or not at all, on rows saved by the previous release). */ +export function readNativeLeaseHolder(value: unknown, token: string | null | undefined): ControllerProcess | null { + if (!token || !value || typeof value !== "object" || (value as { token?: unknown }).token !== token) return null; + return readControllerProcess(value); +} + +export function nativeLeaseHolderStamp(holder: ControllerProcess, scope: NativeTokenLeaseKind): LocalOwnerStamp { + return { + kind: "local", scope, generation: null, host: null, processSpace: holder.hostHash, bootId: holder.bootId, + pid: holder.pid, processStartedAt: holder.processStartedAt, processGroupId: null, + }; +} + +/** + * Called by the shared reconciler only after `proveOwnerGone` proved the + * recorded holder gone. Expires exactly that lease (compare-and-swap on the + * token while it is still outstanding and still held by the proven process), + * after announcing the work the expiry releases. It settles ownership only: + * the finalizer or cleanup that the notification wakes does the rest, under + * its own checks. + */ +export async function settleNativeTokenLease(kind: NativeTokenLeaseKind, db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process") return { settled: false, reason: "owner_not_a_process" }; + const stamp = record.owner.stamp; + const { now } = context; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [row] = await tx.select({ + runId: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, decisionId: nativeRunFinalizations.decisionId, + agentId: heartbeatRuns.agentId, leaseOwner: nativeRunFinalizations.leaseOwner, + leaseHolder: nativeRunFinalizations.leaseHolder, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId))) + .where(and(eq(nativeRunFinalizations.runId, record.recordId), like(nativeRunFinalizations.leaseOwner, `${PREFIX[kind]}%`), + gt(nativeRunFinalizations.leaseExpiresAt, now))) + .for("update", { of: nativeRunFinalizations }); + if (!row?.leaseOwner) return { settled: false, reason: "lease_no_longer_held" }; + const holder = readNativeLeaseHolder(row.leaseHolder, row.leaseOwner); + // The proof judged this exact process; the lease has moved on since. + if (!holder || holder.bootId !== stamp.bootId || holder.pid !== stamp.pid || holder.hostHash !== stamp.processSpace + || holder.processStartedAt !== (stamp.processStartedAt ?? null)) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + // Work intents before the write they announce (sleep-aware background work). + if (kind === "native_finalizer_lease") { + if (row.decisionId) await notifyNativeFinalizationWork(tx, { companyId: row.companyId, runId: row.runId }); + else await notifyNativeResultCommit(tx, { companyId: row.companyId, runId: row.runId, issueId: row.issueId }); + } else { + await notifyRetainedNativeSessionCleanup(tx, { companyId: row.companyId, runId: row.runId, + issueId: row.issueId, agentId: row.agentId }); + } + const updated = await tx.update(nativeRunFinalizations).set({ leaseExpiresAt: now, updatedAt: now }) + .where(and(eq(nativeRunFinalizations.runId, row.runId), eq(nativeRunFinalizations.leaseOwner, row.leaseOwner), + gt(nativeRunFinalizations.leaseExpiresAt, now), + sql`${nativeRunFinalizations.leaseHolder}->>'token' = ${row.leaseOwner}`)) + .returning({ runId: nativeRunFinalizations.runId }); + return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_held" }; + }); +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 988e568c767..8374222b023 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -32,6 +32,12 @@ import { settleNativeAcceptedResultLease, } from "../native-runtime/native-accepted-result-lease-settlement.js"; import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; +import { + nativeLeaseHolderStamp, + readNativeLeaseHolder, + settleNativeTokenLease, + type NativeTokenLeaseKind, +} from "../native-runtime/native-token-lease-settlement.js"; /** * Registry of ownable record kinds (plan section C). @@ -187,6 +193,13 @@ function nrfFenceOwner(row: NrfRow, scope: string): RecordOwner { }; } +/** A finalizer or cleanup lease is owned by the process in `lease_holder`. + * Rows saved before that column keep the token fence (expiry plus takeover). */ +function nrfTokenLeaseOwner(row: NrfRow, scope: NativeTokenLeaseKind): RecordOwner { + const holder = readNativeLeaseHolder(row.leaseHolder, row.leaseOwner); + return holder ? { kind: "process", stamp: nativeLeaseHolderStamp(holder, scope) } : nrfFenceOwner(row, scope); +} + // ---------------------------------------------------------------------------- // heartbeat_runs @@ -382,12 +395,27 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ settle: settleNativeAcceptedResultLease, onUnprovable: requestNativeAcceptedResultLeaseOperatorAction, }, - nrfKind("native_finalizer_lease", "Finalizer lease", "accepted_result", - and(like(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES)), - row => nrfFenceOwner(row, "native_finalizer_lease")), - nrfKind("native_cleanup_lease", "Cleanup lease", "workspace_preservation", - like(nativeRunFinalizations.leaseOwner, "native-cleanup:%"), - row => nrfFenceOwner(row, "native_cleanup_lease")), + // Shared path since Phase 2: the token-only finalizer and cleanup leases + // record their holder process (`lease_holder`, bound to the token); settle + // expires the exact lease of a holder proven gone, so finalization or + // cleanup runs now instead of after the TTL. Outstanding leases only: an + // expired one is free for the next claim. A holder on another host or PID + // space is unprovable and keeps its lease until expiry, when the next claim + // takes it over and the old holder's token compare-and-swap fails: that + // fence is the handoff, never a host difference. + { + ...nrfKind("native_finalizer_lease", "Finalizer lease", "accepted_result", + and(like(nativeRunFinalizations.leaseOwner, "native-finalizer:%"), notInArray(nativeRunFinalizations.phase, TERMINAL_PHASES), + sql`${nativeRunFinalizations.leaseExpiresAt} > now()`), + row => nrfTokenLeaseOwner(row, "native_finalizer_lease")), + settle: (db, record, context) => settleNativeTokenLease("native_finalizer_lease", db, record, context), + }, + { + ...nrfKind("native_cleanup_lease", "Cleanup lease", "workspace_preservation", + and(like(nativeRunFinalizations.leaseOwner, "native-cleanup:%"), sql`${nativeRunFinalizations.leaseExpiresAt} > now()`), + row => nrfTokenLeaseOwner(row, "native_cleanup_lease")), + settle: (db, record, context) => settleNativeTokenLease("native_cleanup_lease", db, record, context), + }, runKind("native_ownership_unverified_hold", "Ownership-unverified hold", "ownership", and(eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "running"), or( and(eq(heartbeatRuns.errorCode, NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE), eq(heartbeatRuns.nativePhase, "terminal_failure")),