From efc2ddc7bf19ec319b21cd294e71e5a4932c16f4 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 21 Oct 2026 15:39:23 -0500 Subject: [PATCH] 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 --- doc/idle-login-cleanup.md | 97 ++++++++ ...-instance-sleep-and-scheduler-inventory.md | 16 +- server/src/__tests__/ai-connections.test.ts | 18 +- .../__tests__/device-login-service.test.ts | 6 +- .../src/__tests__/login-cleanup-work.test.ts | 226 ++++++++++++++++++ .../__tests__/login-session-runtime.test.ts | 19 +- server/src/index.ts | 43 +--- server/src/services/ai-connections.ts | 2 + .../services/delivery-work-notifications.ts | 2 + server/src/services/device-login-reaper.ts | 42 +--- server/src/services/device-login-service.ts | 24 +- .../services/login-cleanup-notifications.ts | 16 ++ server/src/services/login-cleanup-state.ts | 61 +++++ server/src/services/login-cleanup-work.ts | 39 +++ server/src/services/setup-token-reaper.ts | 8 - server/src/services/setup-token-session.ts | 25 +- 16 files changed, 529 insertions(+), 115 deletions(-) create mode 100644 doc/idle-login-cleanup.md create mode 100644 server/src/__tests__/login-cleanup-work.test.ts create mode 100644 server/src/services/login-cleanup-notifications.ts create mode 100644 server/src/services/login-cleanup-state.ts create mode 100644 server/src/services/login-cleanup-work.ts 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; },