diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 63f0814001..fafb7131c0 100644 --- a/src/queue/processors.ts +++ b/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 => { + 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) => { diff --git a/src/signals/focus-manifest-loader.ts b/src/signals/focus-manifest-loader.ts index 6737a4e092..a1e1240000 100644 --- a/src/signals/focus-manifest-loader.ts +++ b/src/signals/focus-manifest-loader.ts @@ -196,7 +196,8 @@ async function readBoundedResponseText(response: Response): Promise(items: T[], limit: number, mapper: (item: T) => Promise): Promise { +/** Bounded-concurrency fan-out: runs `mapper` over `items` with at most `limit` in flight at once (#3899). */ +export async function mapWithConcurrencyLimit(items: T[], limit: number, mapper: (item: T) => Promise): Promise { const results: U[] = new Array(items.length); let nextIndex = 0; const workers = Array.from({ length: Math.min(limit, items.length) }, async () => { diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index f61042d03d..732fe994e5 100644 --- a/test/unit/queue.test.ts +++ b/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({