diff --git a/src/github/backfill.ts b/src/github/backfill.ts index 1cc1dc807..86388f47e 100644 --- a/src/github/backfill.ts +++ b/src/github/backfill.ts @@ -86,6 +86,7 @@ import { import { fetchCachedGitHubGraphQl } from "./graphql-cache"; import { incr } from "../selfhost/metrics"; import { fetchBrokeredInstallationToken, isOrbBrokerMode } from "../orb/broker-client"; +import { mapWithConcurrency } from "../queue/map-with-concurrency"; type GitHubLabelPayload = { name: string; color?: string; @@ -5196,22 +5197,6 @@ function parseNullableInt(value: string | null): number | undefined { return Number.isFinite(parsed) ? parsed : undefined; } -async function mapWithConcurrency(items: T[], concurrency: number, mapper: (item: T, index: number) => 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, index); - } - }), - ); - return results; -} - // Mirror of app.ts's isRateLimitedResponse, reconstructed from the status, rate-limit headers, and body that // a backfill REST/GraphQL failure carries. A 403/429 signals a rate limit ONLY when it has a Retry-After // header, an exhausted x-ratelimit-remaining, or a secondary-limit/abuse body. A bare 403 — "Resource not diff --git a/src/queue/map-with-concurrency.ts b/src/queue/map-with-concurrency.ts index 97e80556a..aa020a27d 100644 --- a/src/queue/map-with-concurrency.ts +++ b/src/queue/map-with-concurrency.ts @@ -1,7 +1,7 @@ export async function mapWithConcurrency( items: T[], concurrency: number, - mapper: (item: T) => Promise, + mapper: (item: T, index: number) => Promise, ): Promise { const results: R[] = new Array(items.length); let nextIndex = 0; @@ -11,7 +11,7 @@ export async function mapWithConcurrency( while (nextIndex < items.length) { const index = nextIndex; nextIndex += 1; - results[index] = await mapper(items[index] as T); + results[index] = await mapper(items[index] as T, index); } }), ); diff --git a/test/unit/map-with-concurrency.test.ts b/test/unit/map-with-concurrency.test.ts new file mode 100644 index 000000000..85eef0d26 --- /dev/null +++ b/test/unit/map-with-concurrency.test.ts @@ -0,0 +1,23 @@ +import { describe, expect, it } from "vitest"; +import { mapWithConcurrency } from "../../src/queue/map-with-concurrency"; + +describe("mapWithConcurrency", () => { + it("passes each mapper the stable item index with concurrent workers", async () => { + const seenByItem: number[] = []; + + const result = await mapWithConcurrency(["a", "b", "c", "d"], 3, async (item, index) => { + seenByItem[index] = index; + await Promise.resolve(); + return `${index}:${item}`; + }); + + expect(seenByItem).toEqual([0, 1, 2, 3]); + expect(result).toEqual(["0:a", "1:b", "2:c", "3:d"]); + }); + + it("keeps one-argument mappers source-compatible", async () => { + const result = await mapWithConcurrency([1, 2, 3], 2, async (item) => item * 2); + + expect(result).toEqual([2, 4, 6]); + }); +});