RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: retire bridge routes and escalate the fifth cleanup failure 

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Harold Kim 2026-09-11 22:29:57 +00:00
parent 4992e90cb0
commit e3b11ec91e
No known key found for this signature in database
GPG Key ID: 4861541D36B2037E
4 changed files with 76 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
packages/plugins/sandbox-providers/daytona/src/plugin.test.ts
View File

@ -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<unknown> | 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<void>(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();
}
});

32
packages/plugins/sandbox-providers/daytona/src/plugin.ts
View File

@ -467,16 +467,17 @@ async function withLivenessTimeout<T>(
* 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<unknown>][] = [
["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<void> {
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<v
console.error(
`Failed to delete Daytona session ${sessionId} during teardown: ${formatErrorMessage(error)}`,
);
} finally {
sandboxHandleSessionStore.clear(scope);
}
}
@ -2012,10 +2014,8 @@ async function closeDaytonaDuplexChannelsForLease(providerLeaseId: string): Prom
const matches = [...daytonaDuplexChannelByRoute.values()].filter(
(entry) => 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({

26
server/src/__tests__/heartbeat-pending-cleanup-sweep.test.ts
View File

@ -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);

34
server/src/services/heartbeat.ts
View File

@ -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<string, unknown>;
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}`));