fix(queue): cap duplicate-sibling live GitHub fetch concurrency (#5835)

src/queue/duplicate-detection.ts

@@ -12,6 +12,10 @@ import { githubRateLimitAdmissionKeyForToken } from "../github/client";
 import { isDuplicateClusterWinnerByClaim, resolveDuplicateClusterWinnerNumber } from "../signals/duplicate-winner";
 import { isDuplicateWinnerEnabledGlobally, resolveDuplicateWinnerEnabled } from "../settings/duplicate-winner-mode";
 import type { PullRequestRecord, RepositorySettings } from "../types";
+import { mapWithConcurrency } from "./map-with-concurrency";
+
+/** Same order of magnitude as processors.ts's other per-item live GitHub fan-outs (#5835). */
+const DUPLICATE_SIBLING_LIVE_RECONCILE_CONCURRENCY = 10;
 
 /**
  * Duplicate-winner adjudication (#dup-winner) seam for the close-reason disposition. Given a PR's open
@@ -93,8 +97,7 @@ export async function reconcileLiveDuplicateSiblings(
   const token = installationToken ?? env.GITHUB_PUBLIC_TOKEN;
   const admissionKey = githubRateLimitAdmissionKeyForToken(env, token, installationId);
   const staleClosed = new Set<number>();
-  await Promise.all(
-    overlapping.map(async (sibling) => {
+  await mapWithConcurrency(overlapping, DUPLICATE_SIBLING_LIVE_RECONCILE_CONCURRENCY, async (sibling) => {
       // #2537: deliberately NOT durable-cached (flagged by the gate's own review) -- despite recomputing every
       // delivery, this reconcile feeds duplicate-winner selection, which can auto-CLOSE the CURRENT PR when
       // duplicateWinnerEnabled. A cached "open" read up to PR_STATE_CACHE_MAX_AGE_MS stale after a missed
@@ -111,8 +114,7 @@ export async function reconcileLiveDuplicateSiblings(
       ).catch(() => undefined);
       if (liveState !== undefined && liveState !== "open")
         staleClosed.add(sibling.number);
-    }),
-  );
+    });
   if (staleClosed.size === 0) return otherOpenPullRequests;
   return otherOpenPullRequests.filter(
     (other) => !staleClosed.has(other.number),

src/queue/map-with-concurrency.ts

@@ -0,0 +1,19 @@
+export async function mapWithConcurrency<T, R>(
+  items: T[],
+  concurrency: number,
+  mapper: (item: T) => Promise<R>,
+): Promise<R[]> {
+  const results: R[] = new Array(items.length);
+  let nextIndex = 0;
+  const workerCount = Math.max(1, Math.min(concurrency, items.length || 1));
+  await Promise.all(
+    Array.from({ length: workerCount }, async () => {
+      while (nextIndex < items.length) {
+        const index = nextIndex;
+        nextIndex += 1;
+        results[index] = await mapper(items[index] as T);
+      }
+    }),
+  );
+  return results;
+}

src/queue/processors.ts

@@ -366,6 +366,8 @@ export {
   linkedIssueDuplicatePullRequestsForGate,
   reconcileLiveDuplicateSiblings,
 } from "./duplicate-detection";
+import { mapWithConcurrency } from "./map-with-concurrency";
+export { mapWithConcurrency } from "./map-with-concurrency";
 // #4013 step 4: same shim shape for the AI-slop-advisory gating/orchestration functions -- imported here
 // for this file's own internal callers, and re-exported so test/unit/advisory-ai-routing-call-sites.test.ts,
 // test/unit/ai-slop.test.ts, and test/unit/gate-check-policy.test.ts's existing
@@ -4914,22 +4916,6 @@ async function countLiveOpenWithConcurrencyUntil(
 // via mapWithConcurrency in addition to the repository query's total row cap.
 const CONTRIBUTOR_CAP_LIVE_CHECK_CONCURRENCY = 10;
 
-export async function mapWithConcurrency<T, R>(items: T[], concurrency: number, mapper: (item: T) => Promise<R>): Promise<R[]> {
-  const results: R[] = new Array(items.length);
-  let nextIndex = 0;
-  const workerCount = Math.max(1, Math.min(concurrency, items.length || 1));
-  await Promise.all(
-    Array.from({ length: workerCount }, async () => {
-      while (nextIndex < items.length) {
-        const index = nextIndex;
-        nextIndex += 1;
-        results[index] = await mapper(items[index] as T);
-      }
-    }),
-  );
-  return results;
-}
-
 /**
  * Install-wide contributor open-item count, LIVE-VERIFIED (#2562 gate-review follow-up): every OTHER counted
  * item is confirmed still-open via a live GET before counting toward the cap (mirrors the existing per-repo

test/unit/reconcile-live-duplicate-siblings.test.ts

@@ -179,4 +179,26 @@ describe("reconcileLiveDuplicateSiblings (#dup-winner / audit #15)", () => {
     const pr = makePr(9, "open", [1]);
     expect(await reconcileLiveDuplicateSiblings(env, null, "owner/repo", pr, siblings, { duplicateWinnerMode: "off" })).toBe(siblings);
   });
+
+  it("bounds live sibling fetches to 10 concurrent in-flight calls (#5835)", async () => {
+    const env = createTestEnv();
+    env.LOOPOVER_DUPLICATE_WINNER = "true";
+    let inFlight = 0;
+    let maxInFlight = 0;
+    vi.stubGlobal("fetch", async (input: RequestInfo | URL) => {
+      const url = String(input);
+      if (!url.includes("/pulls/")) return new Response("not found", { status: 404 });
+      inFlight += 1;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 25));
+      inFlight -= 1;
+      return new Response(JSON.stringify({ state: "open" }), { status: 200 });
+    });
+    const siblings = Array.from({ length: 15 }, (_, index) => makePr(index + 1, "open", [1]));
+    const pr = makePr(99, "open", [1]);
+    const result = await reconcileLiveDuplicateSiblings(env, null, "owner/repo", pr, siblings, settings);
+    expect(result.map((p) => p.number)).toEqual(siblings.map((p) => p.number));
+    expect(maxInFlight).toBeLessThanOrEqual(10);
+    expect(maxInFlight).toBeGreaterThan(1);
+  });
 });