diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 1c6968eae6..c95acd5caa 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1258,27 +1258,44 @@ async function fanOutAgentRegateSweepJobs( } const configured: Array<{ fullName: string; installationId?: number }> = []; let skippedDraining = 0; + let skippedErrored = 0; for (const repo of byKey.values()) { const repoFullName = repo.fullName; - const settings = await resolveRepositorySettings(env, repoFullName); - if ( - !( - isConvergenceRepoAllowed(env, repoFullName) || - isAgentConfigured(settings.autonomy) + // #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; + 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), + }), + ); } - configured.push(repo); } await Promise.all( configured.map((repo, index) => { @@ -1289,15 +1306,27 @@ async function fanOutAgentRegateSweepJobs( ...(typeof repo.installationId === "number" ? { installationId: repo.installationId } : {}), }; const delaySeconds = Math.min(index * 10, 600); - return delaySeconds > 0 - ? env.JOBS.send(message, { delaySeconds }) - : env.JOBS.send(message); + // #audit-sweep-fanout-isolation: one repo's dispatch failure (a transient queue-send error) must not reject + // this Promise.all and, with it, abort every OTHER repo's already-in-flight send AND the audit event below + // that records this fan-out's outcome. Swallow + log per-repo instead; a repo that fails to dispatch here + // simply gets picked up again next tick (it never got its convergence marker stamped, so it stays eligible). + const send = delaySeconds > 0 ? env.JOBS.send(message, { delaySeconds }) : env.JOBS.send(message); + return send.catch((error) => { + console.error( + JSON.stringify({ + level: "error", + event: "sweep_fanout_dispatch_failed", + repository: repo.fullName, + error: errorMessage(error), + }), + ); + }); }), ); await recordAuditEvent(env, { eventType: "agent.sweep.fanout", outcome: "queued", - metadata: { repoCount: configured.length, skippedDraining, requestedBy }, + metadata: { repoCount: configured.length, skippedDraining, skippedErrored, requestedBy }, }); } diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 702d222aa4..990bc79183 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -768,6 +768,61 @@ describe("queue processors", () => { expect(sent).toEqual(expect.arrayContaining([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/advisory-repo", installationId: 9102 })])); }); + it("REGRESSION (#audit-sweep-fanout-isolation): one repo's settings-check failure does not abort the fan-out for every other repo", 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, + }); + await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } }); + await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } }); + const realResolve = repositorySettingsModule.resolveRepositorySettings; + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + const resolveSpy = vi.spyOn(repositorySettingsModule, "resolveRepositorySettings").mockImplementation(async (e, repoFullName) => { + if (repoFullName === "owner/agent-a") throw new Error("D1 read error"); + return realResolve(e, repoFullName); + }); + + await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" }); + + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failure did not block agent-b + expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_fanout_repo_check_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true); + const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.fanout").first<{ outcome: string; metadata_json: string }>(); + expect(fanout?.outcome).toBe("queued"); // the fan-out still completes and records its own outcome + expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 1, skippedErrored: 1 }); + errors.mockRestore(); + resolveSpy.mockRestore(); + }); + + it("REGRESSION (#audit-sweep-fanout-isolation): one repo's dispatch failure does not abort dispatch for every other repo, and the fan-out audit event still records", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ + GITTENSORY_REVIEW_REPOS: "", + JOBS: { + async send(m: import("../../src/types").JobMessage) { + if (m.type === "agent-regate-sweep" && m.repoFullName === "owner/agent-a") throw new Error("queue send error"); + sent.push(m); + }, + } as unknown as Queue, + }); + await upsertRepositoryFromGitHub(env, { name: "agent-a", full_name: "owner/agent-a", private: false, owner: { login: "owner" } }); + await upsertRepositoryFromGitHub(env, { name: "agent-b", full_name: "owner/agent-b", private: false, owner: { login: "owner" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-a", autonomy: { label: "auto" } }); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-b", autonomy: { label: "auto" } }); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await processJob(env, { type: "agent-regate-sweep", requestedBy: "schedule" }); + + expect(sent).toEqual([expect.objectContaining({ type: "agent-regate-sweep", repoFullName: "owner/agent-b" })]); // agent-a's failed send did not block agent-b's + expect(errors.mock.calls.some((call) => String(call[0]).includes("sweep_fanout_dispatch_failed") && String(call[0]).includes("owner/agent-a"))).toBe(true); + const fanout = await env.DB.prepare("select outcome, metadata_json from audit_events where event_type = ?").bind("agent.sweep.fanout").first<{ outcome: string; metadata_json: string }>(); + expect(fanout?.outcome).toBe("queued"); // reached — the dispatch failure did not throw the fan-out itself + expect(JSON.parse(fanout?.metadata_json ?? "{}")).toMatchObject({ repoCount: 2 }); // both PASSED their settings/draining checks regardless of dispatch outcome + errors.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({