From ea1f546a3f14f9c07b56f1a52cda1a9b84659891 Mon Sep 17 00:00:00 2001 From: bitfathers94 <237535319+bitfathers94@users.noreply.github.com> Date: Mon, 27 Jul 2026 15:02:40 +0000 Subject: [PATCH] fix(notifications): rescue notification_deliveries rows stranded at pending by a failed enqueue evaluateAndEnqueueNotificationDeliveries commits each fresh delivery row before enqueuing its notify-deliver job, so a rejected JOBS.send leaves the committed rows at pending. A client retry finds them already present (idempotent) and omits them, so no deliver job is ever re-sent and the notification stays invisible to the recipient until the 90-day retention sweep deletes it. Add a periodic sweep, mirroring sweepStrandedPendingClosures: a bounded lookback window, a grace period so an in-flight row is not mistaken for a lost one, and a minimum re-sweep interval so a row stuck for another reason is retried periodically rather than every tick. A new repository query lists pending rows within the window, the sweep re-enqueues notify-deliver for each and fails open on a DB read error, and it is wired into the periodic queue tick alongside the sibling repair scans. --- src/db/repositories.ts | 27 ++++ src/notifications/stranded-delivery-sweep.ts | 88 +++++++++++ src/queue/job-dispatch.ts | 8 + test/unit/job-dispatch.test.ts | 27 +++- test/unit/stranded-delivery-sweep.test.ts | 154 +++++++++++++++++++ 5 files changed, 300 insertions(+), 4 deletions(-) create mode 100644 src/notifications/stranded-delivery-sweep.ts create mode 100644 test/unit/stranded-delivery-sweep.test.ts diff --git a/src/db/repositories.ts b/src/db/repositories.ts index bf4afb011c..404fe65c52 100644 --- a/src/db/repositories.ts +++ b/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 { + 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 { const db = getDb(env.DB); await db diff --git a/src/notifications/stranded-delivery-sweep.ts b/src/notifications/stranded-delivery-sweep.ts new file mode 100644 index 0000000000..0d51de5e6c --- /dev/null +++ b/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 { + let deliveries: Awaited> = []; + 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 }; +} diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 5185bff16f..37dff44485 100644 --- a/src/queue/job-dispatch.ts +++ b/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 { 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; } diff --git a/test/unit/job-dispatch.test.ts b/test/unit/job-dispatch.test.ts index f2a51b4032..8288a82962 100644 --- a/test/unit/job-dispatch.test.ts +++ b/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[]> { 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); } 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); }); }); diff --git a/test/unit/stranded-delivery-sweep.test.ts b/test/unit/stranded-delivery-sweep.test.ts new file mode 100644 index 0000000000..0294c56134 --- /dev/null +++ b/test/unit/stranded-delivery-sweep.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it, vi } from "vitest"; +import { + STRANDED_NOTIFICATION_GRACE_MS, + STRANDED_NOTIFICATION_LOOKBACK_MS, + STRANDED_NOTIFICATION_REQUEUE_INTERVAL_MS, + sweepStrandedNotificationDeliveries, +} from "../../src/notifications/stranded-delivery-sweep"; +import { createTestEnv } from "../helpers/d1"; + +function envWithQueue(): { env: Env; sent: unknown[] } { + const sent: unknown[] = []; + const env = createTestEnv(); + (env as unknown as { JOBS: { send: (msg: unknown) => Promise } }).JOBS = { + send: async (msg: unknown) => void sent.push(msg), + }; + return { env, sent }; +} + +// Seeds a notification_deliveries row directly so its created_at (and status) can be chosen freely -- the +// repository insert always stamps created_at with the current wall clock, which the grace/lookback windows need +// to be controllable around. +async function seedDelivery(env: Env, opts: { id?: string; status?: string; createdAtMs: number }): Promise { + const id = opts.id ?? crypto.randomUUID(); + await env.DB.prepare( + "INSERT INTO notification_deliveries (id, dedup_key, channel, recipient_login, event_type, repo_full_name, pull_number, title, body, deeplink, actor_login, status, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + ) + .bind( + id, + `dedup:${id}`, + "badge", + "miner", + "pull_request_changes_requested", + "owner/repo", + 7, + "Changes requested on owner/repo#7", + "A reviewer requested changes.", + "https://github.com/owner/repo/pull/7", + "reviewer", + opts.status ?? "pending", + new Date(opts.createdAtMs).toISOString(), + ) + .run(); + return id; +} + +// #9320: a notification_deliveries row committed BEFORE its notify-deliver enqueue is stranded at `pending` +// when that enqueue is lost (queue backpressure / a transient JOBS.send error). A client retry finds the row +// already exists (idempotent) and omits it, so no deliver job is ever re-sent and the notification is invisible +// to the recipient until the 90-day retention sweep silently deletes it. This sweep rescues it. +describe("sweepStrandedNotificationDeliveries (#9320)", () => { + const PAST_GRACE = (now: number) => now - STRANDED_NOTIFICATION_GRACE_MS - 60_000; + + it("re-enqueues notify-deliver for a pending row older than the grace period", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + const id = await seedDelivery(env, { createdAtMs: PAST_GRACE(now) }); + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 1, requeued: 1 }); + expect(sent).toEqual([{ type: "notify-deliver", requestedBy: "notify-evaluate", deliveryId: id }]); + }); + + it("leaves a pending row still inside the grace period alone", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: now - 1_000 }); + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("does not re-enqueue a row that is no longer pending (already delivered in the interim)", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { status: "delivered", createdAtMs: PAST_GRACE(now) }); + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("ignores rows older than the lookback window (already past rescuing by re-enqueue)", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: now - STRANDED_NOTIFICATION_LOOKBACK_MS - 60_000 }); + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + }); + + it("fails open on a DB read error, returning a zero result rather than throwing", async () => { + const { env, sent } = envWithQueue(); + const repositories = await import("../../src/db/repositories"); + vi.spyOn(repositories, "listStrandedPendingNotificationDeliveries").mockRejectedValue(new Error("db down")); + + expect(await sweepStrandedNotificationDeliveries(env)).toEqual({ scanned: 0, requeued: 0 }); + expect(sent).toEqual([]); + vi.restoreAllMocks(); + }); + + it("does not rescue the same delivery every tick — a row stuck for another reason is retried periodically", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: PAST_GRACE(now) }); + + expect((await sweepStrandedNotificationDeliveries(env, now)).requeued).toBe(1); + expect((await sweepStrandedNotificationDeliveries(env, now + 60_000)).requeued).toBe(0); + // Past the interval it tries again, rather than giving up on a row that is genuinely still stranded. + expect((await sweepStrandedNotificationDeliveries(env, now + STRANDED_NOTIFICATION_REQUEUE_INTERVAL_MS + 1_000)).requeued).toBe(1); + expect(sent).toHaveLength(2); + }); + + it("does not count a rescue whose enqueue failed, and re-scans it on the next tick", async () => { + const { env } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: PAST_GRACE(now) }); + (env as unknown as { JOBS: { send: () => Promise } }).JOBS = { send: async () => Promise.reject(new Error("queue down")) }; + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 1, requeued: 0 }); + // Nothing was recorded, so the next tick still sees it as un-rescued rather than believing it handled it. + expect((await sweepStrandedNotificationDeliveries(env, now + 1_000)).scanned).toBe(1); + }); + + it("treats an unreadable rescue history as 'already rescued', so a broken ledger cannot cause a flood", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: PAST_GRACE(now) }); + const repositories = await import("../../src/db/repositories"); + vi.spyOn(repositories, "countRecentAuditEventsForActorAndTarget").mockRejectedValue(new Error("audit read down")); + + expect(await sweepStrandedNotificationDeliveries(env, now)).toEqual({ scanned: 1, requeued: 0 }); + expect(sent).toEqual([]); + vi.restoreAllMocks(); + }); + + it("counts the rescue even when the follow-up audit write fails", async () => { + const { env, sent } = envWithQueue(); + const now = Date.now(); + await seedDelivery(env, { createdAtMs: PAST_GRACE(now) }); + const repositories = await import("../../src/db/repositories"); + vi.spyOn(repositories, "recordAuditEvent").mockRejectedValue(new Error("audit write down")); + + // The job WAS enqueued, which is the part that rescues the delivery; losing the bookkeeping row must not undo it. + expect((await sweepStrandedNotificationDeliveries(env, now)).requeued).toBe(1); + expect(sent).toHaveLength(1); + vi.restoreAllMocks(); + }); + + it("defaults nowMs to the current wall clock when the caller omits it", async () => { + const { env, sent } = envWithQueue(); + await seedDelivery(env, { createdAtMs: PAST_GRACE(Date.now()) }); + + expect((await sweepStrandedNotificationDeliveries(env)).requeued).toBe(1); + expect(sent).toHaveLength(1); + }); +});