diff --git a/src/queue/duplicate-detection.ts b/src/queue/duplicate-detection.ts index 0afacc5090..4f0663adc6 100644 --- a/src/queue/duplicate-detection.ts +++ b/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(); - 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), diff --git a/src/queue/map-with-concurrency.ts b/src/queue/map-with-concurrency.ts new file mode 100644 index 0000000000..97e80556a7 --- /dev/null +++ b/src/queue/map-with-concurrency.ts @@ -0,0 +1,19 @@ +export async function mapWithConcurrency( + items: T[], + concurrency: number, + mapper: (item: T) => Promise, +): Promise { + 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; +} diff --git a/src/queue/processors.ts b/src/queue/processors.ts index d31b268364..806e54702e 100644 --- a/src/queue/processors.ts +++ b/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(items: T[], concurrency: number, mapper: (item: T) => Promise): Promise { - 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 diff --git a/test/unit/reconcile-live-duplicate-siblings.test.ts b/test/unit/reconcile-live-duplicate-siblings.test.ts index 48803e3a19..fd12dcd72f 100644 --- a/test/unit/reconcile-live-duplicate-siblings.test.ts +++ b/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); + }); });