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)));