fix(queue): bound and isolate the notify-evaluate deliver fan-out

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<void> {
   switch (message.type) {
     case "refresh-registry":
@@ -400,15 +408,32 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
       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":

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> = {}): 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<string, unknown>).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);
+  });
+});