diff --git a/server/src/__tests__/environment-lease-owner-stamps.test.ts b/server/src/__tests__/environment-lease-owner-stamps.test.ts index 7e494ec0569..d09ae153c26 100644 --- a/server/src/__tests__/environment-lease-owner-stamps.test.ts +++ b/server/src/__tests__/environment-lease-owner-stamps.test.ts @@ -1,12 +1,19 @@ import { spawn } from "node:child_process"; import { randomUUID } from "node:crypto"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, companies, createDb, environmentLeases, environments } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { currentControllerProcessStamp, currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease-controller.js"; import { environmentService } from "../services/environments.js"; -import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js"; +import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; +import { + createOwnershipReportState, + recheckSettlingKinds, + runOwnershipReconciliationPass, +} from "../services/recovery/ownership-invariant-report.js"; const support = await getEmbeddedPostgresTestSupport(); const suite = support.supported ? describe : describe.skip; @@ -50,6 +57,11 @@ suite("environment lease owner stamps (login/probe lease, teardown claim)", () = return { kind: result.kinds.find(entry => entry.kind === kind)!, violations: result.violations.filter(entry => entry.kind === kind) }; } + async function lease(id: string) { + const [found] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, id)); + return found!; + } + async function claim(metadata: Record) { const id = randomUUID(); await db.insert(environmentLeases).values({ id, companyId, environmentId, status: "pending_cleanup", @@ -107,6 +119,61 @@ suite("environment lease owner stamps (login/probe lease, teardown claim)", () = expect(kind).toMatchObject({ verdicts: { unprovable: 1 }, reasons: { fence_expired_without_provider_proof: 1 } }); }); + it("ends a dead claimant's fence on the next reconciler pass, refunding its counted attempt", async () => { + const attemptId = randomUUID(); + const leaseId = await claim({ pendingCleanupAttemptId: attemptId, pendingCleanupLeaseExpiresAtMs: Date.now() + 14 * 60_000, + pendingCleanupRetryAttempts: 3, pendingCleanupController: { bootId: "earlier-boot", pid: await exitedPid(), + hostHash: currentProcessSpaceHash(), processStartedAt: null, attemptId, countedAttempt: true } }); + // A read-only pass never settles. + await report("environment_teardown_claim"); + expect((await lease(leaseId)).metadata?.pendingCleanupLeaseExpiresAtMs).toBeGreaterThan(Date.now()); + const result = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(result.kinds.find(entry => entry.kind === "environment_teardown_claim")).toMatchObject({ settled: 1 }); + const settled = (await lease(leaseId)).metadata!; + expect(settled.pendingCleanupLeaseExpiresAtMs).toBeLessThanOrEqual(Date.now()); + expect(settled).toMatchObject({ pendingCleanupInFlight: true, pendingCleanupAttemptId: attemptId, + pendingCleanupRetryAttempts: 2, pendingCleanupController: { countedAttempt: false } }); + }); + + it("wakes the cleanup worker for the ended fence, and a malformed flag beside it never blocks the settle", async () => { + const attemptId = randomUUID(); + const malformed = await claim({ pendingCleanupInFlight: "maybe", pendingCleanupAttemptId: randomUUID(), + pendingCleanupRetryAfterMs: Date.now() + 20 * 60_000 }); + const leaseId = await claim({ pendingCleanupAttemptId: attemptId, pendingCleanupLeaseExpiresAtMs: Date.now() + 14 * 60_000, + pendingCleanupController: { bootId: "earlier-boot", pid: await exitedPid(), hostHash: currentProcessSpaceHash(), + processStartedAt: null, attemptId } }); + expect(await nextEnvironmentCleanupAt(db)).toBeGreaterThan(Date.now() + 10 * 60_000); + let wakes = 0; + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.environmentCleanup, () => { wakes += 1; }); + try { + const result = await recheckSettlingKinds(db); + expect(result.settled.map(record => record.recordId)).toEqual([leaseId]); + await vi.waitFor(() => expect(wakes).toBeGreaterThan(0)); + } finally { + unsubscribe(); + } + // The worker's next run follows the ended fence, not the old deadline. + expect(await nextEnvironmentCleanupAt(db)).toBeLessThan(Date.now() + 10 * 60_000); + expect((await lease(malformed)).metadata?.pendingCleanupInFlight).toBe("maybe"); + }); + + it("never ends the fence of a live claimant, another PID space, or a claim held as a fence", async () => { + const live = randomUUID(), foreign = randomUUID(), fenced = randomUUID(); + const expires = Date.now() + 14 * 60_000; + const ids = [ + await claim({ pendingCleanupAttemptId: live, pendingCleanupLeaseExpiresAtMs: expires, + pendingCleanupController: { ...currentControllerProcessStamp(), attemptId: live } }), + await claim({ pendingCleanupAttemptId: foreign, pendingCleanupLeaseExpiresAtMs: expires, + pendingCleanupController: { bootId: "other", pid: await exitedPid(), hostHash: "f".repeat(32), processStartedAt: null, + attemptId: foreign } }), + await claim({ pendingCleanupAttemptId: fenced, pendingCleanupLeaseExpiresAtMs: expires }), + ]; + const result = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState() }); + expect(result.kinds.find(entry => entry.kind === "environment_teardown_claim")).toMatchObject({ settled: 0, + reasons: { current_process: 1, different_process_space: 1, fence_held: 1 } }); + for (const id of ids) expect((await lease(id)).metadata?.pendingCleanupLeaseExpiresAtMs).toBe(expires); + }); + it("names no owner for an in-flight flag without an attempt", async () => { await claim({}); const { kind, violations } = await report("environment_teardown_claim"); diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index fd83dac4a70..7217e636a05 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -21,12 +21,12 @@ import { const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() })); vi.mock("../telemetry.ts", () => ({ getTelemetryClient: () => mockTelemetryClient })); -// A test can hold every caller of the exited-process proof until N callers +// A test can hold every caller of the shared owner proof until N callers // arrive, so overlapping sweeps all read a claim before any of them updates it. const exitProofBarrier = vi.hoisted(() => ({ parties: 0, waiting: [] as Array<() => void> })); -vi.mock("../services/controller-process-identity.js", async importOriginal => { - const actual = await importOriginal(); - return { ...actual, controllerProcessExited: async (...args: Parameters) => { +vi.mock("../services/owner-proof.js", async importOriginal => { + const actual = await importOriginal(); + const proveOwnerGone = async (...args: unknown[]) => { if (exitProofBarrier.parties > 1) { await new Promise(resolve => { exitProofBarrier.waiting.push(resolve); @@ -36,8 +36,9 @@ vi.mock("../services/controller-process-identity.js", async importOriginal => { } }); } - return actual.controllerProcessExited(...args); - } }; + return (actual.proveOwnerGone as (...values: unknown[]) => ReturnType)(...args); + }; + return { ...actual, proveOwnerGone }; }); vi.mock("../middleware/logger.js", () => ({ @@ -423,6 +424,8 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { heartbeatService(db, { environmentRuntime: runtime }).sweepPendingCleanupLeases(), heartbeatService(db, { environmentRuntime: runtime }).sweepPendingCleanupLeases(), ]); + // Both sweeps proved the claimant gone before either settled it. + expect(exitProofBarrier.parties).toBe(0); // One refund (3 -> 2) and one new failed attempt (2 -> 3); never a second refund. expect(teardown).toHaveBeenCalledOnce(); expect(await readAttempts(leaseId)).toBe(3); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index 3ab9a86ace6..b86a6d2848e 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -142,7 +142,7 @@ describe("report-only ownership reconciliation", () => { 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", "legacy_controller_lease"]); + "environment_teardown_claim", "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/environment-cleanup-state.ts b/server/src/services/environment-cleanup-state.ts index 4c6211f6006..c281e19360c 100644 --- a/server/src/services/environment-cleanup-state.ts +++ b/server/src/services/environment-cleanup-state.ts @@ -4,6 +4,45 @@ import { hasDeferredEnvironmentCleanup } from "./environment-cleanup-notificatio export const ENVIRONMENT_CLEANUP_RETRY_MS = 5 * 60_000; +// Escalate and slow cleanup after this many attempts; never abandon a live lease. +export const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5; + +// The reaper stores its retry attempt count under this key in the lease metadata. +export const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts"; + +// Read the stored retry attempt count as a safe value, directly in SQL. A +// provider can write a malformed value under the attempts key. The type guard +// makes any non-number value read as zero. The reader computes as numeric and +// never casts to int, so a finite number outside the 32-bit range (for example +// 1e300) never throws. The reader clamps a negative value to zero and a positive +// value to the attempt cap. One malformed lease therefore never aborts the page +// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`, +// which clamps to the same range. The claim predicate compares the two readers, +// so both must yield the same value for every input. +export function pendingCleanupAttemptsSql() { + return sql` + case + when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number' + then least( + greatest( + floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric), + 0 + ), + ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP} + ) + else 0 + end`; +} + +// Choose the `jsonb_set` target root. A provider can write a scalar or array +// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the +// stored metadata only when its root is an object. A NULL, scalar, or array root +// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also +// covers a NULL root. +export function pendingCleanupMetadataObjectSql() { + return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`; +} + export function recoverableSandboxCreationCondition() { return or(eq(sandboxCreationOperations.state, "pending"), and( eq(sandboxCreationOperations.state, "needs_reconciliation"), diff --git a/server/src/services/environment-teardown-claim-settlement.ts b/server/src/services/environment-teardown-claim-settlement.ts new file mode 100644 index 00000000000..13f99e7892f --- /dev/null +++ b/server/src/services/environment-teardown-claim-settlement.ts @@ -0,0 +1,74 @@ +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases, type Db } from "@paperclipai/db"; +import { + PENDING_CLEANUP_ATTEMPTS_METADATA_KEY, + pendingCleanupAttemptsSql, + pendingCleanupMetadataObjectSql, +} from "./environment-cleanup-state.js"; +import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import type { OwnerProof } from "./owner-proof.js"; +import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js"; + +/** + * Settle of the `environment_teardown_claim` ownership kind (plan Phase 2). + * + * A teardown claim fences other cleanup for 15 minutes while its claimant + * destroys or stops a sandbox. A claimant process that died cannot still be + * running the teardown, so once the shared prover proves the claimant stamped + * for this exact attempt gone, the fence ends now and the pending-cleanup + * sweep may claim the lease again. A dead server is not a failed teardown: a + * sweep claim that counted its attempt gets it back. + * + * The reconciler calls this only for a process owner. A claim held as a fence + * (another claimant, or another host) is never proven gone by a process proof; + * it waits for its deadline, which is the remote handoff. + * + * The update compares and swaps on the exact attempt and claimant the proof + * judged, and it is single-use: it matches only while the fence is still in + * the future and it consumes `countedAttempt`, so an overlapping settle cannot + * refund the attempt twice. It settles ownership only; whether the sandbox + * still owes a stop, a destroy or a retained workspace is unchanged. + */ +export async function settleEnvironmentTeardownClaim(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process") return { settled: false, reason: "claim_held_as_fence" }; + const stamp = record.owner.stamp; + const nowMs = context.now.getTime(); + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [lease] = await tx.select({ metadata: environmentLeases.metadata }).from(environmentLeases) + .where(and(eq(environmentLeases.id, record.recordId), eq(environmentLeases.status, "pending_cleanup"))) + .for("update"); + const metadata = lease?.metadata ?? null; + const attemptId = typeof metadata?.pendingCleanupAttemptId === "string" ? metadata.pendingCleanupAttemptId : null; + const controller = metadata?.pendingCleanupController as Record | undefined; + if (!attemptId || metadata?.pendingCleanupInFlight !== true) return { settled: false, reason: "claim_no_longer_in_flight" }; + if (controller?.attemptId !== attemptId || controller.bootId !== stamp.bootId || controller.pid !== stamp.pid + || controller.hostHash !== stamp.processSpace) { + return { settled: false, reason: "owner_changed_since_proof" }; + } + const expiresAtMs = metadata?.pendingCleanupLeaseExpiresAtMs; + if (typeof expiresAtMs !== "number" || expiresAtMs <= nowMs) return { settled: false, reason: "fence_already_ended" }; + const returnAttempt = controller?.countedAttempt === true; + // Wake the cleanup worker before the write it announces, so it reads the + // ended fence after commit instead of sleeping until the old deadline. + if (record.companyId) await notifyEnvironmentCleanup(tx, { companyId: record.companyId, leaseId: record.recordId }); + const updated = await tx.update(environmentLeases) + .set({ metadata: sql`jsonb_set(jsonb_set(jsonb_set(${pendingCleanupMetadataObjectSql()}, + '{pendingCleanupLeaseExpiresAtMs}', to_jsonb(${nowMs}::bigint)), + array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], + to_jsonb(greatest(${pendingCleanupAttemptsSql()} - ${returnAttempt ? 1 : 0}, 0)::int), true), + '{pendingCleanupController,countedAttempt}', 'false'::jsonb, true)` }) + .where(and( + eq(environmentLeases.id, record.recordId), + eq(environmentLeases.status, "pending_cleanup"), + sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`, + sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`, + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`, + sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${attemptId}`, + sql`coalesce(${environmentLeases.metadata}->'pendingCleanupController'->>'countedAttempt', 'false') = ${returnAttempt ? "true" : "false"}`, + )) + .returning({ id: environmentLeases.id }); + return updated.length ? { settled: true, reason: context.proof.reason } : { settled: false, reason: "claim_no_longer_in_flight" }; + }); +} diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 91751d4c5b6..ea4a088ea48 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -12,8 +12,15 @@ 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, recoveredOrphanFirstAttemptSql } from "../environment-cleanup-state.js"; -import { controllerProcessExited, currentControllerBootId, currentControllerProcess, readControllerProcess } from "../controller-process-identity.js"; +import { + PENDING_CLEANUP_ATTEMPTS_METADATA_KEY, + PENDING_CLEANUP_SWEEP_ATTEMPT_CAP, + pendingCleanupAttemptsSql, + pendingCleanupDeadlineSql, + pendingCleanupMetadataObjectSql, + recoveredOrphanFirstAttemptSql, +} from "../environment-cleanup-state.js"; +import { currentControllerProcess } from "../controller-process-identity.js"; import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { startNativeSessionExecutionLeaseRenewal, type NativeExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; @@ -198,12 +205,8 @@ const ENVIRONMENT_CLEANUP_RETRY_LIMIT = 1_000; const pendingCleanupAttemptsInFlight = new Set(); -// Escalate and slow cleanup after this many attempts; never abandon a live lease. -const PENDING_CLEANUP_SWEEP_ATTEMPT_CAP = 5; - -// The reaper stores its retry state under these keys in the lease metadata. -const PENDING_CLEANUP_ATTEMPTS_METADATA_KEY = "pendingCleanupRetryAttempts"; - +// The reaper stores its retry state in the lease metadata (the attempt cap and +// count key live in environment-cleanup-state.ts). const PENDING_CLEANUP_CAP_WARNED_METADATA_KEY = "pendingCleanupRetryCapWarned"; // The reaper sweeps at most this many orphaned active leases per tick. @@ -226,45 +229,12 @@ const PENDING_CLEANUP_SWEEP_ERROR_KIND = "sweep_failed"; const ORPHANED_ACTIVE_LEASE_SWEEP_ERROR_KIND = "orphaned_active_lease_sweep_failed"; -// Read the stored retry attempt count as a safe value, directly in SQL. A -// provider can write a malformed value under the attempts key. The type guard -// makes any non-number value read as zero. The reader computes as numeric and -// never casts to int, so a finite number outside the 32-bit range (for example -// 1e300) never throws. The reader clamps a negative value to zero and a positive -// value to the attempt cap. One malformed lease therefore never aborts the page -// sweep. This matches the TypeScript reader `readPendingCleanupRetryAttempts`, -// which clamps to the same range. The claim predicate compares the two readers, -// so both must yield the same value for every input. -function pendingCleanupAttemptsSql() { - return sql` - case - when jsonb_typeof(${environmentLeases.metadata} -> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}) = 'number' - then least( - greatest( - floor((${environmentLeases.metadata} ->> ${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY})::numeric), - 0 - ), - ${PENDING_CLEANUP_SWEEP_ATTEMPT_CAP} - ) - else 0 - end`; -} - function pendingCleanupRetryDueSql(explicitRetry = false) { return sql`case when ${explicitRetry} and coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') != 'true' then true else ${pendingCleanupDeadlineSql()} <= ${Date.now()} end`; } -// Choose the `jsonb_set` target root. A provider can write a scalar or array -// metadata root. `jsonb_set` fails on a non-object root, so the reader uses the -// stored metadata only when its root is an object. A NULL, scalar, or array root -// reads as an empty object. `jsonb_typeof(NULL)` is NULL, so the else branch also -// covers a NULL root. -function pendingCleanupMetadataObjectSql() { - return sql`case when jsonb_typeof(${environmentLeases.metadata}) = 'object' then ${environmentLeases.metadata} else '{}'::jsonb end`; -} - // Read the stored cap-warned flag as a safe boolean, directly in SQL. A // malformed value reads as false, so the boolean cast never throws. function pendingCleanupCapWarnedSql() { @@ -1449,62 +1419,6 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD return Math.min(Math.floor(value), PENDING_CLEANUP_SWEEP_ATTEMPT_CAP); } - // A teardown claim fences other sweeps for 15 minutes. After a restart, a - // claim whose server process has exited cannot still be running, so its - // fence ends now. The saved process is bound to its own attempt: a later - // claim from another path that kept the old record is never released. A live - // or unidentified claimer keeps its fence. Pages pass every skipped claim. - async function releaseLostPendingCleanupClaims(): Promise { - const nowMs = Date.now(); - let after: string | null = null; - for (;;) { - const claimed: Array<{ id: string; metadata: Record | null }> = await db - .select({ id: environmentLeases.id, metadata: environmentLeases.metadata }) - .from(environmentLeases) - .where(and( - eq(environmentLeases.status, "pending_cleanup"), - sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`, - sql`jsonb_typeof(${environmentLeases.metadata}->'pendingCleanupLeaseExpiresAtMs') = 'number'`, - sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`, - sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${environmentLeases.metadata}->>'pendingCleanupAttemptId'`, - sql`${environmentLeases.metadata}->'pendingCleanupController'->>'bootId' is distinct from ${currentControllerBootId()}`, - after === null ? undefined : gt(environmentLeases.id, after), - )) - .orderBy(asc(environmentLeases.id)) - .limit(PENDING_CLEANUP_SWEEP_PAGE_SIZE); - for (const row of claimed) { - const metadata = row.metadata ?? {}; - const saved = metadata.pendingCleanupController as Record | undefined; - const controller = readControllerProcess(saved); - const attemptId = saved?.attemptId; - if (!controller || typeof attemptId !== "string" || !await controllerProcessExited(controller)) continue; - // A dead server is not a failed teardown. A sweep claim already counted - // its attempt, so give it back; claims from other paths never counted. - // The update is single-use: it matches only while the fence is still in - // the future and it consumes `countedAttempt`, so an overlapping sweep - // that read the same claim cannot refund the attempt again. - const returnAttempt = saved?.countedAttempt === true; - await db.update(environmentLeases) - .set({ metadata: sql`jsonb_set(jsonb_set(jsonb_set(${pendingCleanupMetadataObjectSql()}, - '{pendingCleanupLeaseExpiresAtMs}', to_jsonb(${nowMs}::bigint)), - array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], - to_jsonb(greatest(${pendingCleanupAttemptsSql()} - ${returnAttempt ? 1 : 0}, 0)::int), true), - '{pendingCleanupController,countedAttempt}', 'false'::jsonb, true)` }) - .where(and( - eq(environmentLeases.id, row.id), - eq(environmentLeases.status, "pending_cleanup"), - sql`${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true'`, - sql`(${environmentLeases.metadata}->>'pendingCleanupLeaseExpiresAtMs')::numeric > ${nowMs}`, - sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`, - sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${attemptId}`, - sql`coalesce(${environmentLeases.metadata}->'pendingCleanupController'->>'countedAttempt', 'false') = ${returnAttempt ? "true" : "false"}`, - )); - } - if (claimed.length < PENDING_CLEANUP_SWEEP_PAGE_SIZE) return; - after = claimed[claimed.length - 1]!.id; - } - } - // Atomically claim one retry attempt on a pending_cleanup lease. The update // only matches when the lease is still pending_cleanup and its stored attempt // count still equals `expectedAttempts`. Two concurrent sweeps read the same @@ -1798,7 +1712,12 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD deferredFlushFailed = true; } - await releaseLostPendingCleanupClaims(); + // A teardown claim whose claimant process is proven gone ends its fence + // now, through the shared ownership path. A claim held as a fence (another + // host, or a claimant without a process stamp) waits for its deadline. + await settleOwnableKind(db, "environment_teardown_claim").catch(() => { + logger.warn({ errorKind: "teardown_claim_settle_failed" }, "teardown claim settle failed; a later sweep retries"); + }); const rows = await db .select() diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 7d06d442344..b2ef3047448 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -42,6 +42,7 @@ import { 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"; +import { settleEnvironmentTeardownClaim } from "../environment-teardown-claim-settlement.js"; /** * Registry of ownable record kinds (plan section C). @@ -593,12 +594,18 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ }, }, runLeaseKind(), - leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership", - and(eq(environmentLeases.status, "pending_cleanup"), sql`(${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean is true`), - teardownClaimOwner), + { + // Shared path since Phase 2: a claimant proven gone ends its fence now. + ...leaseKind("environment_teardown_claim", "Teardown claim (in-flight cleanup)", "ownership", + // Text compare: a malformed flag never makes the whole listing fail. + and(eq(environmentLeases.status, "pending_cleanup"), + sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') = 'true'`), + teardownClaimOwner), + settle: settleEnvironmentTeardownClaim, + }, leaseKind("environment_pending_cleanup", "Pending-cleanup obligation (not in flight)", "ownership", and(eq(environmentLeases.status, "pending_cleanup"), - sql`coalesce((${environmentLeases.metadata}->>'pendingCleanupInFlight')::boolean, false) is false`), + sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') <> 'true'`), () => ({ kind: "none" }), [1, 2, 7]), leaseKind("environment_stop_and_retain_intent", "Stop-and-retain intent", "workspace_preservation", and(eq(environmentLeases.status, "pending_cleanup"), sql`${environmentLeases.metadata} ? 'sandboxStopAndRetain'`),