mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 01:41:42 +02:00
fix: stop idle login cleanup polling (#17106)
**Dependency landed:** PR #17129 is merged. This branch is rebased onto master `0f15f82b19` and preserves its provider-owned completion contract. The shared save path calls `notifyLoginSessionChanged(tx)`; the cleanup module invalidates both existing sandbox cleanup workers without selecting a provider. ## Thinking Path > - Paperclip manages work for AI agents. > - Subscription login can allocate a temporary sandbox. > - Existing cleanup workers scan the database every heartbeat tick, even with no login work. > - Empty scans keep idle databases active. > - This PR uses committed session changes and saved expiry times to schedule cleanup. > - Existing rows still drive recovery after failure and restart. ## Linked Issues or Issue Description **What happened?** On master `898144f4b3`, the enabled heartbeat scheduler runs both login reapers every 30 seconds by default. An instance with no pending logins still queries sessions and leases. **What did you expect to happen?** After startup reconciliation, idle login cleanup should make no recurring SQL or provider calls. New sessions, expiry, failed cleanup and restart must still recover. **Steps to reproduce** Start a claimed instance with heartbeat scheduling enabled and no pending login sessions. Observe repeated adapter-login and setup-token cleanup queries at the heartbeat interval. Related sleep migration: #17094. Searched open issues and PRs. Related setup-token cleanup work: #15774, #14236 and #16648. This PR does not change their release-result or null-lease behavior. ## What Changed - Register separate adapter-login and setup-token workers with the existing delivery coordinator and scheduler. - Notify before writes in the owning stores, the lease-tag write, and AI connection save. Shared save uses a provider-neutral notification; each cleanup worker checks its own durable predicates. The shared transaction protocol retains intent until the outer transaction settles. - Reconstruct session expiry, promotion expiry and concrete cleanup retries from existing rows. Preserve configured heartbeat spacing to bound scan bursts. - Recover tagged orphan leases even with terminal or missing sessions. Protect live acquisitions that have not attached their lease yet. - Exclude local-subscription rows from sandbox reapers. Their existing local-login worker owns cleanup. - Preserve deferred work across drain and shutdown. Add no tables, Cloud alarms or host-sleep exemptions. - Update the inventory and document producer coverage, writer scope and recovery boundaries. ## Verification - Updated head: `1d57e13606`. Seven focused suites passed (282 tests): AI connections, login cleanup, device-login service/reaper, login-session runtime, and setup-token session/reaper. `pnpm -r typecheck` and `pnpm build` passed. The duplicate local `pnpm test:run` was interrupted after CI passed all server shards; it is not counted as a complete local pass. Full-suite evidence comes from the exact-head CI run. Greptile reviewed this exact head at 5/5 with no findings or unresolved threads. [Full CI run 39669285577](https://github.com/paperclipai/paperclip/actions/runs/39669285577) passed on attempt 2. Both aggregate gates passed; no checks remain failed or pending. - Extended the real PostgreSQL AI connection tests for both Anthropic and OpenAI. The tests check that neither cleanup observer wakes inside the completion callback, both wake after commit, and an idempotent save does not emit another wake. - Existing cleanup tests cover idle/history states, producer wake, expiry, promotion, cancellation, retry/restart, orphan recovery, nested commit/rollback, drain, concurrency, and shutdown. - Current-head CI attempt 1: the unchanged `sandbox-callback-bridge.test.ts` HTTP/2 preface test failed because its child process could not bind `127.0.0.1:34033` (`EADDRINUSE`). Its single unchanged rerun passed. The exact test also passed locally without changing code or timeouts. This establishes recovery from the bind collision; it does not establish which process owned the port. No login code ran in that package test. - Previous-head failures remain recorded: the login-runtime transaction fixture was corrected; a workspace-runtime readiness port-bind failure passed in isolation and fresh CI, but its first failure cause remains unconfirmed. An earlier combined AI suite timed out in local-login isolation and passed alone; its timing cause remains unconfirmed. An earlier local full suite was interrupted and was not counted as passing. ## Risks - Process-local signals cover one active server with the instrumented shared database handle. Independent SQL writers or replicas need their own signal integration. - Provider resources that never became persisted tagged leases remain the responsibility of the unchanged general lease/spool recovery. Setup-token acquisition retains its required provider expiry. - Cleanup starts are bounded by the configured heartbeat interval. Writes during a sweep can cause a final follow-up scan before quiet. - Existing login and resource blockers still prevent unsafe host sleep. This is worker polling removal, not provider sleep qualification. ## Model Used OpenAI GPT-6 through Codex, with reasoning, repository tools and test execution. The exact serving model ID and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes:` / `Closes` / `Refs` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub references) - [x] My branch name describes the change and contains no internal/instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
0f15f82b19
commit
efc2ddc7bf
16 files changed
+529
-115
No files matched your search
@@ -22,6 +22,7 @@ import { fetchCompanyQuotaWindows } from "../services/quota-windows.js";
|
||||
import { quotaCredentialRecoveryPath, quotaCredentialRecovery, quotaCredentialHash } from "../services/quota-credential-recovery.js";
|
||||
import { readAiConnectionConfigurationFailure } from "../services/ai-connection-configuration-failure.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { syncConnectionCredentialBindings } from "../services/connection-credential-bindings.js";
|
||||
import * as codexAdapter from "@paperclipai/adapter-codex-local/server";
|
||||
import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
@@ -1685,7 +1686,15 @@ console.log(JSON.stringify({ type: "result", subtype: "success", result: "hello"
|
||||
publicSessionId: randomUUID(), status: "promoting", aiConnection: { ...intent, agentIds: [] },
|
||||
expiresAt: new Date(Date.now() + 60_000),
|
||||
}).returning();
|
||||
const complete = vi.fn(async () => {});
|
||||
const adapterWake = vi.fn(), setupWake = vi.fn();
|
||||
const unsubscribe = [
|
||||
subscribeDeliveryWork(db, DELIVERY_QUEUES.adapterLoginCleanup, adapterWake),
|
||||
subscribeDeliveryWork(db, DELIVERY_QUEUES.setupTokenCleanup, setupWake),
|
||||
];
|
||||
const complete = vi.fn(async () => {
|
||||
expect(adapterWake).not.toHaveBeenCalled();
|
||||
expect(setupWake).not.toHaveBeenCalled();
|
||||
});
|
||||
try {
|
||||
await expect(service.save(companyId, owner, { ...intent, agentIds: [] }, "fixture", { id: session.publicSessionId, complete })).rejects.toThrow("ownership mismatch");
|
||||
await expect(service.save(companyId, "bob", { ...intent, agentIds: [] }, "fixture", { id: session.id, complete })).rejects.toThrow("ownership mismatch");
|
||||
@@ -1697,7 +1706,12 @@ console.log(JSON.stringify({ type: "result", subtype: "success", result: "hello"
|
||||
});
|
||||
expect(await service.save(companyId, owner, { ...intent, agentIds: [] }, "replacement", { id: session.id, complete })).toEqual(saved);
|
||||
expect(complete).toHaveBeenCalledTimes(1);
|
||||
} finally { await db.delete(adapterAuthSessions).where(eq(adapterAuthSessions.id, session.id)); }
|
||||
expect(adapterWake).toHaveBeenCalledTimes(1);
|
||||
expect(setupWake).toHaveBeenCalledTimes(1);
|
||||
} finally {
|
||||
unsubscribe.forEach(stop => stop());
|
||||
await db.delete(adapterAuthSessions).where(eq(adapterAuthSessions.id, session.id));
|
||||
}
|
||||
});
|
||||
it("rolls back credential and connection writes if setup-token completion loses its claim", async () => {
|
||||
const owner = "lost-setup-token-claim";
|
||||
|
||||
@@ -1975,8 +1975,12 @@ describeEmbeddedPostgres("device login service concurrency (embedded postgres)",
|
||||
|
||||
const tagged = await reaperRuntime.listTaggedLeases();
|
||||
const taggedProviderLeaseIds = tagged.map((lease) => lease.providerLeaseId);
|
||||
expect(taggedProviderLeaseIds).toContain("grok-lease-1");
|
||||
expect(taggedProviderLeaseIds).not.toContain("grok-lease-1");
|
||||
expect(taggedProviderLeaseIds).not.toContain("claude-lease-1");
|
||||
await db.update(adapterAuthSessions).set({ status: "failed", providerLeaseId: null })
|
||||
.where(eq(adapterAuthSessions.providerLeaseId, "grok-lease-1"));
|
||||
const orphaned = await reaperRuntime.listTaggedLeases();
|
||||
expect(orphaned.map(lease => lease.providerLeaseId)).toEqual(["grok-lease-1"]);
|
||||
});
|
||||
|
||||
it("carries the CODEX_HOME variable name on the codex_local profile", () => {
|
||||
|
||||
@@ -0,0 +1,226 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { adapterAuthSessions, applyPendingMigrations, companies, createDb, environmentLeases, environments, type Db } from "@paperclipai/db";
|
||||
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { registerLoginCleanupWork } from "../services/login-cleanup-work.js";
|
||||
import { loginCleanupSchedule, orphanLoginLeases } from "../services/login-cleanup-state.js";
|
||||
import { createDbAdapterAuthSessionStore, LOGIN_LEASE_SESSION_TAG_KEY } from "../services/device-login-service.js";
|
||||
import { createDeviceLoginReaper, createProductionLoginSessionReaperRuntime } from "../services/device-login-reaper.js";
|
||||
import { createDbSetupTokenCleanupStore, transitionSetupTokenSessionToStored } from "../services/setup-token-session.js";
|
||||
import { createSetupTokenReaper } from "../services/setup-token-reaper.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const coordinators: ReturnType<typeof createDeliveryWorkCoordinator>[] = [];
|
||||
afterEach(async () => { for (const c of coordinators.splice(0)) await c.stop(); vi.useRealTimers(); });
|
||||
function coordinator(owner: object, canRun = () => true, onError: (error: unknown) => void = vi.fn()) {
|
||||
const c = createDeliveryWorkCoordinator({ owner, canRun, canReconcile: () => true, onError });
|
||||
coordinators.push(c); return c;
|
||||
}
|
||||
function fakeOwner() {
|
||||
const execute = vi.fn(async () => [{ xid: "42" }]);
|
||||
const owner = installDatabaseWorkSignals({ execute, async transaction(fn: (tx: any) => Promise<any>): Promise<any> {
|
||||
return fn({ execute, transaction: owner.transaction });
|
||||
} });
|
||||
return owner;
|
||||
}
|
||||
|
||||
describe("login cleanup lifecycle", () => {
|
||||
it.each(["startup", "commit", "failure"])("retains %s reconciliation across drain without passive SQL", async kind => {
|
||||
vi.useFakeTimers(); const owner = fakeOwner(); let draining = kind === "startup";
|
||||
const c = coordinator(owner, () => !draining);
|
||||
const run = vi.fn(async () => {}), next = vi.fn(async () => null);
|
||||
await registerLoginCleanupWork(c, [{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run, next }], () => true).ready;
|
||||
if (kind !== "startup") {
|
||||
if (kind === "failure") run.mockRejectedValueOnce(new Error("provider unavailable"));
|
||||
else draining = true;
|
||||
await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.adapterLoginCleanup));
|
||||
await vi.advanceTimersByTimeAsync(30_000); draining = true;
|
||||
}
|
||||
const reads = next.mock.calls.length, runs = run.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(90_000);
|
||||
expect(next).toHaveBeenCalledTimes(reads); expect(run).toHaveBeenCalledTimes(runs);
|
||||
draining = false; await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(run).toHaveBeenCalledTimes(runs + 1); expect(c.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("coalesces writes and retains a write during a slow sweep without overlapping", async () => {
|
||||
vi.useFakeTimers(); const owner = fakeOwner(); const c = coordinator(owner);
|
||||
const run = vi.fn(async () => {});
|
||||
await registerLoginCleanupWork(c, [{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run, next: async () => null }], () => true).ready;
|
||||
for (let i = 0; i < 20; i++) {
|
||||
await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.adapterLoginCleanup));
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
}
|
||||
expect(run).toHaveBeenCalledTimes(1);
|
||||
let release!: () => void;
|
||||
run.mockImplementationOnce(() => new Promise<void>(resolve => { release = resolve; }));
|
||||
await vi.advanceTimersByTimeAsync(28_000);
|
||||
await owner.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.adapterLoginCleanup));
|
||||
await vi.advanceTimersByTimeAsync(30_000); expect(run).toHaveBeenCalledTimes(2);
|
||||
release(); await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(run).toHaveBeenCalledTimes(3); expect(c.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("does not block another cleanup worker and awaits in-flight work on shutdown", async () => {
|
||||
vi.useFakeTimers(); const c = coordinator(fakeOwner()); let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const setup = vi.fn(async () => {});
|
||||
const workers = registerLoginCleanupWork(c, [
|
||||
{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => gate, next: async () => null },
|
||||
{ queue: DELIVERY_QUEUES.setupTokenCleanup, run: setup, next: async () => null },
|
||||
], () => true);
|
||||
await vi.advanceTimersByTimeAsync(1); expect(setup).toHaveBeenCalledTimes(1);
|
||||
let stopped = false; const stopping = c.stop().then(() => { stopped = true; });
|
||||
await vi.advanceTimersByTimeAsync(1); expect(stopped).toBe(false);
|
||||
release(); await workers.ready; await stopping; expect(stopped).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_LOGIN_CLEANUP_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("durable login cleanup", () => {
|
||||
let db: ReturnType<typeof createDb>, observed: typeof db;
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let companyId: string, environmentId: string, calls: number, failedDelete: boolean;
|
||||
let store: ReturnType<typeof createDbAdapterAuthSessionStore>;
|
||||
let runtime: ReturnType<typeof createProductionLoginSessionReaperRuntime>;
|
||||
let driver: { releaseRunLease: ReturnType<typeof vi.fn> };
|
||||
const errors: unknown[] = [];
|
||||
beforeAll(async () => {
|
||||
if (external) await applyPendingMigrations(external);
|
||||
else temporary = await startEmbeddedPostgresTestDatabase("login-cleanup-");
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
|
||||
}, 90000);
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
beforeEach(async () => {
|
||||
await db.execute(sql`TRUNCATE TABLE companies, environments CASCADE`);
|
||||
companyId = randomUUID(); environmentId = randomUUID(); calls = 0; failedDelete = false; errors.length = 0;
|
||||
await db.insert(companies).values({ id: companyId, name: "Login cleanup", issuePrefix: "LOG" });
|
||||
await db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" });
|
||||
observed = new Proxy(db, { get(target, key) {
|
||||
if (!["select", "update", "insert", "delete", "execute", "transaction"].includes(String(key))) return Reflect.get(target, key);
|
||||
return (...args: unknown[]) => { calls++; return Reflect.apply(Reflect.get(target, key), target, args); };
|
||||
} });
|
||||
store = createDbAdapterAuthSessionStore(observed);
|
||||
driver = { releaseRunLease: vi.fn(async ({ lease }) => {
|
||||
if (failedDelete) return { cleanupStatus: "failed" };
|
||||
await db.update(environmentLeases).set({ status: "released", cleanupStatus: "success" }).where(eq(environmentLeases.id, lease.id));
|
||||
return { cleanupStatus: "success" };
|
||||
}) };
|
||||
runtime = createProductionLoginSessionReaperRuntime({ db: observed, environmentRuntime: { getDriver: () => driver } as any });
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true });
|
||||
});
|
||||
async function settle(ms = 30_000) {
|
||||
await vi.advanceTimersByTimeAsync(ms);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 });
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 });
|
||||
expect(errors).toEqual([]);
|
||||
}
|
||||
async function quiet(c: ReturnType<typeof coordinator>) {
|
||||
await settle(60_000); expect(c.nextWakeAt()).toBeNull();
|
||||
const before = calls, deletes = driver.releaseRunLease.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(180_000);
|
||||
expect(calls).toBe(before); expect(driver.releaseRunLease).toHaveBeenCalledTimes(deletes);
|
||||
}
|
||||
async function adapterWorker() {
|
||||
const c = coordinator(db, undefined, error => errors.push(error));
|
||||
const reaper = createDeviceLoginReaper({ store, runtime });
|
||||
await registerLoginCleanupWork(c, [{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run: reaper.sweep, next: loginCleanupSchedule(observed).adapter }], () => true).ready;
|
||||
return c;
|
||||
}
|
||||
async function session(at = new Date(Date.now() + 120_000), target = store) {
|
||||
const id = randomUUID();
|
||||
await target.insert({ id, publicSessionId: id, companyId, environmentId, adapterType: "codex_local", startedByUserId: id, expiresAt: at, at: new Date() });
|
||||
return id;
|
||||
}
|
||||
async function lease(sessionId: string) {
|
||||
const id = randomUUID();
|
||||
await db.insert(environmentLeases).values({ id, companyId, environmentId, providerLeaseId: id, metadata: { [LOGIN_LEASE_SESSION_TAG_KEY]: sessionId } });
|
||||
return id;
|
||||
}
|
||||
it("is quiet with configured environments and completed history, excluding local login attempts", async () => {
|
||||
const id = await session(); await store.setStatus({ sessionId: id, status: "authenticated", at: new Date() });
|
||||
await db.insert(adapterAuthSessions).values({ companyId, environmentId, adapterType: "claude_local", startedByUserId: "local", publicSessionId: randomUUID(), connectionMethod: "local_subscription", status: "authenticated", expiresAt: new Date(Date.now() - 1) });
|
||||
const c = await adapterWorker();
|
||||
const setup = createSetupTokenReaper({ store: createDbSetupTokenCleanupStore(observed), leases: { releaseById: vi.fn() } });
|
||||
await registerLoginCleanupWork(c, [{ queue: DELIVERY_QUEUES.setupTokenCleanup, run: setup.sweep, next: loginCleanupSchedule(observed).setupToken }], () => true).ready;
|
||||
await quiet(c);
|
||||
});
|
||||
it("wakes from a real producer, waits for expiry without polling, and settles back to quiet", async () => {
|
||||
const c = await adapterWorker(); await quiet(c);
|
||||
const due = new Date(Date.now() + 120_000); const id = await session(due);
|
||||
await settle(); expect(c.nextWakeAt()).toBe(due.getTime());
|
||||
const before = calls; await vi.advanceTimersByTimeAsync(30_000); expect(calls).toBe(before);
|
||||
await settle(91_000); expect((await store.get(id))?.status).toBe("timed_out"); await quiet(c);
|
||||
});
|
||||
it("respects promotion expiry and disarms cancellation", async () => {
|
||||
const c = await adapterWorker(); const id = await session(new Date(Date.now() + 60_000));
|
||||
const promotion = new Date(Date.now() + 180_000);
|
||||
await store.setStatus({ sessionId: id, status: "promoting", promotionExpiresAt: promotion, at: new Date() });
|
||||
await settle(); expect(c.nextWakeAt()).toBe(promotion.getTime());
|
||||
await store.setStatus({ sessionId: id, status: "cancelled", at: new Date() }); await quiet(c);
|
||||
});
|
||||
it("retries failed deletion only while pending and reconstructs it after restart", async () => {
|
||||
const id = await session(new Date(Date.now() - 1)); const leaseId = await lease(id);
|
||||
await store.recordLeaseAcquired({ sessionId: id, providerLeaseId: leaseId, at: new Date() });
|
||||
failedDelete = true; const first = await adapterWorker();
|
||||
expect((await store.get(id))?.status).toBe("cleanup_pending"); expect(driver.releaseRunLease).toHaveBeenCalledTimes(1);
|
||||
await vi.advanceTimersByTimeAsync(10_000); expect(driver.releaseRunLease).toHaveBeenCalledTimes(1);
|
||||
await first.stop(); failedDelete = false; const second = await adapterWorker();
|
||||
expect((await store.get(id))?.status).toBe("timed_out"); expect(driver.releaseRunLease).toHaveBeenCalledTimes(2); await quiet(second);
|
||||
});
|
||||
it("protects an in-flight lease attachment, then recovers an unbound lease after acquisition fails", async () => {
|
||||
const id = await session(); const leaseId = await lease(id); const c = await adapterWorker();
|
||||
expect(driver.releaseRunLease).not.toHaveBeenCalled(); expect(await orphanLoginLeases(observed)).toEqual([]);
|
||||
await store.setStatus({ sessionId: id, status: "failed", at: new Date() }); await quiet(c);
|
||||
expect(driver.releaseRunLease).toHaveBeenCalledTimes(1);
|
||||
expect(driver.releaseRunLease.mock.calls[0]![0].lease.id).toBe(leaseId);
|
||||
});
|
||||
it("recovers orphan leases without a live session and retains failed orphan cleanup", async () => {
|
||||
await lease(randomUUID()); failedDelete = true; const c = await adapterWorker();
|
||||
expect(driver.releaseRunLease).toHaveBeenCalledTimes(1); expect(c.nextWakeAt()).not.toBeNull();
|
||||
await vi.advanceTimersByTimeAsync(10_000); expect(driver.releaseRunLease).toHaveBeenCalledTimes(1);
|
||||
failedDelete = false; await quiet(c); expect(driver.releaseRunLease).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
it.each([false, true])("keeps nested producer intent until outer settlement (rollback=%s)", async rollback => {
|
||||
const c = await adapterWorker(); await quiet(c);
|
||||
let release!: () => void, written!: () => void, id!: string;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const ready = new Promise<void>(resolve => { written = resolve; });
|
||||
const transaction = db.transaction(async tx => {
|
||||
id = await session(new Date(Date.now() + 120_000), createDbAdapterAuthSessionStore(tx as unknown as Db));
|
||||
written(); await gate; if (rollback) throw new Error("rollback");
|
||||
}).catch(error => { if (!rollback) throw error; });
|
||||
await ready; await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await store.get(id)).toBeNull(); expect(idleWorkSnapshot().active).toBeGreaterThan(0);
|
||||
release(); await transaction; await settle();
|
||||
if (rollback) await quiet(c);
|
||||
else { expect((await store.get(id))?.status).toBe("starting"); await settle(120_000); await quiet(c); }
|
||||
});
|
||||
it.each(["consume", "expire", "cancel"])("setup-token producers wake cleanup on %s", async mode => {
|
||||
const cleanup = createDbSetupTokenCleanupStore(observed), releaseById = vi.fn(async () => {});
|
||||
const c = coordinator(db, undefined, error => errors.push(error));
|
||||
const reaper = createSetupTokenReaper({ store: cleanup, leases: { releaseById } });
|
||||
await registerLoginCleanupWork(c, [{ queue: DELIVERY_QUEUES.setupTokenCleanup, run: reaper.sweep, next: loginCleanupSchedule(observed).setupToken }], () => true).ready;
|
||||
await quiet(c);
|
||||
const identity = { sessionId: randomUUID(), companyId, ownerUserId: "setup", adapterType: "claude_local" };
|
||||
const deadline = Date.now() + 120_000;
|
||||
await cleanup.record({ ...identity, environmentId, leaseId: randomUUID(), state: "starting", deadline, boundAt: null });
|
||||
await settle(); expect(c.nextWakeAt()).toBe(deadline);
|
||||
if (mode === "consume") {
|
||||
expect(await transitionSetupTokenSessionToStored(observed, identity)).toBe(true);
|
||||
await cleanup.consumeStoredClaim(identity);
|
||||
} else if (mode === "cancel") {
|
||||
expect(await cleanup.cancelDurable(identity, ["starting"])).not.toBeNull();
|
||||
} else {
|
||||
await settle(120_000);
|
||||
}
|
||||
await quiet(c);
|
||||
expect(releaseById).toHaveBeenCalledTimes(1); expect(await cleanup.listReapable(Date.now())).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,5 @@
|
||||
import { installDatabaseWorkSignals, databaseWorkPending } from "../../../packages/db/src/work-signals.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import type { AgentAdapterType } from "@paperclipai/shared";
|
||||
@@ -154,7 +156,17 @@ describe("createProductionLoginSessionRuntime lease-acquisition gate", () => {
|
||||
driver: "sandbox",
|
||||
config: { provider: "daytona" },
|
||||
});
|
||||
mockEnvironmentService.updateLeaseMetadata.mockResolvedValue(undefined);
|
||||
const execute = vi.fn(async () => [{ xid: "42" }]);
|
||||
const db = installDatabaseWorkSignals({
|
||||
execute,
|
||||
async transaction<T>(fn: (tx: object) => Promise<T>) { return fn({ execute }); },
|
||||
});
|
||||
const settled = vi.fn();
|
||||
const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.adapterLoginCleanup, settled);
|
||||
mockEnvironmentService.updateLeaseMetadata.mockImplementation(async () => {
|
||||
expect(databaseWorkPending(db, DELIVERY_QUEUES.adapterLoginCleanup)).toBe(true);
|
||||
expect(settled).not.toHaveBeenCalled();
|
||||
});
|
||||
const lease = { id: "lease-1", providerLeaseId: "provider-lease-1", metadata: {} };
|
||||
const acquireRunLease = vi.fn(async () => ({
|
||||
lease,
|
||||
@@ -162,7 +174,7 @@ describe("createProductionLoginSessionRuntime lease-acquisition gate", () => {
|
||||
}));
|
||||
const assertProviderSupportsLoginPty = vi.fn(async () => {});
|
||||
const runtime = createProductionLoginSessionRuntime({
|
||||
db: {} as never,
|
||||
db: db as never,
|
||||
environmentRuntime: { acquireRunLease, getDriver: vi.fn() } as never,
|
||||
assertProviderSupportsLoginPty,
|
||||
});
|
||||
@@ -173,5 +185,8 @@ describe("createProductionLoginSessionRuntime lease-acquisition gate", () => {
|
||||
// The gate passed, so the runtime acquired the provider lease after it.
|
||||
expect(acquireRunLease).toHaveBeenCalledTimes(1);
|
||||
expect(acquired.providerLeaseId).toBe("provider-lease-1");
|
||||
expect(settled).toHaveBeenCalledTimes(1);
|
||||
expect(databaseWorkPending(db, DELIVERY_QUEUES.adapterLoginCleanup)).toBe(false);
|
||||
unsubscribe();
|
||||
});
|
||||
});
|
||||
+7
-36
@@ -1,3 +1,5 @@
|
||||
import { loginCleanupSchedule } from "./services/login-cleanup-state.js";
|
||||
import { registerLoginCleanupWork } from "./services/login-cleanup-work.js";
|
||||
import { timeStartupRecovery } from "./startup-recovery-timing.js";
|
||||
import { skillSourceSyncService } from "./services/skill-source-sync.js";
|
||||
import {
|
||||
@@ -1512,12 +1514,6 @@ async function startServerWithDatabaseTeardown(
|
||||
unregisterTerminalWorkspaceProbe = registerIdleLocalWorkProbe(() => terminalWorkspaceWorker.hasUnrecoveredWork() ? "present" : "none");
|
||||
await terminalWorkspaceWorker.ready;
|
||||
|
||||
// The restart-safe cleanup backstop for adapter login sessions. The
|
||||
// in-process five-minute timer stays the primary control. This reaper runs
|
||||
// on startup and on the scheduler interval. It deletes the login sandbox for
|
||||
// any expired non-terminal session, retries the delete for any terminal
|
||||
// session left in `cleanup_pending`, and deletes a tagged lease that no live
|
||||
// session references.
|
||||
const adapterLoginReaper = createDeviceLoginReaper({
|
||||
store: createDbAdapterAuthSessionStore(db as any),
|
||||
runtime: createProductionLoginSessionReaperRuntime({
|
||||
@@ -1537,20 +1533,7 @@ async function startServerWithDatabaseTeardown(
|
||||
logger.info(result, "adapter login reaper swept login sessions");
|
||||
}
|
||||
};
|
||||
const scheduleAdapterLoginReaperSweep = () => {
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
trackHeartbeatSchedulerWork(adapterLoginReaper
|
||||
.sweep()
|
||||
.then(logAdapterLoginReaperResult)
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "adapter login reaper sweep failed");
|
||||
}));
|
||||
};
|
||||
|
||||
// The restart-safe cleanup backstop for the Claude setup-token login flow. It
|
||||
// runs on startup and on the scheduler interval, so a sandbox lease survives a
|
||||
// server restart and a release failure. It releases any lease whose login
|
||||
// session is terminal, past its deadline, or already consumed.
|
||||
const setupTokenReaper = createProductionSetupTokenReaper({
|
||||
db: db as any,
|
||||
environmentRuntime: environmentRuntimeService(db as any, { pluginWorkerManager }),
|
||||
@@ -1563,15 +1546,6 @@ async function startServerWithDatabaseTeardown(
|
||||
logger.info(result, "setup-token login reaper released leases");
|
||||
}
|
||||
};
|
||||
const scheduleSetupTokenReaperSweep = () => {
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
trackHeartbeatSchedulerWork(setupTokenReaper
|
||||
.sweep()
|
||||
.then(logSetupTokenReaperResult)
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "setup-token login reaper sweep failed");
|
||||
}));
|
||||
};
|
||||
|
||||
const worktreeRunExecutionActivation = await resolveWorktreeRunExecutionActivationState({
|
||||
getExperimental: () => instanceSettingsService(db).getExperimental(),
|
||||
@@ -1766,12 +1740,11 @@ async function startServerWithDatabaseTeardown(
|
||||
}
|
||||
});
|
||||
await step("decision_expiry", () => decisionExecutor.sweepExpired());
|
||||
// A login sandbox that outlived a server restart is deleted before timer
|
||||
// ticks start.
|
||||
await step("adapter_login_reaper", () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult));
|
||||
// A setup-token login sandbox lease that outlived a server restart
|
||||
// releases before timer ticks start.
|
||||
await step("setup_token_reaper", () => setupTokenReaper.sweep().then(logSetupTokenReaperResult));
|
||||
const loginSchedule = loginCleanupSchedule(db, config.heartbeatSchedulerIntervalMs);
|
||||
await step("login_cleanup_workers", () => registerLoginCleanupWork(deliveryWork, [
|
||||
{ queue: DELIVERY_QUEUES.adapterLoginCleanup, run: () => adapterLoginReaper.sweep().then(logAdapterLoginReaperResult), next: loginSchedule.adapter },
|
||||
{ queue: DELIVERY_QUEUES.setupTokenCleanup, run: () => setupTokenReaper.sweep().then(logSetupTokenReaperResult), next: loginSchedule.setupToken },
|
||||
], () => !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));
|
||||
@@ -1808,8 +1781,6 @@ async function startServerWithDatabaseTeardown(
|
||||
scheduleMergedPullRequestConfirmationSweep();
|
||||
scheduleGitHubConnectionEventPoll();
|
||||
scheduleGitHubConnectionContinuitySweep();
|
||||
scheduleAdapterLoginReaperSweep();
|
||||
scheduleSetupTokenReaperSweep();
|
||||
scheduleEnvironmentLeaseCleanupSweep();
|
||||
|
||||
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyLoginSessionChanged } from "./login-cleanup-notifications.js";
|
||||
import { WORKSPACE_RESTORE_LOCK_TIMEOUT_CODE } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
import { withAccountHomeSecretMutationLock } from "@paperclipai/adapter-codex-local/server";
|
||||
import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js";
|
||||
@@ -886,6 +887,7 @@ export function aiConnectionService(db: Db) {
|
||||
})),
|
||||
);
|
||||
}
|
||||
if (loginSession) await notifyLoginSessionChanged(tx);
|
||||
if (loginSession)
|
||||
await tx
|
||||
.update(adapterAuthSessions)
|
||||
|
||||
@@ -7,6 +7,8 @@ export const DELIVERY_QUEUES = {
|
||||
executionContinuation: "execution-continuation",
|
||||
executionStatus: "execution-status",
|
||||
executionDisposition: "execution-disposition",
|
||||
adapterLoginCleanup: "adapter-login-cleanup",
|
||||
setupTokenCleanup: "setup-token-cleanup",
|
||||
localLoginCleanup: "local-login-cleanup",
|
||||
budgetEnforcement: "budget-enforcement",
|
||||
accounting: "cost-accounting",
|
||||
|
||||
@@ -1,11 +1,7 @@
|
||||
import { and, inArray } from "drizzle-orm";
|
||||
import { orphanLoginLeases } from "./login-cleanup-state.js";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { adapterAuthSessions } from "@paperclipai/db";
|
||||
import {
|
||||
ADAPTER_AUTH_ACTIVE_STATUSES,
|
||||
DISPLAYED_CODE_ADAPTER_TYPES,
|
||||
decodePendingTerminal,
|
||||
LOGIN_LEASE_SESSION_TAG_KEY,
|
||||
observeSandboxDelete,
|
||||
terminalCleanupWrite,
|
||||
type AdapterAuthReaperStore,
|
||||
@@ -16,11 +12,6 @@ import { environmentService } from "./environments.js";
|
||||
|
||||
// The restart-safe cleanup backstop for adapter login sessions.
|
||||
//
|
||||
// The in-process five-minute timer in the login-session service stays the
|
||||
// primary control. This reaper is the backstop. It runs on server startup and on
|
||||
// a fixed interval, so sandbox cleanup survives a server restart and a delete
|
||||
// failure. The reaper sweeps three sets:
|
||||
//
|
||||
// 1. The expired non-terminal sessions. It deletes the login sandbox and marks
|
||||
// the session `timed_out`.
|
||||
// 2. The terminal sessions left in the internal `cleanup_pending` state. It
|
||||
@@ -308,35 +299,8 @@ export function createProductionLoginSessionReaperRuntime(
|
||||
return { outcome: "deleted" };
|
||||
},
|
||||
async listTaggedLeases() {
|
||||
// The candidate environments are those with a login session that still
|
||||
// owns a lease. An orphan session row keeps its environment, so this set
|
||||
// covers every orphan lease.
|
||||
const environmentRows = await deps.db
|
||||
.selectDistinct({ environmentId: adapterAuthSessions.environmentId })
|
||||
.from(adapterAuthSessions)
|
||||
.where(
|
||||
and(
|
||||
// The shared table also holds the setup-token rows. Filter by the
|
||||
// closed set of displayed-code adapter types, so the orphan sweep
|
||||
// reads only displayed-code rows.
|
||||
inArray(adapterAuthSessions.adapterType, DISPLAYED_CODE_ADAPTER_TYPES),
|
||||
inArray(adapterAuthSessions.status, [
|
||||
...ADAPTER_AUTH_ACTIVE_STATUSES,
|
||||
"cleanup_pending",
|
||||
]),
|
||||
),
|
||||
);
|
||||
const tagged: TaggedLease[] = [];
|
||||
for (const { environmentId } of environmentRows) {
|
||||
const leases = await environmentsSvc.listLeases(environmentId, { status: "active" });
|
||||
for (const lease of leases) {
|
||||
const sessionId = lease.metadata?.[LOGIN_LEASE_SESSION_TAG_KEY];
|
||||
if (typeof sessionId === "string" && lease.providerLeaseId) {
|
||||
tagged.push({ environmentId, providerLeaseId: lease.providerLeaseId, sessionId });
|
||||
}
|
||||
}
|
||||
}
|
||||
return tagged;
|
||||
const rows = await orphanLoginLeases(deps.db);
|
||||
return rows.map(row => ({ environmentId: row.environmentId!, providerLeaseId: row.providerLeaseId!, sessionId: row.sessionId }));
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,3 +1,5 @@
|
||||
import { writeLoginCleanup } from "./login-cleanup-notifications.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -532,7 +534,7 @@ export function createDbAdapterAuthSessionStore(
|
||||
return {
|
||||
async insert(input) {
|
||||
try {
|
||||
await db.insert(adapterAuthSessions).values({
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.adapterLoginCleanup, tx => tx.insert(adapterAuthSessions).values({
|
||||
id: input.id,
|
||||
companyId: input.companyId,
|
||||
environmentId: input.environmentId,
|
||||
@@ -548,7 +550,7 @@ export function createDbAdapterAuthSessionStore(
|
||||
expiresAt: input.expiresAt,
|
||||
createdAt: input.at,
|
||||
updatedAt: input.at,
|
||||
});
|
||||
}));
|
||||
} catch (error) {
|
||||
if (isUniqueViolation(error)) {
|
||||
throw new AdapterAuthSessionConflictError();
|
||||
@@ -557,23 +559,23 @@ export function createDbAdapterAuthSessionStore(
|
||||
}
|
||||
},
|
||||
async recordLeaseAcquired(input) {
|
||||
await db
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.adapterLoginCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set({ providerLeaseId: input.providerLeaseId, updatedAt: input.at })
|
||||
.where(eq(adapterAuthSessions.id, input.sessionId));
|
||||
.where(eq(adapterAuthSessions.id, input.sessionId)));
|
||||
},
|
||||
async setStatus(input) {
|
||||
await db
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.adapterLoginCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set(buildStatusPatch(input))
|
||||
.where(eq(adapterAuthSessions.id, input.sessionId));
|
||||
.where(eq(adapterAuthSessions.id, input.sessionId)));
|
||||
},
|
||||
async compareAndSetStatus(input) {
|
||||
// The conditional write. It changes the row only when the current status is
|
||||
// one of `expectedStatuses`. The `returning` clause reports the changed
|
||||
// rows, so a lost race returns an empty array. The active-slot partial
|
||||
// unique index still serializes the company slot on the write.
|
||||
const changed = await db
|
||||
const changed = await writeLoginCleanup(db, DELIVERY_QUEUES.adapterLoginCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set(buildStatusPatch(input))
|
||||
.where(
|
||||
@@ -582,7 +584,7 @@ export function createDbAdapterAuthSessionStore(
|
||||
inArray(adapterAuthSessions.status, [...input.expectedStatuses]),
|
||||
),
|
||||
)
|
||||
.returning({ id: adapterAuthSessions.id });
|
||||
.returning({ id: adapterAuthSessions.id }));
|
||||
return changed.length > 0;
|
||||
},
|
||||
async get(sessionId) {
|
||||
@@ -649,6 +651,7 @@ export function createDbAdapterAuthSessionStore(
|
||||
// The shared table also holds the setup-token rows, so every reaper
|
||||
// scan filters by the closed set of displayed-code adapter types.
|
||||
inArray(adapterAuthSessions.adapterType, DISPLAYED_CODE_ADAPTER_TYPES),
|
||||
sql`${adapterAuthSessions.connectionMethod} is distinct from 'local_subscription'`,
|
||||
inArray(adapterAuthSessions.status, [...ADAPTER_AUTH_ACTIVE_STATUSES]),
|
||||
isNotNull(adapterAuthSessions.expiresAt),
|
||||
lte(adapterAuthSessions.expiresAt, nowAt),
|
||||
@@ -668,6 +671,7 @@ export function createDbAdapterAuthSessionStore(
|
||||
.where(
|
||||
and(
|
||||
inArray(adapterAuthSessions.adapterType, DISPLAYED_CODE_ADAPTER_TYPES),
|
||||
sql`${adapterAuthSessions.connectionMethod} is distinct from 'local_subscription'`,
|
||||
eq(adapterAuthSessions.status, "cleanup_pending"),
|
||||
),
|
||||
);
|
||||
@@ -1651,10 +1655,10 @@ export function createProductionLoginSessionRuntime(
|
||||
);
|
||||
// Tag the lease with the session identifier, so the reaper resolves an
|
||||
// orphan lease that no live session references. The tag carries no secret.
|
||||
await environmentsSvc.updateLeaseMetadata(record.lease.id, {
|
||||
await writeLoginCleanup(deps.db, DELIVERY_QUEUES.adapterLoginCleanup, tx => environmentService(tx).updateLeaseMetadata(record.lease.id, {
|
||||
...(record.lease.metadata ?? {}),
|
||||
[LOGIN_LEASE_SESSION_TAG_KEY]: input.sessionId,
|
||||
});
|
||||
}));
|
||||
const sessionHome = sessionLoginHomePath(input.sessionId);
|
||||
const authPath = sessionCredentialPath(input.sessionId);
|
||||
const providerLeaseId = record.lease.providerLeaseId ?? record.lease.id;
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type LoginCleanupQueue = typeof DELIVERY_QUEUES.adapterLoginCleanup | typeof DELIVERY_QUEUES.setupTokenCleanup;
|
||||
|
||||
export function writeLoginCleanup<T>(db: Db, queue: LoginCleanupQueue, write: (tx: Db) => PromiseLike<T>): Promise<T> {
|
||||
return db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, queue);
|
||||
return write(tx as unknown as Db);
|
||||
});
|
||||
}
|
||||
|
||||
export async function notifyLoginSessionChanged(transaction: object): Promise<void> {
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.adapterLoginCleanup);
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.setupTokenCleanup);
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
import { and, eq, inArray, isNotNull, notExists, or, sql } from "drizzle-orm";
|
||||
import { adapterAuthSessions, environmentLeases, type Db } from "@paperclipai/db";
|
||||
import { ADAPTER_AUTH_ACTIVE_STATUSES, DISPLAYED_CODE_ADAPTER_TYPES, LOGIN_LEASE_SESSION_TAG_KEY } from "./device-login-service.js";
|
||||
import { SETUP_TOKEN_ADAPTER_TYPE, SETUP_TOKEN_TERMINAL_STATES } from "./setup-token-session.js";
|
||||
|
||||
export const LOGIN_CLEANUP_RETRY_MS = 30_000;
|
||||
|
||||
export function orphanLoginLeases(db: Db, now = new Date()) {
|
||||
return db.select({
|
||||
environmentId: environmentLeases.environmentId,
|
||||
providerLeaseId: environmentLeases.providerLeaseId,
|
||||
sessionId: sql<string>`${environmentLeases.metadata}->>${LOGIN_LEASE_SESSION_TAG_KEY}`,
|
||||
}).from(environmentLeases).where(and(
|
||||
or(eq(environmentLeases.status, "active"), eq(environmentLeases.cleanupStatus, "failed")),
|
||||
isNotNull(environmentLeases.environmentId),
|
||||
isNotNull(environmentLeases.providerLeaseId),
|
||||
sql`jsonb_typeof(${environmentLeases.metadata}->${LOGIN_LEASE_SESSION_TAG_KEY}) = 'string'`,
|
||||
notExists(db.select({ id: adapterAuthSessions.id }).from(adapterAuthSessions).where(and(
|
||||
eq(adapterAuthSessions.companyId, environmentLeases.companyId),
|
||||
eq(adapterAuthSessions.environmentId, environmentLeases.environmentId),
|
||||
or(
|
||||
eq(adapterAuthSessions.providerLeaseId, environmentLeases.providerLeaseId),
|
||||
and(
|
||||
sql`${adapterAuthSessions.id}::text = ${environmentLeases.metadata}->>${LOGIN_LEASE_SESSION_TAG_KEY}`,
|
||||
inArray(adapterAuthSessions.status, [...ADAPTER_AUTH_ACTIVE_STATUSES]),
|
||||
sql`greatest(${adapterAuthSessions.expiresAt}, case when ${adapterAuthSessions.status} = 'promoting' then ${adapterAuthSessions.promotionExpiresAt} end) > ${now.toISOString()}::timestamptz`,
|
||||
),
|
||||
),
|
||||
))),
|
||||
));
|
||||
}
|
||||
|
||||
export function loginCleanupSchedule(db: Db, retryMs = LOGIN_CLEANUP_RETRY_MS) {
|
||||
const retry = () => Date.now() + retryMs;
|
||||
const next = (value: number | string | null | undefined) => value == null ? null : Math.max(Number(value), retry());
|
||||
return {
|
||||
async adapter() {
|
||||
const [rows, orphans] = await Promise.all([
|
||||
db.select({ at: sql<string | null>`min(extract(epoch from case
|
||||
when ${adapterAuthSessions.status} = 'cleanup_pending' then now()
|
||||
else greatest(${adapterAuthSessions.expiresAt}, case when ${adapterAuthSessions.status} = 'promoting' then ${adapterAuthSessions.promotionExpiresAt} end)
|
||||
end) * 1000)` }).from(adapterAuthSessions).where(and(
|
||||
inArray(adapterAuthSessions.adapterType, DISPLAYED_CODE_ADAPTER_TYPES),
|
||||
sql`${adapterAuthSessions.connectionMethod} is distinct from 'local_subscription'`,
|
||||
inArray(adapterAuthSessions.status, [...ADAPTER_AUTH_ACTIVE_STATUSES, "cleanup_pending"]),
|
||||
)),
|
||||
orphanLoginLeases(db).limit(1),
|
||||
]);
|
||||
return orphans.length ? retry() : next(rows[0]?.at);
|
||||
},
|
||||
async setupToken() {
|
||||
const rows = await db.select({ at: sql<string | null>`min(extract(epoch from case
|
||||
when ${inArray(adapterAuthSessions.status, [...SETUP_TOKEN_TERMINAL_STATES])} or ${adapterAuthSessions.boundAt} is not null then now()
|
||||
else ${adapterAuthSessions.expiresAt} end) * 1000)` }).from(adapterAuthSessions).where(and(
|
||||
eq(adapterAuthSessions.adapterType, SETUP_TOKEN_ADAPTER_TYPE),
|
||||
sql`${adapterAuthSessions.connectionMethod} is distinct from 'local_subscription'`,
|
||||
));
|
||||
return next(rows[0]?.at);
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import type { LoginCleanupQueue } from "./login-cleanup-notifications.js";
|
||||
import { LOGIN_CLEANUP_RETRY_MS } from "./login-cleanup-state.js";
|
||||
|
||||
export function registerLoginCleanupWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
duties: ReadonlyArray<{ queue: LoginCleanupQueue; run: () => Promise<unknown>; next: () => Promise<number | null> }>,
|
||||
canRun: () => boolean,
|
||||
retryMs = LOGIN_CLEANUP_RETRY_MS,
|
||||
) {
|
||||
const workers = duties.map(duty => {
|
||||
let requested = 1;
|
||||
let reconciled = 0;
|
||||
let next: number | null = null;
|
||||
let earliestRunAt = 0;
|
||||
return coordinator.register(duty.queue, {
|
||||
retryMs,
|
||||
pausePassiveMaintenance: true,
|
||||
onSettled: () => { requested += 1; },
|
||||
run: async () => {
|
||||
if (!canRun()) { requested += 1; next = null; return; }
|
||||
const now = Date.now();
|
||||
if (now < earliestRunAt) {
|
||||
if (reconciled !== requested || (next !== null && next <= now)) next = earliestRunAt;
|
||||
return;
|
||||
}
|
||||
const generation = ++requested;
|
||||
earliestRunAt = now + retryMs;
|
||||
await duty.run();
|
||||
const deadline = await duty.next();
|
||||
next = deadline === null ? null : Math.max(deadline, earliestRunAt);
|
||||
reconciled = generation;
|
||||
},
|
||||
hasPending: async () => reconciled !== requested,
|
||||
nextRunAt: () => next,
|
||||
});
|
||||
});
|
||||
return { ready: Promise.all(workers.map(worker => worker.ready)) };
|
||||
}
|
||||
@@ -14,14 +14,6 @@ import {
|
||||
|
||||
// The restart-safe cleanup backstop for the Claude setup-token login flow.
|
||||
//
|
||||
// The in-process five-minute timer in the login-session service stays the
|
||||
// primary control. This reaper is the backstop. It runs on server startup and on
|
||||
// a fixed interval, so a sandbox lease survives a server restart and a release
|
||||
// failure. The reaper reads the durable cleanup store and releases any lease
|
||||
// whose session is terminal, past its deadline, or already consumed. It uses the
|
||||
// same convention as the Codex device-login reaper, so both flows use one reaper
|
||||
// convention.
|
||||
//
|
||||
// The reaper never records secret data. The durable cleanup record holds only
|
||||
// ids, the deadline, the claim marker, and the state. It never holds a URL, a
|
||||
// code, a token, or a raw process chunk.
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { writeLoginCleanup } from "./login-cleanup-notifications.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
// The Claude setup-token login session service. It owns a company-scoped,
|
||||
// owner-bound login session that holds one live setup-token process and one
|
||||
// sandbox lease for the whole login. The login is a two-way round-trip: the
|
||||
@@ -1515,7 +1517,7 @@ export async function transitionSetupTokenSessionToStored(
|
||||
executor: Db,
|
||||
identity: SetupTokenCleanupIdentity,
|
||||
): Promise<boolean> {
|
||||
const changed = await executor
|
||||
const changed = await writeLoginCleanup(executor, DELIVERY_QUEUES.setupTokenCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set({ status: "stored", updatedAt: sql`clock_timestamp()` })
|
||||
.where(
|
||||
@@ -1526,7 +1528,7 @@ export async function transitionSetupTokenSessionToStored(
|
||||
isNull(adapterAuthSessions.boundAt),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
.returning());
|
||||
return changed.length === 1;
|
||||
}
|
||||
|
||||
@@ -1545,7 +1547,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
async record(record): Promise<void> {
|
||||
// The database generates `id`. The store fills `public_session_id` with the
|
||||
// service session id, which the service builds from a CSPRNG at start.
|
||||
await db.insert(adapterAuthSessions).values({
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.setupTokenCleanup, tx => tx.insert(adapterAuthSessions).values({
|
||||
companyId: record.companyId,
|
||||
aiConnection: record.aiConnection,
|
||||
environmentId: record.environmentId,
|
||||
@@ -1556,22 +1558,22 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
status: record.state,
|
||||
expiresAt: new Date(record.deadline),
|
||||
boundAt: record.boundAt === null ? null : new Date(record.boundAt),
|
||||
});
|
||||
}));
|
||||
},
|
||||
|
||||
async markState(identity, state): Promise<void> {
|
||||
// The compare-and-set predicate matches the company, the owner, and the
|
||||
// adapter, so a write never updates a row by the session id alone.
|
||||
await db
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.setupTokenCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set({ status: state, updatedAt: sql`clock_timestamp()` })
|
||||
.where(scopeMatch(identity));
|
||||
.where(scopeMatch(identity)));
|
||||
},
|
||||
|
||||
async remove(identity): Promise<void> {
|
||||
// The delete matches the company, the owner, and the adapter, so it never
|
||||
// removes a row by the session id alone.
|
||||
await db.delete(adapterAuthSessions).where(scopeMatch(identity));
|
||||
await writeLoginCleanup(db, DELIVERY_QUEUES.setupTokenCleanup, tx => tx.delete(adapterAuthSessions).where(scopeMatch(identity)));
|
||||
},
|
||||
|
||||
async listReapable(now): Promise<SetupTokenCleanupRecord[]> {
|
||||
@@ -1585,6 +1587,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
.where(
|
||||
and(
|
||||
eq(adapterAuthSessions.adapterType, SETUP_TOKEN_ADAPTER_TYPE),
|
||||
sql`${adapterAuthSessions.connectionMethod} is distinct from 'local_subscription'`,
|
||||
or(
|
||||
inArray(adapterAuthSessions.status, [...SETUP_TOKEN_TERMINAL_STATES]),
|
||||
lte(adapterAuthSessions.expiresAt, new Date(now)),
|
||||
@@ -1602,7 +1605,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
// database evaluates the current time after any row-lock wait. An expired
|
||||
// claim cannot pass the deadline condition. The write sets `bound_at` one
|
||||
// time and returns the row only on a valid consume.
|
||||
const changed = await db
|
||||
const changed = await writeLoginCleanup(db, DELIVERY_QUEUES.setupTokenCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set({ boundAt: sql`clock_timestamp()`, updatedAt: sql`clock_timestamp()` })
|
||||
.where(
|
||||
@@ -1613,7 +1616,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
isNull(adapterAuthSessions.boundAt),
|
||||
),
|
||||
)
|
||||
.returning();
|
||||
.returning());
|
||||
const row = changed[0];
|
||||
return row ? toCleanupRecord(row) : null;
|
||||
},
|
||||
@@ -1622,7 +1625,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
// One conditional write. The predicate carries the company, the owner, the
|
||||
// adapter, the session id, and one of `cancellableStates`. It never
|
||||
// interrupts a `submitting` write, a `stored` claim, or a terminal row.
|
||||
const changed = await db
|
||||
const changed = await writeLoginCleanup(db, DELIVERY_QUEUES.setupTokenCleanup, tx => tx
|
||||
.update(adapterAuthSessions)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
@@ -1630,7 +1633,7 @@ export function createDbSetupTokenCleanupStore(db: Db): SetupTokenCleanupStore {
|
||||
updatedAt: sql`clock_timestamp()`,
|
||||
})
|
||||
.where(and(scopeMatch(identity), inArray(adapterAuthSessions.status, [...cancellableStates])))
|
||||
.returning();
|
||||
.returning());
|
||||
const row = changed[0];
|
||||
return row ? toCleanupRecord(row) : null;
|
||||
},
|
||||
|
||||
Reference in new issue
Block a user