diff --git a/doc/idle-environment-cleanup.md b/doc/idle-environment-cleanup.md new file mode 100644 index 00000000000..9070703ef8c --- /dev/null +++ b/doc/idle-environment-cleanup.md @@ -0,0 +1,103 @@ +# Idle environment cleanup + +Pending lease cleanup and historical resource verification use the existing +transaction signals, delivery coordinator, and in-process scheduler. They no +longer run unconditionally from the server maintenance tick. The run reaper +performs its inline pending-cleanup sweep only after actually recovering a run +or an orphaned active lease. Explicit user retry keeps its existing direct path. + +## Work and ownership + +| Duty | Truth | Trigger and next deadline | Quiet condition | +| --- | --- | --- | --- | +| Failed teardown | Existing `environment_leases` rows in `pending_cleanup` | Commit signal; later of row backoff and saved retry/ownership deadline | No pending lease or deferred orphan | +| Failed acquisition before a DB row exists | Existing local orphan spool and the producing runtime's buffer | Local buffer registration; five-minute retry while buffered; startup spool load | Durable row owns recovery and spool copy is removed | +| Historical provider verification | Existing failed ephemeral leases and finished custom-image setups without matching cleanup proof | Commit signal; five-minute retry while an unverified candidate remains | No unverified candidates, including retained verified history | + +`environment-cleanup-state.ts` owns the deadline query and the shared safe reader +for retry/ownership metadata. Cleanup claims and the scheduler use that same +reader, including malformed-number handling and the existing maximum horizon. +Startup retains the immediate scan, but never bypasses a saved in-flight claim or +retry cooldown. Subsequent scans respect the five-minute row backoff. Claims, +provider readiness, Stop authority, workspace retention, copyback, receipt +validation, and escalation rules stay with their current owners. + +`environment-cleanup-work.ts` registers two independent workers. Their generation +counters preserve a wake until successful reconciliation, including a first write +during drain, changes during a sweep, and failure followed by drain. The existing +heartbeat interval bounds pending-cleanup starts after bursts of writes. Historical +verification has a separate minimum five-minute spacing, including commit-triggered +passes, and does not delay server readiness. No timer remains once +there is no obligation. A self-generated write can require a final reconciliation +before quiet. A failed sweep retains a retry. Shutdown unsubscribes local buffer +wakes; the delivery coordinator stops subscriptions/timers and joins active work. + +## Producer coverage + +The shared `writeEnvironmentCleanup` helper registers and awaits intent before a +write in a transaction. Existing transaction owners call +`notifyEnvironmentCleanup(tx)` directly. Nested calls retain outer transaction +settlement and uncertainty handling through the shared DB implementation. + +- `environments.ts`: orphan insertion, release, metadata replacement, and run-wide + release. These stores cover normal provider teardown and failed acquisition. +- `environment-runtime.ts`: reusable sandbox cleanup claims and their settlement, + and Stop/export cleanup claims. Renewal of an already active claim retains its + bounded timer; it cannot create new pending work. +- `heartbeat/recovery.ts` and `heartbeat/run-control.ts`: orphan-active recovery, + retry settlement, and explicit recovery of historical leases. +- `environment-task-runtime.ts`: unconfirmed plugin task termination and plugin + replacement orphans, including a registered deferred flusher when the row + cannot be written. The flusher retains an idle hold until the row is durable. +- `sandbox-stop-and-retain.ts`: persisted Stop intent and its settlement. +- `native-workspace-export-resume.ts` and `native-workspace-export-retry.ts`: + terminal-export cleanup, resumed-export cleanup claims, and settlement. +- `workspace-recovery.ts`: verified copyback schedules destruction. +- `native-workspace-sync.ts` and the native session executor's backup stamp: + writes that can invalidate an existing historical row-version proof. +- `environment-custom-images.ts`: provider state, terminal state, and promotion. + Initial active setup insertion is not a historical verification candidate. + +Other lease metadata writes neither create pending cleanup nor change the +row-version/provider identity of a matching historical proof. Changes to an +already pending claim remain covered by its existing deadline. Deletion removes +work; an already scheduled reconciliation observes that removal. Independent SQL +writers and another server process are not covered by these process-local signals. + +## Deferred orphan buffers + +Runtimes that hold an unwritten orphan register their flusher in a registry scoped +to the shared database handle. Registration wakes the worker without contacting +the database. A sweep flushes registered runtimes as well as its own startup spool; +this reaches a producer runtime's buffer even when it differs from the heartbeat's +runtime. The registry retains only runtimes with unresolved buffer ownership. + +Unreadable directories, records, and malformed records retain a retry; a missing +spool directory is an empty spool. A spool read failure still allows other runtimes to flush their buffers and +persisted leases to clean up, then fails the sweep so recovery stays scheduled. +Each concrete cleanup pass reloads the spool, including entries appended by +plugin task replacement after startup and entries beyond buffer capacity. +No empty-spool filesystem or database polling is introduced. Existing limits on a simultaneous database, +spool, and buffer-capacity failure remain; this is not a new durable queue. + +## Verification and boundaries + +The PostgreSQL worker tests exercise quiet configured state and verified history, +real store activation, persisted backoff, failed teardown and restart, nested +commit/rollback, drain/resume, cross-runtime deferred work, failed spool reads, +shutdown joining, provider uncertainty, and historical pagination. Existing +runtime, reaper, Stop, custom-image, and workspace tests cover the underlying +cleanup authority and provider contracts. + +The runtime fixture now gives each simulated instance its own DB handle and spool. +Its journal-write failure injection targets the transaction boundary rather than +the direct update that moved inside that transaction. Fake-clock worker tests wait +for PostgreSQL I/O and distinguish an executing sweep from a retained next wake. + +This does not migrate discovery of orphaned active leases or legacy shutdown +workspace settlement from the broader run-recovery loop. Custom-image *active +session expiry* also remains on its existing tick; only historical verification +moves here. Other maintenance duties, external probes, and connected transports +still affect idle traffic. No Cloud alarm coverage or host-sleep exemption changes. +These tests prove the changed workers become quiet and recover; they do not prove +deployed suspension or fleet savings. diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index d2bbb1ef91c..2e5bb5c9ca5 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -613,12 +613,15 @@ Longer polling intervals alone do not complete an item. artifacts, deliveries, and ownership before deleting resources. Source: [execution-workspaces.ts](../../server/src/services/execution-workspaces.ts). -- [ ] **JOB-17 — Sandbox cleanup and orphan spools.** Current: **30s** scans even - when attempts have longer staleness/backoff. Change: durable cleanup queue with - next retry, immediate failed-release activation, and provider/fleet backstop. - **Complete when:** idle scans stop without leaking paid sandboxes or ignoring - malformed/unpersisted spool entries. Source: - [heartbeat.ts](../../server/src/services/heartbeat.ts), `sweepPendingCleanupLeases`. +- [ ] **JOB-17 — Sandbox cleanup and orphan spools (partially migrated).** + Pending lease cleanup now uses commit signals and saved retry/claim deadlines. + Deferred orphan buffers wake the worker without a DB connection; startup loads + the existing spool. Historical provider verification stops when matching proof + is complete. Empty maintenance passes no longer call these sweeps. Discovery of + orphaned active leases and legacy shutdown workspace settlement still belong + to the broader run-recovery loop. No Cloud sleep exemptions changed. See + [idle-environment-cleanup.md](../idle-environment-cleanup.md) for producer scope, + recovery, tests, and remaining limits. - [x] **JOB-18 — Adapter login and setup-token reapers.** Replaced the default **30s** empty scans with commit-aware workers, session/promotion expiry diff --git a/server/src/__tests__/environment-cleanup-work.test.ts b/server/src/__tests__/environment-cleanup-work.test.ts new file mode 100644 index 00000000000..7f6b1d64c12 --- /dev/null +++ b/server/src/__tests__/environment-cleanup-work.test.ts @@ -0,0 +1,236 @@ +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createSandboxOrphanCleanupSpool } from "../services/sandbox-orphan-cleanup-spool.js"; +import { nextEnvironmentCleanupAt } from "../services/environment-cleanup-state.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { companies, createDb, environmentLeases, environments, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerEnvironmentCleanupWork } from "../services/environment-cleanup-work.js"; +import { registerDeferredEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; +import { environmentService } from "../services/environments.js"; +import { historicalCleanupVerificationService } from "../services/historical-cleanup-verification.js"; +import { heartbeatService } from "../services/heartbeat.js"; +import { environmentRuntimeService } from "../services/environment-runtime.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("environment cleanup scheduling", () => { + let database: Awaited>, db: ReturnType; + let companyId: string, environmentId: string, draining: boolean, failing: boolean; + let coordinator: ReturnType | undefined; + let worker: ReturnType | undefined; + let releases: ReturnType, inspections: ReturnType, reads: ReturnType; + const errors: unknown[] = [], disposers: Array<() => void> = []; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("environment-cleanup-work-"); + db = createDb(database.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + }, 90_000); + beforeEach(async () => { + await db.execute(sql`TRUNCATE companies, environments CASCADE`); + companyId = randomUUID(); environmentId = randomUUID(); draining = false; failing = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Cleanup worker", issuePrefix: "CLN" }); + await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" }); + releases = vi.fn(async () => { if (failing) throw new Error("provider unavailable"); return undefined; }); + inspections = vi.fn(async (row: { providerLeaseId: string | null }) => ({ providerLeaseId: row.providerLeaseId, state: "destroyed" })); + reads = vi.spyOn(db, "select"); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + worker?.stop(); await coordinator?.stop(); coordinator = undefined; worker = undefined; + for (const stop of disposers.splice(0)) stop(); + vi.useRealTimers(); vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + async function start(historical = false, sweepOverride?: (backoffMs: number) => Promise) { + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", retryPendingSandboxTeardown: releases } as any] }); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true, + onError: error => errors.push(error) }); + const history = historical && historicalCleanupVerificationService(db, { inspectLease: inspections, inspectSetup: inspections }); + let historySettled = !history; + worker = registerEnvironmentCleanupWork(db, coordinator, { + sweepPending: sweepOverride ?? (backoffMs => heartbeat.sweepPendingCleanupLeases({ backoffMs })), + historical: history ? { hasPending: history.hasPending, + sweep: async () => { const result = await history.sweep(); historySettled = true; return result; } } : undefined, + canRun: () => !draining, + }); + await worker.ready; + await vi.waitFor(() => expect(historySettled).toBe(true)); + } + async function settle(ms = 30_000, wait = true) { + await vi.advanceTimersByTimeAsync(ms); + if (wait && !draining) { + await vi.waitFor(() => expect(idleWorkSnapshot().active === 0 || coordinator!.nextWakeAt() !== null).toBe(true), { timeout: 5000 }); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(() => expect(idleWorkSnapshot().active === 0 || coordinator!.nextWakeAt() !== null).toBe(true), { timeout: 5000 }); + } + } + const pending = (executor: Db = db) => environmentService(executor).insertPendingCleanupLease({ companyId, environmentId, + provider: "fixture", providerLeaseId: randomUUID(), metadata: { driver: "sandbox" }, failureReason: "fixture" }); + async function quiet() { + for (let pass = 0; pass < 4; pass++) { + await settle(30_000); + if (coordinator!.nextWakeAt() === null) break; + } + expect(coordinator!.nextWakeAt()).toBeNull(); + const count = reads.mock.calls.length, providerCalls = releases.mock.calls.length + inspections.mock.calls.length; + await settle(20 * 60_000); + expect(reads).toHaveBeenCalledTimes(count); + expect(releases.mock.calls.length + inspections.mock.calls.length).toBe(providerCalls); + } + it("stays quiet with an idle environment and verified history, and rechecks changed proof", async () => { + const lease = await pending(); + await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); + await start(true); expect(inspections).toHaveBeenCalledTimes(1); + await quiet(); + await environmentService(db).updateLeaseMetadata(lease.id, { driver: "sandbox", changed: true }); + await settle(); expect(inspections).toHaveBeenCalledTimes(2); + await environmentService(db).updateLeaseMetadata(lease.id, { changed: "again" }); + await settle(90_000); expect(inspections).toHaveBeenCalledTimes(2); + await settle(180_000); expect(inspections).toHaveBeenCalledTimes(3); + await quiet(); + }); + it("wakes on a real orphan insert, waits for its backoff, then disarms after deletion", async () => { + await start(); await quiet(); + const lease = await pending(); await settle(); + expect(releases).not.toHaveBeenCalled(); + const count = reads.mock.calls.length; + await settle(240_000); expect(reads).toHaveBeenCalledTimes(count); + await settle(30_000); expect(releases).toHaveBeenCalledTimes(1); + expect((await environmentService(db).getLeaseById(lease.id))?.status).toBe("expired"); + await quiet(); + }); + it("does not rescan pending cleanup from an otherwise idle run-reaper pass", async () => { + const lease = await pending(); + await db.update(environmentLeases).set({ updatedAt: new Date(Date.now() - 600_000) }).where(eq(environmentLeases.id, lease.id)); + const runtime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", retryPendingSandboxTeardown: releases } as any] }); + await heartbeatService(db, { environmentRuntime: runtime }).reapOrphanedRuns({ staleThresholdMs: 300_000 }); + expect(releases).not.toHaveBeenCalled(); + await start(); expect(releases).toHaveBeenCalledTimes(1); await quiet(); + }); + it("retries a concrete failed teardown and reconstructs its cooldown after restart", async () => { + const lease = await pending(); failing = true; + await start(); expect(releases).toHaveBeenCalledTimes(1); + const due = await nextEnvironmentCleanupAt(db); expect(due).toBeGreaterThan(Date.now()); + worker!.stop(); await coordinator!.stop(); + await start(); expect(releases).toHaveBeenCalledTimes(1); + failing = false; await settle(300_000); expect(releases).toHaveBeenCalledTimes(2); + expect((await environmentService(db).getLeaseById(lease.id))?.status).toBe("expired"); + await quiet(); + }); + it.each([false, true])("respects outer transaction settlement (rollback %s)", async rollback => { + await start(); await quiet(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { written = resolve; }); + const writing = db.transaction(async tx => { await pending(tx as unknown as Db); written(); await gate; if (rollback) throw new Error("rollback"); }) + .catch(error => { if (!rollback) throw error; }); + await ready; await settle(30_000, false); expect(releases).not.toHaveBeenCalled(); + release(); await writing; await settle(300_000); + expect(releases).toHaveBeenCalledTimes(rollback ? 0 : 1); await quiet(); + }); + it("preserves the first write during drain without passive SQL", async () => { + await start(); await quiet(); draining = true; + await pending(); const count = reads.mock.calls.length; + await settle(600_000); expect(reads).toHaveBeenCalledTimes(count); expect(releases).not.toHaveBeenCalled(); + draining = false; await settle(); expect(releases).toHaveBeenCalledTimes(1); await quiet(); + }); + it("wakes for deferred work from another runtime and retries while no durable row exists", async () => { + await start(); await quiet(); + let available = false, stop!: () => void; + const flush = vi.fn(async () => { + if (!available) return { recovered: 0, pending: 1 }; + await pending(); stop(); return { recovered: 1, pending: 0 }; + }); + stop = registerDeferredEnvironmentCleanup(db, flush); disposers.push(stop); + await settle(); expect(flush).toHaveBeenCalledTimes(1); + expect(await db.select().from(environmentLeases)).toEqual([]); + const count = reads.mock.calls.length; + await settle(240_000); expect(reads).toHaveBeenCalledTimes(count); + available = true; await settle(60_000); expect(flush).toHaveBeenCalledTimes(2); + await settle(300_000); expect(releases).toHaveBeenCalledTimes(1); await quiet(); + }); + it("retries a failed spool read without starving other buffers or persisted cleanup", async () => { + const lease = await pending(); + const stop = registerDeferredEnvironmentCleanup(db, async () => { throw new Error("spool unavailable"); }); + disposers.push(stop); + let stopHealthy!: () => void; + let bufferedLeaseId: string | undefined; + const healthy = vi.fn(async () => { + bufferedLeaseId = (await pending()).id; + stopHealthy(); + return { recovered: 1, pending: 0 }; + }); + stopHealthy = registerDeferredEnvironmentCleanup(db, healthy); disposers.push(stopHealthy); + await start(); + expect(healthy).toHaveBeenCalledTimes(1); + expect(releases).toHaveBeenCalledTimes(2); + expect((await environmentService(db).getLeaseById(lease.id))?.status).toBe("expired"); + expect((await environmentService(db).getLeaseById(bufferedLeaseId!))?.status).toBe("expired"); + expect(errors).toHaveLength(1); errors.length = 0; + draining = true; const count = reads.mock.calls.length; + await settle(90_000); expect(reads).toHaveBeenCalledTimes(count); + stop(); draining = false; await quiet(); + }); + it("retains startup recovery for an unread real spool with no buffered or durable row", async () => { + const dir = await mkdtemp(join(tmpdir(), "cleanup-unread-spool-")); + try { + await writeFile(join(dir, "unread.json"), "{ truncated"); + const runtime = environmentRuntimeService(db, { orphanCleanupSpool: createSandboxOrphanCleanupSpool(dir) }); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + await start(false, backoffMs => heartbeat.sweepPendingCleanupLeases({ backoffMs })); + expect(errors).toHaveLength(1); errors.length = 0; + expect(coordinator!.nextWakeAt()).not.toBeNull(); + expect(await db.select().from(environmentLeases)).toEqual([]); + await rm(join(dir, "unread.json")); + await quiet(); + } finally { await rm(dir, { recursive: true, force: true }); } + }); + it("does not delay startup for historical provider checks but joins them on shutdown", async () => { + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const sweep = vi.fn(() => gate); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + worker = registerEnvironmentCleanupWork(db, coordinator, { + sweepPending: async () => {}, historical: { sweep, hasPending: async () => false }, canRun: () => true, + }); + let ready = false, stopped = false; + void worker.ready.then(() => { ready = true; }); + try { + await vi.waitFor(() => expect(ready).toBe(true)); + expect(sweep).toHaveBeenCalledOnce(); + const stopping = coordinator.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1); expect(stopped).toBe(false); + release(); await stopping; expect(stopped).toBe(true); + } finally { release(); } + }); + it("coalesces a write during a slow sweep and waits for it on shutdown", async () => { + await start(); await quiet(); worker!.stop(); await coordinator!.stop(); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let calls = 0; + const starting = start(false, async () => { calls++; if (calls === 1) await gate; }); + await settle(1, false); await pending(); + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await settle(30_000, false); expect(stopped).toBe(false); expect(calls).toBe(1); + release(); await starting; await stopping; expect(stopped).toBe(true); + }); + it("rotates historical pages until all confirmed records are passive", async () => { + for (let i = 0; i < 33; i++) { + const lease = await pending(); await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); + } + await start(true); await settle(300_000); await settle(300_000); expect(inspections).toHaveBeenCalledTimes(33); await quiet(); + }); + it("keeps unknown historical resources scheduled until provider confirmation", async () => { + const lease = await pending(); await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); + inspections.mockResolvedValueOnce({ providerLeaseId: lease.providerLeaseId, state: "unknown" }); + await start(true); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await settle(300_000); expect(inspections).toHaveBeenCalledTimes(2); await quiet(); + }); +}); diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 8eb816fd340..b5f4964700f 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -187,6 +187,7 @@ describe("findReusableSandboxLeaseId", () => { describeEmbeddedPostgres("environmentRuntimeService", () => { let stopDb: (() => Promise) | null = null; let db!: ReturnType; + let connectionString: string; let runtime!: ReturnType; const fixtureRoots: string[] = []; // Give each test its own orphan-cleanup spool directory, so a spool file one @@ -198,13 +199,14 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { beforeAll(async () => { const started = await startEmbeddedPostgresTestDatabase("environment-runtime"); stopDb = started.stop; - db = createDb(started.connectionString); - runtime = environmentRuntimeService(db); + connectionString = started.connectionString; }); beforeEach(async () => { orphanCleanupSpoolDir = await mkdtemp(path.join(os.tmpdir(), "orphan-cleanup-spool-")); process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR = orphanCleanupSpoolDir; + db = createDb(connectionString); + runtime = environmentRuntimeService(db); }); afterEach(async () => { @@ -236,6 +238,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { await db.delete(companySecrets); await db.delete(projects); await db.delete(companies); + await db.$client.end({ timeout: 1 }); }); afterAll(async () => { @@ -1131,7 +1134,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { } if (!injected && fault === "journal write failed") { injected = true; - vi.spyOn(db, "update").mockImplementationOnce(() => { throw new Error("Database unavailable"); }); + vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("Database unavailable")); } throw Object.assign(new Error("Persist observed sandbox identity before cleanup"), { data: { schema: "paperclip/environment-creation-cleanup/v1", cleanup: observed }, diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index 2b5ec8ea3b9..b6451a62f1c 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -6,6 +6,7 @@ import { activityLog, agents, heartbeatRuns, + heartbeatRunEvents, companies, createDb, environmentLeases, @@ -74,6 +75,7 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { await db.delete(activityLog); await db.delete(environmentLeases); await db.delete(costEvents); + await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(environments); await db.delete(agents); @@ -1163,6 +1165,14 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { updatedAt: new Date(Date.now() - 60 * 60 * 1000), }); + // Inline pending cleanup now runs only after actual orphan recovery. Seed + // an abandoned run so this test still reaches the isolated error boundary. + const agentId = randomUUID(), runId = randomUUID(); + await db.insert(agents).values({ id: agentId, companyId, name: "Abandoned worker", + adapterType: "process", status: "idle" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, + status: "running", invocationSource: "manual", startedAt: new Date(Date.now() - 3_600_000) }); + // Make the sweep itself throw. The lease query rejects with an error that // carries a credential-shaped sentinel in both the name and the message. The // reaper's own queries pass through, so only the sweep fails. @@ -1194,7 +1204,7 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { }); // The reaper isolates the sweep, so the reaper itself resolves. - await expect(heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 })).resolves.toBeDefined(); + await expect(heartbeat.reapOrphanedRuns({ staleThresholdMs: 0 })).resolves.toEqual({ reaped: 1, runIds: [runId] }); const sweepCall = vi .mocked(logger.error) diff --git a/server/src/__tests__/plugin-warm-tasks.integration.test.ts b/server/src/__tests__/plugin-warm-tasks.integration.test.ts index d6a9f166683..3846c47dae8 100644 --- a/server/src/__tests__/plugin-warm-tasks.integration.test.ts +++ b/server/src/__tests__/plugin-warm-tasks.integration.test.ts @@ -1,3 +1,4 @@ +import { flushDeferredEnvironmentCleanup, hasDeferredEnvironmentCleanup, registerDeferredEnvironmentCleanup, subscribeDeferredEnvironmentCleanup } from "../services/environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; @@ -127,6 +128,14 @@ const support=await getEmbeddedPostgresTestSupport(); const f=await fixture();await f.nextRun();const next=await f.acquire();f.stopped(); if(mode==="cleanup-error" || mode==="spool") f.failRelease(); const append=vi.fn(async()=>true); + let delayFlush=false,flushCalls=0,releaseFlush!:()=>void; + const flushGate=new Promise(resolve=>{releaseFlush=resolve;}); + const stopGate=mode==="spool" ? registerDeferredEnvironmentCleanup(db,async()=>{ + if(delayFlush && ++flushCalls===1) await flushGate; + return {recovered:0,pending:0}; + }) : undefined; + const wake=vi.fn(); + const unsubscribe=subscribeDeferredEnvironmentCleanup(db,wake); const spool=mode==="spool" ? vi.spyOn(orphanCleanup,"createSandboxOrphanCleanupSpool").mockReturnValue({append,load:async()=>[],remove:async()=>{}}) : undefined; const transaction=mode==="spool" ? vi.spyOn(db,"transaction").mockRejectedValue(Error("database unavailable")) : undefined; const original=db.update.bind(db); @@ -140,12 +149,30 @@ const support=await getEmbeddedPostgresTestSupport(); try { await expect(f.runtime().task({companyId:f.companyId,leaseId:next.lease.id,operation:{kind:"describe"}})) .rejects.toThrow("reconcile the same task"); - } finally {update.mockRestore();transaction?.mockRestore();spool?.mockRestore();} + if(mode==="spool") { + expect(wake).toHaveBeenCalledOnce(); + expect(hasDeferredEnvironmentCleanup(db)).toBe(true); + await expect(flushDeferredEnvironmentCleanup(db)).rejects.toThrow("Deferred orphan cleanup failed"); + expect(hasDeferredEnvironmentCleanup(db)).toBe(true); + } + } finally {update.mockRestore();transaction?.mockRestore();spool?.mockRestore();unsubscribe();} expect(f.calls).toContain("environmentReleaseLease"); if(mode==="spool") expect(append).toHaveBeenCalledWith(expect.objectContaining({providerLeaseId:"task-2",metadata:expect.objectContaining({driver:"plugin",runnerTaskDispatched:false})})); const rows=await db.select().from(environmentLeases).where(eq(environmentLeases.companyId,f.companyId)); expect(rows.filter(row=>row.providerLeaseId === "task-2")).toEqual(mode==="cleanup-error" ? [expect.objectContaining({status:"pending_cleanup",failureReason:"plugin_task_replacement_unowned",metadata:expect.objectContaining({runnerTaskDispatched:false})})] : []); + if(mode==="spool") { + delayFlush=true; + const delayed=flushDeferredEnvironmentCleanup(db); + try { expect(await flushDeferredEnvironmentCleanup(db)).toEqual({recovered:1,pending:0}); } + finally { releaseFlush(); stopGate!(); } + expect(await delayed).toEqual({recovered:1,pending:0}); + expect(hasDeferredEnvironmentCleanup(db)).toBe(false); + const recovered=await db.select().from(environmentLeases).where(eq(environmentLeases.companyId,f.companyId)); + expect(recovered.filter(row=>row.providerLeaseId==="task-2")).toEqual([ + expect.objectContaining({status:"pending_cleanup",failureReason:"plugin_task_replacement_unowned"}), + ]); + } }); it.each(["added","removed"] as const)("retires retained mounts when an authorized project is %s after controller restart",async mode=>{ diff --git a/server/src/__tests__/sandbox-orphan-cleanup-spool.test.ts b/server/src/__tests__/sandbox-orphan-cleanup-spool.test.ts index 74b984114b5..76b86694459 100644 --- a/server/src/__tests__/sandbox-orphan-cleanup-spool.test.ts +++ b/server/src/__tests__/sandbox-orphan-cleanup-spool.test.ts @@ -1,7 +1,8 @@ +import { promises as fs } from "node:fs"; import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createSandboxOrphanCleanupSpool, type DeferredOrphanCleanupRecord, @@ -86,6 +87,21 @@ describe("createSandboxOrphanCleanupSpool", () => { expect(loaded[0]?.providerLeaseId).toBe("sandbox://fake/lease-1"); }); + it("distinguishes unread storage from empty storage without dropping readable records", async () => { + const spool = createSandboxOrphanCleanupSpool(dir); + const record = buildRecord(); + await spool.append(record); + const denied = vi.spyOn(fs, "readdir").mockRejectedValueOnce(Object.assign(new Error("denied"), { code: "EACCES" })); + try { expect(await spool.loadWithStatus!()).toEqual({ records: [], readFailed: true }); } + finally { denied.mockRestore(); } + await writeFile(path.join(dir, "broken.json"), "{ truncated"); + expect(await spool.loadWithStatus!()).toEqual({ records: [record], readFailed: true }); + await rm(path.join(dir, "broken.json")); + expect(await spool.loadWithStatus!()).toEqual({ records: [record], readFailed: false }); + expect(await createSandboxOrphanCleanupSpool(path.join(dir, "missing")).loadWithStatus!()) + .toEqual({ records: [], readFailed: false }); + }); + it("returns an empty list when the spool directory does not exist", async () => { const spool = createSandboxOrphanCleanupSpool(path.join(dir, "missing")); expect(await spool.load()).toEqual([]); diff --git a/server/src/index.ts b/server/src/index.ts index 814c63ebd9b..905605d2e65 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerEnvironmentCleanupWork } from "./services/environment-cleanup-work.js"; import { loginCleanupSchedule } from "./services/login-cleanup-state.js"; import { registerLoginCleanupWork } from "./services/login-cleanup-work.js"; import { timeStartupRecovery } from "./startup-recovery-timing.js"; @@ -1304,8 +1305,8 @@ async function startServerWithDatabaseTeardown( // foreign-company insert tears the provisioned sandbox down. If that teardown // also fails, the acquire records a lease-less `pending_cleanup` lease row. No // other path releases that sandbox, so the master pending-cleanup sweep retries - // the provider teardown and releases the orphan. The sweep runs on startup and - // on the scheduler interval. + // the provider teardown and releases the orphan. The sweep runs on startup, + // committed writes, and deadlines derived from unresolved cleanup. // // This backstop is independent of the heartbeat scheduler toggle. A leaked // provider sandbox costs money whether or not the instance schedules @@ -1315,13 +1316,8 @@ async function startServerWithDatabaseTeardown( // The master pending-cleanup sweep is the single owner of these rows. Its // atomic per-attempt claim makes two overlapping sweeps safe, so the enabled // path can also run the sweep from the orphaned-run reaper without a second - // teardown. The heartbeat scheduler owns the sweep when it is enabled; the - // disabled path creates its own runtime to own the same sweep. - // The interval sweep waits this long after a lease's last write before it - // retries the teardown. The window matches the orphaned-run reaper staleness, - // so a just-failed lease does not draw a retry on every tick. The startup - // sweep passes zero, so a restart retries a stranded orphan at once. - const ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS = 5 * 60 * 1000; + // teardown. The delivery coordinator owns pending-work scheduling whether + // or not the heartbeat scheduler is enabled. const environmentLeaseCleanupHeartbeat = accountingHeartbeat; const chatCompletionDeliveries = chatCompletionDeliveryService(db as any, environmentLeaseCleanupHeartbeat); const connectionDeliveries = connectionIntentDeliveryService(db as any, environmentLeaseCleanupHeartbeat); @@ -1386,31 +1382,21 @@ async function startServerWithDatabaseTeardown( inspectLease: lease => cleanupInspectionRuntime.inspectLeaseResource(lease.id), inspectSetup: setup => environmentCustomImages.inspectSetupResource(setup), }); - let nextHistoricalCleanupCheck = 0; - const scheduleHistoricalCleanupVerification = () => { - if (!isCloudManagedInstance() || isIdleTaskDrainActive() || isWarmStandby() || heartbeatSchedulerStopped || - Date.now() < nextHistoricalCleanupCheck) return; - nextHistoricalCleanupCheck = Date.now() + 5 * 60_000; - // Provider reads never delay readiness or the ordinary cleanup sweep. - trackHeartbeatSchedulerWork(historicalCleanup.sweep().then(result => { - if (result.verified > 0) logger.info(result, "verified historical provider cleanup"); - }).catch(() => logger.warn("historical provider cleanup verification deferred"))); + const runEnvironmentLeaseCleanupSweep = async (backoffMs: number) => { + const result = await environmentLeaseCleanupHeartbeat.sweepPendingCleanupLeases({ backoffMs }); + if (result.destroyed > 0 || result.capped > 0) { + logger.info(result, "environment lease cleanup retried sandbox teardowns"); + } }; - const runEnvironmentLeaseCleanupSweep = (backoffMs: number) => - environmentLeaseCleanupHeartbeat - .sweepPendingCleanupLeases({ backoffMs }) - .then((result) => { - scheduleHistoricalCleanupVerification(); - if (result.destroyed > 0 || result.capped > 0) { - logger.info(result, "environment lease cleanup sweep retried orphan sandbox teardowns"); - } - }) - .catch((err) => { - logger.error({ err }, "environment lease cleanup sweep failed"); - }); - const scheduleEnvironmentLeaseCleanupSweep = () => { - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(runEnvironmentLeaseCleanupSweep(ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS)); + const startEnvironmentCleanup = async () => { + const worker = registerEnvironmentCleanupWork(db, deliveryWork, { + sweepPending: runEnvironmentLeaseCleanupSweep, + historical: isCloudManagedInstance() ? historicalCleanup : undefined, + canRun: () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + minimumSpacingMs: config.heartbeatSchedulerIntervalMs, + }); + server.on("close", worker.stop); + await worker.ready; }; const githubConnectionEvents = githubConnectionEventService(db as any, { wakeup: environmentLeaseCleanupHeartbeat.wakeup, @@ -1790,7 +1776,7 @@ async function startServerWithDatabaseTeardown( ], () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, config.heartbeatSchedulerIntervalMs).ready); // Retry any orphan sandbox teardown left by a failed acquire before a // server restart, so a leaked sandbox does not stay allocated. - await step("environment_lease_cleanup", () => runEnvironmentLeaseCleanupSweep(0)); + await step("environment_lease_cleanup", startEnvironmentCleanup); await step("retention_sweep", runRetentionSweep); }; @@ -1824,7 +1810,6 @@ async function startServerWithDatabaseTeardown( scheduleMergedPullRequestConfirmationSweep(); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); - scheduleEnvironmentLeaseCleanupSweep(); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduled.routines.wake(); @@ -1971,12 +1956,11 @@ async function startServerWithDatabaseTeardown( } else { // The heartbeat scheduler is disabled, but the orphan-sandbox cleanup sweep // is still required. A failed acquire can leak a paid provider sandbox, so - // this path retries the teardown at startup and on the interval, exactly as - // the enabled path does. - await runEnvironmentLeaseCleanupSweep(0); + // this path registers the same startup recovery and pending-work deadlines + // as the enabled path. + await startEnvironmentCleanup(); startHeartbeatSchedulerInterval(() => { scheduleExternalObjectRefreshSweep(new Date()); - scheduleEnvironmentLeaseCleanupSweep(); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); }); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index d0915bae7f0..c4b3b9f2334 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -7,6 +7,8 @@ export const DELIVERY_QUEUES = { executionContinuation: "execution-continuation", executionStatus: "execution-status", executionDisposition: "execution-disposition", + environmentCleanup: "environment-cleanup", + historicalCleanup: "historical-cleanup", adapterLoginCleanup: "adapter-login-cleanup", setupTokenCleanup: "setup-token-cleanup", localLoginCleanup: "local-login-cleanup", diff --git a/server/src/services/environment-cleanup-notifications.ts b/server/src/services/environment-cleanup-notifications.ts new file mode 100644 index 00000000000..003288bb769 --- /dev/null +++ b/server/src/services/environment-cleanup-notifications.ts @@ -0,0 +1,49 @@ +import type { Db } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export async function notifyEnvironmentCleanup(transaction: object): Promise { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.environmentCleanup); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.historicalCleanup); +} + +export function writeEnvironmentCleanup(db: Db, write: (tx: Db) => PromiseLike): Promise { + return db.transaction(async tx => { + await notifyEnvironmentCleanup(tx); + return write(tx as unknown as Db); + }); +} + +type Flush = () => Promise<{ recovered: number; pending: number }>; +const deferred = new WeakMap; listeners: Set<() => void> }>(); +function state(db: object) { + let value = deferred.get(db); + if (!value) { value = { flushers: new Set(), listeners: new Set() }; deferred.set(db, value); } + return value; +} +export function registerDeferredEnvironmentCleanup(db: object, flush: Flush): () => void { + const value = state(db); + value.flushers.add(flush); + for (const wake of value.listeners) wake(); + return () => { value.flushers.delete(flush); }; +} +export function subscribeDeferredEnvironmentCleanup(db: object, wake: () => void): () => void { + const value = state(db); + value.listeners.add(wake); + return () => { value.listeners.delete(wake); }; +} +export function hasDeferredEnvironmentCleanup(db: object): boolean { + return (deferred.get(db)?.flushers.size ?? 0) > 0; +} +export async function flushDeferredEnvironmentCleanup(db: object, additional: Iterable = []): Promise<{ recovered: number; pending: number }> { + let recovered = 0, pending = 0, failures = 0; + for (const flush of new Set([...(deferred.get(db)?.flushers ?? []), ...additional])) { + try { + const result = await flush(); + recovered += result.recovered; pending += result.pending; + } catch { failures++; } + } + // Attempt every runtime before retaining the retry. Provider/spool errors may + // contain credentials, so report only a count rather than their payloads. + if (failures > 0) throw new Error(`Deferred orphan cleanup failed for ${failures} runtime(s)`); + return { recovered, pending }; +} diff --git a/server/src/services/environment-cleanup-state.ts b/server/src/services/environment-cleanup-state.ts new file mode 100644 index 00000000000..177ec9d1906 --- /dev/null +++ b/server/src/services/environment-cleanup-state.ts @@ -0,0 +1,26 @@ +import { eq, sql } from "drizzle-orm"; +import { environmentLeases, type Db } from "@paperclipai/db"; +import { hasDeferredEnvironmentCleanup } from "./environment-cleanup-notifications.js"; + +export const ENVIRONMENT_CLEANUP_RETRY_MS = 5 * 60_000; + +export function pendingCleanupDeadlineSql(now = Date.now()) { + const deadline = sql`case + when ${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true' + and ${environmentLeases.metadata} ? 'pendingCleanupLeaseExpiresAtMs' + then ${environmentLeases.metadata}->'pendingCleanupLeaseExpiresAtMs' + else ${environmentLeases.metadata}->'pendingCleanupRetryAfterMs' end`; + return sql`case when jsonb_typeof(${deadline}) = 'number' then + case when (${deadline})::numeric <= ${now + 30 * 60_000 + 1_000} + then greatest(0, (${deadline})::numeric) else 0 end + else 0 end`; +} + +export async function nextEnvironmentCleanupAt(db: Db, backoffMs = ENVIRONMENT_CLEANUP_RETRY_MS): Promise { + const [row] = await db.select({ next: sql`min(greatest( + extract(epoch from ${environmentLeases.updatedAt}) * 1000 + ${backoffMs}, + ${pendingCleanupDeadlineSql()}))` }).from(environmentLeases).where(eq(environmentLeases.status, "pending_cleanup")); + const saved = row?.next == null ? null : Number(row.next); + const buffered = hasDeferredEnvironmentCleanup(db) ? Date.now() + backoffMs : null; + return saved === null ? buffered : buffered === null ? saved : Math.min(saved, buffered); +} diff --git a/server/src/services/environment-cleanup-work.ts b/server/src/services/environment-cleanup-work.ts new file mode 100644 index 00000000000..2370762e2ce --- /dev/null +++ b/server/src/services/environment-cleanup-work.ts @@ -0,0 +1,51 @@ +import type { Db } from "@paperclipai/db"; +import { ENVIRONMENT_CLEANUP_RETRY_MS, nextEnvironmentCleanupAt } from "./environment-cleanup-state.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, type DeliveryQueue } from "./delivery-work-notifications.js"; +import { subscribeDeferredEnvironmentCleanup } from "./environment-cleanup-notifications.js"; + +export function registerEnvironmentCleanupWork( + db: Db, + coordinator: ReturnType, + input: { + sweepPending: (backoffMs: number) => Promise; + historical?: { sweep: () => Promise; hasPending: () => Promise }; + canRun: () => boolean; + minimumSpacingMs?: number; + }, +) { + const spacing = input.minimumSpacingMs ?? 30_000; + function register(queue: DeliveryQueue, run: () => Promise, deadline: () => Promise, minimumSpacing = spacing) { + let requested = 1, reconciled = 0, next: number | null = null, earliest = 0; + const worker = coordinator.register(queue, { + retryMs: spacing, pausePassiveMaintenance: true, + onSettled: () => { requested++; }, + run: async () => { + if (!input.canRun()) { requested++; next = null; return; } + if (Date.now() < earliest) { next = earliest; return; } + const generation = ++requested; + earliest = Date.now() + minimumSpacing; + await run(); + const due = await deadline(); + next = due === null ? null : Math.max(due, earliest); + reconciled = generation; + }, + hasPending: async () => requested !== reconciled, + nextRunAt: () => next, + }); + return { ready: worker.ready, wake: () => { requested++; worker.wake(); } }; + } + let initial = true; + const pending = register(DELIVERY_QUEUES.environmentCleanup, async () => { + await input.sweepPending(initial ? 0 : ENVIRONMENT_CLEANUP_RETRY_MS); + initial = false; + }, () => nextEnvironmentCleanupAt(db)); + const unsubscribe = subscribeDeferredEnvironmentCleanup(db, pending.wake); + const historical = input.historical; + if (historical) register(DELIVERY_QUEUES.historicalCleanup, historical.sweep, + async () => await historical.hasPending() ? Date.now() + ENVIRONMENT_CLEANUP_RETRY_MS : null, + Math.max(spacing, ENVIRONMENT_CLEANUP_RETRY_MS)); + // The coordinator tracks and joins historical work; provider checks must not + // delay server readiness or starting the heartbeat scheduler. + return { ready: pending.ready, stop: unsubscribe }; +} diff --git a/server/src/services/environment-custom-images.ts b/server/src/services/environment-custom-images.ts index 7f81a935086..324065e84d5 100644 --- a/server/src/services/environment-custom-images.ts +++ b/server/src/services/environment-custom-images.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { and, desc, eq, inArray, lte } from "drizzle-orm"; @@ -594,7 +595,7 @@ export function environmentCustomImageService( ): Promise { const status = normalizePersistedStatus(providerSession.status, fallbackStatus); const now = new Date(); - const row = await db + const row = await writeEnvironmentCleanup(db, tx => tx .update(environmentCustomImageSetupSessions) .set({ providerLeaseId: providerSession.providerLeaseId ?? null, @@ -607,7 +608,7 @@ export function environmentCustomImageService( }) .where(eq(environmentCustomImageSetupSessions.id, session.id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!row) throw notFound("Environment customImage setup session not found"); return toSession(row); } @@ -619,7 +620,7 @@ export function environmentCustomImageService( metadata?: Record | null; }): Promise { const now = new Date(); - const row = await db + const row = await writeEnvironmentCleanup(db, tx => tx .update(environmentCustomImageSetupSessions) .set({ status: input.status, @@ -630,7 +631,7 @@ export function environmentCustomImageService( }) .where(eq(environmentCustomImageSetupSessions.id, input.sessionId)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!row) throw notFound("Environment customImage setup session not found"); return toSession(row); } @@ -1018,6 +1019,7 @@ export function environmentCustomImageService( supersededRows.map((row) => row.id), )); } + await notifyEnvironmentCleanup(tx); await tx .update(environmentCustomImageSetupSessions) .set({ diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 71cca45a3e1..edf646410d4 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,3 +1,4 @@ +import { flushDeferredEnvironmentCleanup, registerDeferredEnvironmentCleanup, writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { executeEnvironmentTask, pluginEnvironmentSupportsTasks, settleDispatchedEnvironmentTask } from "./environment-task-runtime.js"; import type { PluginEnvironmentTaskOperation } from "@paperclipai/plugin-sdk"; import { allowLegacyShutdownWorkspaceCleanup, legacyShutdownWorkspaceResourceProtected } from "./legacy-shutdown-workspace-settlement.js"; @@ -1350,12 +1351,14 @@ function createSandboxEnvironmentDriver( // Keep the token through queue splices, failed flushes and buffer overflow. // Release only when the database durably owns cleanup of this exact orphan. const orphanIdleWork = new Map void>(); + let unregisterDeferredCleanup: (() => void) | undefined; const orphanIdleKey = (record: DeferredOrphanCleanupRecord) => JSON.stringify([ record.companyId, record.provider, record.providerLeaseId ?? record, ]); const trackOrphanIdleWork = (record: DeferredOrphanCleanupRecord) => { const key = orphanIdleKey(record); if (!orphanIdleWork.has(key)) orphanIdleWork.set(key, beginIdleTrackedWork()); + unregisterDeferredCleanup ??= registerDeferredEnvironmentCleanup(db, flushDeferredOrphanCleanups); }; const DEFERRED_ORPHAN_CLEANUP_FAILURE_REASON = "acquire_rejected_teardown_failed"; @@ -1390,10 +1393,15 @@ function createSandboxEnvironmentDriver( }; const flushDeferredOrphanCleanupsOnce = async (): Promise<{ recovered: number; pending: number }> => { - // Other task paths can append after startup; each pass must ingest their records. - for (const record of await orphanCleanupSpool.load()) enqueueDeferredOrphanCleanup(record); + // Other task paths append after startup. Read on each concrete cleanup pass; + // the worker disarms entirely when no pending obligation remains. + const loaded = orphanCleanupSpool.loadWithStatus + ? await orphanCleanupSpool.loadWithStatus() + : { records: await orphanCleanupSpool.load(), readFailed: false }; + for (const record of loaded.records) enqueueDeferredOrphanCleanup(record); if (deferredOrphanCleanups.length === 0) { - return { recovered: 0, pending: 0 }; + if (loaded.readFailed) throw new Error("Orphan cleanup spool has unread records"); + return { recovered: 0, pending: orphanIdleWork.size }; } // New acquisitions can enqueue while this batch is being persisted. const batch = deferredOrphanCleanups.splice(0, deferredOrphanCleanups.length); @@ -1441,7 +1449,9 @@ function createSandboxEnvironmentDriver( ); } } - return { recovered, pending: deferredOrphanCleanups.length }; + if (orphanIdleWork.size === 0) { unregisterDeferredCleanup?.(); unregisterDeferredCleanup = undefined; } + if (loaded.readFailed) throw new Error("Orphan cleanup spool has unread records"); + return { recovered, pending: orphanIdleWork.size }; }; let orphanFlush: Promise<{ recovered: number; pending: number }> | null = null; const flushDeferredOrphanCleanups = (): Promise<{ recovered: number; pending: number }> => { @@ -4021,7 +4031,7 @@ export function environmentRuntimeService( // Persist recovery ownership before any provider work. Re-check scope and // run liveness in this write: the earlier selection cannot fence a resume. // The in-flight lease also excludes concurrent closures and cleanup sweeps. - const [claimed] = await db.update(environmentLeases).set({ + const [claimed] = await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", failureReason: input.failureReason, cleanupStatus: "failed", @@ -4045,7 +4055,7 @@ export function environmentRuntimeService( WHERE ${heartbeatRuns.id} = ${environmentLeases.heartbeatRunId} AND ${heartbeatRuns.status} IN ('queued', 'scheduled_retry', 'running') )`, - )).returning(); + )).returning()); if (!claimed) return null; const lease = toEnvironmentLeaseSnapshot(claimed); let renewalInFlight = false; @@ -4080,11 +4090,11 @@ export function environmentRuntimeService( clearInterval(renewal); // A crash skips this write; the bounded ownership lease lets the sweep // recover. A completed failure becomes eligible for a later explicit retry. - await db.update(environmentLeases).set({ + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ metadata: sql`${environmentLeases.metadata} || '{"pendingCleanupInFlight":false}'::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${attemptId}`, - )).catch(() => { + ))).catch(() => { logger.warn({ leaseId: lease.id }, "scoped cleanup ownership settlement failed"); }); } @@ -4240,7 +4250,7 @@ export function environmentRuntimeService( if (hasStopOnlyCleanup(leaseSnapshot)) { const intentKey = stopOnlyCleanupKey(leaseSnapshot); const attemptId = randomUUID(); - const [claimed] = await db.update(environmentLeases).set({ + const [claimed] = await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", releasedAt: new Date(), metadata: sql`${environmentLeases.metadata} || ${JSON.stringify({ pendingCleanupAttemptId: attemptId, pendingCleanupInFlight: true, pendingCleanupLeaseExpiresAtMs: Date.now() + 15 * 60_000 })}::jsonb`, @@ -4248,7 +4258,7 @@ export function environmentRuntimeService( inArray(environmentLeases.status, ["active", "pending_cleanup"]), sql`coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') = 'false'`, sql`${environmentLeases.metadata}->${intentKey} = ${JSON.stringify(leaseSnapshot.metadata?.[intentKey])}::jsonb`, - )).returning(); + )).returning()); if (!claimed) continue; const snapshot = toEnvironmentLeaseSnapshot(claimed); try { @@ -4419,15 +4429,8 @@ export function environmentRuntimeService( // it down. The dispatcher sums the recovered and pending counts across all // drivers. A driver with no buffer contributes nothing. async flushDeferredOrphanCleanups(): Promise<{ recovered: number; pending: number }> { - let recovered = 0; - let pending = 0; - for (const driver of drivers.values()) { - if (!driver.flushDeferredOrphanCleanups) continue; - const result = await driver.flushDeferredOrphanCleanups(); - recovered += result.recovered; - pending += result.pending; - } - return { recovered, pending }; + return flushDeferredEnvironmentCleanup(db, Array.from(drivers.values()) + .flatMap(driver => driver.flushDeferredOrphanCleanups ? [driver.flushDeferredOrphanCleanups] : [])); }, async destroyReusableSandboxLeases(input: { diff --git a/server/src/services/environment-task-runtime.ts b/server/src/services/environment-task-runtime.ts index e7089e243fd..e58491b01fa 100644 --- a/server/src/services/environment-task-runtime.ts +++ b/server/src/services/environment-task-runtime.ts @@ -1,3 +1,5 @@ +import { beginIdleTrackedWork } from "./task-admission.js"; +import { registerDeferredEnvironmentCleanup, writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { environmentService } from "./environments.js"; import { createSandboxOrphanCleanupSpool, type DeferredOrphanCleanupRecord } from "./sandbox-orphan-cleanup-spool.js"; import { logger } from "../middleware/logger.js"; @@ -245,7 +247,22 @@ export async function executeEnvironmentTask(db: Db, workers: PluginWorkerManage await environmentService(db).insertPendingCleanupLease({...orphan,failureReason:"plugin_task_replacement_unowned"}); } catch { // The existing orphan spool recovers cleanup even when the database is unavailable. - const persisted=await createSandboxOrphanCleanupSpool().append(orphan); + const spool = createSandboxOrphanCleanupSpool(); + const persisted = await spool.append(orphan); + const finish = beginIdleTrackedWork(); + let flushing: Promise<{ recovered: number; pending: number }> | null = null; + const unregister = registerDeferredEnvironmentCleanup(db, () => { + flushing ??= (async () => { + await environmentService(db).insertPendingCleanupLease({...orphan,failureReason:"plugin_task_replacement_unowned"}); + await spool.remove(orphan); + unregister(); + finish(); + return { recovered: 1, pending: 0 }; + })().catch(error => { flushing = null; throw error; }); + // Keep successful completion for a concurrent sweep that already + // captured this callback before it was unregistered. + return flushing; + }); logger.error({providerLeaseId:fresh.providerLeaseId,persisted},"Plugin replacement cleanup requires recovery"); } } @@ -352,9 +369,9 @@ export async function settleDispatchedEnvironmentTask(db: Db, workers: PluginWor await executeEnvironmentTask(db, workers, { ...scope, operation: { kind: "complete" } }); return true; } catch { - await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", failureReason: "plugin_runner_task_cleanup_unconfirmed", releasedAt: new Date(), updatedAt: new Date(), - }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId))); + }).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, lease.companyId)))); throw new Error("Plugin Runner termination is unconfirmed; the saved lease requires cleanup"); } } diff --git a/server/src/services/environments.ts b/server/src/services/environments.ts index 97a04b7732a..ec5abac4cae 100644 --- a/server/src/services/environments.ts +++ b/server/src/services/environments.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; import { randomUUID } from "node:crypto"; @@ -1670,7 +1671,7 @@ export function environmentService(db: Db) { : sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`; const cleanupSettled = options?.expectedPendingCleanupAttemptId && options.cleanupStatus === "success" && status !== "pending_cleanup"; - const row = await db + const row = await writeEnvironmentCleanup(db, tx => tx .update(environmentLeases) .set({ status, @@ -1690,7 +1691,7 @@ export function environmentService(db: Db) { sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`) : undefined)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); return row ? toEnvironmentLease(row) : null; }, @@ -1739,6 +1740,7 @@ export function environmentService(db: Db) { .where(eq(environments.id, input.environmentId)) .for("update"); const environmentIdForRow = environmentRows[0]?.id ?? null; + await notifyEnvironmentCleanup(tx); return tx .insert(environmentLeases) .values({ @@ -1774,7 +1776,7 @@ export function environmentService(db: Db) { id: string, metadata: Record | null, ): Promise => { - const row = await db + const row = await writeEnvironmentCleanup(db, tx => tx .update(environmentLeases) .set({ metadata: withoutCleanupVerification(metadata), @@ -1783,7 +1785,7 @@ export function environmentService(db: Db) { }) .where(eq(environmentLeases.id, id)) .returning() - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); return row ? toEnvironmentLease(row) : null; }, @@ -1792,7 +1794,7 @@ export function environmentService(db: Db) { status: Extract = "released", ): Promise => { const now = new Date(); - const rows = await db + const rows = await writeEnvironmentCleanup(db, tx => tx .update(environmentLeases) .set({ status, @@ -1806,7 +1808,7 @@ export function environmentService(db: Db) { eq(environmentLeases.status, "active"), ), ) - .returning(); + .returning()); return rows.map(toEnvironmentLease); }, }; diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 2d32030510e..73c586bdff3 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,5 @@ +import { pendingCleanupDeadlineSql } from "../environment-cleanup-state.js"; +import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { createHeartbeatAgentStatus } from "./agent-status.js"; import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; @@ -211,20 +213,9 @@ function pendingCleanupAttemptsSql() { } function pendingCleanupRetryDueSql(explicitRetry = false) { - // Ownership renewal and retry cooldown are separate clocks. A late renewal - // cannot change when a completed attempt may retry. Older claims used the - // retry field for both clocks, so retain that fallback until they settle. - const deadline = sql`case - when ${environmentLeases.metadata}->>'pendingCleanupInFlight' = 'true' - and ${environmentLeases.metadata} ? 'pendingCleanupLeaseExpiresAtMs' - then ${environmentLeases.metadata}->'pendingCleanupLeaseExpiresAtMs' - else ${environmentLeases.metadata}->'pendingCleanupRetryAfterMs' end`; return sql`case when ${explicitRetry} and coalesce(${environmentLeases.metadata}->>'pendingCleanupInFlight', 'false') != 'true' then true - when jsonb_typeof(${deadline}) = 'number' - then (${deadline})::numeric <= ${Date.now()} - or (${deadline})::numeric > ${Date.now() + 30 * 60_000 + 1_000} - else true end`; + else ${pendingCleanupDeadlineSql()} <= ${Date.now()} end`; } // Choose the `jsonb_set` target root. A provider can write a scalar or array @@ -1533,7 +1524,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // 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. - const flipped = await db + const flipped = await writeEnvironmentCleanup(db, tx => tx .update(environmentLeases) .set({ status: "pending_cleanup", @@ -1545,7 +1536,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD eq(environmentLeases.status, "active"), ), ) - .returning({ id: environmentLeases.id }); + .returning({ id: environmentLeases.id })); if (flipped.length > 0) recovered += 1; } @@ -1575,6 +1566,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD const now = new Date(); const cutoff = new Date(now.getTime() - backoffMs); + let deferredFlushFailed = false; // Flush the in-process orphan-cleanup buffer first. A failed acquire buffers // an orphan there when every synchronous pending-cleanup write failed after a // failed teardown. The flush re-inserts each buffered record, so a durable @@ -1593,9 +1585,8 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // later tick, and the database rows below still need this sweep. The caught // exception never enters the log, because a write error can carry a // credential in its message, code, cause, or stack. - logger.warn( - "orphan sandbox cleanup buffer flush failed; the sweep continues", - ); + logger.warn("orphan sandbox cleanup buffer flush failed; cleanup will retry"); + deferredFlushFailed = true; } const rows = await db @@ -1781,14 +1772,14 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } // Persist the cooldown independently of process memory. A crash before // this write leaves the bounded in-flight lease for a later sweep. - await db.update(environmentLeases).set({ + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ metadata: sql`${pendingCleanupMetadataObjectSql()} || ${JSON.stringify({ pendingCleanupInFlight: false, pendingCleanupRetryAfterMs: Date.now() + (attempts + 1 >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP ? 30 * 60_000 : Math.min(30 * 60_000, Math.max(30_000, backoffMs))), })}::jsonb`, }).where(and(eq(environmentLeases.id, lease.id), - sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${claimed}`)); + sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${claimed}`))); if (lease.heartbeatRunId) { // Delivery failure must not revert successful provider cleanup. A new // message can still use the persisted receipt on its next admission. @@ -1800,6 +1791,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD } } + if (deferredFlushFailed) throw new Error("Deferred orphan cleanup flush failed"); return { swept: rows.length, destroyed, capped }; } @@ -2504,11 +2496,13 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // Recover an active lease whose run already ended before the same-tick // pending_cleanup sweep, so this tick can stop the recovered sandbox. // Isolate the sweep so its failure never hides the reaper result. + let recoveredActiveLeases = 0; try { await sweepLegacyShutdownWorkspaceSettlements(db); const orphanedActiveLeaseSweep = await sweepOrphanedActiveLeases({ backoffMs: staleThresholdMs, }); + recoveredActiveLeases = orphanedActiveLeaseSweep.recovered; if (orphanedActiveLeaseSweep.recovered > 0) { logger.warn( { recovered: orphanedActiveLeaseSweep.recovered }, @@ -2524,10 +2518,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD ); } - // Retry stranded pending_cleanup leases on the same tick. Isolate the sweep - // so its failure never hides the reaper result. The backoff equals the - // reaper staleness threshold. - try { + if (reaped.length > 0 || recoveredActiveLeases > 0) try { const sweep = await sweepPendingCleanupLeases({ backoffMs: staleThresholdMs, }); diff --git a/server/src/services/heartbeat/run-control.ts b/server/src/services/heartbeat/run-control.ts index cf9410ab0b4..e9b21a98aa3 100644 --- a/server/src/services/heartbeat/run-control.ts +++ b/server/src/services/heartbeat/run-control.ts @@ -1,3 +1,4 @@ +import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { readTaskRetryIntent, reconcileTaskChatRetryReceipts } from "../task-retry.js"; import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, readNativeLocalProcessStop, recordedLocalExecutionStopped } from "../native-local-process-stop.js"; import { isHeartbeatRunTerminalStatus } from "./run-lifecycle.js"; @@ -698,6 +699,7 @@ export function createHeartbeatRunControl(db: Db, dependencies: HeartbeatRunCont or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"])), )).limit(1); if (otherOwner) continue; + await notifyEnvironmentCleanup(tx); await tx.update(environmentLeases).set({ status: "pending_cleanup", updatedAt: new Date() }) .where(eq(environmentLeases.id, lease.id)); } diff --git a/server/src/services/historical-cleanup-verification.ts b/server/src/services/historical-cleanup-verification.ts index 405c2ae69fa..e72f90d1f60 100644 --- a/server/src/services/historical-cleanup-verification.ts +++ b/server/src/services/historical-cleanup-verification.ts @@ -143,7 +143,24 @@ export function historicalCleanupVerificationService(db: Db, inspectors: Inspect } } } - return { sweep(): Promise { + async function hasPending(): Promise { + const [lease] = await db.select({ id: environmentLeases.id }).from(environmentLeases).where(and( + eq(environmentLeases.status, "failed"), eq(environmentLeases.leasePolicy, "ephemeral"), + eq(environmentLeases.cleanupStatus, "success"), + sql`${environmentLeases.releasedAt} IS NOT NULL AND ${environmentLeases.environmentId} IS NOT NULL + AND ${environmentLeases.provider} IS NOT NULL AND ${environmentLeases.provider} <> 'local' + AND ${environmentLeases.providerLeaseId} IS NOT NULL`, + sql`(${sql.raw(verifiedResourceCleanup('environment_leases', 'lease'))}) IS NOT TRUE`, + )).limit(1); + if (lease) return true; + const [setup] = await db.select({ id: environmentCustomImageSetupSessions.id }).from(environmentCustomImageSetupSessions).where(and( + inArray(environmentCustomImageSetupSessions.status, ["promoted", "cancelled", "timed_out", "failed"]), + sql`${environmentCustomImageSetupSessions.finishedAt} IS NOT NULL AND ${environmentCustomImageSetupSessions.providerLeaseId} IS NOT NULL`, + sql`(${sql.raw(verifiedResourceCleanup('environment_custom_image_setup_sessions', 'image_setup'))}) IS NOT TRUE`, + )).limit(1); + return Boolean(setup); + } + return { hasPending, sweep(): Promise { if (!inFlight) inFlight = run().finally(() => { inFlight = undefined; }); return inFlight; } }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 024074bc6f1..69ffd01db7d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import { writeExecutionControl } from "../execution-control-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -12561,7 +12562,8 @@ async function createRunnerdBackendWithinSessionClaim( runnerInstanceId: backup.manifest.runnerInstanceId, completedAt: backup.manifest.completedAt, }); - const updated = await input.db + const leaseId = remoteTarget.leaseId; + const updated = await writeEnvironmentCleanup(input.db, tx => tx .update(environmentLeases) .set({ metadata: { @@ -12570,9 +12572,9 @@ async function createRunnerdBackendWithinSessionClaim( }, updatedAt: new Date(), }) - .where(eq(environmentLeases.id, remoteTarget.leaseId)) + .where(eq(environmentLeases.id, leaseId)) .returning({ id: environmentLeases.id }) - .then((rows) => rows[0] ?? null); + .then((rows) => rows[0] ?? null)); if (!updated) throw new Error("runner_harness_backup_lease_missing"); }; diff --git a/server/src/services/native-runtime/native-workspace-export-resume.ts b/server/src/services/native-runtime/native-workspace-export-resume.ts index 94b3d46c070..f6c150a29a1 100644 --- a/server/src/services/native-runtime/native-workspace-export-resume.ts +++ b/server/src/services/native-runtime/native-workspace-export-resume.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, inArray, ne, sql } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; @@ -55,7 +56,7 @@ export async function preserveNativeWorkspaceExportLease(db: Db, run: typeof hea )).for("update").limit(1); if (!lease || lease.status !== "active") return; const requestId = randomUUID(), now = new Date(); - await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", failureReason: "workspace_export_stop_pending", releasedAt: now, updatedAt: now, metadata: { ...lease.metadata, remoteExecutionTermination: undefined, [NATIVE_WORKSPACE_EXPORT_RESUME_KEY]: { schema: "paperclip.workspace-export-resume.v2", requestId, @@ -64,7 +65,7 @@ export async function preserveNativeWorkspaceExportLease(db: Db, run: typeof hea pluginId: lease.metadata?.pluginId }, pendingCleanupAttemptId: requestId, pendingCleanupInFlight: false, pendingCleanupLeaseExpiresAtMs: 0, pendingCleanupRetryAfterMs: 0, pendingCleanupRetryAttempts: 0, pendingCleanupRetryCapWarned: false }, - }).where(eq(environmentLeases.id, lease.id)); + }).where(eq(environmentLeases.id, lease.id))); } /** Only verified copyback plus commitment permits the original ephemeral @@ -106,7 +107,7 @@ export async function settleNativeWorkspaceExportResume(db: Db, lease: Lease, op const receipt = remoteTerminationReceipt(lease, options.receipt); const stopped = receipt?.state === "stopped"; const now = new Date(); - const [row] = await db.update(environmentLeases).set({ + const [row] = await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: stopped ? options.status ?? "released" : "pending_cleanup", cleanupStatus: stopped ? "success" : "failed", releasedAt: now, lastUsedAt: now, updatedAt: now, metadata: sql`(${environmentLeases.metadata} - 'remoteExecutionTermination' @@ -120,6 +121,6 @@ export async function settleNativeWorkspaceExportResume(db: Db, lease: Lease, op sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.attemptId}`, sql`${environmentLeases.metadata}->'nativeWorkspaceExportResume'->>'requestId' = ${marker.requestId}`, sql`${environmentLeases.metadata}->>'pluginId' = ${marker.pluginId}`, - )).returning(); + )).returning()); return row ?? null; } diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index c8c4ff598e6..4c1fe7c8474 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -1,3 +1,4 @@ +import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; import { isNativeWorkspaceExportRepairCause } from "@paperclipai/shared"; @@ -92,6 +93,7 @@ export async function retryNativeWorkspaceExport(input: { // Persist cleanup authority before any provider call. A lost resume // reply or controller restart must not leave a running sandbox behind // an old stopped receipt. The pending claim fences the cleanup sweeper. + await notifyEnvironmentCleanup(tx); const [lease] = await tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", releasedAt: now, failureReason: "workspace_export_resume_pending", updatedAt: now, metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, @@ -140,6 +142,7 @@ export async function retryNativeWorkspaceExport(input: { await tx.update(issues).set({ executionRunId: current.run.id, updatedAt: now }).where(eq(issues.id, current.issue.id)); // Reclaim the same lease only for control-plane copyback. The normal // finalization settlement will stop/retain it again on success or failure. + await notifyEnvironmentCleanup(tx); await tx.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, failureReason: null, metadata: { ...current.lease.metadata, remoteExecutionTermination: undefined, // Keep stop-only authority through copyback and commitment. A crash diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index ec96d5dead1..bcc3573e4e3 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -1,3 +1,4 @@ +import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { emitStartupReuse, type StartupReuseReason } from "@paperclipai/adapter-utils/startup-reuse"; import type { AdapterRuntimeEvent } from "@paperclipai/adapter-utils"; @@ -647,6 +648,7 @@ async function persistLeaseStamp(input: { .limit(1) .then((rows) => rows[0] ?? null); if (!lease) throw new Error("native_workspace_sync_lease_missing"); + await notifyEnvironmentCleanup(tx); await tx .update(environmentLeases) .set({ diff --git a/server/src/services/sandbox-orphan-cleanup-spool.ts b/server/src/services/sandbox-orphan-cleanup-spool.ts index a6f97b8e8fe..1844ad9f116 100644 --- a/server/src/services/sandbox-orphan-cleanup-spool.ts +++ b/server/src/services/sandbox-orphan-cleanup-spool.ts @@ -50,6 +50,8 @@ export interface SandboxOrphanCleanupSpool { * failure and skips a malformed file. */ load(): Promise; + /** Distinguish empty storage from unread work without discarding readable records. */ + loadWithStatus?(): Promise<{ records: DeferredOrphanCleanupRecord[]; readFailed: boolean }>; /** * Remove one persisted orphan record after its durable `pending_cleanup` row * lands. The removal runs after the database insert succeeds, so a crash @@ -143,6 +145,29 @@ export function createSandboxOrphanCleanupSpool(spoolDir?: string): SandboxOrpha const dir = spoolDir ?? idleOrphanSpoolPath(); registerIdleSpoolDirectory(dir); + async function loadWithStatus() { + let names: string[]; + try { names = await fs.readdir(dir); } + catch (error) { + return { records: [], readFailed: (error as NodeJS.ErrnoException).code !== "ENOENT" }; + } + const records: DeferredOrphanCleanupRecord[] = []; + let readFailed = false; + for (const name of names) { + if (!name.endsWith(RECORD_FILE_SUFFIX)) continue; + try { + const parsed: unknown = JSON.parse(await fs.readFile(path.join(dir, name), "utf8")); + if (isDeferredOrphanCleanupRecord(parsed)) records.push(parsed); + else readFailed = true; + } catch (error) { + // A concurrent successful flush can remove a listed record. Other read + // failures and malformed records remain unresolved cleanup obligations. + if ((error as NodeJS.ErrnoException).code !== "ENOENT") readFailed = true; + } + } + return { records, readFailed }; + } + return { async append(record) { const absPath = path.join(dir, recordFileName(record)); @@ -158,31 +183,8 @@ export function createSandboxOrphanCleanupSpool(spoolDir?: string): SandboxOrpha } }, - async load() { - let names: string[]; - try { - names = await fs.readdir(dir); - } catch { - // The directory does not exist yet, or the read failed. Either way, no - // record waits on the disk. - return []; - } - const records: DeferredOrphanCleanupRecord[] = []; - for (const name of names) { - if (!name.endsWith(RECORD_FILE_SUFFIX)) continue; - try { - const raw = await fs.readFile(path.join(dir, name), "utf8"); - const parsed: unknown = JSON.parse(raw); - if (isDeferredOrphanCleanupRecord(parsed)) { - records.push(parsed); - } - } catch { - // Skip a malformed or unreadable file. A partial file cannot authorize - // a teardown, so the load ignores it and keeps the other records. - } - } - return records; - }, + async load() { return (await loadWithStatus()).records; }, + loadWithStatus, async remove(record) { const absPath = path.join(dir, recordFileName(record)); diff --git a/server/src/services/sandbox-stop-and-retain.ts b/server/src/services/sandbox-stop-and-retain.ts index 01f96262f19..acdb52295f9 100644 --- a/server/src/services/sandbox-stop-and-retain.ts +++ b/server/src/services/sandbox-stop-and-retain.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql, type SQL } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -102,7 +103,7 @@ async function prepareStopAndRetain(db: Db, lease: Lease, fromStatus: "active" | provider: lease.provider, providerLeaseId: lease.providerLeaseId, ...(lease.provider && isBuiltinSandboxProvider(lease.provider) && !lease.metadata?.sandboxProviderPlugin && lease.metadata?.pluginId == null ? { builtinProvider: lease.provider } : { pluginId: lease.metadata?.pluginId }) }; - const [updated] = await db.update(environmentLeases).set({ + const [updated] = await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", failureReason: "sandbox_stop_pending", releasedAt: now, updatedAt: now, metadata: sql`(coalesce(${environmentLeases.metadata}, '{}'::jsonb) - 'remoteExecutionTermination' - 'sandboxStopAndRetainReceipt') || ${JSON.stringify({ [SANDBOX_STOP_AND_RETAIN_KEY]: intent, pendingCleanupAttemptId: requestId, @@ -114,7 +115,7 @@ async function prepareStopAndRetain(db: Db, lease: Lease, fromStatus: "active" | sql`${environmentLeases.provider} is not distinct from ${lease.provider}`, sql`${environmentLeases.providerLeaseId} is not distinct from ${lease.providerLeaseId}`, sql`${environmentLeases.metadata} is not distinct from ${lease.metadata === null ? null : JSON.stringify(lease.metadata)}::jsonb`, - )).returning(); + )).returning()); return updated ?? null; } @@ -125,7 +126,7 @@ export async function settleStopOnlyCleanup(db: Db, lease: Lease, options: { att const receipt = remoteTerminationReceipt(lease, options.receipt), stopped = receipt?.state === "stopped"; const now = new Date(); const removeIntent: SQL = stopped ? sql`- 'sandboxStopAndRetain'` : sql``; - const [updated] = await db.update(environmentLeases).set({ + const [updated] = await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: stopped ? "released" : "pending_cleanup", cleanupStatus: stopped ? "success" : "failed", failureReason: stopped ? null : "sandbox_stop_pending", releasedAt: now, lastUsedAt: now, updatedAt: now, @@ -146,7 +147,7 @@ export async function settleStopOnlyCleanup(db: Db, lease: Lease, options: { att sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.attemptId}`, sql`${environmentLeases.metadata}->'sandboxStopAndRetain' = ${JSON.stringify(lease.metadata?.[SANDBOX_STOP_AND_RETAIN_KEY])}::jsonb`, sql`${environmentLeases.metadata}->>'pluginId' is not distinct from ${intent.pluginId}`, - )).returning(); + )).returning()); return updated ?? null; } diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 984adf7fb63..e2fae8179e5 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -1,3 +1,4 @@ +import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; import { and, asc, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; import { agentWakeupRequests, environmentLeases, heartbeatRuns, issues, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; @@ -298,7 +299,7 @@ export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof hear ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider!), eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), )).limit(1); if (other) return; - await db.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "ephemeral", + await writeEnvironmentCleanup(db, tx => tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "ephemeral", updatedAt: new Date(), metadata: sql`(${environmentLeases.metadata} - 'workspaceRestoreRecovery' - 'remoteExecutionTermination') || ${JSON.stringify({ workspaceRecoveryCleanup: { runId: run.id, descriptorSha256: ref.descriptorSha256 }, pendingCleanupInFlight: false, pendingCleanupRetryAfterMs: 0, pendingCleanupRetryAttempts: 0, pendingCleanupLeaseExpiresAtMs: 0 })}::jsonb`, @@ -306,7 +307,7 @@ export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof hear eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.providerLeaseId, ref.providerLeaseId), eq(environmentLeases.status, "released"), eq(environmentLeases.cleanupStatus, "success"), sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, - sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`)); + sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`))); } export function retainedWorkspaceCandidates(db: Db, cursor?: string) { return db.select({ action: issueRecoveryActions, run: heartbeatRuns }).from(issueRecoveryActions) diff --git a/tests/e2e/chunk-load-recovery.spec.ts b/tests/e2e/chunk-load-recovery.spec.ts index f3f9c4c81d0..637825a52ca 100644 --- a/tests/e2e/chunk-load-recovery.spec.ts +++ b/tests/e2e/chunk-load-recovery.spec.ts @@ -6,9 +6,9 @@ async function json(response: Awaited>) { return response.json(); } -async function failFirstTaskCss(page: Page) { +async function failFirstLazyCss(page: Page) { const counts = { failed: 0, served: 0, documents: 0 }; - await page.route(/\/assets\/issue-detail-[^/]+\.css$/, async (route) => { + await page.route(/\/assets\/CompanyEnvironments-[^/]+\.css$/, async (route) => { if (counts.failed === 0) { counts.failed += 1; await route.abort("failed"); @@ -25,57 +25,45 @@ async function failFirstTaskCss(page: Page) { test.use({ serviceWorkers: "block" }); -for (const entry of ["task", "agents"] as const) { - test(`a failed task-page CSS preload recovers with one reload (${entry} page)`, async ({ page, request }) => { - const company = await json(await request.post("/api/companies", { - data: { name: `Chunk recovery ${randomUUID()}` }, - })); - const issue = await json(await request.post(`/api/companies/${company.id}/issues`, { - data: { title: "Open after a failed stylesheet", status: "backlog" }, - })); - const counts = await failFirstTaskCss(page); +// Task-page styles now live in the initial CSS bundle. Exercise an emitted +// lazy stylesheet so this remains a real network/preload recovery test. +const environmentsPath = "/company/settings/instance/environments"; - if (entry === "task") { - await page.goto(`/${company.issuePrefix}/issues/${issue.identifier}`); - await expect(page.getByRole("heading", { name: issue.title, exact: true })).toBeVisible(); - } else { - await page.goto(`/${company.issuePrefix}/agents`); - await expect(page.getByRole("heading", { name: "Agents", exact: true }).first()).toBeVisible(); - } +test("a failed lazy CSS preload recovers with one reload", async ({ page, request }) => { + const company = await json(await request.post("/api/companies", { + data: { name: `Chunk recovery ${randomUUID()}` }, + })); + const counts = await failFirstLazyCss(page); - await expect(page.getByText("This page hit an error")).toHaveCount(0); - expect(counts.failed).toBe(1); - expect(counts.served).toBeGreaterThan(0); - expect(counts.documents).toBe(2); + await page.goto(`/${company.issuePrefix}${environmentsPath}`); + await expect(page.getByText("Enable Environments in instance experimental settings to manage shared execution targets.") + .or(page.getByTestId("instance-settings-environments-section"))).toBeVisible(); + + await expect(page.getByText("This page hit an error")).toHaveCount(0); + expect(counts.failed).toBe(1); + expect(counts.served).toBeGreaterThan(0); + expect(counts.documents).toBe(2); +}); + +test("a lazy CSS asset that stays broken reloads once, then shows the error", async ({ page, request }) => { + const company = await json(await request.post("/api/companies", { + data: { name: `Chunk recovery ${randomUUID()}` }, + })); + let failed = 0; + let documents = 0; + await page.route(/\/assets\/CompanyEnvironments-[^/]+\.css$/, async (route) => { + failed += 1; + await route.abort("failed"); }); -} - -for (const entry of ["task", "agents"] as const) { - test(`a task-page CSS asset that stays broken reloads once, then shows the error (${entry} page)`, async ({ page, request }) => { - const company = await json(await request.post("/api/companies", { - data: { name: `Chunk recovery ${randomUUID()}` }, - })); - const issue = await json(await request.post(`/api/companies/${company.id}/issues`, { - data: { title: "Open with a broken stylesheet", status: "backlog" }, - })); - let failed = 0; - let documents = 0; - await page.route(/\/assets\/issue-detail-[^/]+\.css$/, async (route) => { - failed += 1; - await route.abort("failed"); - }); - page.on("request", (request) => { - if (request.resourceType() === "document" && request.frame() === page.mainFrame()) documents += 1; - }); - - await page.goto(entry === "task" - ? `/${company.issuePrefix}/issues/${issue.identifier}` - : `/${company.issuePrefix}/agents`); - - await expect(page.getByText("This page hit an error")).toBeVisible(); - await expect(page.getByText(/Unable to preload CSS for .*issue-detail-/)).toBeVisible(); - await page.waitForTimeout(1_000); - expect(documents).toBe(2); - expect(failed).toBe(2); + page.on("request", (request) => { + if (request.resourceType() === "document" && request.frame() === page.mainFrame()) documents += 1; }); -} + + await page.goto(`/${company.issuePrefix}${environmentsPath}`); + + await expect(page.getByText("This page hit an error")).toBeVisible(); + await expect(page.getByText(/Unable to preload CSS for .*CompanyEnvironments-/)).toBeVisible(); + await page.waitForTimeout(1_000); + expect(documents).toBe(2); + expect(failed).toBe(2); +});