From 6fb70f34d82d2d5e61f81293f52485518c636799 Mon Sep 17 00:00:00 2001 From: bitfathers94 <237535319+bitfathers94@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:51:00 +0000 Subject: [PATCH] fix(queue): bound and isolate the notify-evaluate deliver fan-out The notify-deliver enqueue after notify-evaluate ran through an unbounded Promise.all over every resolved delivery (which can outnumber the batch's events one-to-many across channels), rejected the whole fan-out on the first send failure, and left the failure completely unlogged. A retry of that job recovers nothing since evaluateNotificationEvent only returns newly-created rows, so a lost send stayed pending until the 10-minute stranded-delivery sweep caught it. Route the sends through mapWithConcurrency at a bounded, exported NOTIFY_DELIVER_SEND_CONCURRENCY, same posture as the backfill-registered-repos fan-out: every send is attempted exactly once regardless of an earlier one's outcome, a failed send logs a structured notify_deliver_fanout_send_failed line naming the delivery, and any failures still fail the job afterward by throwing an aggregate error naming every lost delivery id. --- src/queue/job-dispatch.ts | 41 ++++++-- test/unit/notifications-events.test.ts | 135 ++++++++++++++++++++++++- 2 files changed, 166 insertions(+), 10 deletions(-) diff --git a/src/queue/job-dispatch.ts b/src/queue/job-dispatch.ts index 796f20bc79..3312fb768c 100644 --- a/src/queue/job-dispatch.ts +++ b/src/queue/job-dispatch.ts @@ -94,6 +94,14 @@ import { // imported back) since processJob was its only caller. const NOTIFY_EVALUATE_EVENT_CONCURRENCY = 5; +// #10022: the evaluate half above is bounded, but the enqueue half that follows it was not -- a batched job +// resolving to many deliveries (one per event per subscribed channel, so deliveries.length can exceed +// events.length) sent every notify-deliver job through an unbounded Promise.all, letting a single job issue as +// many concurrent env.JOBS.send calls as it had deliveries. Same bounded worker-pool shape as +// NOTIFY_EVALUATE_EVENT_CONCURRENCY above, sized independently since the two fan-outs cost different amounts of +// work per item. +export const NOTIFY_DELIVER_SEND_CONCURRENCY = 5; + export async function processJob(env: Env, message: JobMessage): Promise { switch (message.type) { case "refresh-registry": @@ -400,15 +408,32 @@ export async function processJob(env: Env, message: JobMessage): Promise { const deliveries = ( await mapWithConcurrency(events, NOTIFY_EVALUATE_EVENT_CONCURRENCY, (event) => evaluateNotificationEvent(env, event)) ).flat(); - await Promise.all( - deliveries.map((delivery) => - env.JOBS.send({ - type: "notify-deliver", - requestedBy: "notify-evaluate", - deliveryId: delivery.id, - }), - ), + // #10022: bound the send fan-out (deliveries.length can exceed events.length -- one delivery per event + // per resolved channel) and attempt every send exactly once regardless of an earlier one's outcome, same + // posture as the #8355 backfill-registered-repos fan-out above. A retry of this job re-evaluates events + // whose deliveries already exist (evaluateNotificationEvent's created-only return), so a send failure here + // must be logged and named now -- the retry recovers nothing for it. + const sendResults = await mapWithConcurrency( + deliveries, + NOTIFY_DELIVER_SEND_CONCURRENCY, + async (delivery): Promise<{ deliveryId: string; ok: boolean }> => { + try { + await env.JOBS.send({ + type: "notify-deliver", + requestedBy: "notify-evaluate", + deliveryId: delivery.id, + }); + return { deliveryId: delivery.id, ok: true }; + } catch (reason) { + console.error(JSON.stringify({ level: "error", event: "notify_deliver_fanout_send_failed", deliveryId: delivery.id, reason: String(reason) })); + return { deliveryId: delivery.id, ok: false }; + } + }, ); + const failedDeliveryIds = sendResults.filter((result) => !result.ok).map((result) => result.deliveryId); + if (failedDeliveryIds.length > 0) { + throw new Error(`notify-evaluate deliver fan-out: ${failedDeliveryIds.length}/${sendResults.length} delivery send(s) failed: ${failedDeliveryIds.join(", ")}`); + } return; } case "notify-deliver": diff --git a/test/unit/notifications-events.test.ts b/test/unit/notifications-events.test.ts index 7d6c1330e7..8de75b25bd 100644 --- a/test/unit/notifications-events.test.ts +++ b/test/unit/notifications-events.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { detectNotificationEvents } from "../../src/notifications/events"; -import type { GitHubWebhookPayload } from "../../src/types"; +import { NOTIFY_DELIVER_SEND_CONCURRENCY, processJob } from "../../src/queue/job-dispatch"; +import { createTestEnv } from "../helpers/d1"; +import type { DetectedNotificationEvent, GitHubWebhookPayload } from "../../src/types"; const basePayload: GitHubWebhookPayload = { action: "submitted", @@ -204,3 +206,132 @@ describe("detectNotificationEvents — merged PR (#702)", () => { expect(events[0]?.deeplink).toBe("https://github.com/JSONbored/loopover/pull/42"); }); }); + +function detectedEvent(overrides: Partial = {}): DetectedNotificationEvent { + const login = overrides.recipientLogin ?? "miner-1"; + return { + eventType: "pull_request_changes_requested", + recipientLogin: login, + repoFullName: "owner/repo", + pullNumber: 7, + dedupKey: `changes_requested:owner/repo#7:reviewer:${login}`, + deeplink: "https://github.com/owner/repo/pull/7", + actorLogin: "reviewer", + detectedAt: "2026-05-28T12:00:00.000Z", + ...overrides, + }; +} + +describe("processJob notify-evaluate deliver fan-out (#10022)", () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("attempts every OTHER delivery's send even when one rejects, and throws once naming it", async () => { + let callIndex = 0; + let failedDeliveryId: string | undefined; + const sentDeliveryIds: string[] = []; + const env = createTestEnv({ + JOBS: { + async send(message: unknown) { + const deliveryId = (message as { deliveryId?: string }).deliveryId; + const isSecondCall = callIndex === 1; + callIndex += 1; + if (deliveryId) sentDeliveryIds.push(deliveryId); + if (isSecondCall) { + failedDeliveryId = deliveryId; + throw new Error("simulated transient queue-send failure"); + } + return undefined; + }, + } as unknown as Queue, + }); + + const errorLogs: string[] = []; + vi.spyOn(console, "error").mockImplementation((...args: unknown[]) => { + errorLogs.push(String(args[0])); + }); + + const events = ["miner-1", "miner-2", "miner-3"].map((login) => detectedEvent({ recipientLogin: login })); + let thrown: unknown; + try { + await processJob(env, { type: "notify-evaluate", requestedBy: "webhook", events }); + } catch (error) { + thrown = error; + } + + // Every delivery's send was attempted exactly once, regardless of the middle one's rejection. + expect(sentDeliveryIds).toHaveLength(3); + expect(failedDeliveryId).toBeDefined(); + expect(thrown).toBeInstanceOf(Error); + expect((thrown as Error).message).toMatch(/notify-evaluate deliver fan-out: 1\/3 delivery send\(s\) failed:/); + expect((thrown as Error).message).toContain(failedDeliveryId); + + const failureLog = errorLogs.map((line) => JSON.parse(line) as Record).find((log) => log.event === "notify_deliver_fanout_send_failed"); + expect(failureLog).toMatchObject({ level: "error", event: "notify_deliver_fanout_send_failed", deliveryId: failedDeliveryId }); + }); + + it("never issues more concurrent sends than NOTIFY_DELIVER_SEND_CONCURRENCY", async () => { + let inFlight = 0; + let maxInFlight = 0; + const env = createTestEnv({ + JOBS: { + async send() { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + inFlight -= 1; + return undefined; + }, + } as unknown as Queue, + }); + + const events = Array.from({ length: 12 }, (_, index) => detectedEvent({ recipientLogin: `miner-${index}` })); + await expect(processJob(env, { type: "notify-evaluate", requestedBy: "webhook", events })).resolves.toBeUndefined(); + + expect(maxInFlight).toBeGreaterThan(1); + expect(maxInFlight).toBeLessThanOrEqual(NOTIFY_DELIVER_SEND_CONCURRENCY); + }); + + it("does not throw and sends exactly one notify-deliver message per delivery when every send succeeds", async () => { + const sent: Array<{ type: string; requestedBy: string; deliveryId: string }> = []; + const env = createTestEnv({ + JOBS: { + async send(message: unknown) { + sent.push(message as { type: string; requestedBy: string; deliveryId: string }); + return undefined; + }, + } as unknown as Queue, + }); + + const errorLogs: string[] = []; + vi.spyOn(console, "error").mockImplementation((...args: unknown[]) => { + errorLogs.push(String(args[0])); + }); + + const events = ["miner-1", "miner-2", "miner-3"].map((login) => detectedEvent({ recipientLogin: login })); + await expect(processJob(env, { type: "notify-evaluate", requestedBy: "webhook", events })).resolves.toBeUndefined(); + + expect(sent).toHaveLength(3); + for (const message of sent) { + expect(message).toMatchObject({ type: "notify-deliver", requestedBy: "notify-evaluate" }); + expect(typeof message.deliveryId).toBe("string"); + } + expect(errorLogs.some((line) => line.includes("notify_deliver_fanout_send_failed"))).toBe(false); + }); + + it("does not throw when the batch resolves to zero deliveries", async () => { + const sent: unknown[] = []; + const env = createTestEnv({ + JOBS: { + async send(message: unknown) { + sent.push(message); + return undefined; + }, + } as unknown as Queue, + }); + + await expect(processJob(env, { type: "notify-evaluate", requestedBy: "webhook", events: [] })).resolves.toBeUndefined(); + expect(sent).toHaveLength(0); + }); +});