Merge approval queue retry repair into plugin scheduling

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

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.

doc/plans/2026-10-24-recurring-server-sleep-stack.md

@@ -989,3 +989,14 @@ 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.
+
+
+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.

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