diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 58d93cd09c..0f378c2317 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -4117,6 +4117,41 @@ export async function releaseAiReviewLock( await releaseTransientLockIfOwner(env, aiReviewLockKey(repoFullName, prNumber, headSha, mode), ownerToken); } +/** + * The inconclusive-hold shape a pass returns when it lost the {@link claimAiReviewLock} race (#regate-dup-prep): + * another pass already owns this exact (repo, PR, head, mode) lock, so THIS pass defers entirely rather than + * racing it. Shared by both lock-claim sites that guard runAiReviewForAdvisory's expensive section — the + * caller-side claim in maybePublishPrPublicSurface (wraps the cache-read decision itself, so a loser never even + * reaches the cache-miss log) and runAiReviewForAdvisory's own claim (the historical, narrower placement, kept for + * any other/direct caller) — so both produce byte-identical advisory findings and gate disposition instead of two + * hand-written "another pass is running" messages drifting apart over time. `persistable: false` (not merely + * `cacheable: false`): the concurrent pass this call deferred to persists the REAL result within seconds, so this + * placeholder must never be written even non-durably — a later read within the non-cacheable retry cooldown could + * otherwise replay a stale "another pass is running" long after that pass finished. + */ +function aiReviewLockContendedResult( + advisory: Pick>, "findings">, +): Awaited> { + const findings: AdvisoryFinding[] = [ + { + code: "ai_review_inconclusive", + severity: "warning", + title: "AI review already in progress for this PR head", + detail: "Another Gittensory pass is already running the AI review for this exact PR head. This pass is skipping to avoid a duplicate LLM call.", + action: "The gate is held for a human reviewer rather than passed automatically; it re-evaluates once the in-flight review completes or on the next update.", + }, + ]; + advisory.findings.push(...findings); + return { + notes: "AI review is already running for this PR head in another Gittensory pass. Gittensory is holding this PR for manual review until that pass completes.", + reviewerCount: 0, + inlineFindings: [], + findings, + cacheable: false, + persistable: false, + }; +} + /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`; * `check_run` also nests it under `check_suite.head_sha`). Returns "" when absent. The payload type doesn't model * these events, so we narrow off `Record` the same way the `pull_requests[]` read does. */ @@ -6891,6 +6926,15 @@ export async function runAiReviewForAdvisory( // self-host provider's failure log purely for operator correlation; never read by any review logic. Absent // (e.g. a sweep/repair fan-out with no single originating delivery, or a unit test) ⇒ the log line omits it. deliveryId?: string | undefined; + // A {@link claimAiReviewLock} claim the CALLER already acquired (#regate-dup-prep) before its own cache-read + // decision, so the (repo, PR, head, mode) mutex covers that cache-read too — not just this function's + // expensive section. When supplied and `.acquired`, this function trusts it, skips its OWN claim below + // entirely, and — critically — does NOT release it in its `finally` (release stays the claiming caller's job, + // so the lock keeps covering the caller's own post-return cache WRITE too; releasing here the instant this + // function returns would reopen a narrower version of the exact race this lock exists to close). Absent (the + // default, and every existing caller) ⇒ this function claims + releases its own lock exactly as before — + // byte-identical to today. + preAcquiredAiReviewLock?: TransientLockClaim | undefined; }, ): Promise< | { @@ -6974,33 +7018,21 @@ export async function runAiReviewForAdvisory( // return different verdicts. Claim before the expensive section below; a pass that loses the race returns the // same inconclusive-hold shape the "AI produced no usable verdict" path already returns, so the gate is held // (neutral) for a human rather than either pass's independently-decided verdict racing the other's cache write. - const aiReviewLock = await claimAiReviewLock( - env, - args.repoFullName, - args.pr.number, - args.advisory.headSha, - args.settings.aiReviewMode, - ); - if (!aiReviewLock.acquired) { - const findings: AdvisoryFinding[] = [ - { - code: "ai_review_inconclusive", - severity: "warning", - title: "AI review already in progress for this PR head", - detail: "Another Gittensory pass is already running the AI review for this exact PR head. This pass is skipping to avoid a duplicate LLM call.", - action: "The gate is held for a human reviewer rather than passed automatically; it re-evaluates once the in-flight review completes or on the next update.", - }, - ]; - args.advisory.findings.push(...findings); - return { - notes: "AI review is already running for this PR head in another Gittensory pass. Gittensory is holding this PR for manual review until that pass completes.", - reviewerCount: 0, - inlineFindings: [], - findings, - cacheable: false, - persistable: false, - }; - } + // #regate-dup-prep: prefer the caller's OWN claim (args.preAcquiredAiReviewLock) when it already did one — the + // caller wraps its own cache-read decision in the SAME lock key, so claiming again here would be this function + // contending against its own caller's claim (always losing) rather than against a genuinely different pass. + // Absent (every existing/direct caller) ⇒ claim it here exactly as before. + const selfClaimedAiReviewLock = args.preAcquiredAiReviewLock === undefined; + const aiReviewLock = + args.preAcquiredAiReviewLock ?? + (await claimAiReviewLock( + env, + args.repoFullName, + args.pr.number, + args.advisory.headSha, + args.settings.aiReviewMode, + )); + if (!aiReviewLock.acquired) return aiReviewLockContendedResult(args.advisory); try { // BYOK: decrypt the maintainer's provider key only for confirmed contributors when opted in. Falls back to free Workers AI when // no key is configured or the encryption secret is unavailable (getDecryptedRepositoryAiKey → null). @@ -7361,14 +7393,19 @@ export async function runAiReviewForAdvisory( }); return undefined; } finally { - await releaseAiReviewLock( - env, - args.repoFullName, - args.pr.number, - args.advisory.headSha, - args.settings.aiReviewMode, - aiReviewLock.ownerToken, - ); + // #regate-dup-prep: only release a lock THIS call actually claimed. A caller-supplied + // preAcquiredAiReviewLock must keep covering the caller's own post-return work (e.g. persisting the fresh + // review to cache) — releasing it here the instant this function returns would free the lock before that + // write happens, reopening a narrower version of the exact race this lock exists to close. + if (selfClaimedAiReviewLock) + await releaseAiReviewLock( + env, + args.repoFullName, + args.pr.number, + args.advisory.headSha, + args.settings.aiReviewMode, + aiReviewLock.ownerToken, + ); } } @@ -8851,6 +8888,54 @@ async function maybePublishPrPublicSurface( } } if (aiReviewWillRun) { + // Per-(repo, PR, head SHA, mode) advisory lock (#regate-dup-prep), claimed HERE — not just inside + // runAiReviewForAdvisory — so it covers the cache-read DECISION below too, not only the LLM call itself. + // Two near-simultaneous webhook deliveries (or a webhook racing a sweep tick) for the SAME PR at the SAME + // head can both reach this point before either has written a cache entry; without this outer claim both + // would independently run resolveReviewManifestForAiReview, the review-file load, the cache read, log an + // identical "cache miss," and only THEN contend on runAiReviewForAdvisory's own (narrower) internal claim — + // by which point the duplicate prep work (and, depending on timing, a duplicate real LLM call) already + // happened. Claiming before ANY of that means a losing pass defers immediately: it never reads the cache, + // never logs a miss, never loads review files, and never spends the fingerprint computation. A lost race + // returns the shared inconclusive-hold placeholder (same shape runAiReviewForAdvisory's own claim returns) + // rather than duplicating that work anyway — the winning pass (or the next webhook/sweep tick if the + // winner itself fails) is the backstop that populates the cache. Passed into runAiReviewForAdvisory as + // preAcquiredAiReviewLock so that function trusts this claim instead of re-claiming (and losing) against + // itself; released here, AFTER the cache write below, so the lock covers the full read-decide-run-persist + // sequence, not just the read or just the run. + const aiReviewHeadSha = advisory.headSha; + /* v8 ignore next -- defensive: aiReviewWillRun folds in shouldRequirePublicAiReviewForAdvisory's own + * `!args.advisory.headSha` guard, so a truthy aiReviewWillRun always means a truthy headSha; this narrows + * the type for claimAiReviewLock/releaseAiReviewLock (both require a non-nullish headSha) rather than + * guard against a reachable runtime state. */ + if (!aiReviewHeadSha) return; + const aiReviewLock = await claimAiReviewLock( + env, + repoFullName, + pr.number, + aiReviewHeadSha, + settings.aiReviewMode, + ); + if (!aiReviewLock.acquired) { + aiReview = aiReviewLockContendedResult(advisory); + } else { + try { + await aiReviewCacheReadDecideAndRun(aiReviewLock); + } finally { + await releaseAiReviewLock( + env, + repoFullName, + pr.number, + aiReviewHeadSha, + settings.aiReviewMode, + aiReviewLock.ownerToken, + ); + } + } + } + async function aiReviewCacheReadDecideAndRun( + aiReviewLock: TransientLockClaim, + ): Promise { await withReviewPipelineSpan( "selfhost.review.ai", { @@ -9104,6 +9189,10 @@ async function maybePublishPrPublicSurface( reviewSelfHostAiModel, reviewImpactMap, reviewCultureProfile, + // #regate-dup-prep: this call's own advisory lock is already claimed (by aiReviewCacheReadDecideAndRun's + // caller, above) — pass it through so runAiReviewForAdvisory trusts it instead of re-claiming (and + // losing) against itself, and does not release it before the cache write below runs. + preAcquiredAiReviewLock: aiReviewLock, deliveryId: webhook.deliveryId, }); // `persistable === false` (only the lock-contention placeholder — see runAiReviewForAdvisory's return diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index f2644194e3..dba96c75ae 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -5577,6 +5577,96 @@ describe("queue processors", () => { ).resolves.toBeUndefined(); auditSpy.mockRestore(); }); + + it("REPRODUCES the production incident (JSONbored/awesome-claude#4554): two near-simultaneous webhook deliveries for the SAME PR + head SHA spend AI exactly ONCE, not twice (#regate-dup-prep)", async () => { + // Root cause: two GitHub webhook deliveries ~900ms apart for the identical head SHA each independently + // reached the cache-read/cache-miss-log decision in maybePublishPrPublicSurface BEFORE either one ever + // claimed claimAiReviewLock (that claim used to live only deep inside runAiReviewForAdvisory) — so both + // logged an identical "no reusable stored AI review" cache miss and both proceeded to spend a real LLM + // call for byte-identical code. The fix claims the SAME (repo, PR, head, mode) lock in the CALLER, wrapping + // the cache-read decision itself, so the loser of the race defers before ever reading the cache — not + // merely before the LLM call. + // + // A real LLM call has genuine wall-clock latency (seconds), which is exactly the window the second + // delivery's cache-read can land inside, still finding nothing written yet. A synchronous test stub + // resolves instantly, closing that window — Promise.all alone is not enough to force the race + // deterministically (the loser's cache-read can simply land AFTER the winner's entire pass, including its + // cache WRITE, has already completed, producing an incidental cache HIT that proves nothing either way). + // Widen the window explicitly with a real setTimeout inside env.AI.run, the same "hold the window open + // long enough for others to overlap" technique this file already uses for the sweep fan-out concurrency + // regression above (#3899) — so the assertion below is a genuine two-way race, not a scheduling accident. + let aiCalls = 0; + const env = createTestEnv({ + GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem(), + AI: { + run: async () => { + aiCalls += 1; + await new Promise((resolve) => setTimeout(resolve, 20)); // hold the window open long enough for the racing delivery to reach its own cache-read + return { response: JSON.stringify({ assessment: "Looks fine.", blockers: [], nits: [], suggestions: [] }) }; + }, + } as unknown as Ai, + AI_SUMMARIES_ENABLED: "true", + AI_PUBLIC_COMMENTS_ENABLED: "true", + AI_DAILY_NEURON_BUDGET: "100000", + }); + await seedRegateChurnRepo(env); + await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 90, title: "Baseline PR", state: "open", user: { login: "contributor" }, head: { sha: "a90" }, labels: [], body: "Closes #1" }); + await upsertPullRequestDetailSyncState(env, { repoFullName: "JSONbored/gittensory", pullNumber: 90, status: "complete", reviewsSyncedAt: new Date().toISOString() }); + await upsertPullRequestFromGitHub(env, "JSONbored/gittensory", { number: 91, title: "Duplicate-delivery PR", state: "open", user: { login: "contributor" }, head: { sha: "a91" }, labels: [], body: "Closes #1" }); + await upsertPullRequestDetailSyncState(env, { repoFullName: "JSONbored/gittensory", pullNumber: 91, status: "complete", reviewsSyncedAt: new Date().toISOString() }); + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + const method = init?.method ?? "GET"; + if (url.includes("/access_tokens")) return Response.json({ token: "fake-installation-token" }); + if (url.includes("/pulls/90/files") || url.includes("/pulls/91/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]); + if (url.endsWith("/pulls/90")) return Response.json({ number: 90, title: "Baseline PR", state: "open", user: { login: "contributor" }, head: { sha: "a90" }, labels: [], body: "Closes #1", mergeable_state: "clean" }); + if (url.endsWith("/pulls/91")) return Response.json({ number: 91, title: "Duplicate-delivery PR", state: "open", user: { login: "contributor" }, head: { sha: "a91" }, labels: [], body: "Closes #1", mergeable_state: "clean" }); + if (url.includes("/commits/a90/check-runs") || url.includes("/commits/a91/check-runs")) return Response.json({ total_count: 0, check_runs: [] }); + if (url.includes("/commits/a90/status") || url.includes("/commits/a91/status")) return Response.json({ state: "success", statuses: [] }); + if (url.includes("/issues/90/comments") || url.includes("/issues/91/comments")) return method === "POST" || method === "PATCH" ? Response.json({ id: 1 }, { status: 201 }) : Response.json([]); + if (url.includes("/issues/1")) return Response.json({ number: 1, title: "Issue", state: "open", labels: [], user: { login: "reporter" } }); + if (url.includes("/branches/")) return Response.json({ protected: false, protection: { required_status_checks: { contexts: [] } } }); + return Response.json({}); + }); + + // Baseline: measure how many underlying env.AI.run calls a SINGLE genuine review attempt makes (block mode's + // dual-reviewer setup means this is not necessarily 1) — the race assertion below compares in ATTEMPTS, the + // same "attempts, not raw call counts" idiom the #9 low-activity-repo regression above uses, not a hardcoded + // call count. A separate PR/head so this baseline run's own cache write cannot affect the race below. + await processJob(env, { type: "agent-regate-pr", deliveryId: "delivery-baseline", repoFullName: "JSONbored/gittensory", prNumber: 90, installationId: 123 }); + const callsPerAttempt = aiCalls; + expect(callsPerAttempt).toBeGreaterThan(0); + aiCalls = 0; + + // Two DIFFERENT delivery ids (matching the real incident's two distinct webhook deliveries) for the SAME + // repo/PR/head, fired concurrently — neither awaits the other before both are in flight. + await Promise.all([ + processJob(env, { type: "agent-regate-pr", deliveryId: "delivery-a", repoFullName: "JSONbored/gittensory", prNumber: 91, installationId: 123 }), + processJob(env, { type: "agent-regate-pr", deliveryId: "delivery-b", repoFullName: "JSONbored/gittensory", prNumber: 91, installationId: 123 }), + ]); + + // The DISCRIMINATING assertion (fails on unfixed code, verified by temporarily reverting the fix): exactly + // ONE pass logged a genuine cache miss (read the cache, found nothing, ran fresh) — the loser never reached + // the cache-read at all, because the lock now wraps that read too, not just the LLM call. Without the fix + // this is 2: both passes independently reach the cache-read and both log a miss before either one's + // runAiReviewForAdvisory-internal lock (the historical, narrower placement) ever engages. + const missAudit = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and target_key = ?") + .bind("github_app.ai_review_cache_miss", "JSONbored/gittensory#91") + .first<{ n: number }>(); + expect(missAudit?.n).toBe(1); + + // The real review still only spent ONE attempt's worth of AI calls in total — never two full attempts (the + // production incident's duplicate LLM spend), and never zero (the winner completes a real review; the PR + // is not left permanently unreviewed because the other delivery happened to hold the lock). + expect(aiCalls).toBe(callsPerAttempt); + + // The real review was actually published (the winner completed normally) and the PR is left in a normal, + // unbroken state — the loser deferred cleanly rather than crashing processJob or leaving the PR unreviewed. + const publishedReview = await repositoriesModule.getLatestPublishedAiReview(env, "JSONbored/gittensory", 91, "block"); + expect(publishedReview).not.toBeNull(); + const pr91 = await getPullRequest(env, "JSONbored/gittensory", 91); + expect(pr91?.state).toBe("open"); + }); }); });