diff --git a/doc/idle-gateway-credential-expiry.md b/doc/idle-gateway-credential-expiry.md new file mode 100644 index 00000000000..e0486d39886 --- /dev/null +++ b/doc/idle-gateway-credential-expiry.md @@ -0,0 +1,141 @@ +# Gateway credential expiry and idle behavior + +Gateway sessions and named gateway tokens already persist their expiry times. +`cleanupExpiredSessions` remains the owner of physical deletion. Authentication +continues to check expiry, revocation, company scope, run ownership, and active +run state on use. No table, authorization rule, or external alarm is added. + +## Producer inventory + +- `createSession` registers intent before inserting a run-bound session in a + transaction. Its existing audit and returned credential remain unchanged. +- `createNamedGatewayToken` registers intent before inserting an expiring token + in a transaction. A token without an expiry owes no expiry cleanup and does + not notify this worker. +- Last-use updates do not change expiry. Token/session revocation, including + heartbeat completion and connection removal, also leaves the saved expiry + intact. Revoked rows retain the same physical deletion deadline. +- Gateway/company deletion removes dependent rows and can leave one harmless + stale deadline. It cannot create overdue work. No supported service changes + a saved credential expiry after creation. + +The database signal layer owns outer commit, rollback, savepoints, and uncertain +outcomes. The worker supports producers sharing that signal owner. Independent +SQL writers and another process need a separate live ingress contract. Startup +reconstructs existing rows but does not provide continuous replica discovery. + +## Worker contract + +`nextCredentialExpiryAt` reads one earliest non-null expiry from each table using +expiry-first indexes, then compares those at most two values. The migration adds +the session expiry index; the token expiry index already exists. +An empty credential store or retained non-expiring tokens returns no deadline. +`registerGatewayCredentialExpiryWork` supplies cleanup and deadline selection to +the shared `registerDeadlineWork` helper. Provider and authentication behavior +stay in the gateway service; startup only registers the worker. + +The worker starts where the old startup cleanup ran. With heartbeat scheduling +enabled, committed changes and actual expiry times replace the recurring cleanup +call. With heartbeat scheduling disabled, the existing startup-only cleanup is +preserved. Tool action-review recovery is a separate obligation and is unchanged. + +Token and session deletion are each bounded to 500 rows per pass. Session +selection orders by expiry and rechecks expiry when deleting the selected IDs. +A remaining overdue expiry schedules another pass after 30 seconds. Failed reconciliation also +retries at that bound. No unfinished reconciliation or future expiry means no +recurring SQL. A generation retains writes arriving during reconciliation. +Startup restores deadlines; drain retains first writes without passive SQL; +resume completes overdue cleanup. The coordinator owns shutdown and waits for +in-flight work. + +Sequential committed writes are coalesced for up to 30 seconds after a +successful cleanup pass, matching the former heartbeat cleanup cadence. A +previously loaded expiry can preempt that window. A newly reported credential +may physically remain until that bounded follow-up; authentication still rejects +it immediately at expiry. A successful pass with no future expiry and no new +notifications creates no timer. This uses the shared optional notification +coalescing capability, promoted from the later plugin/decision recovery slices; +workers that omit it retain their existing behavior. + +## Evidence and limits + +The concurrent session index runs through the migration executor's existing +per-file top-level path. Initial application and repeat application were tested. +A PostgreSQL plan with 20,000 sessions and 20,000 tokens, including 19,000 +non-expiring tokens, used both expiry indexes and read one row from each. +This establishes bounded lookup on that fixture, not production latency. + +The new PostgreSQL suite exercises real session/token creation, configured idle +state, non-expiring history, expiry ordering, revoked-token cleanup, outer commit +and rollback, startup reconstruction, a backlog larger than one cleanup batch, +and drain/resume. It tests expired authentication before deferred physical +cleanup, through both session and named-gateway entry points. Existing gateway +acceptance and server-startup tests cover the changed integration boundaries. + +Use only disposable migrated databases for the test URL overrides. Their names +must end in `_test`; the suites delete their fixtures. Ordinary CI uses embedded +PostgreSQL. Test credentials are synthetic and no external provider is contacted +by the expiry suite. + +The migration proves worker behavior, not host suspension. Existing Cloud idle +proof remains conservative. An in-process deadline cannot wake stopped compute; +external wake coverage must be verified before changing sleep eligibility. + +## Related unmerged token-lifetime work + +At implementation time, PRs #17221 and #17274 change run-scoped token lifetime +and cleanup eligibility. This slice preserves the current master policy and +changes scheduling only. When combining those changes, the deadline query and +producer signals must use the same token eligibility as cleanup: run-token idle +markers must not schedule the operator-token expiry worker. Keep the run-liveness +reaper with its own owner and retain both regression suites. A conflict-free +text merge alone does not prove this integration. + +## Notification amplification review repair (2026-10-27) + +Ten sequential real session/token creation transactions reproduced ten extra +instance-wide cleanup passes. With coalescing they cause one follow-up pass. +A separate 501-session fixture reproduced the unbounded delete, then passed +with 500 removed initially and the remaining overdue row removed at the saved +30-second retry. A still-valid session survives both passes. The tests also +preserve an already loaded expiry only 500 milliseconds away and verify that a +new one-second session is rejected while physical cleanup waits for coalescing. + +The initial full run exposed four outer-transaction tests whose fake-clock wait +expected the intent hold to release before the new coalescing follow-up. Their +settlement helper now advances the clock through that window while preserving +the original commit/rollback assertions. The complete gateway expiry, gateway +acceptance, custom-image deadline, and startup suites pass: 147 tests in four +files. Server typechecking also passes on the owning branch. This is local +behavioral evidence; combined-stack and fresh remote checks remain required. +The full local test:run remains unverified because of embedded PostgreSQL +startup/capacity limits; the two real gateway suites used separate migrated, +task-owned disposable PostgreSQL databases. + +A later combined run exposed a test-clock overshoot: repeated blind 30-second +advances could expire the future session before its intermediate assertion. +The settlement helper now lets real SQL settle and advances only to an actually +armed follow-up within the coalescing window. All 12 credential expiry tests pass +on the owning gateway branch with the existing exact-deadline and transaction +assertions intact. This follow-up changes the test fixture only. + + +## One-shot cleanup preserves complete backlog removal + +Review found that bounding worker passes also truncated startup cleanup when the +heartbeat scheduler was disabled. That path calls cleanup only once, so it could +leave every expired token or session after the first 500 rows indefinitely. + +`cleanupExpiredSessions` now drains the existing backlog in 500-row SQL batches +by default, using one fixed expiry cutoff for the invocation. The deadline worker +opts into `singleBatch: true` and retains its existing coalescing and next-page +deadline. Future and nonexpiring credentials survive both paths. The return value +still counts deleted sessions, as before. No periodic fallback was added. + +Three real-PostgreSQL regressions reproduced the truncation with 1,001 expired +sessions, tokens, or both, without registering a worker. They pass after the fix. +All 55 expiry/startup cases pass, including the existing worker's 501-row bounded +pass and commit/rollback tests. The original explicit-cleanup compatibility case +and server typecheck also pass. Evidence is in `/tmp/sleep-master-gateway-manual-` +logs: `red`, `green`, `compatibility`, and `types`. This fixes one-shot behavior; +it does not establish provider suspension or complete the stack's CI acceptance. 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 9141878caa8..38dec000a0e 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 @@ -649,8 +649,10 @@ Longer polling intervals alone do not complete an item. - [ ] **JOB-21 — Secret proposals and expired action tokens (partially migrated).** Secret-proposal expiry now uses transaction signals and saved expiry deadlines; - see [the lifecycle contract](../idle-secret-proposal-expiry.md). Tool review - recovery and gateway token cleanup still use **30s** queries. Change: enforce expiry synchronously on use and + see [the lifecycle contract](../idle-secret-proposal-expiry.md). Gateway token + and session expiry now also uses saved deadlines and producer signals; see + [its lifecycle contract](../idle-gateway-credential-expiry.md). Tool review + recovery still uses **30s** queries. Change: enforce expiry synchronously on use and register required scrub/cleanup deadlines. **Complete when:** idle cleanup does not extend authorization or secret-retention periods. Sources: [secret-proposals.ts](../../server/src/services/secret-proposals.ts), diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 20221274e70..009401a1d93 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -18,7 +18,7 @@ Do not mark an item done from a timer removal or a green check alone. | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | -| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret-proposal expiry in progress; tool reviews and gateway tokens pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged after CI and passing reviews; gateway credential expiry in progress; tool reviews pending | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 7f067cbfc1c..77fb0c62fa1 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -501,6 +501,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0332_little_night_nurse.sql", "0345_luxuriant_carmella_unuscione.sql", "0369_late_logan.sql", + "0381_mean_scream.sql", ]); async function applyPendingMigrationsManually( diff --git a/packages/db/src/migrations/0381_mean_scream.sql b/packages/db/src/migrations/0381_mean_scream.sql new file mode 100644 index 00000000000..8827365ffdd --- /dev/null +++ b/packages/db/src/migrations/0381_mean_scream.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_gateway_sessions_expiry_idx" ON "tool_gateway_sessions" USING btree ("expires_at"); diff --git a/packages/db/src/migrations/meta/0375_snapshot.json b/packages/db/src/migrations/meta/0381_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0375_snapshot.json rename to packages/db/src/migrations/meta/0381_snapshot.json index 7bd1861bd2c..23802aaeece 100644 --- a/packages/db/src/migrations/meta/0375_snapshot.json +++ b/packages/db/src/migrations/meta/0381_snapshot.json @@ -1,6 +1,6 @@ { - "id": "3e17203b-ec95-41e9-8363-685982450199", - "prevId": "945c31ed-f39f-4272-a91d-85de925f5361", + "id": "2e93deb7-f24c-42d1-8c8e-30b21cff90bc", + "prevId": "bfda87fa-662b-47bb-87c0-dabc814aebd7", "version": "7", "dialect": "postgresql", "tables": { @@ -47782,6 +47782,18 @@ "primaryKey": false, "notNull": false }, + "git_authorization_grant_id": { + "name": "git_authorization_grant_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "git_authorized_by_user_id": { + "name": "git_authorized_by_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "pause_reason": { "name": "pause_reason", "type": "text", @@ -47920,6 +47932,19 @@ ], "onDelete": "no action", "onUpdate": "no action" + }, + "projects_git_authorization_grant_id_connection_grants_id_fk": { + "name": "projects_git_authorization_grant_id_connection_grants_id_fk", + "tableFrom": "projects", + "tableTo": "connection_grants", + "columnsFrom": [ + "git_authorization_grant_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" } }, "compositePrimaryKeys": {}, @@ -55197,6 +55222,21 @@ "method": "btree", "with": {} }, + "tool_gateway_sessions_expiry_idx": { + "name": "tool_gateway_sessions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_gateway_sessions_run_idx": { "name": "tool_gateway_sessions_run_idx", "columns": [ @@ -55942,6 +55982,12 @@ "notNull": true, "default": "''" }, + "responsible_user_id": { + "name": "responsible_user_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, "owner_note": { "name": "owner_note", "type": "text", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 96c505941fd..176a963f7a4 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2654,6 +2654,13 @@ "when": 1793060791724, "tag": "0380_broken_aqueduct", "breakpoints": true + }, + { + "idx": 381, + "version": "7", + "when": 1793060791725, + "tag": "0381_mean_scream", + "breakpoints": true } ] } diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index f4aede1b4cd..825eb91589f 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -714,6 +714,7 @@ export const toolGatewaySessions = pgTable( uniqueIndex("tool_gateway_sessions_token_hash_uq").on(table.tokenHash), index("tool_gateway_sessions_company_agent_idx").on(table.companyId, table.agentId), index("tool_gateway_sessions_company_expires_idx").on(table.companyId, table.expiresAt), + index("tool_gateway_sessions_expiry_idx").on(table.expiresAt), index("tool_gateway_sessions_run_idx").on(table.companyId, table.runId), index("tool_gateway_sessions_issue_idx").on(table.companyId, table.issueId), index("tool_gateway_sessions_gateway_idx").on(table.companyId, table.gatewayId), diff --git a/server/src/__tests__/gateway-credential-expiry.test.ts b/server/src/__tests__/gateway-credential-expiry.test.ts new file mode 100644 index 00000000000..caa76674104 --- /dev/null +++ b/server/src/__tests__/gateway-credential-expiry.test.ts @@ -0,0 +1,269 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, companies, createDb, heartbeatRuns, toolAccessAuditEvents, toolGatewaySessions, + toolMcpGateways, toolMcpGatewayTokens, toolProfiles } from "@paperclipai/db"; +import { createToolGatewayService } from "../services/tool-gateway.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerGatewayCredentialExpiryWork } from "../services/gateway-credential-expiry-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_GATEWAY_EXPIRY_TEST_DATABASE_URL"); +const supported = external ? true : (await getEmbeddedPostgresTestSupport()).supported; + +describe.skipIf(!supported)("gateway credential expiry", () => { + let db: ReturnType; + let temporary: Awaited> | undefined; + let gateway: ReturnType; + let coordinator: ReturnType | undefined; + let companyId: string, agentId: string, runId: string, namedId: string; + let draining: boolean; + const errors: unknown[] = []; + + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("gateway-expiry-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 90_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + companyId = randomUUID(); agentId = randomUUID(); runId = randomUUID(); draining = false; errors.length = 0; + await db.insert(companies).values({ id: companyId, name: "Credential expiry", issuePrefix: "EXP" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Agent", role: "engineer", adapterType: "process" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "assignment", status: "running" }); + const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "expiry", name: "Expiry", defaultAction: "deny" }).returning(); + gateway = createToolGatewayService(db, { toolActionSigningSecret: "expiry-test-only" }); + namedId = (await gateway.createNamedGateway({ companyId, body: { name: "Expiry", profileId: profile.id } })).id; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await db.delete(activityLog); + await db.delete(toolAccessAuditEvents); + await db.delete(toolGatewaySessions); + await db.delete(toolMcpGatewayTokens); + await db.delete(toolMcpGateways); + await db.delete(toolProfiles); + await db.delete(heartbeatRuns); + await db.delete(agents); + await db.delete(companies); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerGatewayCredentialExpiryWork(coordinator, gateway, () => !draining).ready; + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(async () => { + // Let real SQL settle before advancing only to an actually armed follow-up. + // Repeated blind clock jumps can expire the future credential under test. + if (idleWorkSnapshot().active > 0) { + await db.execute(sql`select pg_sleep(0.01)`); + const next = coordinator!.nextWakeAt(); + if (next !== null && next <= Date.now() + 30_000) { + await vi.advanceTimersByTimeAsync(Math.max(0, next - Date.now())); + } + } + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + } + async function quiet() { + await settle(); expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select"), deletes = vi.spyOn(db, "delete"); + await settle(60 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled(); expect(deletes).not.toHaveBeenCalled(); + reads.mockRestore(); selects.mockRestore(); deletes.mockRestore(); + } + const token = (expiresAt?: Date) => gateway.createNamedGatewayToken({ companyId, gatewayId: namedId, body: { name: "Expiry", expiresAt } }); + const session = (service = gateway) => service.createSession({ companyId, agentId, runId, ttlMs: 5 * 60_000 }); + + it("stays quiet with configuration and a nonexpiring token, then expires real tokens and sessions by deadline", async () => { + await start(); await quiet(); + const retained = await token(); await quiet(); + const active = await session(); + const due = new Date(Date.now() + 3 * 60_000); + const expiring = await token(due); + await settle(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); + const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select"); + await settle(30_000); + expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled(); + reads.mockRestore(); selects.mockRestore(); + await gateway.revokeNamedGatewayToken({ companyId, tokenId: expiring.id }); + await settle(2 * 60_000); + expect((await db.select().from(toolMcpGatewayTokens)).map(row => row.id)).toEqual([retained.id]); + expect(coordinator!.nextWakeAt()).toBe(active.expiresAt.getTime()); + await settle(2 * 60_000); + expect(await db.select().from(toolGatewaySessions)).toEqual([]); + await quiet(); + }); + + it.each([ + { kind: "session", rollback: false }, { kind: "session", rollback: true }, + { kind: "token", rollback: false }, { kind: "token", rollback: true }, + ])("respects $kind outer transaction settlement (rollback $rollback)", async ({ kind, rollback }) => { + await start(); await quiet(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const staged = new Promise(resolve => { written = resolve; }); + const rollbackError = new Error("rollback fixture"); + const writing = db.transaction(async tx => { + const scoped = createToolGatewayService(tx as unknown as typeof db, { toolActionSigningSecret: "expiry-test-only" }); + if (kind === "session") await session(scoped); + else await scoped.createNamedGatewayToken({ companyId, gatewayId: namedId, + body: { name: "Outer transaction", expiresAt: new Date(Date.now() + 5 * 60_000) } }); + written(); await gate; if (rollback) throw rollbackError; + }).catch(error => { if (error !== rollbackError) throw error; }); + await Promise.race([staged, writing]); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(await db.select().from(toolGatewaySessions)).toEqual([]); + expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]); + } finally { release(); } + await writing; await settle(); + if (rollback) await quiet(); + else { + expect(coordinator!.nextWakeAt()).toBe(await gateway.nextCredentialExpiryAt()); + await settle(5 * 60_000); await quiet(); + } + }); + + it("coalesces sequential credential writes within the former cleanup cadence", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions"); + const next = vi.spyOn(gateway, "nextCredentialExpiryAt"); + await start(); + const firstPassAt = Date.now(); + for (let i = 0; i < 10; i++) { + if (i % 2 === 0) await session(); + else await token(new Date(Date.now() + 5 * 60_000)); + await vi.advanceTimersByTimeAsync(1); + // Let each old scan finish so this measures sequential writes, not overlapping calls. + await db.execute(sql`select pg_sleep(0.01)`); + } + expect(cleanup).toHaveBeenCalledTimes(1); + expect(next).toHaveBeenCalledTimes(1); + expect(coordinator!.nextWakeAt()).toBe(firstPassAt + 30_000); + await vi.advanceTimersByTimeAsync(30_000); + await vi.waitFor(() => expect(next).toHaveBeenCalledTimes(2)); + await db.execute(sql`select pg_sleep(0.01)`); + expect(cleanup).toHaveBeenCalledTimes(2); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now()); + await settle(5 * 60_000); await quiet(); + }); + + it("keeps a loaded expiry deadline ahead of credential notification coalescing", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const deadline = Date.now() + 500; + await token(new Date(deadline)); + await start(); + expect(coordinator!.nextWakeAt()).toBe(deadline); + await session(); + await vi.advanceTimersByTimeAsync(1); + await db.execute(sql`select pg_sleep(0.01)`); + expect(coordinator!.nextWakeAt()).toBe(deadline); + await vi.advanceTimersByTimeAsync(499); + await vi.waitFor(async () => expect(await db.select().from(toolMcpGatewayTokens)).toEqual([])); + expect(await db.select().from(toolGatewaySessions)).toHaveLength(1); + await settle(5 * 60_000); await quiet(); + }); + + it("rejects a newly expired credential while physical cleanup is coalesced", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions"); + await start(); + const at = Date.now(); + const active = await gateway.createSession({ companyId, agentId, runId, ttlMs: 1000 }); + await vi.advanceTimersByTimeAsync(1000); + await db.execute(sql`select pg_sleep(0.01)`); + expect(cleanup).toHaveBeenCalledTimes(1); + expect(await db.select().from(toolGatewaySessions)).toHaveLength(1); + await expect(gateway.listToolsForSession(active.token)).rejects.toMatchObject({ status: 401, reasonCode: "session_expired" }); + expect(coordinator!.nextWakeAt()).toBe(at + 30_000); + await vi.advanceTimersByTimeAsync(29_000); + await vi.waitFor(async () => expect(await db.select().from(toolGatewaySessions)).toEqual([])); + await quiet(); + }); + + it.each(["sessions", "tokens", "both"] as const)("drains the full %s backlog without a registered scheduler", async kind => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + const now = new Date(); + const futureSession = await session(); + const futureToken = await token(new Date(now.getTime() + 60_000)); + const nonexpiringToken = await token(); + if (kind !== "tokens") await db.insert(toolGatewaySessions).values(Array.from({ length: 1001 }, () => ({ + companyId, agentId, runId, tokenHash: randomUUID(), expiresAt: new Date(now.getTime() - 1), + }))); + if (kind !== "sessions") await db.insert(toolMcpGatewayTokens).values(Array.from({ length: 1001 }, () => ({ + companyId, gatewayId: namedId, name: "Expired", tokenHash: randomUUID(), expiresAt: new Date(now.getTime() - 1), + }))); + await expect(gateway.cleanupExpiredSessions({ now })).resolves.toEqual({ deletedCount: kind === "tokens" ? 0 : 1001 }); + expect((await db.select().from(toolGatewaySessions)).map(row => row.id)).toEqual([futureSession.id]); + expect((await db.select().from(toolMcpGatewayTokens)).map(row => row.id).sort()).toEqual([futureToken.id, nonexpiringToken.id].sort()); + expect(coordinator).toBeUndefined(); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it("bounds session cleanup and schedules the remaining overdue batch", async () => { + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: false }); + await db.insert(toolGatewaySessions).values(Array.from({ length: 501 }, () => ({ + companyId, agentId, runId, tokenHash: randomUUID(), expiresAt: new Date(Date.now() - 1), + }))); + const future = await session(); + const cleanup = vi.spyOn(gateway, "cleanupExpiredSessions"); + await start(); + expect(cleanup).toHaveResolvedWith({ deletedCount: 500 }); + expect(await db.select().from(toolGatewaySessions)).toHaveLength(2); + expect(coordinator!.nextWakeAt()).toBe(Date.now() + 30_000); + await vi.advanceTimersByTimeAsync(29_999); + expect(cleanup).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1); + await vi.waitFor(async () => expect((await db.select().from(toolGatewaySessions)).map(row => row.id)).toEqual([future.id])); + await settle(5 * 60_000); await quiet(); + }); + + it("reconstructs saved token expiry on startup", async () => { + const due = new Date(Date.now() + 5 * 60_000); + await token(due); + await start(); await settle(); + expect(coordinator!.nextWakeAt()).toBe(due.getTime()); + await settle(5 * 60_000); + expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]); + await quiet(); + }); + + it("drains an overdue persisted backlog across bounded batches", async () => { + await db.insert(toolMcpGatewayTokens).values(Array.from({ length: 501 }, (_, index) => ({ + companyId, gatewayId: namedId, name: `Expired ${index}`, tokenHash: randomUUID(), + tokenPrefix: "expired", expiresAt: new Date(Date.now() - 1), + }))); + await start(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await quiet(); + expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]); + }); + + it("retains cleanup through drain without permitting expired authentication", async () => { + await start(); await quiet(); draining = true; + const active = await session(); + const expiredToken = await token(new Date(Date.now() + 3 * 60_000)); + await settle(); + const reads = vi.spyOn(db, "execute"), selects = vi.spyOn(db, "select"); + await settle(10 * 60_000); + expect(reads).not.toHaveBeenCalled(); expect(selects).not.toHaveBeenCalled(); + reads.mockRestore(); selects.mockRestore(); + expect(await db.select().from(toolGatewaySessions)).toHaveLength(1); + await expect(gateway.listToolsForSession(active.token)).rejects.toMatchObject({ status: 401, reasonCode: "session_expired" }); + await expect(gateway.initializeNamedGatewayProtocol({ gatewayId: namedId, bearerToken: expiredToken.token })) + .rejects.toMatchObject({ status: 401 }); + draining = false; await settle(); + expect(await db.select().from(toolGatewaySessions)).toEqual([]); + expect(await db.select().from(toolMcpGatewayTokens)).toEqual([]); + await quiet(); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index f3d7171c03a..9788b8599ff 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -71,6 +71,7 @@ const { toolGateway: { sweepActionReviews: vi.fn(async () => ({ scanned: 0 })), cleanupExpiredSessions: vi.fn(async () => ({ deletedCount: 0 })), + nextCredentialExpiryAt: vi.fn(async () => null), }, toolActionDeliveries: { hasPending: vi.fn(async () => false), sweepPending: vi.fn(async () => ({ scanned: 0, delivered: 0 })) }, }, diff --git a/server/src/__tests__/tool-gateway.test.ts b/server/src/__tests__/tool-gateway.test.ts index a1f0b553d17..4e9ba662bdf 100644 --- a/server/src/__tests__/tool-gateway.test.ts +++ b/server/src/__tests__/tool-gateway.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createHash, randomUUID } from "node:crypto"; import { createServer, type IncomingMessage } from "node:http"; import express from "express"; @@ -65,8 +66,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const externalTestDatabaseUrl = process.env.PAPERCLIP_TOOL_GATEWAY_TEST_DATABASE_URL; -const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const externalTestDatabase = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_GATEWAY_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabase ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const testToolActionSigningSecret = "test-tool-action-signing-secret"; @@ -554,8 +555,9 @@ describeEmbeddedPostgres("tool gateway acceptance", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-"); - db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); + if (externalTestDatabase) { db = createDb(externalTestDatabase); return; } + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-"); + db = createDb(tempDb.connectionString); }, 90_000); afterEach(async () => { diff --git a/server/src/index.ts b/server/src/index.ts index ca51efe4345..4116e46bd93 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerGatewayCredentialExpiryWork } from "./services/gateway-credential-expiry-work.js"; import { registerSecretProposalExpiryWork } from "./services/secret-proposal-expiry-work.js"; import { registerCustomImageCleanupWork } from "./services/custom-image-cleanup-work.js"; import { registerWorkspaceRecoveryWork } from "./services/workspace-recovery-work.js"; @@ -1464,7 +1465,10 @@ async function startServerWithDatabaseTeardown( }; await timeStartupRecovery("tool_reviews", () => app.locals.toolGateway.sweepActionReviews()).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); - await timeStartupRecovery("gateway_sessions", () => app.locals.toolGateway.cleanupExpiredSessions()).catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed")); + await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled + ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready + : app.locals.toolGateway.cleanupExpiredSessions()).catch((err: unknown) => logger.error({ err }, "startup gateway token cleanup failed")); await timeStartupRecovery("delivery_workers", () => Promise.all([budgetEnforcementWorker, accountingWorker, fastResponseWorker, ...deliveryWorkers].map(worker => worker.ready))); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); @@ -1891,7 +1895,6 @@ async function startServerWithDatabaseTeardown( })); trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews().catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); - trackHeartbeatSchedulerWork(app.locals.toolGateway.cleanupExpiredSessions().catch((err: unknown) => logger.error({ err }, "gateway token cleanup failed"))); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { diff --git a/server/src/services/deadline-work.ts b/server/src/services/deadline-work.ts index 9f1de1b2a24..044c3cbfce4 100644 --- a/server/src/services/deadline-work.ts +++ b/server/src/services/deadline-work.ts @@ -9,11 +9,14 @@ export function registerDeadlineWork( next: () => Promise; canRun: () => boolean; retryMs: number; + // Optional batching latency for notifications; already-known deadlines still preempt it. + notificationCoalesceMs?: number; }, ) { let requested = 1, reconciled = 0; let next: number | null = null; let earliest = 0; + let coalesceUntil = 0; return coordinator.register(queue, { retryMs: task.retryMs, pausePassiveMaintenance: true, @@ -21,6 +24,10 @@ export function registerDeadlineWork( run: async () => { if (!task.canRun()) { requested++; return; } if (Date.now() < earliest) { next = earliest; return; } + if (Date.now() < coalesceUntil && (next === null || next > Date.now())) { + next = next === null ? coalesceUntil : Math.min(next, coalesceUntil); + return; + } const generation = ++requested; earliest = Date.now() + task.retryMs; await task.run(); @@ -28,6 +35,7 @@ export function registerDeadlineWork( // Back off failed or still-overdue work, not a successfully read future deadline. if (deadline === null || deadline > Date.now()) earliest = 0; next = deadline === null ? null : Math.max(earliest, deadline); + coalesceUntil = Date.now() + (task.notificationCoalesceMs ?? 0); reconciled = generation; }, hasPending: async () => requested !== reconciled, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index b6bed1d0f25..191b8aef36b 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + gatewayCredentialExpiry: "gateway-credential-expiry", secretProposalExpiry: "secret-proposal-expiry", customImageExpiry: "custom-image-expiry", historicalCleanup: "historical-cleanup", diff --git a/server/src/services/gateway-credential-expiry-work.ts b/server/src/services/gateway-credential-expiry-work.ts new file mode 100644 index 00000000000..2cf6097b6d8 --- /dev/null +++ b/server/src/services/gateway-credential-expiry-work.ts @@ -0,0 +1,18 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { createToolGatewayService } from "./tool-gateway.js"; + +export function registerGatewayCredentialExpiryWork( + coordinator: ReturnType, + gateway: Pick, "cleanupExpiredSessions" | "nextCredentialExpiryAt">, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.gatewayCredentialExpiry, { + retryMs: 30_000, + notificationCoalesceMs: 30_000, + canRun, + run: () => gateway.cleanupExpiredSessions({ singleBatch: true }), + next: () => gateway.nextCredentialExpiryAt(), + }); +} diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index d2d703dfcd6..c75413e8529 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -9210,45 +9210,48 @@ export function createToolGatewayService( const token = generateNamedGatewayToken(tokenId); const tokenPrefix = `pcgw_${tokenId.slice(0, 8)}`; const now = new Date(); - const [row] = await db - .insert(toolMcpGatewayTokens) - .values({ - id: tokenId, - companyId: input.companyId, - gatewayId: input.gatewayId, - name: input.body.name, - tokenHash: hashGatewayToken(token), - tokenPrefix, - subjectType: input.body.subjectType ?? "gateway_client", - subjectId: input.body.subjectId ?? null, - clientLabel: input.body.clientLabel, - ownerNote: input.body.ownerNote, - allowedActions: input.body.allowedActions ?? [ - "tools/list", - "tools/call", - "resources/list", - "resources/read", - "prompts/list", - "prompts/get", - ], - expiresAt: input.body.expiresAt ?? null, - expiryOverrideReason: input.body.expiryOverrideReason ?? null, - expiryOverrideByAgentId: - input.actor?.agentId && input.body.expiryOverrideReason - ? input.actor.agentId - : null, - expiryOverrideByUserId: - input.actor?.userId && input.body.expiryOverrideReason - ? input.actor.userId - : null, - expiryOverrideAt: input.body.expiryOverrideReason ? now : null, - responsibleUserId: input.managementContext?.actor.type === "agent" ? input.managementContext.responsibleUserId : null, - createdByAgentId: input.actor?.agentId ?? null, - createdByUserId: input.actor?.userId ?? null, - createdAt: now, - updatedAt: now, - }) - .returning(); + const [row] = await db.transaction(async tx => { + if (input.body.expiresAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.gatewayCredentialExpiry); + return tx + .insert(toolMcpGatewayTokens) + .values({ + id: tokenId, + companyId: input.companyId, + gatewayId: input.gatewayId, + name: input.body.name, + tokenHash: hashGatewayToken(token), + tokenPrefix, + subjectType: input.body.subjectType ?? "gateway_client", + subjectId: input.body.subjectId ?? null, + clientLabel: input.body.clientLabel, + ownerNote: input.body.ownerNote, + allowedActions: input.body.allowedActions ?? [ + "tools/list", + "tools/call", + "resources/list", + "resources/read", + "prompts/list", + "prompts/get", + ], + expiresAt: input.body.expiresAt ?? null, + expiryOverrideReason: input.body.expiryOverrideReason ?? null, + expiryOverrideByAgentId: + input.actor?.agentId && input.body.expiryOverrideReason + ? input.actor.agentId + : null, + expiryOverrideByUserId: + input.actor?.userId && input.body.expiryOverrideReason + ? input.actor.userId + : null, + expiryOverrideAt: input.body.expiryOverrideReason ? now : null, + responsibleUserId: input.managementContext?.actor.type === "agent" ? input.managementContext.responsibleUserId : null, + createdByAgentId: input.actor?.agentId ?? null, + createdByUserId: input.actor?.userId ?? null, + createdAt: now, + updatedAt: now, + }) + .returning(); + }); if (input.managementContext) await logActivity(db, { companyId: input.companyId, ...input.managementContext.audit, action: "tool_gateway.token_created", entityType: "tool_mcp_gateway", entityId: input.gatewayId, details: { tokenId: row.id, subjectType: row.subjectType, allowedActions: row.allowedActions } }); @@ -9373,18 +9376,21 @@ export function createToolGatewayService( expiresAt: new Date(now.getTime() + sessionTtlMs(input.ttlMs)), }; - await db.insert(toolGatewaySessions).values({ - id: session.id, - companyId: session.companyId, - agentId: input.agentId, - runId: input.runId, - issueId: session.issueId, - projectId: session.projectId, - tokenHash: hashGatewayToken(token), - expiresAt: session.expiresAt, - createdAt: session.createdAt, - updatedAt: session.createdAt, - } as any); + await db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.gatewayCredentialExpiry); + await tx.insert(toolGatewaySessions).values({ + id: session.id, + companyId: session.companyId, + agentId: input.agentId, + runId: input.runId, + issueId: session.issueId, + projectId: session.projectId, + tokenHash: hashGatewayToken(token), + expiresAt: session.expiresAt, + createdAt: session.createdAt, + updatedAt: session.createdAt, + } as any); + }); await writeAudit({ session, @@ -11860,23 +11866,50 @@ export function createToolGatewayService( return { ...sessionView, revokedAt: session!.revokedAt ?? now }; }, - async cleanupExpiredSessions(input: { now?: Date } = {}) { + async nextCredentialExpiryAt(): Promise { + const [row] = await db.execute<{ at: string | Date | null }>(sql` + select min(expires_at) as at from ( + (select ${toolMcpGatewayTokens.expiresAt} as expires_at from ${toolMcpGatewayTokens} + where ${toolMcpGatewayTokens.expiresAt} is not null + order by ${toolMcpGatewayTokens.expiresAt} limit 1) + union all + (select ${toolGatewaySessions.expiresAt} as expires_at from ${toolGatewaySessions} + order by ${toolGatewaySessions.expiresAt} limit 1) + ) expiries + `); + return row?.at == null ? null : new Date(row.at).getTime(); + }, + + async cleanupExpiredSessions(input: { now?: Date; singleBatch?: boolean } = {}) { const now = input.now ?? new Date(); - const expiredTokens = await db.select({ id: toolMcpGatewayTokens.id }) - .from(toolMcpGatewayTokens) - .where(lte(toolMcpGatewayTokens.expiresAt, now)) - .orderBy(asc(toolMcpGatewayTokens.expiresAt), asc(toolMcpGatewayTokens.id)) - .limit(500); - if (expiredTokens.length > 0) { - await db.delete(toolMcpGatewayTokens) + let deletedCount = 0; + while (true) { + const expiredTokens = await db.select({ id: toolMcpGatewayTokens.id }) + .from(toolMcpGatewayTokens) + .where(lte(toolMcpGatewayTokens.expiresAt, now)) + .orderBy(asc(toolMcpGatewayTokens.expiresAt), asc(toolMcpGatewayTokens.id)) + .limit(500); + if (expiredTokens.length > 0) { + await db.delete(toolMcpGatewayTokens) .where(and(inArray(toolMcpGatewayTokens.id, expiredTokens.map((token) => token.id)), lte(toolMcpGatewayTokens.expiresAt, now))) .returning({ id: toolMcpGatewayTokens.id }); + } + const expiredSessions = await db.select({ id: toolGatewaySessions.id }) + .from(toolGatewaySessions) + .where(lte(toolGatewaySessions.expiresAt, now)) + .orderBy(asc(toolGatewaySessions.expiresAt), asc(toolGatewaySessions.id)) + .limit(500); + if (expiredSessions.length > 0) { + const rows = await db.delete(toolGatewaySessions) + .where(and(inArray(toolGatewaySessions.id, expiredSessions.map((session) => session.id)), lte(toolGatewaySessions.expiresAt, now))) + .returning({ id: toolGatewaySessions.id }); + deletedCount += rows.length; + } + // The worker schedules the next page; one-shot startup/manual callers + // must finish the backlog without relying on another scheduled pass. + if (input.singleBatch || (expiredTokens.length < 500 && expiredSessions.length < 500)) break; } - const rows = await db - .delete(toolGatewaySessions) - .where(lte(toolGatewaySessions.expiresAt, now)) - .returning({ id: toolGatewaySessions.id }); - return { deletedCount: rows.length }; + return { deletedCount }; }, async listRuntimeSlots(companyId?: string) {