perf(queue): parallelize regate-sweep per-repo settings resolution

src/queue/processors.ts

@@ -387,6 +387,7 @@ import {
   loadRepoFocusManifest,
   loadRepoFocusManifests,
   loadRepoReviewContext,
+  mapWithConcurrencyLimit,
 } from "../signals/focus-manifest-loader";
 import { resolveRepositorySettings } from "../settings/repository-settings";
 import { getLastRepoDocRefreshAttemptedAtBulk, performRepoDocRefresh } from "../github/repo-doc-refresh-runner";
@@ -1243,6 +1244,16 @@ async function fanOutRepoSignalSnapshotJobs(
   });
 }
 
+// Bounded concurrency for the per-repo settings+drain-state resolution below (#3899) — matches
+// REPO_FOCUS_MANIFEST_MAX_CONCURRENT_LOADS, the same "many small per-repo D1/KV reads" shape.
+export const SWEEP_FANOUT_RESOLUTION_CONCURRENCY = 4;
+
+type SweepFanoutResolutionOutcome =
+  | { kind: "ineligible" }
+  | { kind: "draining" }
+  | { kind: "configured"; repo: { fullName: string; installationId?: number } }
+  | { kind: "errored" };
+
 // #777 scheduled re-gate sweep. The cron (index.ts) enqueues one fan-out job hourly; this enqueues a per-repo
 // sweep job for every repo that opted the agent in (an acting autonomy level). Mirrors the signal-snapshot
 // fan-out so each repo's sweep runs as its own bounded, retryable queue message.
@@ -1281,46 +1292,54 @@ async function fanOutAgentRegateSweepJobs(
       ...(typeof repo?.installationId === "number" ? { installationId: repo.installationId } : {}),
     });
   }
+  // #3899: resolve every repo's settings + drain-state CONCURRENTLY (bounded), not one at a time. Each repo
+  // costs resolveRepositorySettings's own 3 parallel round-trips plus a 4th getLatestRegatedAt read; awaiting
+  // that serially per repo made this whole prefix scale linearly with repo count, before the per-repo dispatch
+  // below (already parallel) even started. Reuses the same bounded worker-pool helper loadRepoFocusManifests
+  // already relies on for the same "many small per-repo D1/KV reads" shape.
+  const outcomes = await mapWithConcurrencyLimit(
+    [...byKey.values()],
+    SWEEP_FANOUT_RESOLUTION_CONCURRENCY,
+    async (repo): Promise<SweepFanoutResolutionOutcome> => {
+      const repoFullName = repo.fullName;
+      // #audit-sweep-fanout-isolation: one repo's settings/draining-check failure (a transient D1 read error, say)
+      // must not throw and abort resolution for every OTHER repo's independent worker — return an "errored"
+      // outcome for just this repo (it gets picked up again next tick) instead of rejecting.
+      try {
+        const settings = await resolveRepositorySettings(env, repoFullName);
+        if (
+          !(
+            isConvergenceRepoAllowed(env, repoFullName) ||
+            isAgentConfigured(settings.autonomy)
+          )
+        )
+          return { kind: "ineligible" };
+        // In-flight guard (#audit-sweep-fanout): skip a repo whose prior sweep is still draining — its per-PR jobs are
+        // mid-flight and stamping last_regated_at as they run, so the freshest stamp being within the sweep window
+        // means a sweep is active. Re-arming now would enqueue duplicate per-PR jobs for the not-yet-drained
+        // candidates, so this is what finally stops the 2-min cron piling a second full sweep on an unfinished one.
+        if (isRegateSweepDraining(await getLatestRegatedAt(env, repoFullName), now)) return { kind: "draining" };
+        return { kind: "configured", repo };
+      } catch (error) {
+        console.error(
+          JSON.stringify({
+            level: "error",
+            event: "sweep_fanout_repo_check_failed",
+            repository: repoFullName,
+            error: errorMessage(error),
+          }),
+        );
+        return { kind: "errored" };
+      }
+    },
+  );
   const configured: Array<{ fullName: string; installationId?: number }> = [];
   let skippedDraining = 0;
   let skippedErrored = 0;
-  for (const repo of byKey.values()) {
-    const repoFullName = repo.fullName;
-    // #audit-sweep-fanout-isolation: one repo's settings/draining-check failure (a transient D1 read error, say)
-    // must not throw out of this loop and abort the fan-out for EVERY OTHER already-iterated-and-pending repo —
-    // it previously did, since an uncaught throw here escapes the whole function before the dispatch loop below
-    // ever runs. Skip just this repo (it gets picked up again next tick) and keep going.
-    try {
-      const settings = await resolveRepositorySettings(env, repoFullName);
-      if (
-        !(
-          isConvergenceRepoAllowed(env, repoFullName) ||
-          isAgentConfigured(settings.autonomy)
-        )
-      )
-        continue;
-      // In-flight guard (#audit-sweep-fanout): skip a repo whose prior sweep is still draining — its per-PR jobs are
-      // mid-flight and stamping last_regated_at as they run, so the freshest stamp being within the sweep window
-      // means a sweep is active. Re-arming now would enqueue duplicate per-PR jobs for the not-yet-drained
-      // candidates, so this is what finally stops the 2-min cron piling a second full sweep on an unfinished one.
-      if (
-        isRegateSweepDraining(await getLatestRegatedAt(env, repoFullName), now)
-      ) {
-        skippedDraining += 1;
-        continue;
-      }
-      configured.push(repo);
-    } catch (error) {
-      skippedErrored += 1;
-      console.error(
-        JSON.stringify({
-          level: "error",
-          event: "sweep_fanout_repo_check_failed",
-          repository: repoFullName,
-          error: errorMessage(error),
-        }),
-      );
-    }
+  for (const outcome of outcomes) {
+    if (outcome.kind === "configured") configured.push(outcome.repo);
+    else if (outcome.kind === "draining") skippedDraining += 1;
+    else if (outcome.kind === "errored") skippedErrored += 1;
   }
   await Promise.all(
     configured.map((repo, index) => {

src/signals/focus-manifest-loader.ts

@@ -196,7 +196,8 @@ async function readBoundedResponseText(response: Response): Promise<string | nul
   }
 }
 
-async function mapWithConcurrencyLimit<T, U>(items: T[], limit: number, mapper: (item: T) => Promise<U>): Promise<U[]> {
+/** Bounded-concurrency fan-out: runs `mapper` over `items` with at most `limit` in flight at once (#3899). */
+export async function mapWithConcurrencyLimit<T, U>(items: T[], limit: number, mapper: (item: T) => Promise<U>): Promise<U[]> {
   const results: U[] = new Array(items.length);
   let nextIndex = 0;
   const workers = Array.from({ length: Math.min(limit, items.length) }, async () => {

test/unit/queue.test.ts

@@ -53,7 +53,7 @@ import {
   markAiReviewPublished,
   recordReviewSuppression,
 } from "../../src/db/repositories";
-import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, enrichOpenPullRequestsWithChangedFiles, processJob, reconcileLiveDuplicateSiblings, releaseAiReviewLock, releasePrActuationLock } from "../../src/queue/processors";
+import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, enrichOpenPullRequestsWithChangedFiles, processJob, reconcileLiveDuplicateSiblings, releaseAiReviewLock, releasePrActuationLock, SWEEP_FANOUT_RESOLUTION_CONCURRENCY } from "../../src/queue/processors";
 import type { PullRequestRecord } from "../../src/types";
 import { aiReviewCacheInputFingerprint } from "../../src/review/ai-review-cache-input";
 import { fingerprint as reviewMemoryFingerprint } from "../../src/review/review-memory-match";
@@ -825,6 +825,37 @@ describe("queue processors", () => {
     errors.mockRestore();
   });
 
+  it("REGRESSION (#3899): resolves multiple repos' settings/drain-state CONCURRENTLY, bounded by SWEEP_FANOUT_RESOLUTION_CONCURRENCY", async () => {
+    const sent: import("../../src/types").JobMessage[] = [];
+    const env = createTestEnv({
+      GITTENSORY_REVIEW_REPOS: "",
+      JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue,
+    });
+    const repoNames = ["r1", "r2", "r3", "r4", "r5", "r6"];
+    for (const name of repoNames) {
+      await upsertRepositoryFromGitHub(env, { name, full_name: `owner/${name}`, private: false, owner: { login: "owner" } });
+      await upsertRepositorySettings(env, { repoFullName: `owner/${name}`, autonomy: { label: "auto" } });
+    }
+    const realResolve = repositorySettingsModule.resolveRepositorySettings;
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const resolveSpy = vi.spyOn(repositorySettingsModule, "resolveRepositorySettings").mockImplementation(async (e, repoFullName) => {
+      inFlight += 1;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await new Promise((resolve) => setTimeout(resolve, 5)); // hold the window open long enough for others to overlap
+      const result = await realResolve(e, repoFullName);
+      inFlight -= 1;
+      return result;
+    });
+
+    await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" });
+
+    expect(maxInFlight).toBeGreaterThan(1); // proves real overlap — not the old strictly-sequential loop
+    expect(maxInFlight).toBeLessThanOrEqual(SWEEP_FANOUT_RESOLUTION_CONCURRENCY); // proves BOUNDED, not unlimited fan-out
+    expect(sent.filter((m) => m.type === "agent-regate-sweep").length).toBe(repoNames.length); // every repo still dispatched
+    resolveSpy.mockRestore();
+  });
+
   it("agent re-gate sweep recomputes stale open PR verdicts as an advisory audit, never publishing (#777)", async () => {
     const sent: import("../../src/types").JobMessage[] = [];
     const env = createTestEnv({