diff --git a/doc/idle-login-cleanup.md b/doc/idle-login-cleanup.md new file mode 100644 index 00000000000..813cf34eb85 --- /dev/null +++ b/doc/idle-login-cleanup.md @@ -0,0 +1,97 @@ +# Idle login cleanup + +Adapter device-login and Claude setup-token cleanup no longer scan on each +heartbeat scheduler tick (30 seconds by default). Two independent workers use +existing commit-aware delivery signals and the in-process work scheduler. No +schema or queue table is added. + +## Ownership and triggers + +| Duty | Existing durable truth | Trigger / deadline | Quiet condition | +| --- | --- | --- | --- | +| Displayed-code session expiry | `adapter_auth_sessions` | Insert or state change; session expiry, extended by an active promotion claim | No active expiring sessions | +| Displayed-code failed deletion | Session `cleanup_pending` | State change; 30-second retry while pending | Provider confirms deletion and status settles | +| Displayed-code orphan cleanup | Tagged `environment_leases` | Lease tag write, session transition, startup; retry while an orphan remains | No unreferenced active/failed-cleanup tagged leases | +| Setup-token expiry / consumed claims / terminal cleanup | Existing setup-token rows in `adapter_auth_sessions` | Record, state, claim, cancellation or deletion commit; expiry or retry | Cleanup rows removed after release | + +`login-cleanup-state.ts` owns the database predicates and reconstructed deadlines. +`login-cleanup-work.ts` registers the independent workers. Domain stores register +intent before each write via `writeLoginCleanup`. The AI connection save path +also signals its session mutation in its existing transaction through +`notifyLoginSessionChanged`. That helper invalidates both sandbox cleanup workers +without choosing a provider. Each worker inspects its own durable predicates; the +unaffected worker becomes quiet after reconciliation. This bounded extra scan +occurs only after a session write, not on a recurring idle cadence. Login flows +still own completion through the callback introduced in PR #17129. Nested writes use +the shared outer-transaction settlement protocol, including rollback and uncertain +commit reconciliation. Notification means inspect durable state, not perform one +unit of work. + +Each worker coalesces bursts and starts at most once per configured heartbeat +interval (30 seconds by default). This keeps the old scan budget and cleanup +resolution. Retry intervals use that same configuration. A future deadline beyond that +spacing is scheduled directly. A due or failed obligation retries after 30 seconds; +an empty worker has no recurring scan or timer. A write during a sweep retains a +new generation, including writes made by cleanup itself. A final follow-up scan +can therefore occur before the worker becomes quiet. + +Local-subscription sessions share the table but are excluded from these sandbox +cleanup predicates. Their separate local-login worker owns filesystem cleanup. +Completed displayed-code session history is passive. A terminal setup-token row +is still work until its cleanup record is removed. + +## Acquisition and orphan recovery + +The displayed-code service persists a session and expiry before acquiring a +sandbox. Its production runtime signals the lease-tag write. An unreferenced +lease with that tag must not be deleted while its owning session is active and +unexpired, including a live promotion claim. The session deadline covers an +attachment that never completes. A failed session transition makes the orphan +eligible immediately, subject to worker spacing. + +The orphan query examines tagged lease rows directly, so a terminal or missing +session cannot hide a failed cleanup by removing its environment from the scan. +A failed orphan delete remains visible and keeps its retry armed. Existing bound +session cleanup, promotion locking, conditional status updates and provider +confirmation semantics remain with their existing owners. + +This is not a new provider-orphan discovery mechanism. A provider resource that +never became a persisted tagged lease still belongs to the existing general +lease cleanup / failed-acquisition spool recovery. Setup-token acquisition still +requires provider-backed expiry before creating its cleanup record. The general +sandbox cleanup worker and its startup recovery remain unchanged. Related +pre-existing setup-token release-outcome and null-lease fixes are tracked in +public PRs #15774 and #14236 (issue #16648); this change does not replace them. + +## Lifecycle and sleep boundary + +Workers register during startup housekeeping. They reconstruct current work even +when no producer notification survived a restart. Failed scans keep a retry. +Startup, a notification, or a failed attempt skipped during drain keeps its +requested generation until reconciliation succeeds after resume. Passive waiting +uses memory only. Warm standby and shutdown do not dispatch cleanup. Coordinator +shutdown removes subscriptions, stops timers and waits for in-flight work. + +This supports one active server and writers using the shared database handle and +instrumented stores. Independent SQL writers and other replicas do not generate +these process-local signals and require their own integration. + +No Cloud wake coverage or host-sleep exemptions are added. Outstanding login +sessions and unresolved resources retain the existing durable idle-safety blockers. +A local future deadline is not an external alarm. These tests establish worker +quiet and reactivation, not measured provider suspension or fleet savings. + +## Verification + +`server/src/__tests__/login-cleanup-work.test.ts` exercises empty/configured/history +states, real store reactivation, expiry, promotion deadlines, cancellation, +failed-delete retry and restart, live lease attachment, orphan recovery, nested +commit/rollback, setup-token consumption, burst coalescing, concurrent writes, +drain/resume and shutdown. Real PostgreSQL tests count calls through the worker's +database handle and provider release seam while fake time crosses former polling +periods. Existing device-login, setup-token and AI connection regression suites +cover the surrounding login contracts. + +The fake-clock database fixture warms its connections and disables pool idle/lifetime +expiry so time jumps test the worker rather than racing socket connection timers. +Database suspension and reconnect behavior require separate qualification. 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 e77a8719c71..d2bbb1ef91c 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 @@ -620,12 +620,16 @@ Longer polling intervals alone do not complete an item. malformed/unpersisted spool entries. Source: [heartbeat.ts](../../server/src/services/heartbeat.ts), `sweepPendingCleanupLeases`. -- [ ] **JOB-18 — Adapter login and setup-token reapers.** Current: **30s** - durable-session/lease scans, alongside local one-shot timers. Change: register - actual expiry and cleanup retry times. **Complete when:** abandoned logins and - interrupted teardown still release sandboxes across restart, without scans - when no obligations exist. Source: - [index.ts](../../server/src/index.ts), adapter and setup-token reaper wiring. +- [x] **JOB-18 — Adapter login and setup-token reapers.** Replaced the default + **30s** empty scans with commit-aware workers, session/promotion expiry + deadlines, and retries for concrete failed cleanup. Existing session and lease + rows remain authoritative; no queue table was added. Startup reconstructs work, + and drain retains deferred generations. Tagged orphan recovery protects live + acquisitions and includes terminal/missing-session orphans. Local-subscription + rows remain owned by their separate worker. General failed-acquisition spool + recovery and host-sleep blockers are unchanged. See + [idle-login-cleanup.md](../idle-login-cleanup.md) for producer coverage, + verification and remaining provider-orphan boundaries. - [ ] **JOB-19 — Custom-image setup expiry.** Current: **30s** active-session expiry scans. Change: register `expiresAt` and cleanup attempts. **Complete diff --git a/server/src/__tests__/ai-connections.test.ts b/server/src/__tests__/ai-connections.test.ts index 49b9cd7d81d..5f09e9bea31 100644 --- a/server/src/__tests__/ai-connections.test.ts +++ b/server/src/__tests__/ai-connections.test.ts @@ -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"; diff --git a/server/src/__tests__/device-login-service.test.ts b/server/src/__tests__/device-login-service.test.ts index 13566ca3858..8f41acddc23 100644 --- a/server/src/__tests__/device-login-service.test.ts +++ b/server/src/__tests__/device-login-service.test.ts @@ -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", () => { diff --git a/server/src/__tests__/login-cleanup-work.test.ts b/server/src/__tests__/login-cleanup-work.test.ts new file mode 100644 index 00000000000..c21d04c97d3 --- /dev/null +++ b/server/src/__tests__/login-cleanup-work.test.ts @@ -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[] = []; +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): Promise { + 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(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(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, observed: typeof db; + let temporary: Awaited> | undefined; + let companyId: string, environmentId: string, calls: number, failedDelete: boolean; + let store: ReturnType; + let runtime: ReturnType; + let driver: { releaseRunLease: ReturnType }; + 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) { + 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(resolve => { release = resolve; }); + const ready = new Promise(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([]); + }); +}); diff --git a/server/src/__tests__/login-session-runtime.test.ts b/server/src/__tests__/login-session-runtime.test.ts index d5cfb03faae..532526a397e 100644 --- a/server/src/__tests__/login-session-runtime.test.ts +++ b/server/src/__tests__/login-session-runtime.test.ts @@ -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(fn: (tx: object) => Promise) { 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(); }); }); diff --git a/server/src/index.ts b/server/src/index.ts index a7b0d17fabe..ed7417f4915 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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; diff --git a/server/src/services/ai-connections.ts b/server/src/services/ai-connections.ts index 0177f6352ac..434887f4dc2 100644 --- a/server/src/services/ai-connections.ts +++ b/server/src/services/ai-connections.ts @@ -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) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 26f737b39cd..f91df0318f1 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", + adapterLoginCleanup: "adapter-login-cleanup", + setupTokenCleanup: "setup-token-cleanup", localLoginCleanup: "local-login-cleanup", budgetEnforcement: "budget-enforcement", accounting: "cost-accounting", diff --git a/server/src/services/device-login-reaper.ts b/server/src/services/device-login-reaper.ts index f49d5308932..f19151fb60c 100644 --- a/server/src/services/device-login-reaper.ts +++ b/server/src/services/device-login-reaper.ts @@ -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 })); }, }; } diff --git a/server/src/services/device-login-service.ts b/server/src/services/device-login-service.ts index a76d46f179d..b409bef2c4f 100644 --- a/server/src/services/device-login-service.ts +++ b/server/src/services/device-login-service.ts @@ -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; diff --git a/server/src/services/login-cleanup-notifications.ts b/server/src/services/login-cleanup-notifications.ts new file mode 100644 index 00000000000..d6279b7f156 --- /dev/null +++ b/server/src/services/login-cleanup-notifications.ts @@ -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(db: Db, queue: LoginCleanupQueue, write: (tx: Db) => PromiseLike): Promise { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, queue); + return write(tx as unknown as Db); + }); +} + +export async function notifyLoginSessionChanged(transaction: object): Promise { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.adapterLoginCleanup); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.setupTokenCleanup); +} diff --git a/server/src/services/login-cleanup-state.ts b/server/src/services/login-cleanup-state.ts new file mode 100644 index 00000000000..5129111a59f --- /dev/null +++ b/server/src/services/login-cleanup-state.ts @@ -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`${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`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`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); + }, + }; +} diff --git a/server/src/services/login-cleanup-work.ts b/server/src/services/login-cleanup-work.ts new file mode 100644 index 00000000000..d3d7e44cf46 --- /dev/null +++ b/server/src/services/login-cleanup-work.ts @@ -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, + duties: ReadonlyArray<{ queue: LoginCleanupQueue; run: () => Promise; next: () => Promise }>, + 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)) }; +} diff --git a/server/src/services/setup-token-reaper.ts b/server/src/services/setup-token-reaper.ts index 0a6926bcb3b..a1105eaea97 100644 --- a/server/src/services/setup-token-reaper.ts +++ b/server/src/services/setup-token-reaper.ts @@ -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. diff --git a/server/src/services/setup-token-session.ts b/server/src/services/setup-token-session.ts index bf324479e48..bd291a7b8cf 100644 --- a/server/src/services/setup-token-session.ts +++ b/server/src/services/setup-token-session.ts @@ -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 { - 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 { // 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 { // 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 { // 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 { @@ -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; },