From e3b11ec91e881d158bdf4e51c4976eb985958a18 Mon Sep 17 00:00:00 2001 From: Harold Kim Date: Fri, 11 Sep 2026 22:29:57 +0000 Subject: [PATCH] fix: retire bridge routes and escalate the fifth cleanup failure Co-Authored-By: Paperclip --- .../daytona/src/plugin.test.ts | 24 ++++++++++--- .../sandbox-providers/daytona/src/plugin.ts | 32 ++++++++--------- .../heartbeat-pending-cleanup-sweep.test.ts | 26 ++++++++++++-- server/src/services/heartbeat.ts | 34 ++++++++++--------- 4 files changed, 76 insertions(+), 40 deletions(-) diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index f711241313..7d1eb6ea9f 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -253,7 +253,7 @@ describe("Daytona sandbox provider plugin", () => { expect(manifest.version).toBe("0.1.7"); }); - it("opens a duplex channel, forwards a host write, and closes it on lease release", async () => { + it.each([false, true])("closes duplex routes on lease release even when bridge drain hangs: %s", async (hangDrain) => { process.env.DAYTONA_API_KEY = "host-key"; // A fake PTY handle records each host write, drives the data stream on demand, // and records the kill and the disconnect. @@ -297,6 +297,8 @@ describe("Daytona sandbox provider plugin", () => { }, } as unknown as PluginContext); + let finishBlocked: (() => void) | undefined; + let blocked: Promise | undefined; try { await plugin.definition.onEnvironmentAcquireLease?.({ driverKey: "daytona", @@ -306,7 +308,7 @@ describe("Daytona sandbox provider plugin", () => { agentId: "agent-1", executionWorkspaceId: "workspace-1", adapterType: "codex_local", - config: { image: "node:20", timeoutMs: 300000, reuseLease: true }, + config: { image: "node:20", timeoutMs: 300000, livenessTimeoutMs: 5, reuseLease: true }, }); const open = await plugin.definition.onDuplexChannelOpen?.({ @@ -368,14 +370,24 @@ describe("Daytona sandbox provider plugin", () => { }, ]); - // Lease release closes the channel: it kills the child and releases the - // pseudo-terminal socket. + if (hangDrain) { + sandbox.process.executeCommand.mockImplementationOnce(async () => { + await new Promise(resolve => { finishBlocked = resolve; }); + return { exitCode: 0, result: "done", artifacts: { stdout: "done" } }; + }); + blocked = plugin.definition.onEnvironmentExecute?.({ driverKey: "daytona", companyId: "company-1", + environmentId: "env-1", config: { timeoutMs: 300000 }, bypassSession: true, + lease: { providerLeaseId: "sandbox-123", metadata: {} }, command: "printf", args: ["done"] }); + await vi.waitFor(() => expect(finishBlocked).toBeTypeOf("function")); + } + sandbox.stop.mockImplementation(async () => { expect(disconnected).toBe(1); }); + // Route invalidation must happen before provider stop, including after drain timeout. await plugin.definition.onEnvironmentReleaseLease?.({ driverKey: "daytona", companyId: "company-1", environmentId: "env-1", providerLeaseId: "sandbox-123", - config: { image: "node:20", timeoutMs: 300000, reuseLease: true }, + config: { image: "node:20", timeoutMs: 300000, livenessTimeoutMs: 5, reuseLease: true }, }); expect(killed).toBeGreaterThanOrEqual(1); expect(disconnected).toBe(1); @@ -390,6 +402,8 @@ describe("Daytona sandbox provider plugin", () => { }); expect(inputs.length).toBe(inputsBefore); } finally { + finishBlocked?.(); + await blocked; restore(); } }); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index f78ce34c1f..78edc09c8c 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -467,16 +467,17 @@ async function withLivenessTimeout( * the receipt boundary; timing out this drain is never termination evidence. */ async function drainSandboxBeforeTermination(sandbox: Sandbox, scope: SandboxScope) { const timeoutMs = Math.max(1, Math.min(scope.config.livenessTimeoutMs || 10_000, 10_000)); - try { - await withLivenessTimeout("sandbox.activityDrain", timeoutMs, - () => sandboxHandleActivityGates.waitForIdle(scope)); - await withLivenessTimeout("sandbox.sessionTeardown", timeoutMs, - () => teardownSession(sandbox, scope)); - await withLivenessTimeout("sandbox.channelTeardown", timeoutMs, - () => closeDaytonaDuplexChannelsForLease(scope.providerLeaseId)); - } catch { - // Do not log provider exception text: it can contain credentials. - console.warn("Sandbox bridge drain failed; continuing provider termination."); + const steps: [string, () => Promise][] = [ + ["sandbox.activityDrain", () => sandboxHandleActivityGates.waitForIdle(scope)], + ["sandbox.sessionTeardown", () => teardownSession(sandbox, scope)], + ["sandbox.channelTeardown", () => closeDaytonaDuplexChannelsForLease(scope.providerLeaseId)], + ]; + for (const [operation, action] of steps) { + try { await withLivenessTimeout(operation, timeoutMs, action); } + catch { + // Each cleanup is independent; one hung bridge must not retain other routes. + console.warn("Sandbox bridge cleanup failed; continuing provider termination."); + } } } @@ -1572,6 +1573,9 @@ async function getOrCreateSession(sandbox: Sandbox, scope: SandboxScope): Promis async function teardownSession(sandbox: Sandbox, scope: SandboxScope): Promise { const sessionId = sandboxHandleSessionStore.get(scope); if (!sessionId) return; + // Retire the captured identity before awaiting the provider. A late response + // must not clear a new session created after this lease is resumed. + sandboxHandleSessionStore.clear(scope); try { // Wrap the session delete in a short `session.close` provider span. The // host maps the name to `sandbox.daytona.session.close`. @@ -1585,8 +1589,6 @@ async function teardownSession(sandbox: Sandbox, scope: SandboxScope): Promise entry.providerLeaseId === providerLeaseId, ); - for (const entry of matches) { - forgetDaytonaDuplexChannel(entry); - await entry.session.close().catch(() => undefined); - } + for (const entry of matches) forgetDaytonaDuplexChannel(entry); + await Promise.allSettled(matches.map(entry => entry.session.close())); } const plugin = definePlugin({ diff --git a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts index 8f6a856b57..ce58d45b7c 100644 --- a/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts +++ b/server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts @@ -267,6 +267,26 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1); }); + it("does not let malformed provider metadata defer cleanup forever", async () => { + const { companyId, environmentId } = await seedCompanyAndEnvironment(); + await insertPendingCleanupLease({ companyId, environmentId, + updatedAt: new Date(Date.now() - 60 * 60_000), metadata: { pendingCleanupRetryAfterMs: 1e300 } }); + const destroy = vi.fn(async () => null); + await heartbeatService(db, { environmentRuntime: fakeRuntime(destroy) }).sweepPendingCleanupLeases(); + expect(destroy).toHaveBeenCalledTimes(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, + updatedAt: new Date(Date.now() - 60 * 60_000), metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 } }); + const runtime = fakeRuntime(vi.fn(async () => null)); + await heartbeatService(db, { environmentRuntime: runtime }).sweepPendingCleanupLeases(); + const [saved] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, leaseId)); + expect(saved.metadata?.[CAP_WARNED_KEY]).toBe(true); + expect(Number(saved.metadata?.pendingCleanupRetryAfterMs)).toBeGreaterThan(Date.now() + 29 * 60_000); + }); + it("retries after a restart and cooldown when the provider recovers", async () => { const { companyId, environmentId } = await seedCompanyAndEnvironment(); const leaseId = await insertOrphanEphemeralLease({ companyId, environmentId, @@ -820,7 +840,7 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { .mocked(logger.warn) .mock.calls.filter( (call) => - call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup", + call[1] === "environment lease needs operator attention; automatic cleanup continues with backoff", ); expect(capWarnings.length).toBeLessThanOrEqual(1); }); @@ -1117,7 +1137,7 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { .mocked(logger.warn) .mock.calls.filter( (call) => - call[1] === "environment lease reached the pending_cleanup retry cap; left for manual cleanup", + call[1] === "environment lease needs operator attention; automatic cleanup continues with backoff", ); expect(capWarnings.length).toBe(0); @@ -1127,7 +1147,7 @@ describeEmbeddedPostgres("heartbeat sweepPendingCleanupLeases", () => { companyId, environmentId, updatedAt: new Date(Date.now() - 60 * 60 * 1000), - metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 1 }, + metadata: { [ATTEMPTS_KEY]: ATTEMPT_CAP - 2 }, }); await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); const belowCapMetadata = await readMetadata(belowCapLeaseId); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 803d25821d..19fbe25668 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -682,6 +682,7 @@ function pendingCleanupRetryDueSql() { return sql`case when jsonb_typeof(${environmentLeases.metadata}->'pendingCleanupRetryAfterMs') = 'number' then (${environmentLeases.metadata}->>'pendingCleanupRetryAfterMs')::numeric <= ${Date.now()} + or (${environmentLeases.metadata}->>'pendingCleanupRetryAfterMs')::numeric > ${Date.now() + 30 * 60_000 + 1_000} else true end`; } @@ -17789,20 +17790,7 @@ export function heartbeatService( const metadata = { ...(row.metadata ?? {}) } as Record; const attempts = readPendingCleanupRetryAttempts(metadata); - if (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) { - capped += 1; - // Warn once, then continue automatic cleanup with backoff. The atomic claim - // keeps the warning to one log line even when two sweeps overlap. - if (metadata[PENDING_CLEANUP_CAP_WARNED_METADATA_KEY] !== true) { - const warned = await claimPendingCleanupCapWarning(row.id); - if (warned) { - logger.warn( - { leaseId: row.id, environmentId: row.environmentId, attempts }, - "environment lease needs operator attention; automatic cleanup continues with backoff", - ); - } - } - } + const environment = row.environmentId ? await environmentsSvc.getById(row.environmentId) @@ -17911,13 +17899,27 @@ export function heartbeatService( "pending_cleanup lease retry failed", ); } + if (attempts + 1 >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP) { + capped += 1; + // Warn once, then continue automatic cleanup with backoff. The atomic claim + // keeps the warning to one log line even when two sweeps overlap. + if (metadata[PENDING_CLEANUP_CAP_WARNED_METADATA_KEY] !== true) { + const warned = await claimPendingCleanupCapWarning(row.id); + if (warned) { + logger.warn( + { leaseId: row.id, environmentId: row.environmentId, attempts }, + "environment lease needs operator attention; automatic cleanup continues with backoff", + ); + } + } + } // 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({ metadata: sql`${pendingCleanupMetadataObjectSql()} || ${JSON.stringify({ pendingCleanupInFlight: false, - pendingCleanupRetryAfterMs: Date.now() + (attempts >= PENDING_CLEANUP_SWEEP_ATTEMPT_CAP - ? 30 * 60_000 : Math.max(30_000, backoffMs)), + 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}`));