queue: dedupe backfill mapWithConcurrency helper

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<T, R>(items: T[], concurrency: number, mapper: (item: T, index: number) => 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, 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

src/queue/map-with-concurrency.ts

@@ -1,7 +1,7 @@
 export async function mapWithConcurrency<T, R>(
   items: T[],
   concurrency: number,
-  mapper: (item: T) => Promise<R>,
+  mapper: (item: T, index: number) => Promise<R>,
 ): Promise<R[]> {
   const results: R[] = new Array(items.length);
   let nextIndex = 0;
@@ -11,7 +11,7 @@ export async function mapWithConcurrency<T, R>(
       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);
       }
     }),
   );

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]);
+  });
+});