fix(notifications): rescue notification_deliveries rows stranded at pending by a failed enqueue

src/db/repositories.ts

@@ -2313,6 +2313,33 @@ export async function getNotificationDeliveryById(env: Env, id: string): Promise
   return row ? toNotificationDeliveryRecord(row) : null;
 }
 
+/** #9320: `notification_deliveries` rows stranded at `status: "pending"` — a notify-deliver job that may have
+ *  been lost to a failed/partial enqueue. Bounded on both ends: `createdAt` older than `olderThanIso` (the
+ *  grace cutoff, so a row merely waiting its turn behind a busy queue is not mistaken for a lost one) and no
+ *  older than `notBeforeIso` (the lookback floor, so the scan stays cheap and skips rows the 90-day retention
+ *  sweep is about to delete anyway). Oldest first, so the most-stranded rows are rescued first under the limit. */
+export async function listStrandedPendingNotificationDeliveries(
+  env: Env,
+  olderThanIso: string,
+  notBeforeIso: string,
+  limit: number,
+): Promise<NotificationDeliveryRecord[]> {
+  const db = getDb(env.DB);
+  const rows = await db
+    .select()
+    .from(notificationDeliveries)
+    .where(
+      and(
+        eq(notificationDeliveries.status, "pending"),
+        lt(notificationDeliveries.createdAt, olderThanIso),
+        gte(notificationDeliveries.createdAt, notBeforeIso),
+      ),
+    )
+    .orderBy(asc(notificationDeliveries.createdAt), asc(notificationDeliveries.id))
+    .limit(limit);
+  return rows.map(toNotificationDeliveryRecord);
+}
+
 export async function markNotificationDeliveryDelivered(env: Env, id: string): Promise<void> {
   const db = getDb(env.DB);
   await db

src/notifications/stranded-delivery-sweep.ts

@@ -0,0 +1,88 @@
+import { countRecentAuditEventsForActorAndTarget, listStrandedPendingNotificationDeliveries, recordAuditEvent } from "../db/repositories";
+import { errorMessage } from "../utils/json";
+
+/**
+ * #9320 — rescue a `notification_deliveries` row stranded at `status: "pending"` by a failed/partial enqueue.
+ *
+ * `evaluateAndEnqueueNotificationDeliveries` commits each fresh delivery row (idempotent on
+ * `UNIQUE(dedup_key, channel)`) BEFORE it enqueues the matching `notify-deliver` job. If that `Promise.all`
+ * enqueue rejects for even one delivery (queue backpressure, a transient `JOBS.send` error), the whole call
+ * throws — but the rows already committed stay at `pending`. A client retry re-submits the same events, yet the
+ * idempotency check now finds those rows already exist (`created: false`) and omits them, so no `notify-deliver`
+ * job is ever (re-)sent. The row is invisible to the recipient (`buildNotificationFeed` surfaces only
+ * `delivered`/`read`) until the 90-day retention sweep silently deletes it.
+ *
+ * This mirrors `sweepStrandedPendingClosures` (#9031): a deadline something re-checks, rather than a sequence
+ * that depends on one message surviving. `deliverNotification` is already idempotent (it no-ops a row that is no
+ * longer `pending`), so replaying `notify-deliver` for a row delivered in the interim is harmless by
+ * construction — but the sweep's own query filters on `status: "pending"` so a delivered row is never even
+ * re-enqueued, rather than relying solely on that downstream no-op.
+ */
+
+export const STRANDED_NOTIFICATION_REQUEUED_EVENT = "notification.delivery.requeued";
+
+/** Grace past a row's creation before the sweep steps in, so a delivery merely waiting its turn behind a busy
+ *  queue is never mistaken for a lost one. */
+export const STRANDED_NOTIFICATION_GRACE_MS = 10 * 60 * 1000;
+
+/** How far back to scan. A pending row older than this is past rescuing by re-enqueue — the retention sweep is
+ *  about to delete it — and the scan stays cheap by not carrying that history forever. */
+export const STRANDED_NOTIFICATION_LOOKBACK_MS = 3 * 24 * 60 * 60 * 1000;
+
+/** Minimum spacing between rescues for the same delivery, so a row stuck for a reason the re-enqueue cannot
+ *  fix gets retried periodically instead of on every single sweep tick. */
+export const STRANDED_NOTIFICATION_REQUEUE_INTERVAL_MS = 60 * 60 * 1000;
+
+/** Bounded rows-per-tick, so one sweep can never fan out an unbounded number of re-enqueues. */
+export const STRANDED_NOTIFICATION_SCAN_LIMIT = 500;
+
+export type StrandedNotificationSweepResult = { scanned: number; requeued: number };
+
+export async function sweepStrandedNotificationDeliveries(
+  env: Env,
+  nowMs: number = Date.now(),
+): Promise<StrandedNotificationSweepResult> {
+  let deliveries: Awaited<ReturnType<typeof listStrandedPendingNotificationDeliveries>> = [];
+  try {
+    deliveries = await listStrandedPendingNotificationDeliveries(
+      env,
+      new Date(nowMs - STRANDED_NOTIFICATION_GRACE_MS).toISOString(),
+      new Date(nowMs - STRANDED_NOTIFICATION_LOOKBACK_MS).toISOString(),
+      STRANDED_NOTIFICATION_SCAN_LIMIT,
+    );
+  } catch (error) {
+    console.warn(JSON.stringify({ level: "warn", event: "stranded_notification_sweep_scan_failed", message: errorMessage(error).slice(0, 160) }));
+    return { scanned: 0, requeued: 0 };
+  }
+
+  let requeued = 0;
+  for (const delivery of deliveries) {
+    // Spacing gate: skip a delivery re-enqueued within the interval. An unreadable ledger falls back to
+    // "already rescued" (1) so a broken audit read can never turn the sweep into a per-tick flood.
+    const recent = await countRecentAuditEventsForActorAndTarget(
+      env,
+      "loopover",
+      STRANDED_NOTIFICATION_REQUEUED_EVENT,
+      delivery.id,
+      new Date(nowMs - STRANDED_NOTIFICATION_REQUEUE_INTERVAL_MS).toISOString(),
+    ).catch(() => 1);
+    if (recent > 0) continue;
+
+    const sent = await env.JOBS.send({ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: delivery.id })
+      .then(() => true)
+      .catch(() => false);
+    if (!sent) continue;
+    requeued += 1;
+    // Best-effort ledger, exactly like the enqueue it accompanies: losing this bookkeeping row must not undo a
+    // rescue that already reached the queue — the next tick's spacing gate simply re-evaluates from scratch.
+    await recordAuditEvent(env, {
+      eventType: STRANDED_NOTIFICATION_REQUEUED_EVENT,
+      actor: "loopover",
+      targetKey: delivery.id,
+      outcome: "queued",
+      detail: "notify-deliver re-enqueued for a delivery stranded at pending past its grace period",
+      metadata: { deliveryId: delivery.id, recipientLogin: delivery.recipientLogin, channel: delivery.channel },
+    }).catch(() => undefined);
+  }
+  return { scanned: deliveries.length, requeued };
+}

src/queue/job-dispatch.ts

@@ -57,6 +57,7 @@ import { sweepStaleApprovalQueue } from "../services/agent-approval-queue";
 import { reconcileMissingPrOutcomes } from "../review/pr-outcome-reconciler";
 import { sweepStrandedPendingClosures } from "../review/pending-closure-watchdog";
 import { reconcileSurfaceWithoutDisposition } from "../review/surface-disposition-reconciler";
+import { sweepStrandedNotificationDeliveries } from "../notifications/stranded-delivery-sweep";
 // The 15 handlers below have no reason to move -- each is only reachable via this dispatcher (or, for
 // mapWithConcurrency, ALSO used by other still-in-processors.ts code), so they stay put and are exported
 // there purely for this one-directional import-back (processors.ts itself never calls processJob).
@@ -312,6 +313,13 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
         if (orphanedDispositions && orphanedDispositions.requeued > 0) {
           console.log(JSON.stringify({ event: "surface_without_disposition_reconciled", ...orphanedDispositions }));
         }
+        // #9320 re-enqueues a notify-deliver for a notification_deliveries row stranded at `pending` when its
+        // enqueue was lost to queue backpressure, which otherwise leaves the notification invisible to the
+        // recipient until the 90-day retention sweep silently deletes it.
+        const strandedNotifications = await sweepStrandedNotificationDeliveries(env).catch(() => null);
+        if (strandedNotifications && strandedNotifications.requeued > 0) {
+          console.log(JSON.stringify({ event: "stranded_notification_deliveries_requeued", ...strandedNotifications }));
+        }
         await fanOutAgentRegateSweepJobs(env, message.requestedBy);
         return;
       }

test/unit/job-dispatch.test.ts

@@ -187,47 +187,66 @@ describe("agent-regate-sweep runs the durability repair scans (#9026, #9031, #89
     outcomes?: unknown;
     stranded?: unknown;
     orphaned?: unknown;
+    notifications?: unknown;
     outcomesThrows?: boolean;
     strandedThrows?: boolean;
     orphanedThrows?: boolean;
+    notificationsThrows?: boolean;
   }): Promise<Record<string, unknown>[]> {
     const logs: string[] = [];
     vi.spyOn(console, "log").mockImplementation((...args: unknown[]) => void logs.push(String(args[0])));
     const reconciler = await import("../../src/review/pr-outcome-reconciler");
     const watchdog = await import("../../src/review/pending-closure-watchdog");
     const surfaceReconciler = await import("../../src/review/surface-disposition-reconciler");
+    const notificationSweep = await import("../../src/notifications/stranded-delivery-sweep");
     const outcomeSpy = vi.spyOn(reconciler, "reconcileMissingPrOutcomes");
     const strandedSpy = vi.spyOn(watchdog, "sweepStrandedPendingClosures");
     const orphanedSpy = vi.spyOn(surfaceReconciler, "reconcileSurfaceWithoutDisposition");
+    const notificationSpy = vi.spyOn(notificationSweep, "sweepStrandedNotificationDeliveries");
     if (stubs.outcomesThrows) outcomeSpy.mockRejectedValue(new Error("db down"));
     else outcomeSpy.mockResolvedValue((stubs.outcomes ?? { scanned: 0, backfilled: 0 }) as never);
     if (stubs.strandedThrows) strandedSpy.mockRejectedValue(new Error("db down"));
     else strandedSpy.mockResolvedValue((stubs.stranded ?? { scanned: 0, requeued: 0 }) as never);
     if (stubs.orphanedThrows) orphanedSpy.mockRejectedValue(new Error("db down"));
     else orphanedSpy.mockResolvedValue((stubs.orphaned ?? { scanned: 0, requeued: 0 }) as never);
+    if (stubs.notificationsThrows) notificationSpy.mockRejectedValue(new Error("db down"));
+    else notificationSpy.mockResolvedValue((stubs.notifications ?? { scanned: 0, requeued: 0 }) as never);
 
     await processJob(createTestEnv(), { type: "agent-regate-sweep", requestedBy: "schedule" });
     return logs.map((line) => JSON.parse(line) as Record<string, unknown>);
   }
 
   it("reports what each scan repaired", async () => {
-    const logs = await sweepWith({ outcomes: { scanned: 5, backfilled: 3 }, stranded: { scanned: 2, requeued: 1 }, orphaned: { scanned: 3, requeued: 2 } });
+    const logs = await sweepWith({
+      outcomes: { scanned: 5, backfilled: 3 },
+      stranded: { scanned: 2, requeued: 1 },
+      orphaned: { scanned: 3, requeued: 2 },
+      notifications: { scanned: 4, requeued: 3 },
+    });
     expect(logs.find((log) => log.event === "pr_outcomes_reconciled")).toMatchObject({ scanned: 5, backfilled: 3 });
     expect(logs.find((log) => log.event === "pending_closure_verifications_requeued")).toMatchObject({ scanned: 2, requeued: 1 });
     expect(logs.find((log) => log.event === "surface_without_disposition_reconciled")).toMatchObject({ scanned: 3, requeued: 2 });
+    expect(logs.find((log) => log.event === "stranded_notification_deliveries_requeued")).toMatchObject({ scanned: 4, requeued: 3 });
   });
 
   it("stays quiet when a scan looked but repaired nothing", async () => {
-    const logs = await sweepWith({ outcomes: { scanned: 9, backfilled: 0 }, stranded: { scanned: 4, requeued: 0 }, orphaned: { scanned: 6, requeued: 0 } });
+    const logs = await sweepWith({
+      outcomes: { scanned: 9, backfilled: 0 },
+      stranded: { scanned: 4, requeued: 0 },
+      orphaned: { scanned: 6, requeued: 0 },
+      notifications: { scanned: 7, requeued: 0 },
+    });
     expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false);
     expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false);
     expect(logs.some((log) => log.event === "surface_without_disposition_reconciled")).toBe(false);
+    expect(logs.some((log) => log.event === "stranded_notification_deliveries_requeued")).toBe(false);
   });
 
-  it("completes the tick even when all three scans throw", async () => {
-    const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true, orphanedThrows: true });
+  it("completes the tick even when all scans throw", async () => {
+    const logs = await sweepWith({ outcomesThrows: true, strandedThrows: true, orphanedThrows: true, notificationsThrows: true });
     expect(logs.some((log) => log.event === "pr_outcomes_reconciled")).toBe(false);
     expect(logs.some((log) => log.event === "pending_closure_verifications_requeued")).toBe(false);
     expect(logs.some((log) => log.event === "surface_without_disposition_reconciled")).toBe(false);
+    expect(logs.some((log) => log.event === "stranded_notification_deliveries_requeued")).toBe(false);
   });
 });