From d6b27a4b485abc780b4d13172faf28ecb8df9eb2 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 23:50:36 -0500 Subject: [PATCH 1/2] fix(server): bound plugin refreshes and finish teardown after errors Co-Authored-By: Paperclip --- doc/idle-plugin-jobs.md | 33 +++++++++- ...2026-10-24-recurring-server-sleep-stack.md | 13 ++++ .../__tests__/plugin-scheduled-work.test.ts | 61 ++++++++++++++++--- server/src/app.ts | 8 +-- server/src/services/plugin-job-scheduler.ts | 2 +- server/src/services/plugin-loader.ts | 7 +-- server/src/shutdown.test.ts | 24 ++++++++ server/src/shutdown.ts | 15 +++++ 8 files changed, 143 insertions(+), 20 deletions(-) diff --git a/doc/idle-plugin-jobs.md b/doc/idle-plugin-jobs.md index 71f1f163d97..3fb6575c54d 100644 --- a/doc/idle-plugin-jobs.md +++ b/doc/idle-plugin-jobs.md @@ -37,7 +37,7 @@ including a worker becoming available without a database mutation. queued, pending, or running records. Manual and scheduled dispatch share this boundary. Worker RPC calls run outside transactions. - Active calls consume concurrency slots and remain tracked until settlement. - Their completion wakes waiting work without an interval. Slow calls do not + Their completion requests a coalesced refresh for waiting work. Slow calls do not prevent other available slots from being used. - Completion and the scheduled job's next timestamp commit together. Failed persistence retries the retained settlement, not the plugin side effect. @@ -48,7 +48,16 @@ including a worker becoming available without a database mutation. admitted work, then flushes retained completion results without another RPC. If persistence still fails, stop rejects and retains the result for another attempt; it does not report successful shutdown. Existing calls retain their - RPC timeout. Application teardown must complete before the pool is ended. + RPC timeout. A failed stop still allows application and plugin-worker cleanup; + teardown then retries the retained settlement once before reporting any remaining + error, while the database pool is open. + +Producer and lifecycle wakes share a 30-second minimum refresh spacing, matching +the former scan budget. Sequential run-status writes coalesce instead of each +causing a full inventory scan. This is a bound on requested refreshes, not a +recurring timer: settled workers without a future obligation remain disarmed. +Manual dispatch still uses its direct claim path. New schedules and completion- +released capacity can wait up to that spacing before scheduled dispatch. ## Limits @@ -98,3 +107,23 @@ external effect that the database has never acknowledged. The final focused run passes 62 cases across plugin scheduling, loader recovery, warm standby, and the delivery coordinator. Server typechecking passes. Remote checks and a fresh review remain required for the repaired head. + + +A second review reproduced 16 full inventory reads during fifteen sequential run +lifecycles on forty manual jobs. Restoring the prior 30-second spacing bounds that +burst to one read, then reconciliation returns to zero SQL. Shutdown regressions +also reproduced skipped cleanup after a settlement error. The shared teardown +boundary now always attempts remaining cleanup and retries a failed stop once; +both application teardown and plugin-loader shutdown use it. A PostgreSQL trigger +case verifies the retained result is saved after cleanup removes the write fault, +without a second worker RPC. Persistent failures remain visible. + +The first expanded timing run failed fifteen assertions because the test assumed +one second was sufficient to settle a coalesced wake. Tests now wait for the actual +null/future deadline before measuring three minutes of SQL silence. The corrected +run passed all 75 cases; the additional post-cleanup retry case is verified separately +below. No production timeout or quiet assertion was relaxed. + +The post-cleanup retry run passes all 48 shutdown/plugin cases (28 plugin and +20 shutdown). Together with the unchanged consumer selection this covers 76 +unique cases. Server typecheck and build pass on the final round-two repair. 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 3a446cfdd4a..c0633d791f7 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -976,3 +976,16 @@ updates take the job lock. All 62 focused lifecycle/consumer cases and server typechecking pass. #17510 at 400f4c0fc8 cleared the security gate and previous CI failures, but environment-cleanup pagination has a 32-versus-33 assertion race in server shard 1. Diagnose and repair it before accepting the stack. + + +Plugin round-two review repair restores a 30-second minimum between requested +full refreshes without adding idle polling. A burst regression failed at 16 +reads before repair and passes at at most one afterward. Failed plugin stop +no longer skips application/worker teardown; retained settlement is retried once +after cleanup while the pool is open, with persistent errors still reported. +The final 48 shutdown/plugin cases and earlier 75-case consumer selection pass +(76 unique cases), with server types/build passing. Fresh remote gates remain. +Approval #17510 has passing latest-head CI and Greptile 5/5 at 0c7109c103, but +Paperclip review identified persistent-failure backoff and duplicated PR matching +that still require repair. Tool health remains the next new migration after +these owning-PR repairs. diff --git a/server/src/__tests__/plugin-scheduled-work.test.ts b/server/src/__tests__/plugin-scheduled-work.test.ts index d61f2160eb5..48fb5c41b84 100644 --- a/server/src/__tests__/plugin-scheduled-work.test.ts +++ b/server/src/__tests__/plugin-scheduled-work.test.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "../shutdown.js"; import { EventEmitter } from "node:events"; import { randomUUID } from "node:crypto"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -61,7 +62,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu }, { timeout: 5000 }); } async function quiet() { - await settled(); expect(coordinator!.nextWakeAt()).toBeNull(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 5000 }); const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); await vi.advanceTimersByTimeAsync(180_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); @@ -80,6 +85,23 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } it("does not query repeatedly after an empty startup reconciliation", async () => { await start(); await quiet(); }); + it("bounds full refreshes while many job status writes arrive sequentially", async () => { + await pluginJobStore(db).syncJobDeclarations(pluginId, Array.from({ length: 40 }, (_, i) => ({ jobKey: `manual-${i}`, schedule: "" }))); + const jobs = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId)); + await start(); await quiet(); + const reads = vi.spyOn(db, "select"); + for (const job of jobs.slice(0, 15)) { + const run = await pluginJobStore(db).createRun({ jobId: job.id, pluginId, trigger: "manual" }); + await pluginJobStore(db).markRunning(run.id); + await pluginJobStore(db).completeRun(run.id, { status: "succeeded" }); + await vi.advanceTimersByTimeAsync(1100); + } + expect(reads.mock.calls.length).toBeLessThanOrEqual(1); + reads.mockRestore(); + await quiet(); + expect(await runs()).toHaveLength(15); + expect(rpc).not.toHaveBeenCalled(); + }); it("keeps configured manual-only and invalid schedules quiet", async () => { await job("manual", ""); await job("invalid", "not cron"); await start(); await quiet(); expect(rpc).not.toHaveBeenCalled(); }); @@ -147,7 +169,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); expect(rpc).toHaveBeenCalledTimes(1); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); expect(rpc).toHaveBeenCalledTimes(1); await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0; await completed(); expect(rpc).toHaveBeenCalledTimes(1); }); @@ -155,7 +177,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); draining = true; await coordinator!.stop(); await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); errors.length = 0; @@ -163,11 +185,28 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect((await runs())[0].status).toBe("succeeded"); expect(rpc).toHaveBeenCalledTimes(1); }); + it("finishes service cleanup and retries a failed settlement before closing the pool", async () => { + await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); + await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); + await due(); await start(); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); + draining = true; await coordinator!.stop(); errors.length = 0; + const cleanup = vi.fn(async () => { + expect((await runs())[0].status).toBe("running"); + await db.execute(sql`drop trigger plugin_settlement_failure on plugin_job_runs`); + }); + const stop = vi.fn(() => scheduler!.stop()); + await finishServiceShutdown(stop, cleanup); + expect(cleanup).toHaveBeenCalledOnce(); + expect(stop).toHaveBeenCalledTimes(2); + expect((await runs())[0].status).toBe("succeeded"); + expect(rpc).toHaveBeenCalledTimes(1); + }); it("reports shutdown persistence failure and retains the result for another stop attempt", async () => { await db.execute(sql`create function plugin_settlement_failure() returns trigger language plpgsql as $$ begin if new.status in ('succeeded', 'failed') then raise exception 'settlement unavailable'; end if; return new; end $$`); await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure()`); await due(); await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); draining = true; await coordinator!.stop(); errors.length = 0; try { await expect(scheduler!.stop()).rejects.toThrow("Plugin job completion persistence failed during shutdown"); @@ -183,7 +222,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create trigger plugin_settlement_failure before update on plugin_job_runs for each row execute function plugin_settlement_failure(${sql.raw("'" + blocked.id + "'")})`); try { await start(); - await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThan(0); }, { timeout: 5000 }); const independent = await due("independent"); await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); @@ -244,7 +283,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu it("does no SQL while waiting for a future configured deadline", async () => { const row = await job(); const at = new Date(Date.now() + 600_000); await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at); - await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + }, { timeout: 5000 }); const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); await vi.advanceTimersByTimeAsync(180_000); expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); @@ -254,7 +297,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu const row = await job(); const at = new Date(Date.now() + 600_000); await pluginJobStore(db).updateRunTimestamps(row.id, new Date(0), at); await start(); await settled(); await scheduler!.stop(); await coordinator!.stop(); - await start(); await settled(); expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + await start(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(coordinator!.nextWakeAt()).toBe(at.getTime()); + }, { timeout: 5000 }); await vi.advanceTimersByTimeAsync(at.getTime() - Date.now()); await completed(); expect(rpc).toHaveBeenCalledTimes(1); }); it("rechecks a paused definition after waiting for its lock", async () => { diff --git a/server/src/app.ts b/server/src/app.ts index d3dfddd0a9f..2205bd47203 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "./shutdown.js"; import { promptInjectionPreventionRoutes } from "./routes/prompt-injection-prevention.js"; import { voiceSessionRoutes, voiceWebhookRoutes } from "./routes/voice-sessions.js"; import { eq } from "drizzle-orm"; @@ -1455,10 +1456,7 @@ export async function createApp( let appServicesShutdown: Promise | null = null; const shutdownAppServices = (): Promise => { if (appServicesShutdown) return appServicesShutdown; - appServicesShutdown = (async () => { - // The scheduler tick queries the database. Stop it here, inside the - // awaited teardown, so no tick runs after the caller ends the pool. - await scheduler.stop(); + appServicesShutdown = finishServiceShutdown(() => scheduler.stop(), async () => { await agentLifecycle.stop(); await publicMcpEvents?.stop(); await dotMcpEvents?.stop(); @@ -1493,7 +1491,7 @@ export async function createApp( // caller stops the database and the provider. A lease release that // fails stays a durable record for the startup reaper. await setupTokenLoginService?.shutdown(); - })(); + }); return appServicesShutdown; }; app.locals.paperclipShutdown = shutdownAppServices; diff --git a/server/src/services/plugin-job-scheduler.ts b/server/src/services/plugin-job-scheduler.ts index d9a4c3c8e0b..827600f019b 100644 --- a/server/src/services/plugin-job-scheduler.ts +++ b/server/src/services/plugin-job-scheduler.ts @@ -45,7 +45,7 @@ function nextCron(schedule: string, now: Date): Date | null { } export function createPluginJobScheduler(options: PluginJobSchedulerOptions): PluginJobScheduler { - const { db, jobStore, workerManager, tickIntervalMs = 1_000, + const { db, jobStore, workerManager, tickIntervalMs = 30_000, jobTimeoutMs = 5 * 60_000, maxConcurrentJobs = 10 } = options; const log = logger.child({ service: "plugin-job-scheduler" }); const activeJobs = new Set(); diff --git a/server/src/services/plugin-loader.ts b/server/src/services/plugin-loader.ts index 283cbc6c401..d00f683a84b 100644 --- a/server/src/services/plugin-loader.ts +++ b/server/src/services/plugin-loader.ts @@ -1,3 +1,4 @@ +import { finishServiceShutdown } from "../shutdown.js"; /** * PluginLoader — discovery, installation, and runtime activation of plugins. * @@ -2143,11 +2144,7 @@ export function pluginLoader( const { workerManager, jobScheduler } = runtimeServices; - // 1. Stop the job scheduler tick loop - await jobScheduler.stop(); - - // 2. Stop all worker processes - await workerManager.stopAll(); + await finishServiceShutdown(() => jobScheduler.stop(), () => workerManager.stopAll()); log.info("plugin-loader: all plugins shut down"); }, diff --git a/server/src/shutdown.test.ts b/server/src/shutdown.test.ts index c9d118a8095..754a99fb1e1 100644 --- a/server/src/shutdown.test.ts +++ b/server/src/shutdown.test.ts @@ -5,6 +5,7 @@ import { coordinateHeartbeatSchedulerShutdown, drainRunExecutionFinalizersForShutdown, finalizeServerShutdown, + finishServiceShutdown, loadWithoutCoordinatedShutdownSignalHooks, } from "./shutdown.js"; @@ -523,3 +524,26 @@ describe("coordinateHeartbeatSchedulerShutdown", () => { }); }); }); + +describe("service shutdown failure isolation", () => { + it("awaits remaining cleanup before reporting a scheduler failure", async () => { + const failure = new Error("plugin result could not be saved"); + const gate = deferred(); + const cleanup = vi.fn(() => gate.promise); + let finished = false; + const stopping = finishServiceShutdown(async () => { throw failure; }, cleanup) + .then(() => { finished = true; return undefined; }, error => { finished = true; return error; }); + try { + await vi.waitFor(() => expect(cleanup).toHaveBeenCalledOnce()); + expect(finished).toBe(false); + } finally { gate.resolve(); } + expect(await stopping).toBe(failure); + }); + it("reports both failures after attempting cleanup", async () => { + const stopFailure = new Error("scheduler failure"), cleanupFailure = new Error("worker failure"); + const cleanup = vi.fn(async () => { throw cleanupFailure; }); + await expect(finishServiceShutdown(async () => { throw stopFailure; }, cleanup)) + .rejects.toMatchObject({ errors: [stopFailure, cleanupFailure] }); + expect(cleanup).toHaveBeenCalledOnce(); + }); +}); diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index cda7499392b..ea927cbd6e1 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -272,3 +272,18 @@ export async function coordinateHeartbeatSchedulerShutdown< waitedForSchedulerIdle: true, }; } + +export async function finishServiceShutdown( + stop: () => Promise, + cleanup: () => Promise, +): Promise { + const failures: unknown[] = []; + let retryStop = false; + try { await stop(); } catch { retryStop = true; } + try { await cleanup(); } catch (error) { failures.push(error); } + if (retryStop) { + try { await stop(); } catch (error) { failures.unshift(error); } + } + if (failures.length === 1) throw failures[0]; + if (failures.length > 1) throw new AggregateError(failures, "Service shutdown failed"); +} From 0d3100665d0d5ddd93a8755b32e0484edb9fc0bf Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 24 Oct 2026 23:55:39 -0500 Subject: [PATCH 2/2] fix(server): back off failed approval queue projections Co-Authored-By: Paperclip --- doc/idle-approval-queue-seeding.md | 25 +++++- ...2026-10-24-recurring-server-sleep-stack.md | 11 +++ .../approval-queue-seeding-work.test.ts | 63 +++++++++++++ .../server-startup-feedback-export.test.ts | 7 +- .../services/approval-queue-seeding-work.ts | 89 +++++++++++-------- .../src/services/decision-queue-seed-rules.ts | 6 ++ server/src/services/decision-queues.ts | 4 +- 7 files changed, 159 insertions(+), 46 deletions(-) create mode 100644 server/src/services/decision-queue-seed-rules.ts diff --git a/doc/idle-approval-queue-seeding.md b/doc/idle-approval-queue-seeding.md index 8e788fdc03a..f12452282cc 100644 --- a/doc/idle-approval-queue-seeding.md +++ b/doc/idle-approval-queue-seeding.md @@ -37,7 +37,8 @@ changes retain their explicit signals. The queue service owns seed matching, creation, membership uniqueness, and audit writes. The feed and worker use the same writer through a small normalized source -contract. A worker does not construct a presentation-layer AttentionItem. Queue +contract. The candidate lookup and writer share the pull-request product SQL +predicate, so the matching policy has one owner. A worker does not construct a presentation-layer AttentionItem. Queue reads keep their existing source authorization checks. A queue-rule update and seeding serialize on the existing queue row. Existing memberships are not pruned when an approval resolves. This preserves current behavior; pruning is a separate @@ -77,8 +78,11 @@ hide later candidates. Atomic existing membership uniqueness prevents duplicate rows and duplicate insertion audit events. After a sweep, another lookup decides whether actual projection debt remains. -Only that debt, a newer committed signal, or a failed attempt retains a one-second -retry. The generation check follows the final awaited lookup, so a commit arriving +Unprocessed pages and newer committed signals retain a one-second continuation. +Failed scopes back off exponentially from one second to one minute. New signals +do not reset a failing scope's backoff; other companies can proceed independently. +The next retry is registered with the shared scheduler, with no SQL during the +wait. A failed scope remains tracked until its owed memberships are reconciled. The generation check follows the final awaited lookup, so a commit arriving during an empty read cannot be erased. Once reconciled, there is no deadline and no recurring worker SQL. The shared coordinator preserves requested work during drain. Passive pending checks inspect memory only; startup reconstructs from rows. Shutdown aborts between rows @@ -143,3 +147,18 @@ inspections after only 32 had completed. The test advanced fake time before the previous asynchronous page had necessarily published its next deadline. It now waits for all 33 inspections, then retains the zero-further-work assertions. All 14 environment cleanup cases pass on real PostgreSQL; production code is unchanged. + + +Further review reproduced 120 failed transactions in two simulated minutes for a +persistently failing projection. Per-scope backoff now bounds that test to at most +eight attempts, even with new signals every second. Removing the fault recovers +the membership. A real coordinator/producer case proves thirty seconds without +SQL during backoff, another company's successful queue seeding during that wait, +and eventual recovery without losing the original work. All 25 worker cases pass. +The first consumer command omitted the external DB setting for queue routes and +skipped those six cases; the corrected command executes and passes all six. +The shared PR predicate is exercised by both those writer tests and worker tests. +Server typecheck and build pass; fresh remote review/checks remain required. + +The final startup, approval-retention, and attention consumer run passes all 93 +cases. With the 25 worker and six queue cases, 124 affected tests pass. 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 eae91729127..2906ef0bc64 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -946,3 +946,14 @@ Plugin scheduling is published as #17518 at 8b4f551dc2. Its Actions checks passe but review found retained completion writes missing from shutdown and a stale deadline update race. Repair those and preserve independent job progress when one completion write fails before treating the slice as ready. + + +Approval queue follow-up: persistent projection failure reproduced 120 failed +transactions in two simulated minutes. Per-scope exponential backoff now caps +retries at one minute without discarding debt or letting repeated signals reset +the budget. A coordinator regression proves SQL silence during backoff and real +producer progress for another company. The candidate and seed writer share the +PR-product predicate. All 124 affected worker/queue/startup/retention/attention +cases pass, along with server types/build. One initial consumer command skipped +six queue tests because its DB variable was missing; the corrected run executes +and passes them. Fresh remote checks and review remain required after publication. diff --git a/server/src/__tests__/approval-queue-seeding-work.test.ts b/server/src/__tests__/approval-queue-seeding-work.test.ts index f3961b2c9e1..dae46ff9c64 100644 --- a/server/src/__tests__/approval-queue-seeding-work.test.ts +++ b/server/src/__tests__/approval-queue-seeding-work.test.ts @@ -282,6 +282,68 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu draining = false; await seeded(approval.id); await quiet(); }); + it("backs off persistent projection failures without forgetting the owed membership", async () => { + const service = approvalQueueSeedingService(db); + await service.sweep(); + const approval = await pending(); + service.request(new Set([companyId])); + await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin raise exception 'projection unavailable'; end $$`); + await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail()`); + const transactions = vi.spyOn(db, "transaction"); + try { + for (let second = 0; second < 120; second++) { + service.request(new Set([companyId])); + await service.sweep().catch(() => undefined); + await vi.advanceTimersByTimeAsync(1000); + } + expect(transactions.mock.calls.length).toBeLessThanOrEqual(8); + expect(transactions.mock.calls.length).toBeGreaterThan(1); + expect(await memberships()).toEqual([]); + expect(await service.next()).not.toBeNull(); + } finally { + transactions.mockRestore(); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); + } + await vi.advanceTimersByTimeAsync(60_000); + await service.sweep(); + expect(await memberships()).toEqual([expect.objectContaining({ sourceId: approval.id })]); + expect(await service.next()).toBeNull(); + }); + + it("keeps backoff SQL-quiet while another company's committed work proceeds", async () => { + const service = approvalQueueSeedingService(db), sweep = service.sweep; + let running = false; + service.sweep = async signal => { running = true; try { await sweep(signal); } finally { running = false; } }; + await start(service); await quiet(); + await db.execute(sql`create function approval_queue_fail() returns trigger language plpgsql as $$ begin + if new.company_id::text = TG_ARGV[0] then raise exception 'projection unavailable'; end if; return new; end $$`); + await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + companyId + "'")})`); + const approval = await pending(); + for (let count = 1; count <= 7; count++) { + await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(errors.length).toBeGreaterThanOrEqual(count); }, { timeout: 5000 }); + await vi.waitFor(() => expect(running).toBe(false)); + } + const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "transaction"); + await vi.advanceTimersByTimeAsync(30_000); + expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); + reads.mockRestore(); writes.mockRestore(); + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Healthy tenant", issuePrefix: `H${otherId.slice(0, 6)}` }); + const [otherIssue] = await db.insert(issues).values({ companyId: otherId, title: "Other PR", status: "todo" }).returning(); + await workProductService(db).createForIssue(otherIssue.id, otherId, { type: "pull_request", provider: "github", title: "Other PR", status: "open" }); + const otherApproval = await approvalRecords(db).create(otherId, { type: "request_board_approval", payload: {} }); + await issueApprovalService(db).link(otherIssue.id, otherApproval.id); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await db.select().from(decisionQueueItems).where(eq(decisionQueueItems.companyId, otherId))) + .toEqual([expect.objectContaining({ sourceId: otherApproval.id })]); + }, { timeout: 5000 }); + expect(await memberships()).toEqual([]); + await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); errors.length = 0; + await vi.advanceTimersByTimeAsync(60_000); + await seeded(approval.id); await quiet(); + }); + it("drains multiple pages without duplicating memberships or audit events", async () => { const target = await issue(); await product(target.id); const rows = await db.insert(approvals).values(Array.from({ length: 101 }, () => ({ companyId, type: "request_board_approval", payload: {} }))).returning(); @@ -309,6 +371,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await db.execute(sql`create trigger approval_queue_fail before insert on decision_queue_items for each row execute function approval_queue_fail(${sql.raw("'" + last + "'")})`); const service = approvalQueueSeedingService(db); await expect(service.sweep()).rejects.toThrow("Approval queue seeding failed"); + await vi.advanceTimersByTimeAsync(1000); await service.sweep(); expect((await memberships()).map(row => row.sourceId)).toEqual([last]); await db.execute(sql`drop trigger approval_queue_fail on decision_queue_items`); await service.sweep(); expect(await memberships()).toHaveLength(101); expect(await service.next()).toBeNull(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index e95f4ca9289..5a1bcf2ca6c 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -331,7 +331,7 @@ vi.mock("../services/join-request-retention.js", async importOriginal => ({ vi.mock("../services/approval-queue-seeding-work.js", async importOriginal => ({ ...await importOriginal(), - approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(async () => null) })), + approvalQueueSeedingService: vi.fn(() => ({ sweep: vi.fn(async () => undefined), next: vi.fn(() => null), request: vi.fn() })), })); vi.mock("../services/approval-retention.js", async importOriginal => ({ @@ -545,7 +545,10 @@ describe("startServer feedback export wiring", () => { if (enabled) { const service = vi.mocked(approvalQueueSeedingService).mock.results.at(-1)!.value; await registration![1].run(new AbortController().signal); - expect(service.sweep).toHaveBeenCalled(); expect(service.next).toHaveBeenCalled(); + expect(service.sweep).toHaveBeenCalled(); + expect(registration![1].nextRunAt()).toBeNull(); + expect(await registration![1].hasPending()).toBe(false); + expect(service.next).toHaveBeenCalled(); } }); diff --git a/server/src/services/approval-queue-seeding-work.ts b/server/src/services/approval-queue-seeding-work.ts index 86cb4bec400..c934bb7c5ad 100644 --- a/server/src/services/approval-queue-seeding-work.ts +++ b/server/src/services/approval-queue-seeding-work.ts @@ -1,6 +1,7 @@ import { and, asc, eq, gt, sql, type SQL } from "drizzle-orm"; -import { approvals, companies, issues, type Db } from "@paperclipai/db"; +import { approvals, companies, issues, issueWorkProducts, type Db } from "@paperclipai/db"; import { executionIssueCondition } from "./issue-visibility.js"; +import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js"; import { decisionQueueService } from "./decision-queues.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; @@ -10,12 +11,12 @@ const linkedIssue = sql`(select linked.issue_id from issue_approvals lin order by linked.issue_id limit 1)`; export function approvalQueueSeedingService(db: Db) { - const pending = new Map([[undefined, { generation: 0 }]]); + const pending = new Map([[undefined, { generation: 0, failures: 0, retryAt: 0 }]]); function request(companyIds: ReadonlySet | undefined) { for (const companyId of companyIds ?? []) { const existing = pending.get(companyId); if (existing) existing.generation++; - else pending.set(companyId, { generation: 0 }); + else pending.set(companyId, { generation: 0, failures: 0, retryAt: 0 }); } } function candidates(database: Db, companyId: string | undefined, condition?: SQL) { @@ -24,47 +25,56 @@ export function approvalQueueSeedingService(db: Db) { .where(and(eq(approvals.status, "pending"), eq(companies.status, "active"), companyId === undefined ? undefined : eq(approvals.companyId, companyId), condition, sql`exists (select 1 from ${issues} where ${issues.id} = ${linkedIssue} and ${issues.companyId} = ${approvals.companyId} and ${executionIssueCondition()} - and exists (select 1 from issue_work_products product where product.company_id = ${approvals.companyId} - and product.issue_id = ${issues.id} and product.type = 'pull_request'))`, + and exists (select 1 from ${issueWorkProducts} where ${pullRequestSeedProductCondition(approvals.companyId)} + and ${issueWorkProducts.issueId} = ${issues.id}))`, sql`not exists (select 1 from decision_queues queue where queue.company_id = ${approvals.companyId} and queue.key = 'prs' and (not queue.seed_rules_enabled or exists (select 1 from decision_queue_items item where item.queue_id = queue.id and item.company_id = ${approvals.companyId} and item.source_kind = 'approval' and item.source_id = ${approvals.id}::text)))`, )).orderBy(asc(approvals.id)); } async function sweep(signal?: AbortSignal) { - const scope = pending.entries().next().value; + const scope = [...pending.entries()].find(([, state]) => state.retryAt <= Date.now()); if (!scope) return; const [companyId, state] = scope; - const generation = state.generation; - const rows = await candidates(db, companyId, state.cursor ? gt(approvals.id, state.cursor) : undefined).limit(100); - const errors: unknown[] = []; - for (const row of rows) { - if (signal?.aborted) break; - try { - await db.transaction(async tx => { - const database = tx as unknown as Db; - const [current] = await candidates(database, row.companyId, eq(approvals.id, row.id)).for("update", { of: approvals }); - if (!current) return; - await decisionQueueService(database).materializeSeedSources(current.companyId, [{ - sourceKind: "approval", sourceId: current.id, issueId: current.issueId, - }]); - }); - } catch (error) { errors.push(error); } - state.cursor = row.id; + try { + const generation = state.generation; + const rows = await candidates(db, companyId, state.cursor ? gt(approvals.id, state.cursor) : undefined).limit(100); + const errors: unknown[] = []; + for (const row of rows) { + if (signal?.aborted) break; + try { + await db.transaction(async tx => { + const database = tx as unknown as Db; + const [current] = await candidates(database, row.companyId, eq(approvals.id, row.id)).for("update", { of: approvals }); + if (!current) return; + await decisionQueueService(database).materializeSeedSources(current.companyId, [{ + sourceKind: "approval", sourceId: current.id, issueId: current.issueId, + }]); + }); + } catch (error) { errors.push(error); } + state.cursor = row.id; + } + if (rows.length < 100) state.cursor = undefined; + const remaining = await candidates(db, companyId).limit(1); + if (!signal?.aborted && generation === state.generation && remaining.length === 0) pending.delete(companyId); + if (errors.length === 1) throw errors[0]; + if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed"); + state.retryAt = 0; + } catch (error) { + state.failures = Math.min(state.failures + 1, 7); + state.retryAt = Date.now() + Math.min(60_000, 1_000 * 2 ** (state.failures - 1)); + throw error; + } finally { + if (pending.has(companyId)) { + pending.delete(companyId); + pending.set(companyId, state); + } } - if (rows.length < 100) state.cursor = undefined; - const remaining = await candidates(db, companyId).limit(1); - if (!signal?.aborted && generation === state.generation && remaining.length === 0) { - pending.delete(companyId); - } else { - pending.delete(companyId); - pending.set(companyId, state); - } - if (errors.length === 1) throw errors[0]; - if (errors.length) throw new AggregateError(errors, "Approval queue seeding failed"); } - async function next() { - return pending.size ? Date.now() : null; + function next() { + let earliest: number | null = null; + for (const state of pending.values()) earliest = earliest === null ? state.retryAt : Math.min(earliest, state.retryAt); + return earliest; } return { sweep, next, request }; } @@ -75,19 +85,20 @@ export function registerApprovalQueueSeedingWork( canRun: () => boolean, ) { let earliest = 0; - let next: number | null = null; return coordinator.register(DELIVERY_QUEUES.approvalQueueSeeding, { retryMs: 1_000, pausePassiveMaintenance: true, onSettled: service.request, run: async signal => { if (!canRun()) return; - if (Date.now() < earliest) { next = earliest; return; } + if (Date.now() < earliest) return; earliest = Date.now() + 1_000; await service.sweep(signal); - next = await service.next(); }, - hasPending: async () => (await service.next()) !== null, - nextRunAt: () => next, + hasPending: async () => (service.next() ?? Infinity) <= Date.now(), + nextRunAt: () => { + const next = service.next(); + return next === null ? null : Math.max(earliest, next); + }, }); } diff --git a/server/src/services/decision-queue-seed-rules.ts b/server/src/services/decision-queue-seed-rules.ts new file mode 100644 index 00000000000..c46b0cde464 --- /dev/null +++ b/server/src/services/decision-queue-seed-rules.ts @@ -0,0 +1,6 @@ +import { and, eq, type SQLWrapper } from "drizzle-orm"; +import { issueWorkProducts } from "@paperclipai/db"; + +export function pullRequestSeedProductCondition(companyId: string | SQLWrapper) { + return and(eq(issueWorkProducts.companyId, companyId), eq(issueWorkProducts.type, "pull_request")); +} diff --git a/server/src/services/decision-queues.ts b/server/src/services/decision-queues.ts index e89ac14ace6..b0c871aa085 100644 --- a/server/src/services/decision-queues.ts +++ b/server/src/services/decision-queues.ts @@ -1,3 +1,4 @@ +import { pullRequestSeedProductCondition } from "./decision-queue-seed-rules.js"; import { notifyApprovalQueueWork } from "./approval-work-notifications.js"; import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -402,8 +403,7 @@ export function decisionQueueService(db: Db) { .select({ issueId: issueWorkProducts.issueId }) .from(issueWorkProducts) .where(and( - eq(issueWorkProducts.companyId, companyId), - eq(issueWorkProducts.type, "pull_request"), + pullRequestSeedProductCondition(companyId), inArray(issueWorkProducts.issueId, issueIds), )).then((rows) => rows.map((row) => row.issueId)));