diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 3c455240488..731b9d044bc 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -613,6 +613,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0432_strong_mole_man.sql", "0435_premium_puff_adder.sql", "0436_legal_reavers.sql", + "0447_legacy_controller_process.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0447_legacy_controller_process.sql b/packages/db/src/migrations/0447_legacy_controller_process.sql new file mode 100644 index 00000000000..7222b093614 --- /dev/null +++ b/packages/db/src/migrations/0447_legacy_controller_process.sql @@ -0,0 +1,2 @@ +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "controller_process" jsonb;--> statement-breakpoint +CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_legacy_controller_lease_idx" ON "heartbeat_runs" USING btree ("controller_lease_expires_at") WHERE "heartbeat_runs"."runtime_mode" = 'legacy' AND "heartbeat_runs"."status" = 'running' AND "heartbeat_runs"."controller_boot_id" is not null; diff --git a/packages/db/src/migrations/meta/0441_snapshot.json b/packages/db/src/migrations/meta/0447_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0441_snapshot.json rename to packages/db/src/migrations/meta/0447_snapshot.json index 6e0bbf97914..4e82c37d4bb 100644 --- a/packages/db/src/migrations/meta/0441_snapshot.json +++ b/packages/db/src/migrations/meta/0447_snapshot.json @@ -1,6 +1,6 @@ { - "id": "a2ec96c6-3cf1-4c51-8ee8-b35f090bfd42", - "prevId": "0c990bdc-6df0-44ca-8f21-28124f7b39e7", + "id": "76f06e29-53a8-4ecc-809a-d52b6bbb6520", + "prevId": "782688f1-7843-4a93-8ea3-2cec362d269e", "version": "7", "dialect": "postgresql", "tables": { @@ -34856,6 +34856,12 @@ "primaryKey": false, "notNull": false }, + "controller_process": { + "name": "controller_process", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, "execution_stage": { "name": "execution_stage", "type": "text", @@ -34880,6 +34886,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", @@ -35119,6 +35131,22 @@ "method": "btree", "with": {} }, + "heartbeat_runs_legacy_controller_lease_idx": { + "name": "heartbeat_runs_legacy_controller_lease_idx", + "columns": [ + { + "expression": "controller_lease_expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'legacy' AND \"heartbeat_runs\".\"status\" = 'running' AND \"heartbeat_runs\".\"controller_boot_id\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_cost_accounting_pending_idx": { "name": "heartbeat_runs_cost_accounting_pending_idx", "columns": [ @@ -44760,6 +44788,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", @@ -44924,6 +44958,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 b2011fce23e..5ee6e128f06 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -3116,6 +3116,13 @@ "when": 1794105010331, "tag": "0446_native_lease_holder", "breakpoints": true + }, + { + "idx": 447, + "version": "7", + "when": 1794105389611, + "tag": "0447_legacy_controller_process", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 239d0cad70e..34cb7475ee4 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -92,6 +92,11 @@ export const heartbeatRuns = pgTable( // Legacy controller lease. A PID alone is not an identity across containers. controllerBootId: uuid("controller_boot_id"), controllerLeaseExpiresAt: timestamp("controller_lease_expires_at", { withTimezone: true }), + // Server process that claimed the legacy controller lease ({bootId, pid, + // hostHash, processStartedAt}). It names the controller only while its + // bootId equals controller_boot_id; revocation swaps the boot id, so a + // revoked lease never keeps it. Null on rows saved before it existed. + controllerProcess: jsonb("controller_process").$type>(), executionStage: text("execution_stage"), processPid: integer("process_pid"), processGroupId: integer("process_group_id"), @@ -151,6 +156,11 @@ export const heartbeatRuns = pgTable( treeHoldSuppressionIdx: index("heartbeat_runs_tree_hold_suppression_idx") .on(table.companyId, sql`(${table.resultJson}->'treeHoldSuppression'->>'holdId')`) .where(sql`${table.status} = 'cancelled' and ${table.resultJson} ? 'treeHoldSuppression'`), + // The `legacy_controller_lease` ownership kind (reconciler listing and + // settle watch): outstanding legacy controller leases of running runs. + legacyControllerLeaseIdx: index("heartbeat_runs_legacy_controller_lease_idx") + .on(table.controllerLeaseExpiresAt) + .where(sql`${table.runtimeMode} = 'legacy' AND ${table.status} = 'running' AND ${table.controllerBootId} is not null`), costAccountingPendingIdx: index("heartbeat_runs_cost_accounting_pending_idx").on(table.updatedAt, table.id).where(sql`${table.costAccountingPending} = true`), costDataQualityIdx: index("heartbeat_runs_cost_data_quality_idx") .on(table.companyId, sql`coalesce(${table.finishedAt}, ${table.createdAt}) desc`, table.id) diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 3e5c761c7dc..0e76b9eceb3 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -140,8 +140,9 @@ 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", "native_finalizer_lease", "native_cleanup_lease", "environment_run_lease_active", "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", "environment_run_lease_active", + "wake_request_claimed", "legacy_controller_lease"]); 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 d141b7082ea..91751d4c5b6 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -32,6 +32,7 @@ import { pendingCaseHandoffForRun } from "./case-handoff-lifecycle.js"; import { retireInheritedRemoteRunnerIdentity } from "../native-runtime/native-inherited-runner-retry.js"; import { legacyControllerBootId, + legacyControllerProvenGone, liveLegacyControllerExpiresAt, revokeExpiredLegacyControllerWithState, } from "../legacy-controller-lease.js"; @@ -2590,8 +2591,15 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD scope.onDeferredDeadline?.(nativeControllerLeaseExpiresAt.getTime()); continue; } + // The shared prover (Phase 2): a controller of another boot whose + // recorded process is proven gone holds no lease and needs no age guard. + const legacyControllerGone = run.runtimeMode === "legacy" && run.controllerBootId !== legacyControllerBootId + && await legacyControllerProvenGone(run); const legacyLease = await liveLegacyControllerExpiresAt(db, run); - if (legacyLease) { scope.onDeferredDeadline?.(legacyLease.getTime()); continue; } + if (legacyLease && !(legacyControllerGone + && (await settleOwnableKind(db, "legacy_controller_lease", { runIds: [run.id] })).length > 0)) { + scope.onDeferredDeadline?.(legacyLease.getTime()); continue; + } // A legacy controller proves it is alive only by renewing its lease, and // the check above found no live lease (a missing expiry also lands here; @@ -2609,7 +2617,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD !run.processPid && !run.processGroupId; // Apply staleness threshold to avoid false positives - if (staleThresholdMs > 0 && !priorBootSetupExpired) { + if (staleThresholdMs > 0 && !priorBootSetupExpired && !legacyControllerGone) { const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; if (now.getTime() - refTime < staleThresholdMs) { scope.onDeferredDeadline?.(refTime + staleThresholdMs); @@ -2680,10 +2688,19 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD (!monitorNextCheckAt || monitorNextCheckAt.getTime() <= now.getTime()); const revokedState = await revokeExpiredLegacyControllerWithState(db, run); if (!revokedState) continue; + // The scan snapshot can predate a child the old controller saved just + // before it died. The child checks above used that snapshot, so a proven + // gone controller is never enough here: when revocation reports a child + // the snapshot did not, re-check on the next pass from the fresh row. + if (legacyControllerGone && ((revokedState.processPid ?? null) !== (run.processPid ?? null) + || (revokedState.processGroupId ?? null) !== (run.processGroupId ?? null))) { + scope.onDeferredDeadline?.(Date.now() + 1_000); + continue; + } // The earlier snapshot may predate a final renewal that admitted the // adapter. Revocation leaves a fresh foreign lease, so a later pass // applies the age guard to that dispatched run. - if (priorBootSetupExpired && staleThresholdMs > 0 && (revokedState.executionStage !== "preparing" + if (priorBootSetupExpired && !legacyControllerGone && staleThresholdMs > 0 && (revokedState.executionStage !== "preparing" || revokedState.processPid || revokedState.processGroupId)) { const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; scope.onDeferredDeadline?.(Math.max(refTime + staleThresholdMs, Date.now() + 60_000)); diff --git a/server/src/services/legacy-controller-lease.test.ts b/server/src/services/legacy-controller-lease.test.ts index 46d76b61915..283b345ff50 100644 --- a/server/src/services/legacy-controller-lease.test.ts +++ b/server/src/services/legacy-controller-lease.test.ts @@ -1,5 +1,5 @@ import { randomUUID } from "node:crypto"; -import { eq, sql } from "drizzle-orm"; +import { and, eq, ne, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, closeLivenessDb, companies, createDb, heartbeatRuns, livenessDb, subscribeDatabaseWork } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; @@ -8,8 +8,24 @@ import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; import { heartbeatService } from "./heartbeat.js"; import { hasLiveLegacyController, LEGACY_CONTROLLER_LEASE_MS, LEGACY_CONTROLLER_RENEW_MS, LEGACY_CONTROLLER_RENEWAL_RETRY_DELAY_MS, legacyControllerBootId, legacyControllerClaim, LegacyControllerLeaseLostError, - renewLegacyControllerLease, revokeExpiredLegacyController, revokeExpiredLegacyControllerWithState, - watchLegacyControllerLease } from "./legacy-controller-lease.js"; + legacyControllerOwnerStamp, renewLegacyControllerLease, revokeExpiredLegacyController, + revokeExpiredLegacyControllerWithState, watchLegacyControllerLease } from "./legacy-controller-lease.js"; +import { currentProcessSpaceHash, currentProcessStartedAtMs } from "./controller-process-identity.js"; +import { proveOwnerGone } from "./owner-proof.js"; +import * as legacyControllerLease from "./legacy-controller-lease.js"; + +/** Runs just before recovery's revocation, to model a controller that saved + * its child's PID after recovery's scan and then died. Null in other cases. */ +const beforeRevocation: { hook: null | ((runId: string) => Promise) } = { hook: null }; +vi.mock("./legacy-controller-lease.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, revokeExpiredLegacyControllerWithState: async (...args: Parameters) => { + if (beforeRevocation.hook) await beforeRevocation.hook(args[1].id); + return actual.revokeExpiredLegacyControllerWithState(...args); + } }; +}); +import { createOwnershipReportState, runOwnershipReconciliationPass } from "./recovery/ownership-invariant-report.js"; +import { OWNABLE_KINDS } from "./recovery/ownership-registry.js"; type WatchArgs = Parameters; // A renewal test double: each attempt runs in a transaction on the liveness pool. @@ -249,6 +265,119 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu it("leaves native controller ownership to the native coordinator", () => { expect(legacyControllerClaim("native")).toEqual({}); }); + + // Phase 2: the claim records its controller process, and a controller of + // another boot proven gone by the shared prover holds no lease and needs no + // age guard. Proof is by identity, never by lease age. + it("stamps the claiming process, bound to the lease's boot id", async () => { + const run = await seed(); + expect(run.controllerProcess).toMatchObject({ bootId: legacyControllerBootId, pid: process.pid, + hostHash: currentProcessSpaceHash() }); + expect(legacyControllerOwnerStamp(run)).toMatchObject({ bootId: legacyControllerBootId, pid: process.pid }); + // Revocation swaps the boot id, so the revoked lease names nobody. + expect(legacyControllerOwnerStamp({ ...run, controllerBootId: randomUUID() })).toBeNull(); + }); + + type Controller = { bootId: string; pid: number; hostHash: string; processStartedAt: string | null }; + const deadController = (hostHash = currentProcessSpaceHash()): Controller => ({ bootId: randomUUID(), + pid: 2_000_000_001, hostHash, processStartedAt: new Date(Date.now() - 60_000).toISOString() }); + /** A run whose adapter was dispatched by `controller`, with its lease still live. */ + async function dispatchedBy(controller: Controller | null) { + const run = await seed(); + await db.update(heartbeatRuns).set({ controllerBootId: controller?.bootId ?? randomUUID(), + controllerProcess: controller, executionStage: "dispatching" }).where(eq(heartbeatRuns.id, run.id)); + return run.id; + } + const saved = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + /** The startup reaper with the default age guard, over this case's run only: + * earlier cases leave running rows in the shared database. */ + async function reapOnly(id: string) { + await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }) + .where(and(eq(heartbeatRuns.status, "running"), ne(heartbeatRuns.id, id))); + await heartbeatService(db).reapOrphanedRuns({ staleThresholdMs: 5 * 60_000 }); + } + + it("settles a hard-killed controller's run at once, before its lease expires and without the age guard", async () => { + // Crash point: the controller claimed the run, dispatched the adapter, and + // was killed. The next boot proves it gone and does not wait 60 s plus 5 min. + const id = await dispatchedBy(deadController()); + await reapOnly(id); + expect(await saved(id)).toMatchObject({ status: "failed", errorCode: "process_lost" }); + }); + + it("never fails a run whose dead controller saved a live child after recovery's scan", async () => { + // Race: the scan read the run before the old controller saved its child's + // PID and died. The controller is proven gone, but its child still runs. + const id = await dispatchedBy(deadController()); + beforeRevocation.hook = async runId => { + await db.update(heartbeatRuns).set({ processPid: process.pid, processHostHash: currentProcessSpaceHash(), + processStartedAt: new Date((await currentProcessStartedAtMs()) ?? Date.now()) }).where(eq(heartbeatRuns.id, runId)); + }; + try { + await reapOnly(id); + } finally { beforeRevocation.hook = null; } + expect((await saved(id)).status).not.toBe("failed"); + }); + + it("settles a dead controller's expired lease without waiting out the age guard", async () => { + const id = await dispatchedBy(deadController()); + await expire(id); + await reapOnly(id); + expect(await saved(id)).toMatchObject({ status: "failed", errorCode: "process_lost" }); + }); + + it.each([ + { label: "a live earlier-boot controller", controller: async (): Promise => ({ bootId: randomUUID(), + pid: process.pid, hostHash: currentProcessSpaceHash(), + processStartedAt: new Date((await currentProcessStartedAtMs())!).toISOString() }) }, + { label: "a dead PID in another PID space", controller: async () => deadController("another-pid-space") }, + { label: "a lease saved by the previous release", controller: async () => null }, + ])("keeps the run and lease of $label", async ({ controller }) => { + const id = await dispatchedBy(await controller()); + const before = await saved(id); + await reapOnly(id); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + const after = await saved(id); + expect({ status: after.status, errorCode: after.errorCode, expiresAt: after.controllerLeaseExpiresAt }) + .toEqual({ status: "running", errorCode: null, expiresAt: before.controllerLeaseExpiresAt }); + }); + + it("keeps an expired lease of a live earlier-boot controller behind the age guard", async () => { + const id = await dispatchedBy({ bootId: randomUUID(), pid: process.pid, hostHash: currentProcessSpaceHash(), + processStartedAt: new Date((await currentProcessStartedAtMs())!).toISOString() }); + await expire(id); + await reapOnly(id); + expect(await saved(id)).toMatchObject({ status: "running", errorCode: null }); + }); + + it("settles a dead controller's lease in the reconciler pass, never in a read-only pass", async () => { + const id = await dispatchedBy(deadController()); + const kind = OWNABLE_KINDS.find(candidate => candidate.id === "legacy_controller_lease")!; + const [record] = await kind.listForRuns!(db, [id]); + expect(record!.owner.kind === "process" && await proveOwnerGone(record!.owner.stamp)) + .toEqual({ verdict: "gone", reason: "pid_dead" }); + const before = await saved(id); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect((await saved(id)).controllerLeaseExpiresAt).toEqual(before.controllerLeaseExpiresAt); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + const after = await saved(id); + expect(after.controllerLeaseExpiresAt!.getTime()).toBeLessThanOrEqual(Date.now()); + // Ownership only: live-run recovery, woken by the settle, settles the run. + expect(after.status).toBe("running"); + expect(await kind.listForRuns!(db, [id])).toEqual([]); + }); + + it("settles only the exact controller it proved", async () => { + const id = await dispatchedBy(deadController()); + const kind = OWNABLE_KINDS.find(candidate => candidate.id === "legacy_controller_lease")!; + const [record] = await kind.listForRuns!(db, [id]); + const before = await saved(id); + await db.update(heartbeatRuns).set({ controllerProcess: { ...before.controllerProcess, pid: 2_000_000_002 } }) + .where(eq(heartbeatRuns.id, id)); + expect(await kind.settle!(db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } })) + .toEqual({ settled: false, reason: "owner_changed_since_proof" }); + expect((await saved(id)).controllerLeaseExpiresAt).toEqual(before.controllerLeaseExpiresAt); + }); }); describe("legacy controller lease loss attribution", () => { diff --git a/server/src/services/legacy-controller-lease.ts b/server/src/services/legacy-controller-lease.ts index ae7656429b9..20ee6002d43 100644 --- a/server/src/services/legacy-controller-lease.ts +++ b/server/src/services/legacy-controller-lease.ts @@ -3,7 +3,13 @@ import { and, eq, gt, lte, sql } from "drizzle-orm"; import { heartbeatRuns, livenessDb, type Db } from "@paperclipai/db"; import type { LegacyControllerLeaseLossCause, LegacyControllerLeaseLossTiming } from "./run-cancellation.js"; import { randomUUID } from "node:crypto"; -import { currentControllerBootId } from "./controller-process-identity.js"; +import { + currentControllerBootId, + currentControllerProcessStamp, + readControllerProcess, +} from "./controller-process-identity.js"; +import { proveOwnerGone, type LocalOwnerStamp, type OwnerProof } from "./owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js"; // A boot UUID has meaning across containers; a numeric PID does not. Legacy // leases stamp the single process boot id shared with native claims. @@ -34,6 +40,9 @@ export function legacyControllerClaim(runtimeMode: string) { return { legacyProcessSettlement: null, controllerBootId: legacyControllerBootId, + // This process, so a hard-killed controller is proven gone on the shared + // ownership path (no I/O: the claim is on the dispatch path). + controllerProcess: currentControllerProcessStamp(), controllerLeaseExpiresAt: sql`clock_timestamp() + interval '60 seconds'`, executionStage: "preparing", }; @@ -142,6 +151,74 @@ export async function revokeExpiredLegacyControllerWithState( return revoked ?? null; } +/** + * The recorded controller process of a legacy lease (plan Phase 2), or null. + * `controller_process` names the controller only while its boot id equals + * `controller_boot_id`: revocation swaps in a random boot id, so a revoked + * lease never keeps it. Rows saved by the previous release have none; their + * lease stays a boot-id fence (expiry, revocation and the age guard). + */ +export function legacyControllerOwnerStamp( + run: Pick, +): LocalOwnerStamp | null { + const controller = readControllerProcess(run.controllerProcess); + if (!controller || !run.controllerBootId || controller.bootId !== run.controllerBootId) return null; + return { kind: "local", scope: "legacy_controller_lease", generation: null, host: null, + processSpace: controller.hostHash, bootId: controller.bootId, pid: controller.pid, + processStartedAt: controller.processStartedAt, processGroupId: null }; +} + +/** True only when the shared prover proves the recorded legacy controller gone. */ +export async function legacyControllerProvenGone( + run: Pick, +): Promise { + if (run.runtimeMode === "native") return false; + const stamp = legacyControllerOwnerStamp(run); + return stamp !== null && (await proveOwnerGone(stamp)).verdict === "gone"; +} + +/** + * Settle of the `legacy_controller_lease` ownership kind. The shared + * reconciler calls it only after `proveOwnerGone` proved the recorded + * controller gone. It expires exactly that controller's outstanding lease + * (compare-and-swap on run, boot id and recorded pid) after waking live-run + * recovery, which then revokes it and settles the run as a lost process under + * its own checks (a live child process still keeps it running). It settles + * ownership only. + */ +export async function settleLegacyControllerLease(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process" || !record.companyId) return { settled: false, reason: "owner_not_a_process" }; + const stamp = record.owner.stamp; + if (!stamp.bootId) return { settled: false, reason: "owner_not_a_process" }; + const companyId = record.companyId; + const bootId = stamp.bootId; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [row] = await tx.select({ agentId: heartbeatRuns.agentId, controllerProcess: heartbeatRuns.controllerProcess, + controllerBootId: heartbeatRuns.controllerBootId }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, record.recordId), eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"), + eq(heartbeatRuns.controllerBootId, bootId), gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`))) + .for("update"); + if (!row) return { settled: false, reason: "lease_no_longer_held" }; + const current = legacyControllerOwnerStamp(row); + // The proof judged this exact process; the lease has moved on since. + if (!current || current.pid !== stamp.pid || current.processSpace !== stamp.processSpace + || (current.processStartedAt ?? null) !== (stamp.processStartedAt ?? null)) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + // Work intent before the write it announces (sleep-aware background work). + await notifyLiveRunRecovery(tx, { companyId, runId: record.recordId, agentId: row.agentId }); + const updated = await tx.update(heartbeatRuns).set({ controllerLeaseExpiresAt: sql`clock_timestamp()` }) + .where(and(eq(heartbeatRuns.id, record.recordId), eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.controllerBootId, bootId), + gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`))) + .returning({ id: heartbeatRuns.id }); + return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_held" }; + }); +} + export async function revokeExpiredLegacyController(db: Db, run: Run): Promise { return (await revokeExpiredLegacyControllerWithState(db, run)) !== null; } diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index c4fc7da85b0..fa76be3bbe1 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -32,6 +32,7 @@ import { settleNativeAcceptedResultLease, } from "../native-runtime/native-accepted-result-lease-settlement.js"; import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; +import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js"; import { nativeLeaseHolderStamp, readNativeLeaseHolder, @@ -208,7 +209,7 @@ function nrfTokenLeaseOwner(row: NrfRow, scope: NativeTokenLeaseKind): RecordOwn type RunRow = Pick; + | "runnerProfileJson" | "controllerProcess">; const runColumns = { id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: heartbeatRuns.issueId, status: heartbeatRuns.status, @@ -217,6 +218,7 @@ const runColumns = { processStartedAt: heartbeatRuns.processStartedAt, processHostHash: heartbeatRuns.processHostHash, updatedAt: heartbeatRuns.updatedAt, runnerProfileJson: heartbeatRuns.runnerProfileJson, + controllerProcess: heartbeatRuns.controllerProcess, }; function runKind(id: string, title: string, obligation: Obligation, where: SQL | undefined, @@ -245,6 +247,17 @@ function runProcessStamp(row: RunRow, scope: string): LocalOwnerStamp { }; } +const legacyControllerLeaseWhere = and(eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"), + isNotNull(heartbeatRuns.controllerBootId), sql`${heartbeatRuns.controllerLeaseExpiresAt} > now()`); + +function legacyControllerLeaseOwner(row: RunRow): RecordOwner { + const stamp = legacyControllerOwnerStamp(row); + if (stamp) return { kind: "process", stamp }; + return { kind: "fence", stamp: { kind: "remote", scope: "legacy_controller_lease", generation: null, + fenceToken: row.controllerBootId }, + fence: { currentFenceToken: row.controllerBootId, currentGeneration: null, leaseExpiresAt: row.controllerLeaseExpiresAt } }; +} + // ---------------------------------------------------------------------------- // environment_leases @@ -775,10 +788,27 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ })) }; }, }, - runKind("legacy_controller_lease", "Legacy controller lease", "ownership", - and(eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "running"), isNotNull(heartbeatRuns.controllerBootId)), - row => ({ kind: "process", stamp: { kind: "local", scope: "legacy_controller_lease", generation: null, host: null, - processSpace: null, bootId: row.controllerBootId, pid: null, processStartedAt: null } })), + { + // Shared path since Phase 2: the claim records its controller process + // (`controller_process`, bound to the boot id); settle expires the lease of + // a controller proven gone, and live-run recovery settles the run now + // instead of after lease expiry plus the age guard. Outstanding leases + // only. Rows saved before that column, and revoked leases, stay boot-id + // fences: alive while leased, then handled by the existing revocation. + ...runKind("legacy_controller_lease", "Legacy controller lease", "ownership", legacyControllerLeaseWhere, + legacyControllerLeaseOwner), + listForRuns: async (db, runIds) => { + const records: InProgressRecord[] = []; + for (let index = 0; index < runIds.length; index += 500) { + const rows = await db.select(runColumns).from(heartbeatRuns) + .where(and(legacyControllerLeaseWhere, inArray(heartbeatRuns.id, [...runIds.slice(index, index + 500)]))); + records.push(...(rows as RunRow[]).map(row => ({ recordId: row.id, companyId: row.companyId, runId: row.id, + issueId: row.issueId, owner: legacyControllerLeaseOwner(row), updatedAt: row.updatedAt }))); + } + return records; + }, + settle: settleLegacyControllerLease, + }, runKind("detached_process_marker", "Detached-process marker", "ownership", and(eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.errorCode, "process_detached")), row => row.processPid ? { kind: "process", stamp: runProcessStamp(row, "detached_process_marker") } : { kind: "none" },