diff --git a/server/src/__tests__/environment-run-lease-ownership.test.ts b/server/src/__tests__/environment-run-lease-ownership.test.ts new file mode 100644 index 00000000000..32d517d41aa --- /dev/null +++ b/server/src/__tests__/environment-run-lease-ownership.test.ts @@ -0,0 +1,288 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agents, + companies, + completionContracts, + createDb, + environmentLeases, + environments, + heartbeatRuns, + issues, + nativeRunFinalizations, + nativeRunResults, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { + currentControllerBootId, + currentControllerProcess, + currentControllerProcessStamp, + currentProcessSpaceHash, +} from "../services/controller-process-identity.js"; +import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease-controller.js"; +import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js"; +import { settleEnvironmentRunLease } from "../services/environment-run-lease-settlement.js"; +import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.js"; +import { environmentService } from "../services/environments.js"; +import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; +import { + createOwnershipReportState, + runOwnershipReconciliationPass, + settleOwnableKind, +} from "../services/recovery/ownership-invariant-report.js"; + +const support = await getEmbeddedPostgresTestSupport(); +const suite = support.supported ? describe : describe.skip; + +/** A PID in this PID space whose process has exited. */ +async function exitedPid(): Promise { + const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" }); + const pid = child.pid!; + await new Promise(resolve => { child.once("exit", () => resolve()); child.kill("SIGKILL"); }); + return pid; +} + +suite("environment_run_lease_active on the shared ownership path", () => { + let db!: ReturnType; + let temporary: Awaited> | null = null; + let companyId = ""; + let agentId = ""; + let sandboxEnvironmentId = ""; + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("paperclip-environment-run-lease-ownership-"); + db = createDb(temporary.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + sandboxEnvironmentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Paperclip", issuePrefix: "ERL", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Coder", role: "engineer", status: "idle", + adapterType: "codex_local", adapterConfig: {}, runtimeConfig: {}, permissions: {} }); + await db.insert(environments).values({ id: sandboxEnvironmentId, name: "Fake Sandbox", driver: "sandbox", + status: "active", config: { provider: "fake", image: "ubuntu:24.04" } }); + }, 30_000); + + afterEach(async () => { + vi.unstubAllEnvs(); + await db.delete(environmentLeases); + await db.delete(heartbeatRuns); + }); + + afterAll(async () => { + await temporary?.cleanup(); + }); + + async function run(status: string, extra: Partial = {}) { + const id = randomUUID(); + await db.insert(heartbeatRuns).values({ id, companyId, agentId, invocationSource: "on_demand", status, + nextEventSeq: 1, ...extra }); + return id; + } + + /** The state a crash between the terminal run write and the lease release + * leaves behind: a terminal run whose lease is still active. */ + async function lease(runId: string, holder: Record | null, sandbox = false) { + const id = randomUUID(); + await db.insert(environmentLeases).values({ id, companyId, heartbeatRunId: runId, status: "active", + leasePolicy: "ephemeral", + ...(sandbox ? { environmentId: sandboxEnvironmentId, provider: "fake", providerLeaseId: `sandbox://fake/${id}` } : {}), + metadata: holder ? { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: holder } : {} }); + return id; + } + + async function row(id: string) { + const [found] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)); + return found!; + } + + async function deadHolder() { + return { bootId: "earlier-boot", pid: await exitedPid(), hostHash: currentProcessSpaceHash(), processStartedAt: null }; + } + + it("stamps the lease holder in the write that makes a lease active, with no extra I/O", async () => { + const runId = await run("running"); + const acquired = await environmentService(db).acquireLease({ companyId, environmentId: sandboxEnvironmentId, + heartbeatRunId: runId, provider: "fake", providerLeaseId: "sandbox://fake/stamped", metadata: { driver: "sandbox" } }); + expect(acquired.metadata?.[ENVIRONMENT_LEASE_CONTROLLER_KEY]).toEqual(currentControllerProcessStamp()); + expect(acquired.metadata?.driver).toBe("sandbox"); + }); + + it("settles a lease whose holder died after the run went terminal, on the next pass", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, await deadHolder()); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({ + settled: 1, reasons: { pid_dead: 1 } }); + expect(await row(leaseId)).toMatchObject({ status: "pending_cleanup", failureReason: "orphaned_active_lease_recovered" }); + }); + + it("stops and keeps a dead holder's sandbox instead of destroying it", async () => { + const runId = await run("succeeded"); + const leaseId = await lease(runId, await deadHolder(), true); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + const after = await row(leaseId); + expect(after).toMatchObject({ status: "pending_cleanup", failureReason: "sandbox_stop_pending" }); + expect(after.metadata?.sandboxStopAndRetain).toMatchObject({ leaseId, runId, builtinProvider: "fake", + recoveredOrphanedRunLease: true, ownerProof: "pid_dead" }); + }); + + it("stops a recovered orphan at once, even when the lease was used moments ago; a failed stop backs off", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, await deadHolder(), true); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(await nextEnvironmentCleanupAt(db)).toBeLessThanOrEqual(Date.now()); + const stop = vi.fn(async () => { throw new Error("provider unavailable"); }); + const heartbeat = heartbeatService(db, { environmentRuntime: { + retryPendingSandboxTeardown: stop } as unknown as HeartbeatEnvironmentRuntime }); + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60_000 }); + expect(stop).toHaveBeenCalledOnce(); + // The failed first attempt counted; the next one waits out the backoff. + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 5 * 60_000 }); + expect(stop).toHaveBeenCalledOnce(); + expect((await row(leaseId)).status).toBe("pending_cleanup"); + }); + + it("reads one bounded page per recovery call, so kept leases never block the leases behind them", async () => { + const foreign = { bootId: "other", pid: 1, hostHash: "another-pid-space", processStartedAt: null }; + const keptRuns = await Promise.all(Array.from({ length: 100 }, () => run("failed"))); + await db.insert(environmentLeases).values(keptRuns.map((runId, index) => ({ + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, companyId, heartbeatRunId: runId, + status: "active", metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: foreign } }))); + const orphanId = "ffffffff-0000-4000-8000-000000000000"; + await db.insert(environmentLeases).values({ id: orphanId, companyId, heartbeatRunId: await run("failed"), status: "active", + metadata: { [ENVIRONMENT_LEASE_CONTROLLER_KEY]: await deadHolder() } }); + const heartbeat = heartbeatService(db); + expect(await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: 0 }); + expect((await row(orphanId)).status).toBe("active"); + expect(await heartbeat.sweepOrphanedActiveLeases({ backoffMs: 0 })).toEqual({ recovered: 1 }); + expect((await row(orphanId)).status).toBe("pending_cleanup"); + }); + + it("waits for native settlement only while the lease's environment exists to release it", async () => { + const issueId = randomUUID(); + await db.insert(issues).values({ id: issueId, companyId, title: "Native settlement" }); + const contractId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "result-worker-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "result-worker-v1", objective: "Finish", criteria: [] }, + canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" }); + const native = { runtimeMode: "native", nativeIssueId: issueId, issueId, completionContractId: contractId }; + const kept = await run("succeeded", native); + const orphaned = await run("succeeded", native); + const goneEnvironmentId = randomUUID(); + await db.insert(environments).values({ id: goneEnvironmentId, name: "Deleted", driver: "sandbox", status: "active", + config: { provider: "fake", image: "ubuntu:24.04" } }); + for (const runId of [kept, orphaned]) { + const resultId = randomUUID(); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: "fixture", schemaStatus: "accepted", resultJson: {}, canonicalSha256: "a".repeat(64) }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "committed", resultId }); + } + const keptLease = await lease(kept, await deadHolder(), true); + const orphanedLease = await lease(orphaned, await deadHolder(), true); + await db.update(environmentLeases).set({ environmentId: goneEnvironmentId }).where(eq(environmentLeases.id, orphanedLease)); + await db.delete(environments).where(eq(environments.id, goneEnvironmentId)); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect((await row(keptLease)).status).toBe("active"); + expect((await row(orphanedLease)).status).toBe("pending_cleanup"); + await db.delete(nativeRunFinalizations); + await db.delete(nativeRunResults); + await db.delete(heartbeatRuns); + await db.delete(completionContracts); + await db.delete(issues).where(eq(issues.id, issueId)); + }); + + it("keeps a lease while its run is active, whatever its holder", async () => { + const runId = await run("running"); + const leaseId = await lease(runId, await deadHolder()); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + await settleOwnableKind(db, "environment_run_lease_active", { currentProcessIdle: true, skipRunIds: new Set() }); + expect((await row(leaseId)).status).toBe("active"); + }); + + it("keeps a lease whose holder is alive: another live PID, or another PID space", async () => { + const runId = await run("failed"); + const self = await currentControllerProcess(); + const alive = await lease(runId, { ...self, bootId: "another-boot" }); + const foreign = await lease(runId, { bootId: "other", pid: 1, hostHash: "another-pid-space", processStartedAt: null }); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({ + settled: 0, reasons: { pid_alive: 1, different_process_space: 1 } }); + expect((await row(alive)).status).toBe("active"); + expect((await row(foreign)).status).toBe("active"); + }); + + it("leaves a current-process lease to this process: settled only when it is not executing the run", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, currentControllerProcessStamp()); + // The periodic pass cannot see this process's executions. + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect((await row(leaseId)).status).toBe("active"); + // The executor is still in its finally: it releases the lease itself. + expect(await settleOwnableKind(db, "environment_run_lease_active", { runIds: [runId], + skipRunIds: new Set([runId]), currentProcessIdle: true })).toEqual([]); + expect((await row(leaseId)).status).toBe("active"); + // Without the idle claim, the current process stays alive. + expect(await settleOwnableKind(db, "environment_run_lease_active", { runIds: [runId] })).toEqual([]); + const settled = await settleOwnableKind(db, "environment_run_lease_active", { runIds: [runId], + skipRunIds: new Set(), currentProcessIdle: true }); + expect(settled).toEqual([{ kind: "environment_run_lease_active", recordId: leaseId, runId, reason: "current_process_idle" }]); + expect((await row(leaseId)).status).toBe("pending_cleanup"); + }); + + it("never settles on a read-only pass", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, await deadHolder()); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false }); + expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({ + settled: 0, verdicts: { gone: 1 } }); + expect((await row(leaseId)).status).toBe("active"); + }); + + it("does not settle a lease re-stamped by a new holder after the proof", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, await deadHolder()); + const original = (await row(leaseId)).metadata!; + const [record] = await OWNABLE_KINDS.find(kind => kind.id === "environment_run_lease_active")!.listForRuns!(db, [runId]); + await db.update(environmentLeases).set({ metadata: { ...original, + [ENVIRONMENT_LEASE_CONTROLLER_KEY]: currentControllerProcessStamp() } }).where(eq(environmentLeases.id, leaseId)); + expect(await settleEnvironmentRunLease(db, record!, { now: new Date(), proof: { verdict: "gone", reason: "pid_dead" } })) + .toEqual({ settled: false, reason: "owner_changed_since_proof" }); + expect((await row(leaseId)).status).toBe("active"); + }); + + describe("leases saved before holders were stamped", () => { + it("keeps the run fact alone for one release on a single-host server", async () => { + const runId = await run("failed"); + const leaseId = await lease(runId, null); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect((await row(leaseId)).status).toBe("pending_cleanup"); + }); + + it("never releases one whose run this boot owns (a hot-restart adoption may be releasing it)", async () => { + const runId = await run("failed", { controllerBootId: currentControllerBootId() }); + const leaseId = await lease(runId, null); + await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect((await row(leaseId)).status).toBe("active"); + // This process knows it is not executing the run: the local path settles it. + expect(await settleOwnableKind(db, "environment_run_lease_active", { runIds: [runId], + skipRunIds: new Set([randomUUID()]), currentProcessIdle: true })).toHaveLength(1); + expect((await row(leaseId)).status).toBe("pending_cleanup"); + }); + + it("names no owner on an ephemeral server, where the host cannot be known", async () => { + vi.stubEnv("PAPERCLIP_EPHEMERAL_SERVER", "true"); + const runId = await run("failed"); + const leaseId = await lease(runId, null); + const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(report.kinds.find(kind => kind.kind === "environment_run_lease_active")).toMatchObject({ + settled: 0, verdicts: { unprovable: 1 }, reasons: { no_owner_stamp: 1 } }); + expect(report.violations.some(violation => violation.recordId === leaseId && violation.rule === "I1_owned")).toBe(true); + expect((await row(leaseId)).status).toBe("active"); + }); + }); +}); diff --git a/server/src/__tests__/heartbeat-orphaned-active-lease-sweep.test.ts b/server/src/__tests__/heartbeat-orphaned-active-lease-sweep.test.ts index 5d2fdb4e334..e28dd459ea7 100644 --- a/server/src/__tests__/heartbeat-orphaned-active-lease-sweep.test.ts +++ b/server/src/__tests__/heartbeat-orphaned-active-lease-sweep.test.ts @@ -191,7 +191,9 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { return (await db.update(environmentLeases).set({ status: "expired", cleanupStatus: "success" }) .where(eq(environmentLeases.id, leaseId)).returning())[0]; }); - const runtime = { destroyRunLease } as unknown as HeartbeatEnvironmentRuntime; + const retryPendingSandboxTeardown = vi.fn(async ({ lease }: { lease: { providerLeaseId: string | null } }) => + ({ providerLeaseId: lease.providerLeaseId, state: "stopped" })); + const runtime = { destroyRunLease, retryPendingSandboxTeardown } as unknown as HeartbeatEnvironmentRuntime; try { const restarted = heartbeatService(db, { environmentRuntime: runtime }); await restarted.sweepOrphanedActiveLeases({ backoffMs: 0 }); @@ -206,8 +208,17 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { const afterExport = heartbeatService(db, { environmentRuntime: runtime }); await afterExport.sweepOrphanedActiveLeases({ backoffMs: 0 }); await afterExport.sweepPendingCleanupLeases({ backoffMs: 0 }); - expect(destroyRunLease).toHaveBeenCalledOnce(); - await expect(readFile(output)).rejects.toMatchObject({ code: "ENOENT" }); + if (status === "active") { + // The ownership settle stops the orphaned sandbox and keeps it; it + // never destroys it (workspace preservation decides that). + expect(destroyRunLease).not.toHaveBeenCalled(); + expect(retryPendingSandboxTeardown).toHaveBeenCalledOnce(); + expect(await readFile(output, "utf8")).toBe(nonce); + expect(await leaseRow(leaseId)).toMatchObject({ status: "released", cleanupStatus: "success" }); + } else { + expect(destroyRunLease).toHaveBeenCalledOnce(); + await expect(readFile(output)).rejects.toMatchObject({ code: "ENOENT" }); + } } finally { await rm(sandbox, { recursive: true, force: true }); } @@ -400,13 +411,13 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { }); it("test_defers_a_guarded_lease_instead_of_leaving_it_at_the_front_of_the_page", async () => { - const { companyId, agentId, environmentId } = await seedCompanyAgentAndEnvironment(); - const runId = await insertHeartbeatRun({ companyId, agentId, status: "failed" }); + const { companyId, environmentId } = await seedCompanyAgentAndEnvironment(); const sharedProviderLeaseId = "sandbox://fake/shared-resource"; + // A run-less (login/probe) lease still pages by idle age. const orphanedLeaseId = await insertActiveLease({ companyId, environmentId, - heartbeatRunId: runId, + heartbeatRunId: null, updatedAt: oldEnough(), providerLeaseId: sharedProviderLeaseId, }); @@ -520,9 +531,12 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { .then((rows) => rows[0] ?? null); return row ? { ...row, status: "expired" as const } : null; }); + const retryPendingSandboxTeardown = vi.fn(async ({ lease }: { lease: { providerLeaseId: string | null } }) => + ({ providerLeaseId: lease.providerLeaseId, state: "stopped" })); const heartbeat = heartbeatService(db, { environmentRuntime: { destroyRunLease, + retryPendingSandboxTeardown, } as unknown as HeartbeatEnvironmentRuntime, }); @@ -531,9 +545,12 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { // pending_cleanup teardown both run in this one call. await heartbeat.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }); - expect(destroyRunLease).toHaveBeenCalledTimes(1); + // The orphaned run's sandbox is stopped and kept, never destroyed. + expect(destroyRunLease).not.toHaveBeenCalled(); + expect(retryPendingSandboxTeardown).toHaveBeenCalledTimes(1); const row = await leaseRow(leaseId); - expect(row?.status).toBe("expired"); + expect(row?.status).toBe("released"); + expect(row?.metadata?.sandboxStopAndRetainReceipt).toMatchObject({ leaseId, method: "builtin.stopLease" }); }); it("test_stops_the_recovered_sandbox_on_the_startup_call", async () => { @@ -556,9 +573,12 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { .then((rows) => rows[0] ?? null); return row ? { ...row, status: "expired" as const } : null; }); + const retryPendingSandboxTeardown = vi.fn(async ({ lease }: { lease: { providerLeaseId: string | null } }) => + ({ providerLeaseId: lease.providerLeaseId, state: "stopped" })); const heartbeat = heartbeatService(db, { environmentRuntime: { destroyRunLease, + retryPendingSandboxTeardown, } as unknown as HeartbeatEnvironmentRuntime, }); @@ -567,9 +587,12 @@ describeEmbeddedPostgres("heartbeat sweepOrphanedActiveLeases", () => { // and the recovered lease reaches the provider teardown in this one call. await heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 }); - expect(destroyRunLease).toHaveBeenCalledTimes(1); + // The orphaned run's sandbox is stopped and kept, never destroyed. + expect(destroyRunLease).not.toHaveBeenCalled(); + expect(retryPendingSandboxTeardown).toHaveBeenCalledTimes(1); const row = await leaseRow(leaseId); - expect(row?.status).toBe("expired"); + expect(row?.status).toBe("released"); + expect(row?.metadata?.sandboxStopAndRetainReceipt).toMatchObject({ leaseId, method: "builtin.stopLease" }); }); it("test_logs_a_distinct_error_kind_and_no_exception_field_when_the_sweep_fails", async () => { diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 1d26ce393a7..3e5c761c7dc 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -140,8 +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", "native_finalizer_lease", "native_cleanup_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", "environment_run_lease_active", "wake_request_claimed"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); @@ -150,7 +150,7 @@ describe("report-only ownership reconciliation", () => { const before = await snapshot(); const state = createOwnershipReportState(); - const first = await runOwnershipReconciliationPass(db, { state }); + const first = await runOwnershipReconciliationPass(db, { state, settle: false }); expect(kindReport(first, "native_executor_claim")).toMatchObject({ inProgress: 1, verdicts: { alive: 1, gone: 0, unprovable: 0 }, reasons: { current_process: 1 } }); @@ -178,13 +178,13 @@ describe("report-only ownership reconciliation", () => { "I7_scope", "record_cap", "wake_coverage"]); expect(first.awaitingSecondProof).toBe(1); - const second = await runOwnershipReconciliationPass(db, { state }); + const second = await runOwnershipReconciliationPass(db, { state, settle: false }); expect(rules(second, "environment_run_lease_active")).toEqual(["I2_settled"]); expect(second.newViolations.map(violation => `${violation.rule}:${violation.kind}`)).toEqual(["I2_settled:environment_run_lease_active"]); // A stuck record already reported for I2 needs no further pass. expect(second.awaitingSecondProof).toBe(0); - expect((await runOwnershipReconciliationPass(db, { state })).awaitingSecondProof).toBe(0); + expect((await runOwnershipReconciliationPass(db, { state, settle: false })).awaitingSecondProof).toBe(0); // Report-only: nothing was settled, released or retried. expect(await snapshot()).toEqual(before); diff --git a/server/src/services/environment-cleanup-state.ts b/server/src/services/environment-cleanup-state.ts index 444bc090600..4c6211f6006 100644 --- a/server/src/services/environment-cleanup-state.ts +++ b/server/src/services/environment-cleanup-state.ts @@ -16,6 +16,17 @@ export function recoverableSandboxCreationCondition() { )); } +/** A run lease the ownership settle moved to cleanup, before its first + * cleanup attempt. Its holder is proven gone, so the first stop or teardown is + * due now regardless of how recently the lease was used; a failed attempt + * counts and the normal retry backoff applies from then on. */ +export function recoveredOrphanFirstAttemptSql() { + return sql`(${environmentLeases.failureReason} = 'orphaned_active_lease_recovered' + or ${environmentLeases.metadata}->'sandboxStopAndRetain'->>'recoveredOrphanedRunLease' = 'true') + and coalesce(${environmentLeases.metadata}->>'pendingCleanupRetryAttempts', '0') = '0' + and coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`; +} + export function pendingCleanupDeadlineSql(now = Date.now()) { const deadline = sql`case when ${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true' @@ -29,9 +40,10 @@ export function pendingCleanupDeadlineSql(now = Date.now()) { } export async function nextEnvironmentCleanupAt(db: Db, backoffMs = ENVIRONMENT_CLEANUP_RETRY_MS): Promise { - const [row] = await db.select({ next: sql`min(greatest( + const [row] = await db.select({ next: sql`min(case when ${recoveredOrphanFirstAttemptSql()} + then ${pendingCleanupDeadlineSql()} else greatest( extract(epoch from ${environmentLeases.updatedAt}) * 1000 + ${backoffMs}, - ${pendingCleanupDeadlineSql()}))` }).from(environmentLeases).where(eq(environmentLeases.status, "pending_cleanup")); + ${pendingCleanupDeadlineSql()}) end)` }).from(environmentLeases).where(eq(environmentLeases.status, "pending_cleanup")); const [creation] = await db.select({ next: sql`min(extract(epoch from ${sandboxCreationOperations.nextAttemptAt}) * 1000)` }) .from(sandboxCreationOperations).where(recoverableSandboxCreationCondition()); const deadlines = [row?.next, creation?.next].filter((value): value is string => value != null).map(Number); diff --git a/server/src/services/environment-lease-controller.ts b/server/src/services/environment-lease-controller.ts new file mode 100644 index 00000000000..5cda7d383bc --- /dev/null +++ b/server/src/services/environment-lease-controller.ts @@ -0,0 +1,25 @@ +import { currentControllerProcessStamp, readControllerProcess, type ControllerProcess } from "./controller-process-identity.js"; + +/** + * The server process that holds an environment lease (plan section A). + * + * A run lease is owned by its run while the run is active. Once the run is + * terminal, the release is still owed by the process that executes the run: + * it releases the lease from its own `finally`, with the run's resource + * disposition. This stamp names that process, so the shared ownership prover + * can tell a lease whose holder is releasing it from one whose holder died + * between the terminal run write and the release. + * + * The stamp is written in the same write that makes the lease active and does + * no I/O (`currentControllerProcessStamp`). + */ +export const ENVIRONMENT_LEASE_CONTROLLER_KEY = "leaseController"; + +export function withEnvironmentLeaseController(metadata: Record | null): Record { + return { ...(metadata ?? {}), [ENVIRONMENT_LEASE_CONTROLLER_KEY]: currentControllerProcessStamp() }; +} + +export function readEnvironmentLeaseController(metadata: unknown): ControllerProcess | null { + if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) return null; + return readControllerProcess((metadata as Record)[ENVIRONMENT_LEASE_CONTROLLER_KEY]); +} diff --git a/server/src/services/environment-run-lease-settlement.ts b/server/src/services/environment-run-lease-settlement.ts new file mode 100644 index 00000000000..0c1596bebc7 --- /dev/null +++ b/server/src/services/environment-run-lease-settlement.ts @@ -0,0 +1,153 @@ +import { and, eq, inArray, ne, notInArray, or, sql } from "drizzle-orm"; +import { environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { currentControllerBootId } from "./controller-process-identity.js"; +import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import { readEnvironmentLeaseController } from "./environment-lease-controller.js"; +import { allowLegacyShutdownWorkspaceCleanup } from "./legacy-shutdown-workspace-settlement.js"; +import { nativeWorkspaceExportPendingForLease } from "./native-runtime/native-workspace-finalization-state.js"; +import type { OwnerProof } from "./owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js"; +import { isBuiltinSandboxProvider } from "./sandbox-provider-runtime.js"; +import { prepareSandboxStopAndRetain } from "./sandbox-stop-and-retain.js"; + +/** Failure reason written on a run lease the shared ownership path settled. */ +export const ORPHANED_RUN_LEASE_FAILURE_REASON = "orphaned_active_lease_recovered"; + +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; +const TERMINAL_NATIVE_PHASES = ["committed", "terminal_failure", "applied"]; + +type LeaseRow = typeof environmentLeases.$inferSelect; + +/** + * Settle of the `environment_run_lease_active` ownership kind (plan Phase 2). + * + * A run's executor writes the terminal run status and then releases the run's + * environment lease from its `finally`. A crash between the two writes, or a + * release that skipped the lease, leaves the lease `active` with no holder. + * Previously a sweep released such a lease once it had been idle for five + * minutes. The shared reconciler now calls this only after the lease holder + * (the run while it is active, then the stamped lease holder) is proven gone. + * + * The settle compares and swaps on the exact lease row the proof judged: the + * same run, status `active`, and the same metadata (which holds the holder + * stamp). It settles ownership only: + * - a sandbox is stopped and kept (the stop-and-retain intent), never + * destroyed: deciding whether its files are still needed is the workspace + * preservation obligation, not this one; + * - a lease without a remote sandbox moves to `pending_cleanup`, as before; + * - a lease still needed by another obligation is left alone: a pending native + * workspace export, a legacy shutdown fence, another live lease on the same + * provider resource, or an accepted native result whose settlement can + * release the lease (its environment still exists) with the run's own + * resource disposition. + */ +export async function settleEnvironmentRunLease(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (!record.runId) return { settled: false, reason: "lease_has_no_run" }; + const runId = record.runId; + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [lease] = await tx.select().from(environmentLeases) + .where(and(eq(environmentLeases.id, record.recordId), eq(environmentLeases.status, "active"), + eq(environmentLeases.heartbeatRunId, runId))) + .for("update"); + if (!lease) return { settled: false, reason: "lease_no_longer_active" }; + if (!holderUnchanged(lease, record)) return { settled: false, reason: "owner_changed_since_proof" }; + const [run] = await tx.select({ status: heartbeatRuns.status, controllerBootId: heartbeatRuns.controllerBootId }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + if (run && !TERMINAL_RUN_STATUSES.includes(run.status)) return { settled: false, reason: "owner_run_active" }; + if (record.owner.kind === "run" && !context.proof.localExecutionsKnown + && await currentBootOwnsRun(tx, lease, run?.controllerBootId ?? null)) { + // A lease saved before holders were stamped: this boot may be releasing + // it right now (a hot-restart adoption), so only an earlier boot's + // lease is released by the run fact alone, unless the caller named every + // run this process executes. + return { settled: false, reason: "current_boot_owns_run" }; + } + const kept = await keptByAnotherObligation(tx, lease); + if (kept) return { settled: false, reason: kept }; + + const recovery = { recoveredOrphanedRunLease: true, ownerProof: context.proof.reason }; + if (await stopsASandbox(tx, lease)) { + const stopped = await prepareSandboxStopAndRetain(tx, lease, recovery); + return stopped ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_active" }; + } + // Keep the row's updatedAt, so the pending_cleanup sweep accepts it in the + // same tick instead of waiting out its own retry backoff. + const flipped = await writeEnvironmentCleanup(tx, write => write.update(environmentLeases) + .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON }) + .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.status, "active"), + eq(environmentLeases.heartbeatRunId, runId), + sql`${environmentLeases.metadata} is not distinct from ${lease.metadata === null ? null : JSON.stringify(lease.metadata)}::jsonb`)) + .returning({ id: environmentLeases.id }), { companyId: lease.companyId, runId, leaseId: lease.id }); + return flipped.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "lease_no_longer_active" }; + }); +} + +/** The proof judged a holder stamp (or its absence, for a legacy row); a lease + * re-stamped since then belongs to someone else. */ +function holderUnchanged(lease: LeaseRow, record: InProgressRecord): boolean { + const holder = readEnvironmentLeaseController(lease.metadata); + if (record.owner.kind === "run") return holder === null; + if (record.owner.kind !== "process") return false; + const stamp = record.owner.stamp; + return holder !== null && holder.bootId === stamp.bootId && holder.pid === stamp.pid + && holder.hostHash === stamp.processSpace && holder.processStartedAt === (stamp.processStartedAt ?? null); +} + +async function currentBootOwnsRun(tx: Db, lease: LeaseRow, runControllerBootId: string | null): Promise { + const bootId = currentControllerBootId(); + if (runControllerBootId === bootId) return true; + const [native] = await tx.select({ controllerBootId: nativeRunFinalizations.controllerBootId }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.runId, lease.heartbeatRunId!), + eq(nativeRunFinalizations.companyId, lease.companyId))); + return native?.controllerBootId === bootId; +} + +async function keptByAnotherObligation(tx: Db, lease: LeaseRow): Promise { + if (await nativeWorkspaceExportPendingForLease(tx, lease)) return "workspace_export_pending"; + if (!(await allowLegacyShutdownWorkspaceCleanup(tx, lease))) return "legacy_shutdown_fence"; + // One provider resource id names one physical sandbox; another live lease + // on it still owns that sandbox. + if (lease.provider && lease.providerLeaseId) { + const [other] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider), + eq(environmentLeases.providerLeaseId, lease.providerLeaseId), + inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"]))).limit(1); + if (other) return "provider_resource_shared"; + } + // An accepted native result is finalized and then releases this lease with + // the run's resource disposition (native environment settlement). + const [accepted] = await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), + eq(nativeRunResults.schemaStatus, "accepted"))) + .where(and(eq(nativeRunFinalizations.runId, lease.heartbeatRunId!), eq(nativeRunFinalizations.companyId, lease.companyId), + or(notInArray(nativeRunFinalizations.phase, TERMINAL_NATIVE_PHASES), eq(nativeRunFinalizations.phase, "committed"), + and(eq(nativeRunFinalizations.phase, "terminal_failure"), + sql`${nativeRunFinalizations.failureDetail} ? 'workspaceFinalizeAttempt'`)))).limit(1); + // Native settlement releases through the lease's environment; with the + // environment row gone it skips the lease, so this settle must not wait. + if (accepted && lease.environmentId) { + const [environment] = await tx.select({ id: environments.id }).from(environments) + .where(eq(environments.id, lease.environmentId)); + if (environment) return "native_result_settles_lease"; + } + return null; +} + +/** A sandbox lease whose stop can be dispatched from the recorded lease alone + * (the same pins the stop-only cleanup requires). */ +async function stopsASandbox(tx: Db, lease: LeaseRow): Promise { + if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId) return false; + let driver = typeof lease.metadata?.driver === "string" ? lease.metadata.driver : null; + if (!driver && lease.environmentId) { + const [environment] = await tx.select({ driver: environments.driver }).from(environments) + .where(eq(environments.id, lease.environmentId)); + driver = environment?.driver ?? null; + } + if ((driver ?? "sandbox") !== "sandbox") return false; + const plugin = lease.metadata?.pluginId; + if (isBuiltinSandboxProvider(lease.provider)) return plugin == null && !lease.metadata?.sandboxProviderPlugin; + return typeof plugin === "string" && plugin.length > 0; +} diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 3e50b257a80..ba7c75b7391 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,3 +1,4 @@ +import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "./environment-lease-controller.js"; import { RetainedSandboxAbsentError, resumeReplyConfirmsAbsence } from "./retained-sandbox-absent-error.js"; import { readNativeWorkspaceMigration } from "./native-runtime/native-workspace-migration.js"; import { ReusableSandboxResumeError } from "./reusable-sandbox-resume-error.js"; @@ -3203,6 +3204,8 @@ const INTERNAL_PLUGIN_SANDBOX_CONFIG_KEYS = new Set([ "sandboxLeaseAcquisition", "nativeHarnessBackup", "nativeWorkspaceSync", + // The host process that holds the lease (ownership proof); never provider config. + ENVIRONMENT_LEASE_CONTROLLER_KEY, ]); // Drop the host-internal and per-lease runtime keys from a sandbox config diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index d3588319726..295921d2cbb 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -6,6 +6,7 @@ import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notific import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { notifyDeferredExecutionWork, notifyDeferredWakesForIssue } from "./deferred-execution-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; +import { withEnvironmentLeaseController } from "./environment-lease-controller.js"; import { randomUUID } from "node:crypto"; import { SANDBOX_PLATFORM_PROBE_METADATA_KEY } from "./environment-custom-image-runtime.js"; @@ -1568,7 +1569,8 @@ export function environmentService(db: Db) { releasedAt: null, failureReason: null, cleanupStatus: null, - metadata: withoutReusableLeaseHandoff(withoutCleanupVerification(input.metadata)), + // The holder stamp is written with the row that makes the lease active. + metadata: withEnvironmentLeaseController(withoutReusableLeaseHandoff(withoutCleanupVerification(input.metadata))), createdAt: now, updatedAt: now, }; @@ -1660,7 +1662,8 @@ export function environmentService(db: Db) { expiresAt: input.expiresAt ?? null, failureReason: null, cleanupStatus: null, - metadata: preserveReusableLeaseHandoff(previous?.metadata ?? null, withoutCleanupVerification(input.metadata)), + metadata: withEnvironmentLeaseController( + preserveReusableLeaseHandoff(previous?.metadata ?? null, withoutCleanupVerification(input.metadata))), updatedAt: now, }) .where( diff --git a/server/src/services/heartbeat/live-recovery-work.ts b/server/src/services/heartbeat/live-recovery-work.ts index 58dd3f5f8de..775ccb57139 100644 --- a/server/src/services/heartbeat/live-recovery-work.ts +++ b/server/src/services/heartbeat/live-recovery-work.ts @@ -57,7 +57,10 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType>'{legacyShutdownWorkspaceSettlement,state}' = 'pending'`, )).orderBy(asc(environmentLeases.id)).limit(100); rows = candidates.map(row => ({ id: row.id, companyId: row.companyId, runId: row.runId, - dueAt: kind === "orphanLease" ? row.updatedAt.getTime() + 5 * 60_000 + dueAt: kind === "orphanLease" + // A run lease is judged by its holder proof, not by age. A run-less + // (login/probe) lease keeps its idle backoff until its kind moves. + ? row.runId ? row.updatedAt.getTime() : row.updatedAt.getTime() + 5 * 60_000 : Date.parse(String((row.metadata?.legacyShutdownWorkspaceSettlement as Record)?.deadlineAt)) || Date.now(), })); } else { diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 5858b719e0a..d141b7082ea 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,4 +1,5 @@ import { settleOwnableKind } from "../recovery/ownership-invariant-report.js"; +import { ORPHANED_RUN_LEASE_FAILURE_REASON } from "../environment-run-lease-settlement.js"; import { readNativeStartupCleanup, readNativeStartupContext } from "../native-runtime/native-startup-cleanup.js"; import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js"; @@ -11,7 +12,7 @@ import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-wor import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-native-session-cleanup-work.js"; import { reconcileRetainedNativeSessionCleanups } from "../native-runtime/retained-native-session-cleanup.js"; import { createScheduledRetryCancellationRecovery } from "./scheduled-retry-cancellation.js"; -import { pendingCleanupDeadlineSql } from "../environment-cleanup-state.js"; +import { pendingCleanupDeadlineSql, recoveredOrphanFirstAttemptSql } from "../environment-cleanup-state.js"; import { controllerProcessExited, currentControllerBootId, currentControllerProcess, readControllerProcess } from "../controller-process-identity.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js"; @@ -50,6 +51,7 @@ import { eq, gt, inArray, + isNotNull, isNull, lte, ne, @@ -206,6 +208,9 @@ const PENDING_CLEANUP_CAP_WARNED_METADATA_KEY = "pendingCleanupRetryCapWarned"; // The reaper sweeps at most this many orphaned active leases per tick. const ORPHANED_ACTIVE_LEASE_SWEEP_PAGE_SIZE = 20; +// One recovery call settles at most this many terminal runs' active leases. +const ORPHANED_RUN_LEASE_PAGE_SIZE = 100; + // A provider or plugin destroy rejection can carry a bearer credential, a // signed URL, or provider response detail in its name, code, message, cause, or // stack. The exception fields cross the server boundary, so they are not a @@ -1608,6 +1613,9 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // it. The bump pushes the row past the fixed-size page, so the next tick // reaches the rows behind it. It costs the row one extra backoff wait, // which is safe because the guard means a physical sandbox still exists. + // Where the next unscoped orphaned run-lease page starts, per company scope. + const orphanRunLeaseCursors = new Map(); + async function deferOrphanedActiveLease(leaseId: string): Promise { const now = new Date(); await db @@ -1621,57 +1629,86 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ); } - // An active lease is reachable only while its heartbeat run keeps the - // running status. The reaper writes the run status and the lease release as - // two separate statements, so a restart between them can leave a terminal - // run and an active lease. `releaseRunLeases` also skips a lease whose - // environment row is gone, so that lease stays active too. No later query - // finds either lease, because every production query selects an active - // lease by its environment or by its heartbeat run id, never by age. This - // sweep finds both stranded classes and moves each lease to pending_cleanup, - // so the existing pending_cleanup sweep tears the sandbox down from the - // data already on the lease row. + // A run's active lease outlives a crash between the terminal run write and + // the lease release, or a release that skipped the lease (for example when + // its environment row is gone). No production query finds such a lease by + // its run again. The shared ownership path settles it once its holder is + // proven gone: the stamped lease holder, or this process when it is not + // executing the run (`environment-run-lease-settlement.ts`). Proof replaces + // the earlier "run terminal plus five minutes" release. + // + // Workspace preservation is a separate obligation and runs first: a lease + // whose native workspace export is still pending is preserved, never + // released, while its run is not executing here. + // + // A login or probe lease has no run. It keeps the idle backoff below until + // its kind moves to the shared path. async function sweepOrphanedActiveLeases(opts: { + /** Idle time before a run-less (login/probe) lease is recovered. */ backoffMs: number; companyId?: string; leaseId?: string; onDeferredDeadline?: (at: number) => void; }): Promise<{ recovered: number }> { - const cutoff = new Date(Date.now() - opts.backoffMs); + const localRuns = { has: (runId: string) => activeRunExecutions.has(runId) + || adapterExecutionControls.has(runId) || runningProcesses.has(runId) }; + let recovered = 0; - const rows = await db + // One bounded page per call. An unscoped call continues after the last + // page it read, so leases kept by a live holder or another obligation + // never block the leases behind them. + const cursorKey = opts.leaseId ? null : opts.companyId ?? "*"; + const after = cursorKey === null ? undefined : orphanRunLeaseCursors.get(cursorKey); + const scoped = await db .select({ lease: environmentLeases }) .from(environmentLeases) - .leftJoin( - heartbeatRuns, - eq(environmentLeases.heartbeatRunId, heartbeatRuns.id), - ) - .where( - and( - eq(environmentLeases.status, "active"), - opts.companyId ? eq(environmentLeases.companyId, opts.companyId) : undefined, - opts.leaseId ? eq(environmentLeases.id, opts.leaseId) : undefined, - or( - isNull(environmentLeases.heartbeatRunId), - inArray(heartbeatRuns.status, [ - ...HEARTBEAT_RUN_TERMINAL_STATUSES, - ]), - ), - lte(environmentLeases.updatedAt, cutoff), - ), - ) + .leftJoin(heartbeatRuns, eq(environmentLeases.heartbeatRunId, heartbeatRuns.id)) + .where(and( + eq(environmentLeases.status, "active"), + opts.companyId ? eq(environmentLeases.companyId, opts.companyId) : undefined, + opts.leaseId ? eq(environmentLeases.id, opts.leaseId) : undefined, + after ? gt(environmentLeases.id, after) : undefined, + isNotNull(environmentLeases.heartbeatRunId), + or(isNull(heartbeatRuns.id), inArray(heartbeatRuns.status, [...HEARTBEAT_RUN_TERMINAL_STATUSES])), + )) + .orderBy(asc(environmentLeases.id)) + .limit(ORPHANED_RUN_LEASE_PAGE_SIZE); + if (cursorKey !== null) { + if (scoped.length < ORPHANED_RUN_LEASE_PAGE_SIZE) orphanRunLeaseCursors.delete(cursorKey); + else orphanRunLeaseCursors.set(cursorKey, scoped[scoped.length - 1]!.lease.id); + } + for (const { lease } of scoped) { + if (lease.heartbeatRunId && !localRuns.has(lease.heartbeatRunId) + && await nativeWorkspaceExportPendingForLease(db, lease) + && await preserveOrphanedNativeWorkspace(db, lease, { onDeferredDeadline: opts.onDeferredDeadline })) recovered += 1; + } + + const runIds = [...new Set(scoped.flatMap(({ lease }) => lease.heartbeatRunId ? [lease.heartbeatRunId] : []))]; + if (runIds.length > 0) { + const settled = await settleOwnableKind(db, "environment_run_lease_active", { + runIds, skipRunIds: localRuns, currentProcessIdle: true, + }); + recovered += settled.filter(record => scoped.some(({ lease }) => lease.id === record.recordId)).length; + // A lease kept for a live holder needs no timer here: the holder releases + // it, and the ownership settle watch re-proves holders that might die + // without a signal. A lease kept by another obligation waits for that + // obligation's own signal. + } + + const cutoff = new Date(Date.now() - opts.backoffMs); + const runless = await db + .select({ lease: environmentLeases }) + .from(environmentLeases) + .where(and( + eq(environmentLeases.status, "active"), + opts.companyId ? eq(environmentLeases.companyId, opts.companyId) : undefined, + opts.leaseId ? eq(environmentLeases.id, opts.leaseId) : undefined, + isNull(environmentLeases.heartbeatRunId), + lte(environmentLeases.updatedAt, cutoff), + )) .orderBy(asc(environmentLeases.updatedAt)) .limit(ORPHANED_ACTIVE_LEASE_SWEEP_PAGE_SIZE); - - let recovered = 0; - for (const { lease } of rows) { - if (await nativeWorkspaceExportPendingForLease(db, lease)) { - if (lease.heartbeatRunId && !activeRunExecutions.has(lease.heartbeatRunId) - && !adapterExecutionControls.has(lease.heartbeatRunId) - && await preserveOrphanedNativeWorkspace(db, lease, { onDeferredDeadline: opts.onDeferredDeadline })) recovered += 1; - if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); - continue; - } + for (const { lease } of runless) { if (!(await allowLegacyShutdownWorkspaceCleanup(db, lease))) { if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); continue; @@ -1683,46 +1720,26 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD const [otherOwner] = await db .select({ id: environmentLeases.id }) .from(environmentLeases) - .where( - and( - ne(environmentLeases.id, lease.id), - eq(environmentLeases.provider, lease.provider), - eq(environmentLeases.providerLeaseId, lease.providerLeaseId), - inArray(environmentLeases.status, [ - "active", - "retained", - "pending_cleanup", - ]), - ), - ) + .where(and( + ne(environmentLeases.id, lease.id), + eq(environmentLeases.provider, lease.provider), + eq(environmentLeases.providerLeaseId, lease.providerLeaseId), + inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"]), + )) .limit(1); if (otherOwner) { - // Defer this row so the fixed-size page reaches the rows behind - // it next tick, instead of re-selecting the same guarded rows - // forever. + // Defer this row so the fixed-size page reaches the rows behind it. if (!opts.leaseId) await deferOrphanedActiveLease(lease.id); continue; } } - - // Keep the row's existing updatedAt value. The select above already - // proved the row is older than the backoff cutoff, so the - // pending_cleanup sweep can accept the same row in this same tick. A - // fresh timestamp here would push the row inside that sweep's own - // backoff window and delay the teardown by one full tick. + // Keep the row's existing updatedAt value, so the pending_cleanup sweep + // can accept the same row in this same tick. const flipped = await writeEnvironmentCleanup(db, tx => tx .update(environmentLeases) - .set({ - status: "pending_cleanup", - failureReason: "orphaned_active_lease_recovered", - }) - .where( - and( - eq(environmentLeases.id, lease.id), - eq(environmentLeases.status, "active"), - ), - ) - .returning({ id: environmentLeases.id }), { companyId: lease.companyId, runId: lease.heartbeatRunId ?? undefined, leaseId: lease.id }); + .set({ status: "pending_cleanup", failureReason: ORPHANED_RUN_LEASE_FAILURE_REASON }) + .where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.status, "active"))) + .returning({ id: environmentLeases.id }), { companyId: lease.companyId, leaseId: lease.id }); if (flipped.length > 0) recovered += 1; } @@ -1792,7 +1809,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD opts?.explicitRetry ? eq(environmentLeases.heartbeatRunId, opts.explicitRetry.runId) : undefined, opts?.environmentRetry ? eq(environmentLeases.environmentId, opts.environmentRetry.environmentId) : undefined, pendingCleanupRetryDueSql(manualRetry), - backoffMs > 0 ? lte(environmentLeases.updatedAt, cutoff) : undefined, + backoffMs > 0 ? or(lte(environmentLeases.updatedAt, cutoff), recoveredOrphanFirstAttemptSql()) : undefined, ), ) .orderBy(asc(environmentLeases.updatedAt)) diff --git a/server/src/services/owner-proof.ts b/server/src/services/owner-proof.ts index d118e751fb4..f190bdc9eda 100644 --- a/server/src/services/owner-proof.ts +++ b/server/src/services/owner-proof.ts @@ -74,6 +74,9 @@ export type OwnerProof = { reason: OwnerProofReason; /** The local proof judged a row saved before its PID space was recorded. */ legacyHostFallback?: true; + /** The caller named every run this process executes, so a record of another + * run is not held here (`settleOwnableKind` with `currentProcessIdle`). */ + localExecutionsKnown?: true; }; export type OwnerProofReason = @@ -85,6 +88,10 @@ export type OwnerProofReason = | "provider_stopped" | "deployment_replaced" | "operator_attested" + /** This process owns the record and is not executing its run. Only a + * caller that knows this process's executions may conclude it + * (`settleOwnableKind` with `currentProcessIdle`); the prover never does. */ + | "current_process_idle" // alive | "current_process" | "pid_alive" diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts index 490e7bcb918..74b293430de 100644 --- a/server/src/services/recovery/ownership-invariant-report.ts +++ b/server/src/services/recovery/ownership-invariant-report.ts @@ -140,7 +140,7 @@ const INTENT_LOOKBACK_MS = 60 * 60_000; * PostgreSQL case-insensitive regular expression. */ const TEMPORARY_REFUSAL_PATTERN = "budget|paus|lock|busy|unavailable|retry|admission|hold|drain|restart|capacity|rate.?limit|timeout"; -type Classification = { verdict: OwnerProof["verdict"]; reason: string; legacyHostFallback?: true }; +type Classification = { verdict: OwnerProof["verdict"]; reason: string; legacyHostFallback?: true; localExecutionsKnown?: true }; async function classifyRunOwners(db: Db, records: InProgressRecord[]): Promise> { const runIds = [...new Set(records.flatMap(record => record.owner.kind === "run" ? [record.owner.runId] : []))]; @@ -198,6 +198,11 @@ export type SettledRecord = { kind: string; recordId: string; runId: string | nu */ export async function settleOwnableKind(db: Db, kindId: string, options: { now?: Date; runIds?: readonly string[]; skipRunIds?: { has(runId: string): boolean }; probe?: ProcessProbe; + /** The caller's `skipRunIds` names every run this process is executing. A + * run record this process owns whose run is not among them has no holder + * here, so its owner is gone (`current_process_idle`). The periodic pass + * cannot know this process's executions and never concludes it. */ + currentProcessIdle?: boolean; } = {}): Promise { return (await settleKind(db, kindId, options)).settled; } @@ -221,6 +226,7 @@ export async function recheckSettlingKinds(db: Db, options: { now?: Date; probe? async function settleKind(db: Db, kindId: string, options: { now?: Date; runIds?: readonly string[]; skipRunIds?: { has(runId: string): boolean }; probe?: ProcessProbe; + currentProcessIdle?: boolean; }, explicitKind?: OwnableKind): Promise<{ settled: SettledRecord[]; watched: number }> { const kind = explicitKind ?? OWNABLE_KINDS.find(candidate => candidate.id === kindId); if (!kind?.listInProgress || !kind.settle) throw new Error(`ownership kind ${kindId} has no settle path`); @@ -246,7 +252,11 @@ async function settleKind(db: Db, kindId: string, options: { const settled: SettledRecord[] = []; let watched = 0; for (const record of unique) { - const proof = await classify(record.owner, runProofs, options.probe, now); + let proof = await classify(record.owner, runProofs, options.probe, now); + if (options.currentProcessIdle && record.runId && proof.verdict === "alive" && proof.reason === "current_process") { + proof = { verdict: "gone", reason: "current_process_idle" }; + } + if (options.currentProcessIdle) proof = { ...proof, localExecutionsKnown: true }; const outcome = await settleOrSurface(db, kind, record, proof, now); if (outcome === "settled") { settled.push({ kind: kind.id, recordId: record.recordId, runId: record.runId, reason: proof.reason }); diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 8374222b023..e0fbe89ce7d 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -38,6 +38,9 @@ import { settleNativeTokenLease, type NativeTokenLeaseKind, } from "../native-runtime/native-token-lease-settlement.js"; +import { ephemeralServerEnabled } from "../ephemeral-server-mode.js"; +import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../environment-lease-controller.js"; +import { settleEnvironmentRunLease } from "../environment-run-lease-settlement.js"; /** * Registry of ownable record kinds (plan section C). @@ -266,6 +269,56 @@ function leaseKind(id: string, title: string, obligation: Obligation, where: SQL }; } +const TERMINAL_RUN_STATUSES = ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]; + +type RunLeaseRow = LeaseRow & { runStatus: string | null }; + +/** + * Owner of an active run lease. While the run is active, the run owns it. + * Once the run is terminal, the release is owed by the process that holds the + * lease (`metadata.leaseController`, stamped when the lease became active). + * A lease saved before that stamp existed keeps, for one release, the run + * fact alone on a single-host server (the settle still refuses a run this + * boot owns); an ephemeral server cannot tell which host holds it, so it + * names no owner and stays visible as an I1 gap. + */ +function runLeaseOwner(row: RunLeaseRow): RecordOwner { + if (row.runStatus !== null && !TERMINAL_RUN_STATUSES.includes(row.runStatus)) return runOwner(row.heartbeatRunId); + const holder = controllerStamp(json(row.metadata)[ENVIRONMENT_LEASE_CONTROLLER_KEY], "environment_run_lease_active"); + if (holder.kind !== "none") return holder; + return ephemeralServerEnabled() ? { kind: "none" } : runOwner(row.heartbeatRunId); +} + +/** Shared path since Phase 2: replaces the "run terminal plus five minutes" + * orphaned active lease sweep. */ +function runLeaseKind(): OwnableKind { + const where = and(eq(environmentLeases.status, "active"), isNotNull(environmentLeases.heartbeatRunId)); + const select = (db: Db) => db.select({ id: environmentLeases.id, companyId: environmentLeases.companyId, + heartbeatRunId: environmentLeases.heartbeatRunId, issueId: environmentLeases.issueId, + metadata: environmentLeases.metadata, updatedAt: environmentLeases.updatedAt, runStatus: heartbeatRuns.status }) + .from(environmentLeases).leftJoin(heartbeatRuns, eq(heartbeatRuns.id, environmentLeases.heartbeatRunId)); + const record = (row: RunLeaseRow): InProgressRecord => ({ recordId: row.id, companyId: row.companyId, + runId: row.heartbeatRunId, issueId: row.issueId, owner: runLeaseOwner(row), updatedAt: row.updatedAt }); + return { + id: "environment_run_lease_active", title: "Run lease (active)", obligation: "ownership", storage: "database", + invariantChecks: [1, 2], settle: settleEnvironmentRunLease, + listInProgress: async (db, limit, offset) => { + const rows = await select(db).where(where).orderBy(asc(environmentLeases.id)).limit(limit + 1).offset(offset); + const result = page(rows, limit); + return { truncated: result.truncated, records: result.rows.map(record) }; + }, + listForRuns: async (db, runIds) => { + const records: InProgressRecord[] = []; + for (let index = 0; index < runIds.length; index += 500) { + const rows = await select(db).where(and(where, + inArray(environmentLeases.heartbeatRunId, [...runIds.slice(index, index + 500)]))); + records.push(...rows.map(record)); + } + return records; + }, + }; +} + /** `ControllerProcess` metadata ({bootId, pid, hostHash, processStartedAt}). */ function controllerStamp(value: unknown, scope: string): RecordOwner { const controller = readControllerProcess(value); @@ -498,9 +551,7 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ })) }; }, }, - leaseKind("environment_run_lease_active", "Run lease (active)", "ownership", - and(eq(environmentLeases.status, "active"), isNotNull(environmentLeases.heartbeatRunId)), - row => runOwner(row.heartbeatRunId)), + runLeaseKind(), leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership", and(eq(environmentLeases.status, "pending_cleanup"), sql`(${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean is true`), row => controllerStamp(json(row.metadata).pendingCleanupController, "environment_teardown_claim")), diff --git a/server/src/services/sandbox-stop-and-retain.ts b/server/src/services/sandbox-stop-and-retain.ts index 0bacfbe699d..c17c6a10965 100644 --- a/server/src/services/sandbox-stop-and-retain.ts +++ b/server/src/services/sandbox-stop-and-retain.ts @@ -74,9 +74,11 @@ export function readStopOnlyCleanup(lease: Lease) { } /** Persist before provider dispatch so an older worker or a restart cannot turn - * an explicit stop into ordinary destructive release. Invalid pins stay pending. */ -export async function prepareSandboxStopAndRetain(db: Db, lease: Lease) { - return prepareStopAndRetain(db, lease, "active"); + * an explicit stop into ordinary destructive release. Invalid pins stay pending. + * `recovery` marks an ownership settle on the intent; the pending-cleanup + * sweep dispatches such a stop's first attempt without its retry backoff. */ +export async function prepareSandboxStopAndRetain(db: Db, lease: Lease, recovery?: Record) { + return prepareStopAndRetain(db, lease, "active", recovery); } /** Historical release rows can lack a stop receipt. Re-inspect the exact