diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index 7bd1c6fb487..70040517547 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { @@ -38,6 +39,7 @@ vi.mock("../middleware/logger.js", () => ({ import { logger } from "../middleware/logger.ts"; import { heartbeatService, type HeartbeatEnvironmentRuntime } from "../services/heartbeat.ts"; import { hasPendingRunCleanup } from "../services/execution-blocker.js"; +import { currentControllerProcess } from "../services/controller-process-identity.js"; import { environmentStopErrorData, JsonRpcCallError, PLUGIN_RPC_ERROR_CODES, PluginEnvironmentStopError } from "@paperclipai/plugin-sdk"; const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); @@ -147,8 +149,9 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { environmentId: string | null; updatedAt: Date; metadata?: Record; + id?: string; }): Promise { - const id = randomUUID(); + const id = input.id ?? randomUUID(); await db.insert(environmentLeases).values({ id, companyId: input.companyId, @@ -345,6 +348,70 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { expect(teardown).toHaveBeenCalledTimes(expired ? 1 : 0); }); + describe("a teardown claim whose server stopped mid-attempt", () => { + async function exitedPid(): Promise { + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await new Promise(resolve => child.once("exit", resolve)); + return child.pid!; + } + function claimMetadata(controller: Record, attemptId = "lost-claim") { + return { pendingCleanupAttemptId: attemptId, pendingCleanupInFlight: true, [ATTEMPTS_KEY]: 1, + pendingCleanupLeaseExpiresAtMs: Date.now() + 14 * 60_000, pendingCleanupController: controller }; + } + async function sweepClaimedBy(controller: Record, attemptId = "lost-claim", + before?: (scope: { companyId: string; environmentId: string }) => Promise) { + const scope = await seedCompanyAndEnvironment(); + await before?.(scope); + const leaseId = await insertOrphanEphemeralLease({ ...scope, updatedAt: new Date(Date.now() - 60_000), + metadata: claimMetadata(controller, attemptId) }); + const teardown = vi.fn(async () => {}); + const service = heartbeatService(db, { environmentRuntime: { + retryPendingSandboxTeardown: teardown, + } as unknown as HeartbeatEnvironmentRuntime }); + const result = await service.sweepPendingCleanupLeases(); + return { leaseId, teardown, result }; + } + async function exitedClaimer(attemptId = "lost-claim") { + return { ...await currentControllerProcess(), bootId: randomUUID(), pid: await exitedPid(), attemptId }; + } + + it("retries now instead of waiting for the claim to expire", async () => { + const { teardown, result } = await sweepClaimedBy(await exitedClaimer()); + expect(teardown).toHaveBeenCalledTimes(1); + expect(result.destroyed).toBe(1); + }); + it("keeps the fence of a claim held by this live server", async () => { + const { leaseId, teardown } = await sweepClaimedBy({ ...await currentControllerProcess(), attemptId: "lost-claim" }); + expect(teardown).not.toHaveBeenCalled(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(lease!.metadata).toMatchObject({ pendingCleanupInFlight: true, pendingCleanupAttemptId: "lost-claim" }); + }); + it("keeps the fence of a newer claim that kept an earlier attempt's server record", async () => { + // Another claim path replaced the attempt but left the exited server's record. + const { teardown } = await sweepClaimedBy(await exitedClaimer("earlier-attempt"), "newer-live-claim"); + expect(teardown).not.toHaveBeenCalled(); + }); + it("keeps the fence of a claim from another host", async () => { + const { teardown } = await sweepClaimedBy({ ...await exitedClaimer(), hostHash: "f".repeat(32) }); + expect(teardown).not.toHaveBeenCalled(); + }); + it("keeps the fence of a claim without a recorded server identity", async () => { + const { teardown } = await sweepClaimedBy({}); + expect(teardown).not.toHaveBeenCalled(); + }); + it("pages past more fenced claims than one sweep page to find a lost one", async () => { + const foreign = { ...await exitedClaimer(), hostHash: "f".repeat(32) }; + const { teardown, result } = await sweepClaimedBy(await exitedClaimer(), "lost-claim", async scope => { + for (let index = 0; index < SWEEP_PAGE_SIZE + 5; index++) { + await insertOrphanEphemeralLease({ ...scope, updatedAt: new Date(Date.now() - 60_000), + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, metadata: claimMetadata(foreign) }); + } + }); + expect(teardown).toHaveBeenCalledTimes(1); + expect(result.destroyed).toBe(1); + }); + }); + it("starts escalation and the long cooldown on the fifth failed attempt", async () => { const { companyId, environmentId } = await seedCompanyAndEnvironment(); const leaseId = await insertPendingCleanupLease({ companyId, environmentId, diff --git a/server/src/__tests__/sandbox-creation-operations.test.ts b/server/src/__tests__/sandbox-creation-operations.test.ts index 9ba022747b3..e4792ce6ce0 100644 --- a/server/src/__tests__/sandbox-creation-operations.test.ts +++ b/server/src/__tests__/sandbox-creation-operations.test.ts @@ -1,4 +1,4 @@ -import { execFile } from "node:child_process"; +import { execFile, spawn } from "node:child_process"; import { createServer } from "node:http"; import { promisify } from "node:util"; import { fileURLToPath } from "node:url"; @@ -17,6 +17,7 @@ import { HttpError } from "../errors.js"; import * as secretOwner from "../services/secrets.js"; import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js"; +import { readProcessStartedAt } from "../services/hot-restart.js"; import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); @@ -334,4 +335,107 @@ describe("durable sandbox creation through the runtime entry point", () => { expect(await nextEnvironmentCleanupAt(db)).toBeNull(); await expect(acquire()).rejects.toThrow("requires reconciliation"); expect(creates).toBe(1); }); + + describe("after the creating server exits mid-acquire", () => { + async function exitedPid(): Promise { + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await new Promise(resolve => child.once("exit", resolve)); + return child.pid!; + } + // The provider allocated the sandbox; the server exited before committing the lease. + async function interruptedCreation(controller: (current: Record) => Promise>) { + loseReply = true; + await expect(acquire()).rejects.toThrow("Lost provider reply"); + const [operation] = await db.select().from(sandboxCreationOperations); + expect(operation).toMatchObject({ state: "pending" }); + expect(operation!.nextAttemptAt.getTime()).toBeGreaterThan(Date.now() + 25 * 60_000); + const current = operation!.metadata.creationController as Record; + await db.update(sandboxCreationOperations).set({ metadata: { ...operation!.metadata, creationController: await controller(current) } }) + .where(eq(sandboxCreationOperations.id, operation!.id)); + return operation!; + } + const inspections = () => vi.mocked(workers.call).mock.calls.filter(value => value[1] === "environmentInspectLease"); + + it("destroys the sandbox on the next sweep instead of waiting for the settle deadline", async () => { + const pid = await exitedPid(); + const operation = await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), pid })); + await runtime().flushDeferredOrphanCleanups(); + const [lease] = await db.select().from(environmentLeases); + expect(lease).toMatchObject({ status: "pending_cleanup", companyId, providerLeaseId: resources.get(operation.id), + failureReason: "interrupted_sandbox_creation" }); + expect(await db.select().from(sandboxCreationOperations)).toEqual([expect.objectContaining({ state: "cleanup_pending", leaseId: lease!.id })]); + await runtime().retryPendingSandboxTeardown({ environment: null, lease: (await environmentService(db).getLeaseById(lease!.id))! }); + expect(resources.size).toBe(0); expect(destroys).toBe(1); expect(creates).toBe(1); + }); + it("pages past more unprovable creations than one page to find an abandoned one", async () => { + const pid = await exitedPid(); + const operation = await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), pid })); + const { id: _id, runId: _runId, ...row } = operation; + // Sorted ahead of the abandoned intent: creators on another host cannot be judged. + await db.insert(sandboxCreationOperations).values(Array.from({ length: 105 }, (_, index) => ({ ...row, + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, runId: randomUUID(), + metadata: { ...row.metadata, creationController: { ...(row.metadata.creationController as object), bootId: randomUUID(), pid, hostHash: "f".repeat(32) } } }))); + await runtime().flushDeferredOrphanCleanups(); + expect(await db.select().from(environmentLeases)).toEqual([expect.objectContaining({ status: "pending_cleanup", + providerLeaseId: resources.get(operation.id) })]); + }); + it("treats a restarted server that reused the creator PID as a different process", async () => { + await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), + processStartedAt: new Date(Date.parse(String(current.processStartedAt)) - 60_000).toISOString() })); + await runtime().flushDeferredOrphanCleanups(); + expect(await db.select().from(environmentLeases)).toEqual([expect.objectContaining({ status: "pending_cleanup" })]); + }); + it("leaves a creation owned by this live server untouched until its deadline", async () => { + await interruptedCreation(async current => current); + await runtime().flushDeferredOrphanCleanups(); + expect(inspections()).toHaveLength(0); + expect(await db.select().from(environmentLeases)).toEqual([]); + expect(await db.select().from(sandboxCreationOperations)).toEqual([expect.objectContaining({ state: "pending" })]); + expect(resources.size).toBe(1); expect(destroys).toBe(0); + }); + it("leaves a creation owned by another live server process untouched", async () => { + const pid = process.ppid; + const startedAt = await readProcessStartedAt(pid); + await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), pid, processStartedAt: startedAt })); + await runtime().flushDeferredOrphanCleanups(); + expect(inspections()).toHaveLength(0); + expect(await db.select().from(environmentLeases)).toEqual([]); + expect(resources.size).toBe(1); + }); + it("does not judge a creator on another host", async () => { + const pid = await exitedPid(); + await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), pid, hostHash: "f".repeat(32) })); + await runtime().flushDeferredOrphanCleanups(); + expect(inspections()).toHaveLength(0); + expect(await db.select().from(environmentLeases)).toEqual([]); + expect(resources.size).toBe(1); + }); + it("keeps ownership when an early lookup finds no sandbox yet", async () => { + const pid = await exitedPid(); + const operation = await interruptedCreation(async current => ({ ...current, bootId: randomUUID(), pid })); + const resource = resources.get(operation.id)!; + resources.delete(operation.id); // Not materialized at the provider yet. + await runtime().flushDeferredOrphanCleanups(); + expect(inspections()).toHaveLength(1); + const [pending] = await db.select().from(sandboxCreationOperations); + expect(pending).toMatchObject({ state: "pending" }); + expect(pending!.nextAttemptAt.getTime()).toBeGreaterThan(Date.now()); + // A late allocation is still found and removed by the retained owner. + resources.set(operation.id, resource); + await due(); await runtime().flushDeferredOrphanCleanups(); + expect(await db.select().from(environmentLeases)).toEqual([expect.objectContaining({ status: "pending_cleanup", providerLeaseId: resource })]); + }); + it("never touches a sandbox whose lease was committed", async () => { + const { lease } = await acquire(); + const pid = await exitedPid(); + const [operation] = await db.select().from(sandboxCreationOperations); + expect(operation).toMatchObject({ state: "attached" }); + await db.update(sandboxCreationOperations).set({ metadata: { ...operation!.metadata, + creationController: { ...(operation!.metadata.creationController as object), bootId: randomUUID(), pid } } }); + await runtime().flushDeferredOrphanCleanups(); + expect(inspections()).toHaveLength(0); + expect(await environmentService(db).getLeaseById(lease.id)).toMatchObject({ status: "active" }); + expect(resources.size).toBe(1); expect(destroys).toBe(0); + }); + }); }); diff --git a/server/src/services/controller-process-identity.ts b/server/src/services/controller-process-identity.ts new file mode 100644 index 00000000000..f752827934e --- /dev/null +++ b/server/src/services/controller-process-identity.ts @@ -0,0 +1,91 @@ +import { createHash, randomUUID } from "node:crypto"; +import { readFileSync, readlinkSync } from "node:fs"; +import { hostname, uptime } from "node:os"; +import { readProcessStartedAt } from "./hot-restart.js"; + +/** + * Identity of the server process that started a provider side effect, saved + * with its durable record. After a restart, recovery uses it to prove that the + * starter has exited, so the side effect cannot still be in flight there. + */ +export type ControllerProcess = { bootId: string; pid: number; hostHash: string; processStartedAt: string | null }; + +const bootId = randomUUID(); +const UNKNOWN_PROCESS_SPACE = "unknown"; + +/** + * Names the space in which PIDs are comparable: one kernel boot and, on Linux, + * one PID namespace. Containers that share a hostname but not a PID namespace + * get different values. When the space cannot be read, no record is judged. + * A redeploy onto another host or namespace also never matches; such records + * keep their original deadlines. + */ +function processSpaceHash(): string { + try { + const parts = [hostname()]; + if (process.platform === "linux") { + parts.push(readFileSync("/proc/sys/kernel/random/boot_id", "utf8").trim(), readlinkSync("/proc/self/ns/pid")); + } else { + // No PID namespaces here. The boot time keeps a rebooted host distinct; + // rounding can only make a record unmatched, which fails closed. + parts.push(String(Math.round((Date.now() - uptime() * 1000) / 600_000))); + } + if (parts.some(part => !part)) return UNKNOWN_PROCESS_SPACE; + return createHash("sha256").update(parts.join("\0")).digest("hex").slice(0, 32); + } catch { + return UNKNOWN_PROCESS_SPACE; + } +} + +const hostHash = processSpaceHash(); +let processStartedAt: Promise | undefined; + +async function observedProcessStartedAt(pid: number): Promise { + try { + const startedAt = Date.parse(await readProcessStartedAt(pid) ?? ""); + return Number.isFinite(startedAt) ? startedAt : null; + } catch { + return null; + } +} + +export function currentControllerBootId(): string { + return bootId; +} + +export async function currentControllerProcess(): Promise { + processStartedAt ??= observedProcessStartedAt(process.pid) + .then(value => value === null ? null : new Date(value).toISOString()); + return { bootId, pid: process.pid, hostHash, processStartedAt: await processStartedAt }; +} + +export function readControllerProcess(value: unknown): ControllerProcess | null { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const record = value as Record; + if (typeof record.bootId !== "string" || typeof record.hostHash !== "string" || + typeof record.pid !== "number" || !Number.isInteger(record.pid) || record.pid <= 0) return null; + return { bootId: record.bootId, pid: record.pid, hostHash: record.hostHash, + processStartedAt: typeof record.processStartedAt === "string" ? record.processStartedAt : null }; +} + +function processExists(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException | undefined)?.code === "EPERM"; + } +} + +/** True only with proof, in this process's own PID space, that the recorded + * server process has exited. Another space, an unreadable space, this server, + * or an ambiguous live PID returns false. */ +export async function controllerProcessExited(controller: ControllerProcess): Promise { + if (controller.bootId === bootId || hostHash === UNKNOWN_PROCESS_SPACE || controller.hostHash !== hostHash) return false; + if (!processExists(controller.pid)) return true; + // The PID now belongs to a live process, possibly this restarted server. + // Only a definitely different start time proves the recorded process is gone. + const recorded = Date.parse(controller.processStartedAt ?? ""); + const observed = await observedProcessStartedAt(controller.pid); + return Number.isFinite(recorded) && observed !== null && Math.abs(observed - recorded) >= 1_000; +} diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index cd00e460eb0..04d40f10e29 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -11,6 +11,7 @@ import { retainedNativeSessionCleanupWork } from "../native-runtime/retained-nat 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 { 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"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; @@ -1417,6 +1418,49 @@ 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; + await db.update(environmentLeases) + .set({ metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, '{pendingCleanupLeaseExpiresAtMs}', to_jsonb(${nowMs}::bigint))` }) + .where(and( + eq(environmentLeases.id, row.id), + eq(environmentLeases.status, "pending_cleanup"), + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`, + sql`${environmentLeases.metadata}->'pendingCleanupController'->>'attemptId' = ${attemptId}`, + )); + } + 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 @@ -1436,13 +1480,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ): Promise { const now = new Date(); const attemptId = randomUUID(); + const controller = await currentControllerProcess(); const claimed = await db .update(environmentLeases) .set({ metadata: sql`jsonb_set(${pendingCleanupMetadataObjectSql()}, array[${PENDING_CLEANUP_ATTEMPTS_METADATA_KEY}], to_jsonb(${expectedAttempts + 1}::int), true) || ${JSON.stringify({ ...(manualAttempt ? { pendingCleanupManualAttemptId: randomUUID() } : {}), pendingCleanupAttemptId: attemptId, pendingCleanupInFlight: true, - pendingCleanupLeaseExpiresAtMs: Date.now() + 15 * 60_000, + pendingCleanupLeaseExpiresAtMs: Date.now() + 15 * 60_000, pendingCleanupController: { ...controller, attemptId }, })}::jsonb`, lastUsedAt: now, updatedAt: now, @@ -1691,6 +1736,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD deferredFlushFailed = true; } + await releaseLostPendingCleanupClaims(); + const rows = await db .select() .from(environmentLeases) diff --git a/server/src/services/sandbox-creation-operations.ts b/server/src/services/sandbox-creation-operations.ts index d17300c7d4e..0b5b7bee8c7 100644 --- a/server/src/services/sandbox-creation-operations.ts +++ b/server/src/services/sandbox-creation-operations.ts @@ -1,6 +1,7 @@ import { collectSecretRefPaths, isUuidSecretRef, parseSecretRefBindingObject, readConfigValueAtPath } from "./json-schema-secret-refs.js"; import { createHash, randomUUID } from "node:crypto"; -import { and, eq, lte, or, sql } from "drizzle-orm"; +import { and, asc, eq, gt, lte, or, sql } from "drizzle-orm"; +import { controllerProcessExited, currentControllerBootId, currentControllerProcess, readControllerProcess } from "./controller-process-identity.js"; import { companies, environments, sandboxCreationOperations, type Db } from "@paperclipai/db"; import { readEnvironmentCreationCleanupError, type PluginEnvironmentLease, type PluginEnvironmentLeaseObservation } from "@paperclipai/plugin-sdk"; import { assertEnvironmentCompanyBinding, environmentService } from "./environments.js"; @@ -10,6 +11,46 @@ import { notifyEnvironmentCleanup, writeEnvironmentCleanup } from "./environment export type SandboxCreationOperation = typeof sandboxCreationOperations.$inferSelect; +// A live creator may still be waiting for the provider, and a provider may still +// materialize an aborted request. Absence is conclusive only after this delay. +const CREATION_SETTLE_FLOOR_MS = 30 * 60_000; +const CREATION_PROMOTION_PAGE_SIZE = 100; + +/** After a restart, creation intents from an exited server cannot still be in + * flight there. Reconcile them now instead of at the settle deadline. Absence + * stays inconclusive until that deadline; see recoverSandboxCreations. */ +async function promoteAbandonedSandboxCreations(db: Db): Promise { + const now = new Date(); + // Page past live, foreign, and unidentified creators so they cannot hide an + // abandoned intent behind them. + let after: string | null = null; + for (;;) { + const waiting: Array> = await db + .select({ id: sandboxCreationOperations.id, metadata: sandboxCreationOperations.metadata }) + .from(sandboxCreationOperations).where(and( + eq(sandboxCreationOperations.state, "pending"), gt(sandboxCreationOperations.nextAttemptAt, now), + sql`${sandboxCreationOperations.metadata}->'creationController'->>'bootId' is distinct from ${currentControllerBootId()}`, + sql`coalesce(${sandboxCreationOperations.metadata}->>'creationControllerLost', '') <> 'true'`, + after === null ? undefined : gt(sandboxCreationOperations.id, after), + )).orderBy(asc(sandboxCreationOperations.id)).limit(CREATION_PROMOTION_PAGE_SIZE); + for (const operation of waiting) { + const controller = readControllerProcess(operation.metadata.creationController); + if (!controller || !await controllerProcessExited(controller)) continue; + await db.update(sandboxCreationOperations).set({ nextAttemptAt: now, updatedAt: now, + metadata: sql`coalesce(${sandboxCreationOperations.metadata}, '{}'::jsonb) || '{"creationControllerLost":true}'::jsonb` }) + .where(and(eq(sandboxCreationOperations.id, operation.id), eq(sandboxCreationOperations.state, "pending"), + gt(sandboxCreationOperations.nextAttemptAt, now))); + } + if (waiting.length < CREATION_PROMOTION_PAGE_SIZE) return; + after = waiting[waiting.length - 1]!.id; + } +} + +function creationAbsenceInconclusive(operation: SandboxCreationOperation): boolean { + const settleAt = Date.parse(typeof operation.metadata.creationSettleAt === "string" ? operation.metadata.creationSettleAt : ""); + return operation.metadata.creationControllerLost === true && Number.isFinite(settleAt) && Date.now() < settleAt; +} + export function recoverableTaskCreation(operation: Pick): boolean { const metadata = operation.metadata; return metadata.runnerTaskLease === true && metadata.creationOperationVersion === 1 && @@ -52,6 +93,8 @@ export async function createOwnedSandbox( } const id = randomUUID(); const runId = owner.heartbeatRunId ?? randomUUID(); + const settleAt = new Date(Date.now() + Math.max(CREATION_SETTLE_FLOOR_MS, (owner.timeoutMs ?? 0) + 60_000)); + const creationController = await currentControllerProcess(); const operation = await db.transaction(async tx => { await notifyEnvironmentCleanup(tx, { companyId: owner.companyId, runId: owner.heartbeatRunId ?? undefined }); await assertEnvironmentCompanyBinding(tx as unknown as Db, owner.environmentId, owner.companyId); @@ -59,8 +102,9 @@ export async function createOwnedSandbox( id, companyId: owner.companyId, environmentId: owner.environmentId, originalEnvironmentId: owner.environmentId, runId, requestKey: owner.requestKey ?? "run", issueId: owner.issueId, executionWorkspaceId: owner.executionWorkspaceId, provider: owner.provider, pluginId: owner.pluginId, driverKey: owner.driverKey, - config: owner.config, metadata: { ...owner.metadata, creationCredentialHashes: credentialHashes }, - nextAttemptAt: new Date(Date.now() + Math.max(30 * 60_000, (owner.timeoutMs ?? 0) + 60_000)), + config: owner.config, metadata: { ...owner.metadata, creationCredentialHashes: credentialHashes, + creationController, creationSettleAt: settleAt.toISOString() }, + nextAttemptAt: settleAt, }).onConflictDoNothing().returning(); if (inserted) return inserted; const [existing] = await tx.select().from(sandboxCreationOperations).where(and( @@ -152,6 +196,7 @@ export async function recoverSandboxCreations(db: Db, inspect: (operation: SandboxCreationOperation) => Promise, onTaskRecovered?: (operation: SandboxCreationOperation) => void | Promise, ): Promise { + await promoteAbandonedSandboxCreations(db); const due = await db.select().from(sandboxCreationOperations).where(and( recoverableSandboxCreationCondition(), lte(sandboxCreationOperations.nextAttemptAt, new Date()), )).limit(100); @@ -187,6 +232,10 @@ export async function recoverSandboxCreations(db: Db, cleanup.runId !== candidate.runId || cleanup.attemptId !== candidate.id || cleanup.providerLeaseId !== observation!.providerLeaseId)) { throw new Error("Sandbox creation recovery returned foreign ownership"); } + // An early lookup after a lost creator may precede provider materialization. + if (existingCompany && !cleanup && creationAbsenceInconclusive(candidate)) { + throw new Error("Sandbox creation has not settled yet"); + } await db.transaction(async tx => { const [company] = await tx.select({ id: companies.id }).from(companies).where(eq(companies.id, candidate.companyId)).for("key share"); await tx.select({ id: environments.id }).from(environments).where(eq(environments.id, candidate.originalEnvironmentId)).for("update");