diff --git a/src/queue/concurrency.ts b/src/queue/concurrency.ts new file mode 100644 index 0000000000..f5e711693a --- /dev/null +++ b/src/queue/concurrency.ts @@ -0,0 +1,30 @@ +// Bounded fan-out helper, shared by the queue modules (#5835). It lived in processors.ts, but +// duplicate-detection.ts needs it too and deliberately does not import processors.ts -- that file's own header +// records why: importing back would make the two circularly dependent (it inlines an admission-key wrapper for +// exactly the same reason). A neutral module both can depend on is the way to share this without reintroducing +// the cycle, rather than growing a third private copy of the same worker-pool loop. + +/** + * Map `items` through `mapper` with at most `concurrency` in flight at once, preserving input order in the + * result. Unlike `Promise.all(items.map(...))`, the number of simultaneously-running mappers never exceeds + * `concurrency` -- which is what keeps a large input from bursting that many concurrent GitHub REST calls out + * of a single webhook delivery and draining the installation's shared rate-limit bucket. + * + * A `concurrency` below 1 is clamped to 1 (never zero workers, which would hang), and never exceeds the item + * count (no idle workers). An empty input resolves immediately to an empty array. + */ +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/duplicate-detection.ts b/src/queue/duplicate-detection.ts index bcbb8b2012..51412a7397 100644 --- a/src/queue/duplicate-detection.ts +++ b/src/queue/duplicate-detection.ts @@ -7,6 +7,7 @@ // processors.ts circularly dependent on each other. import { createInstallationToken } from "../github/app"; +import { mapWithConcurrency } from "./concurrency"; import { fetchLivePullRequestState } from "../github/backfill"; import { githubRateLimitAdmissionKeyForToken } from "../github/client"; import { isDuplicateClusterWinnerByClaim, resolveDuplicateClusterWinnerNumber } from "../signals/duplicate-winner"; @@ -53,6 +54,15 @@ export function dupWinnerLinkedDuplicateWinnerNumber( return winner === null || winner === prNumber ? null : winner; } +// A popular linked issue can accumulate dozens of duplicate PRs, and this reconcile fires one live +// fetchLivePullRequestState per overlapping sibling. An unbounded `Promise.all` over that list bursts every one +// of those REST calls simultaneously out of a single webhook delivery, against the ONE installation-wide ~5000/hr +// bucket every managed repo shares. 10 mirrors the two established caps for exactly this "many live per-item +// GitHub reads from one delivery" shape -- CONTRIBUTOR_CAP_LIVE_CHECK_CONCURRENCY and +// GLOBAL_OPEN_ITEM_LIVE_CHECK_CONCURRENCY (both `processors.ts`, the file this one was extracted from) -- so the +// three bounded fan-outs stay consistent rather than each picking their own number (#5835). +const DUPLICATE_SIBLING_LIVE_CHECK_CONCURRENCY = 10; + /** * Live-reconcile the duplicate cluster's open siblings before the winner is elected (#dup-winner / audit #15). * @@ -91,8 +101,10 @@ 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_CHECK_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 @@ -109,7 +121,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( diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 905e965628..66a3f6ad9b 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -323,6 +323,7 @@ import { buildAiReviewDiff, buildSecretScanDiff, buildUnifiedReviewDiff, totalAd export { buildAiReviewDiff, buildSecretScanDiff } from "../review/review-diff"; import { estimateReviewEffort } from "../review/review-effort"; import { buildUnifiedCommentBody } from "../review/unified-comment-bridge"; +import { mapWithConcurrency } from "./concurrency"; import { isRetryableJobError, RetryableJobError } from "./retryable"; import { claimPrActuationLock, @@ -4890,21 +4891,11 @@ 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; -} +// Moved to ./concurrency (#5835) so duplicate-detection.ts can bound its own live-sibling fan-out with the same +// worker pool -- that file deliberately does not import this one (its header records why: it would make the two +// circularly dependent), so the helper now lives in a neutral module both can depend on. Re-exported here so +// this file's existing surface is unchanged. +export { mapWithConcurrency } from "./concurrency"; /** * Install-wide contributor open-item count, LIVE-VERIFIED (#2562 gate-review follow-up): every OTHER counted diff --git a/test/unit/queue-concurrency.test.ts b/test/unit/queue-concurrency.test.ts new file mode 100644 index 0000000000..f92fed21e8 --- /dev/null +++ b/test/unit/queue-concurrency.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from "vitest"; + +import { mapWithConcurrency } from "../../src/queue/concurrency"; + +/** Runs `mapper` over `items` while recording the peak number of simultaneously in-flight mappers. */ +async function withPeakInFlight(items: T[], concurrency: number, work: (item: T) => Promise) { + let inFlight = 0; + let peak = 0; + const results = await mapWithConcurrency(items, concurrency, async (item) => { + inFlight += 1; + peak = Math.max(peak, inFlight); + try { + return await work(item); + } finally { + inFlight -= 1; + } + }); + return { results, peak }; +} + +/** Resolves on a later microtask turn, so overlapping mappers actually interleave. */ +const tick = () => new Promise((resolve) => setTimeout(resolve, 0)); + +describe("mapWithConcurrency() (#5835)", () => { + it("never runs more than `concurrency` mappers at once", async () => { + const items = Array.from({ length: 25 }, (_, index) => index); + const { results, peak } = await withPeakInFlight(items, 4, async (item) => { + await tick(); + return item * 2; + }); + expect(peak).toBeLessThanOrEqual(4); + // ...and it really does run them concurrently, rather than trivially serialising. + expect(peak).toBe(4); + expect(results).toEqual(items.map((item) => item * 2)); + }); + + it("preserves input order in the result even when mappers finish out of order", async () => { + // The first item resolves LAST, so an order-preserving implementation cannot just push as results arrive. + const { results } = await withPeakInFlight([0, 1, 2, 3], 4, async (item) => { + await new Promise((resolve) => setTimeout(resolve, item === 0 ? 5 : 0)); + return `item-${item}`; + }); + expect(results).toEqual(["item-0", "item-1", "item-2", "item-3"]); + }); + + it("never spawns more workers than there are items", async () => { + const { peak, results } = await withPeakInFlight([1, 2], 10, async (item) => { + await tick(); + return item; + }); + expect(peak).toBeLessThanOrEqual(2); + expect(results).toEqual([1, 2]); + }); + + it("resolves to an empty array on empty input, without hanging or calling the mapper", async () => { + // `items.length || 1` keeps the worker count at 1 rather than 0 here -- a zero-worker pool would resolve + // Promise.all([]) immediately, which is fine, but the guard also protects the `concurrency <= 0` case below. + let called = 0; + const results = await mapWithConcurrency([], 5, async (item) => { + called += 1; + return item; + }); + expect(results).toEqual([]); + expect(called).toBe(0); + }); + + it("clamps a zero/negative concurrency to a single worker instead of hanging forever", async () => { + // Math.max(1, ...) is load-bearing: 0 workers would never drain the queue and the promise would never settle. + for (const concurrency of [0, -3]) { + const { results, peak } = await withPeakInFlight([1, 2, 3], concurrency, async (item) => { + await tick(); + return item; + }); + expect(peak).toBe(1); + expect(results).toEqual([1, 2, 3]); + } + }); + + it("propagates a mapper rejection", async () => { + await expect( + mapWithConcurrency([1, 2, 3], 2, async (item) => { + if (item === 2) throw new Error("boom"); + return item; + }), + ).rejects.toThrow("boom"); + }); +}); diff --git a/test/unit/reconcile-live-duplicate-siblings.test.ts b/test/unit/reconcile-live-duplicate-siblings.test.ts index f8036cd6b1..6352a4bd82 100644 --- a/test/unit/reconcile-live-duplicate-siblings.test.ts +++ b/test/unit/reconcile-live-duplicate-siblings.test.ts @@ -154,4 +154,35 @@ describe("reconcileLiveDuplicateSiblings (#dup-winner / audit #15)", () => { expect(mockedToken).toHaveBeenCalledWith(env, 4242); expect(result.map((p) => p.number)).toEqual([]); }); + + it("REGRESSION: bounds the live per-sibling fan-out instead of bursting one fetch per sibling at once (#5835)", async () => { + // A popular linked issue can hold dozens of duplicate PRs. The old `Promise.all(overlapping.map(...))` fired + // one live REST call per sibling SIMULTANEOUSLY out of a single webhook delivery, against the ~5000/hr bucket + // every managed repo shares through one installation. The fan-out is now capped at 10, matching the two + // sibling caps in processors.ts (CONTRIBUTOR_CAP_/GLOBAL_OPEN_ITEM_LIVE_CHECK_CONCURRENCY). + const env = createTestEnv(); + env.LOOPOVER_DUPLICATE_WINNER = "true"; + + let inFlight = 0; + let peakInFlight = 0; + let totalFetches = 0; + vi.stubGlobal("fetch", async () => { + inFlight += 1; + peakInFlight = Math.max(peakInFlight, inFlight); + totalFetches += 1; + await new Promise((resolve) => setTimeout(resolve, 0)); + inFlight -= 1; + return new Response(JSON.stringify({ state: "open" }), { status: 200 }); + }); + + const siblings = Array.from({ length: 40 }, (_, index) => makePr(index + 1, "open", [1])); + const pr = makePr(999, "open", [1]); + const result = await reconcileLiveDuplicateSiblings(env, null, "owner/repo", pr, siblings); + + expect(totalFetches).toBe(40); // every sibling is still verified -- the cap bounds concurrency, not coverage + expect(peakInFlight).toBeLessThanOrEqual(10); + expect(peakInFlight).toBe(10); // and it really does saturate the cap, rather than serialising + // All 40 came back live-open, so none is dropped: same result the unbounded version produced. + expect(result).toBe(siblings); + }); });