From c9c6928b2d2e3f06ff364f8002bd89b49de6d449 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Tue, 14 Jul 2026 12:59:31 -0700 Subject: [PATCH] fix(queue): cap backlog-convergence re-review attempts per head SHA sweepRepoBacklogConvergence re-selects any open PR whose public review surface never converged to its current head (lastPublishedSurfaceSha != headSha) -- the same signal the main sweep's outage-repair path already guards with a per-SHA attempt cap. This sweep had no such guard: when a PR's gate-check finalize silently failed while its comment still published, the marker never advanced and this sweep re-dispatched a full agent-regate-pr job for it every ~30 minutes indefinitely. Confirmed live on the three PRs an ops-anomaly detector flagged: the same sticky PR comment was edited 11-100 times over 1.3-23.5 hours, long after the head SHA had stopped changing (in one case, zero pushes for the entire window). Extract the existing repair-attempt-cap check (previously inlined in surfaceRepairPriorityPullNumbers) into isRegateRepairExhausted and share it from both sweeps against the SAME per-(repo, PR, headSha) budget -- deliberately not an independent counter, since both sweeps re-select on the identical signal and would otherwise double the wasted spend the cap exists to prevent. Passing repairHeadSha on backlog-convergence's dispatched jobs reuses the existing execution-time (not dispatch-time) attempt recording, so a deferred or dropped job still doesn't count against the cap. --- src/queue/processors.ts | 102 ++++++++++++++++++++++---------------- test/unit/queue-2.test.ts | 78 +++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 43 deletions(-) diff --git a/src/queue/processors.ts b/src/queue/processors.ts index b89c4b9aa9..905e965628 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1052,6 +1052,49 @@ function regateRepairTargetKey(repoFullName: string, prNumber: number, headSha: return `${repoFullName}#${prNumber}#${headSha}`; } +/** + * True when `pr`'s current head SHA has already exhausted REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA repair attempts + * within the lookback window. Records (at most once per SHA) the exhausted audit event + error-level log as a + * side effect the first time a SHA crosses the cap. Shared by both surfaceRepairPriorityPullNumbers and + * sweepRepoBacklogConvergence (#orb-retry-storm, backlog-convergence half): both sweeps re-select on the + * identical lastPublishedSurfaceSha-mismatch signal, so they share one attempt budget per SHA rather than each + * independently hammering the same stuck PR. + */ +async function isRegateRepairExhausted(env: Env, repoFullName: string, pr: Pick): Promise { + const headSha = pr.headSha!; + const sinceIso = new Date(Date.now() - REGATE_REPAIR_ATTEMPT_LOOKBACK_MS).toISOString(); + const targetKey = regateRepairTargetKey(repoFullName, pr.number, headSha); + const attempts = await countRecentAuditEventsForActorAndTarget(env, "loopover", REGATE_REPAIR_ATTEMPT_EVENT_TYPE, targetKey, sinceIso); + if (attempts < REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA) return false; + const alreadyFlagged = await countRecentAuditEventsForActorAndTarget(env, "loopover", REGATE_REPAIR_EXHAUSTED_EVENT_TYPE, targetKey, sinceIso); + if (alreadyFlagged === 0) { + await recordAuditEvent(env, { + eventType: REGATE_REPAIR_EXHAUSTED_EVENT_TYPE, + actor: "loopover", + targetKey, + outcome: "denied", + detail: `re-gate repair exhausted after ${attempts} attempt(s) for the same head SHA; falling back to ordinary staleness cadence`, + metadata: { repoFullName, prNumber: pr.number, headSha, attempts }, + }); + // level:"error" is deliberate, not a code failure: this line only fires once the cap above already + // stopped the wasteful repair loop, so its OWN existence is the operator-visible signal (via the + // structured log → Sentry forwarder, forwardStructuredLogToSentry) that a PR kept failing repair for the + // same head SHA — the same "surface an anomaly at error level" convention selfhost_ai_provider_failed / + // selfhost_ai_providers_exhausted already use in src/selfhost/ai.ts. + console.error( + JSON.stringify({ + level: "error", + event: "regate_repair_exhausted", + repo: repoFullName, + pullNumber: pr.number, + headSha, + attempts, + }), + ); + } + return true; +} + async function surfaceRepairPriorityPullNumbers( env: Env, repoFullName: string, @@ -1080,53 +1123,12 @@ async function surfaceRepairPriorityPullNumbers( }), ); } - const sinceIso = new Date(Date.now() - REGATE_REPAIR_ATTEMPT_LOOKBACK_MS).toISOString(); await Promise.all( [...priorityPullNumbers].map(async (prNumber) => { const pr = pulls.find((candidate) => candidate.number === prNumber); /* v8 ignore next -- priorityPullNumbers is only ever populated (both loops above) from a `pr` in `pulls` that already had a truthy headSha, so this lookup always succeeds with one; the guard only satisfies Array#find's `| undefined` return type. */ if (!pr?.headSha) return; - const targetKey = regateRepairTargetKey(repoFullName, pr.number, pr.headSha); - const attempts = await countRecentAuditEventsForActorAndTarget( - env, - "loopover", - REGATE_REPAIR_ATTEMPT_EVENT_TYPE, - targetKey, - sinceIso, - ); - if (attempts < REGATE_REPAIR_MAX_ATTEMPTS_PER_SHA) return; - priorityPullNumbers.delete(prNumber); - const alreadyFlagged = await countRecentAuditEventsForActorAndTarget( - env, - "loopover", - REGATE_REPAIR_EXHAUSTED_EVENT_TYPE, - targetKey, - sinceIso, - ); - if (alreadyFlagged > 0) return; - await recordAuditEvent(env, { - eventType: REGATE_REPAIR_EXHAUSTED_EVENT_TYPE, - actor: "loopover", - targetKey, - outcome: "denied", - detail: `re-gate repair exhausted after ${attempts} attempt(s) for the same head SHA; falling back to ordinary staleness cadence`, - metadata: { repoFullName, prNumber: pr.number, headSha: pr.headSha, attempts }, - }); - // level:"error" is deliberate, not a code failure: this line only fires once the cap above already - // stopped the wasteful repair loop, so its OWN existence is the operator-visible signal (via the - // structured log → Sentry forwarder, forwardStructuredLogToSentry) that a PR kept failing repair for the - // same head SHA — the same "surface an anomaly at error level" convention selfhost_ai_provider_failed / - // selfhost_ai_providers_exhausted already use in src/selfhost/ai.ts. - console.error( - JSON.stringify({ - level: "error", - event: "regate_repair_exhausted", - repo: repoFullName, - pullNumber: pr.number, - headSha: pr.headSha, - attempts, - }), - ); + if (await isRegateRepairExhausted(env, repoFullName, pr)) priorityPullNumbers.delete(prNumber); }), ); return [...priorityPullNumbers]; @@ -1724,7 +1726,15 @@ export async function sweepRepoBacklogConvergence( const sweepInstallationId = repo?.installationId ?? null; if (sweepInstallationId == null) return; const openPullRequests = await listOpenPullRequests(env, repoFullName); - const candidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests }); + const allCandidates = selectBacklogConvergenceCandidates({ pulls: openPullRequests }); + // #orb-retry-storm (backlog-convergence half): needsSurfaceConvergence re-fires on the exact same + // lastPublishedSurfaceSha-mismatch signal as the main sweep's outage-repair priority path, but this sweeper + // had no memory of prior attempts at all -- a PR whose gate-check finalize kept failing silently got a fresh + // full re-review dispatched every ~30 minutes indefinitely. Share the same per-SHA attempt budget as the main + // sweep (isRegateRepairExhausted) rather than adding an independent cap, since both sweeps competing for the + // same stuck PR would otherwise double the wasted spend the cap exists to prevent. + const exhaustedFlags = await Promise.all(allCandidates.map((pr) => isRegateRepairExhausted(env, repoFullName, pr))); + const candidates = allCandidates.filter((_pr, index) => !exhaustedFlags[index]); if (candidates.length === 0) return; // Stamp the backlog-convergence draining marker for EVERY candidate NOW, at dispatch — not in the downstream // per-PR job (#4502, mirrors #audit-sweep-dispatch-stamp). This makes getLatestBacklogConvergenceRegatedAt @@ -1752,6 +1762,12 @@ export async function sweepRepoBacklogConvergence( repoFullName, prNumber: pr.number, installationId: sweepInstallationId, + // #orb-retry-storm: selectBacklogConvergenceCandidates only returns PRs needsSurfaceConvergence already + // confirmed have a truthy headSha, so this is unconditional (unlike the main sweep's priority-repair + // dispatch, which mixes repair and ordinary candidates). Passing it lets regatePullRequest record the + // attempt at execution time against the SAME shared per-SHA budget isRegateRepairExhausted checked + // above -- jobs deferred or dropped before they run still don't count against the cap. + repairHeadSha: pr.headSha!, ...(pr.createdAt ? { prCreatedAt: pr.createdAt } : {}), }; const delaySeconds = Math.min(index * 10, 600); diff --git a/test/unit/queue-2.test.ts b/test/unit/queue-2.test.ts index 6306762608..02837fc596 100644 --- a/test/unit/queue-2.test.ts +++ b/test/unit/queue-2.test.ts @@ -1933,6 +1933,84 @@ describe("queue processors", () => { } }, 60_000); + it("REGRESSION (#orb-retry-storm, backlog-convergence half): a PR whose head SHA already exhausted the shared repair-attempt budget is not re-dispatched by the backlog-convergence sweep", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertInstallation(env, { action: "created", installation: { id: 9412, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo4", full_name: "owner/agent-repo4", private: false, owner: { login: "owner" } }, 9412); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo4", autonomy: { merge: "auto" }, reviewCheckMode: "required", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + // Never marked surface-published for this head -- needsSurfaceConvergence is true, a genuine backlog-convergence candidate. + await upsertPullRequestFromGitHub(env, "owner/agent-repo4", { number: 1, title: "Stuck convergence", state: "open", user: { login: "c" }, head: { sha: "stuck-sha" }, labels: [], body: "" }); + const targetKey = "owner/agent-repo4#1#stuck-sha"; + // Pre-seed the SAME shared budget the main sweep's repair path charges against -- this is the whole point of + // sharing isRegateRepairExhausted rather than giving backlog-convergence its own independent counter. + for (let i = 0; i < 5; i += 1) { + await repositoriesModule.recordAuditEvent(env, { + eventType: "agent.sweep.regate.repair_attempt", + actor: "loopover", + targetKey, + outcome: "queued", + detail: "prior attempt", + metadata: {}, + }); + } + vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); + const errors = vi.spyOn(console, "error").mockImplementation(() => undefined); + + try { + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo4" }); + + const fanned = sent.filter((job): job is Extract => job.type === "agent-regate-pr"); + expect(fanned.every((job) => job.deliveryId !== "backlog-convergence:owner/agent-repo4#1")).toBe(true); + const exhausted = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") + .bind("agent.sweep.regate.repair_exhausted", targetKey) + .first<{ n: number }>(); + expect(exhausted?.n).toBe(1); + const exhaustedLogs = errors.mock.calls.filter(([line]) => typeof line === "string" && line.includes("regate_repair_exhausted")); + expect(exhaustedLogs).toHaveLength(1); + } finally { + errors.mockRestore(); + } + }, 60_000); + + it("a fresh backlog-convergence candidate under the cap is dispatched with repairHeadSha set, and executing it charges the SAME shared attempt budget the main sweep's repair path reads", async () => { + const sent: import("../../src/types").JobMessage[] = []; + const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); + await upsertInstallation(env, { action: "created", installation: { id: 9413, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo5", full_name: "owner/agent-repo5", private: false, owner: { login: "owner" } }, 9413); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo5", autonomy: { merge: "auto" }, aiReviewMode: "off", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo5", { number: 2, title: "Fresh convergence", state: "open", user: { login: "c" }, head: { sha: "fresh-sha" }, base: { ref: "main" }, labels: [], body: "" }); + const targetKey = "owner/agent-repo5#2#fresh-sha"; + vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); + vi.stubGlobal("fetch", async (input: RequestInfo | URL) => { + const url = input.toString(); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (/\/pulls\/2(?:\?|$)/.test(url)) return Response.json({ number: 2, title: "Fresh convergence", state: "open", user: { login: "c" }, head: { sha: "fresh-sha" }, mergeable_state: "clean", labels: [], body: "" }); + if (url.includes("/pulls/2/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]); + return Response.json({}); + }); + + await processJob(env, { type: "backlog-convergence-sweep", requestedBy: "test", repoFullName: "owner/agent-repo5" }); + + const fanned = sent.filter((job): job is Extract => job.type === "agent-regate-pr"); + const dispatched = fanned.find((job) => job.deliveryId === "backlog-convergence:owner/agent-repo5#2"); + expect(dispatched).toMatchObject({ repoFullName: "owner/agent-repo5", prNumber: 2, repairHeadSha: "fresh-sha" }); + + const attemptsBefore = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") + .bind("agent.sweep.regate.repair_attempt", targetKey) + .first<{ n: number }>(); + expect(attemptsBefore?.n).toBe(0); + + await processJob(env, dispatched!); + + // The SAME event type/target key the main sweep's repair path reads -- proves the two sweeps share one + // budget for this SHA instead of each independently retrying it. + const attemptsAfter = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") + .bind("agent.sweep.regate.repair_attempt", targetKey) + .first<{ n: number }>(); + expect(attemptsAfter?.n).toBe(1); + }, 60_000); + it("REGRESSION (#5385-sentry, GITTENSORY-1E gate-finding): a retryable GitHub rate-limit error surfacing from real post-readiness review work STILL records the repair attempt before propagating for the queue's own retry", async () => { // LoopOver review finding on PR #5482: the original fix recorded the attempt AFTER reReviewStoredPullRequest // returns, so a retryable error thrown from a genuinely-executed (post-readiness) pass never got charged --