fix(server): back off failed approval queue projections

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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

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<typeof import("../services/approval-queue-seeding-work.js")>(),
-  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();
     }
   });
 

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<string>`(select linked.issue_id from issue_approvals lin
   order by linked.issue_id limit 1)`;
 
 export function approvalQueueSeedingService(db: Db) {
-  const pending = new Map<string | undefined, { cursor?: string; generation: number }>([[undefined, { generation: 0 }]]);
+  const pending = new Map<string | undefined, { cursor?: string; generation: number; failures: number; retryAt: number }>([[undefined, { generation: 0, failures: 0, retryAt: 0 }]]);
   function request(companyIds: ReadonlySet<string> | 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);
+    },
   });
 }

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"));
+}

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