diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 7f03845843..343c32cd9c 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1704,7 +1704,7 @@ async function maybeRunAgentMaintenance( // critical section (extracted below so the try/finally doesn't force-reindent that whole block); a pass that // loses the race defers cleanly — the next webhook/sweep tick is the backstop. Lightweight stand-in for the // per-PR SubmissionLock Durable Object noted as a longer-term TODO in env.d.ts. - if (!(await claimAgentMaintenanceLock(env, repoFullName, pr.number))) return; + if (!(await claimPrActuationLock(env, repoFullName, pr.number))) return; try { await runAgentMaintenancePlanAndExecute(env, { installationId, @@ -1718,7 +1718,7 @@ async function maybeRunAgentMaintenance( liveFacts: args.liveFacts, }); } finally { - await releaseAgentMaintenanceLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number); } } @@ -2305,17 +2305,23 @@ async function putTransientKey( } } -// Per-PR actuation mutex (#2135). Two DIFFERENT webhook deliveries for the same PR (e.g. a `reopened` event and -// a concurrent `check_suite completed` event) can be dequeued by separate workers at nearly the same time; both -// would read the same stale-but-still-"current" state, both pass their own freshness checks, and both -// independently fire a mutating call. This is a lightweight interim mutex (a full per-PR Durable Object / -// SubmissionLock is a separate, more-involved follow-up — see the TODO in env.d.ts) built on the SAME transient -// cache used for CI-completion coalescing above, claimed ATOMICALLY (see claimTransientLock) so two racing -// deliveries can never both win the claim — a short TTL, best-effort release. A lock-contended caller fails -// OPEN (returns false / skips this pass) rather than blocking — the delivery holding the lock is evaluating -// the SAME PR, and the periodic sweep is the backstop if this specific trigger is dropped. A cache adapter with -// no claim() primitive gets NO exclusivity at all (every call proceeds) rather than a get-then-set pair that -// only *looks* atomic — see claimTransientLock's doc comment for why that fallback was removed. +// ONE shared per-PR actuation mutex (#2129/#2135) for every mutating PR pass: the sweep/webhook-driven +// maintenance plan-and-execute, the draft-dodge close, and the reopen-reclose. These are three INDEPENDENTLY +// triggered webhook/sweep paths for the SAME PR (e.g. a `reopened` event and a concurrent `check_suite +// completed` event, or a sweep tick racing either) that can be dequeued by separate workers at nearly the same +// time; each would read its own stale-but-still-"current" state, each would pass its own freshness checks, and +// each could independently fire a mutating call for the same PR. A single lock namespace is deliberate: separate +// per-path locks (the original design) do not exclude each other, so a maintenance pass and a draft-dodge close +// could still race — the whole point of this mutex is to make "does something else already own this PR" one +// question with one answer, not one question per code path (review round 4). This is a lightweight interim +// mutex (a full per-PR Durable Object / SubmissionLock is a separate, more-involved follow-up — see the TODO in +// env.d.ts) built on the SAME transient cache used for CI-completion coalescing above, claimed ATOMICALLY (see +// claimTransientLock) so two racing deliveries can never both win the claim — a short TTL, best-effort release. +// A lock-contended caller fails OPEN (returns false / skips this pass) rather than blocking — the delivery +// holding the lock is evaluating the SAME PR, and the periodic sweep is the backstop if this specific trigger is +// dropped. A cache adapter with no claim() primitive gets NO exclusivity at all (every call proceeds) rather +// than a get-then-set pair that only *looks* atomic — see claimTransientLock's doc comment for why that fallback +// was removed. // // KNOWN LIMITATION: the lock value is a constant, not a per-holder ownership token, so release does not verify // it still owns the key — if a holder ran past the TTL, a later claimer's live lock could be deleted by the @@ -2323,7 +2329,8 @@ async function putTransientKey( // token + a conditional (check-then-delete) release would close this properly, but needs a new atomic // compare-and-delete primitive on the cache adapter — tracked alongside the Durable Object follow-up above. The // TTL is set generously long specifically so this window is practically unreachable: the guarded operations -// (a handful of sequential GitHub API calls) should never legitimately run anywhere near this long. +// (a handful of sequential GitHub API calls, or a maintenance pass's plan-and-execute) should never legitimately +// run anywhere near this long. const PR_ACTUATION_LOCK_TTL_SECONDS = 600; function prActuationLockKey(repoFullName: string, prNumber: number): string { return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`; @@ -2520,53 +2527,14 @@ async function claimTransientLock( } } -// Per-PR advisory lock around maybeRunAgentMaintenance's plan-and-execute critical section (#2129). The TTL is a -// crash-safety backstop only — the normal path releases explicitly in a finally block within a few seconds — so -// it is sized well above any realistic pass duration (matches CI_COALESCE_WINDOW_SECONDS, an already-vetted -// value for a comparable-scale operation in this file), not to bound throughput. -const AGENT_MAINTENANCE_LOCK_TTL_SECONDS = 60; - -function agentMaintenanceLockKey(repoFullName: string, prNumber: number): string { - return `agent-maintenance-lock:${repoFullName.toLowerCase()}#${prNumber}`; -} - -/** - * Claim the per-PR advisory lock. Returns false when another pass already holds it (caller must skip this pass - * — the next webhook/sweep tick is the backstop). A missing cache or cache hiccup fails OPEN (returns true — - * the lock is a defense-in-depth serializer, not the primary safety gate, and must never itself block actuation). - */ -export async function claimAgentMaintenanceLock( - env: Env, - repoFullName: string, - prNumber: number, -): Promise { - return claimTransientLock( - env, - agentMaintenanceLockKey(repoFullName, prNumber), - AGENT_MAINTENANCE_LOCK_TTL_SECONDS, - ); -} - -/** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */ -export async function releaseAgentMaintenanceLock( - env: Env, - repoFullName: string, - prNumber: number, -): Promise { - try { - await env.SELFHOST_TRANSIENT_CACHE?.del?.( - agentMaintenanceLockKey(repoFullName, prNumber), - ); - } catch { - // best-effort; the TTL is the backstop if release fails - } -} - // Per-(repo, PR, head SHA) advisory lock around runAiReviewForAdvisory's expensive grounding/RAG/enrichment/LLM // section (#confirmed-bug: a webhook pass and an agent-regate-pr sweep pass can independently reach this same // code for the SAME PR at the SAME head SHA, both miss the cache, and both fire a real LLM call — which can // return DIFFERENT verdicts). The TTL is a crash-safety backstop only (see AI_REVIEW_LOCK_TTL_SECONDS below), not -// a throughput bound — same philosophy as AGENT_MAINTENANCE_LOCK_TTL_SECONDS (#2129/#2368). +// a throughput bound — same philosophy as PR_ACTUATION_LOCK_TTL_SECONDS (#2129/#2368). Deliberately its OWN lock +// namespace, not the shared pr-actuation-lock above: this guards an expensive read-and-cache (dedup a redundant +// LLM call for the identical head+mode), not a GitHub-mutating actuation, so it has different scoping (keyed by +// head SHA + mode, not just PR) and a much longer TTL (an LLM call legitimately runs far longer than a close). const AI_REVIEW_LOCK_TTL_SECONDS = 1_800; // 30 minutes — see justification below. function aiReviewLockKey(repoFullName: string, prNumber: number, headSha: string, mode: string): string { @@ -4642,7 +4610,7 @@ export async function runAiReviewForAdvisory( })) ) return undefined; - // Per-(repo, PR, head SHA, mode) advisory lock (#confirmed-bug, mirrors #2129/#2368's claimAgentMaintenanceLock): + // Per-(repo, PR, head SHA, mode) advisory lock (#confirmed-bug, mirrors #2129/#2368's claimPrActuationLock): // a webhook pass and an agent-regate-pr sweep pass can independently reach this point for the SAME PR at the // SAME head, both miss the cache (neither has written yet), and both fire a real, wasteful LLM call that can // return different verdicts. Claim before the expensive section below; a pass that loses the race returns the diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 46dc72421f..a1de5c16f0 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -45,7 +45,7 @@ import { upsertRepositoryFromGitHub, putCachedAiReview, } from "../../src/db/repositories"; -import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock, releaseAiReviewLock, releasePrActuationLock } from "../../src/queue/processors"; +import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, processJob, releaseAiReviewLock, releasePrActuationLock } from "../../src/queue/processors"; import { aiReviewCacheInputFingerprint } from "../../src/review/ai-review-cache-input"; import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; import { normalizeRegistryPayload } from "../../src/registry/normalize"; @@ -3460,109 +3460,6 @@ describe("queue processors", () => { expect(denied?.n).toBe(1); }); - it("claimAgentMaintenanceLock claims when free, denies when held (per-PR, not per-repo), and release frees it again (#2129)", async () => { - const env = createTestEnv({}); - // First claim for this PR succeeds — no prior pass in-flight. - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - // A second, concurrent pass for the SAME PR (regardless of what triggered it — webhook or sweep) is denied - // while the first is still in-flight — exactly the race #2129 describes, since job-coalesce keys never - // match across trigger shapes but this lock is keyed purely on repo+PR. - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false); - // A DIFFERENT PR in the same repo is unaffected — the lock is per-PR, not a repo-wide serializer. - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 8)).toBe(true); - // Release (the finally block's job) frees the PR — a subsequent pass can claim it again. - await releaseAgentMaintenanceLock(env, "owner/agent-repo", 7); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - }); - - it("claimAgentMaintenanceLock fails OPEN on a broken transient cache — never itself blocks actuation (#2129)", async () => { - const env = createTestEnv({ - SELFHOST_TRANSIENT_CACHE: { - get: async () => { throw new Error("cache read error"); }, - set: async () => { throw new Error("cache write error"); }, - del: async () => { throw new Error("cache delete error"); }, - }, - }); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - await expect(releaseAgentMaintenanceLock(env, "owner/agent-repo", 7)).resolves.toBeUndefined(); - }); - - it("claimAgentMaintenanceLock fails OPEN when the atomic claim primitive itself throws (#2368)", async () => { - const env = createTestEnv({ - SELFHOST_TRANSIENT_CACHE: { - get: async () => null, - set: async () => undefined, - claim: async () => { throw new Error("redis unavailable"); }, - }, - }); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - }); - - it("REGRESSION (#2368): claimAgentMaintenanceLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => { - // #2368: a get-then-set pair has a window between the read and the write where two concurrent callers can - // both observe an absent key and both claim it — exactly what the per-PR lock exists to prevent. This test - // races two claims for the same PR via Promise.all (both kick off before either resolves) against the - // default test cache's claim(), which mirrors createRedisCache's atomic SET NX: the check-and-set happens - // with no `await` boundary in between, so it is impossible for both callers to see "unclaimed". - const env = createTestEnv({}); - const [first, second] = await Promise.all([ - claimAgentMaintenanceLock(env, "owner/agent-repo", 7), - claimAgentMaintenanceLock(env, "owner/agent-repo", 7), - ]); - expect([first, second].filter(Boolean)).toHaveLength(1); - }); - - it("REGRESSION (#2368): claimAgentMaintenanceLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => { - const calls: string[] = []; - const env = createTestEnv({ - SELFHOST_TRANSIENT_CACHE: { - get: async () => { calls.push("get"); return null; }, - set: async () => { calls.push("set"); }, - claim: async () => { calls.push("claim"); return true; }, - }, - }); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available - }); - - it("claimAgentMaintenanceLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#confirmed-bug, review round 2)", async () => { - // A prior version of this helper fell back to a get-then-set pair (even with an extra write-then-verify - // re-read) when claim() wasn't available. That is NOT a real exclusivity guarantee: caller A can write its - // own token, read it straight back, and return true entirely before caller B's later write/read also - // completes and also returns true -- both callers "win". Rather than pretend to serialize via a check that - // silently fails under exactly the concurrent load this lock exists to guard against, a cache without - // claim() now gets NO exclusivity at all -- every call proceeds, sequential or concurrent, even for a key a - // previous call already "set" via get/set. - const values = new Map(); - const env = createTestEnv({ - SELFHOST_TRANSIENT_CACHE: { - get: async (key: string) => values.get(key) ?? null, - set: async (key: string, value: string) => { values.set(key, value); }, - }, - }); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true); - }); - - it("REGRESSION (#confirmed-bug, review round 2): claimAgentMaintenanceLock does not falsely deny — and does not falsely claim exclusivity for — two genuinely concurrent callers when the cache has no claim()", async () => { - // Documents the corrected, honest contract under the exact interleaving the gate flagged: with no atomic - // claim() primitive, BOTH concurrent callers proceed (true), because this helper no longer attempts a - // get/set-based serialization that can't actually provide exclusivity. - const values = new Map(); - const yieldThenRun = (fn: () => T): Promise => new Promise((resolve) => queueMicrotask(() => resolve(fn()))); - const env = createTestEnv({ - SELFHOST_TRANSIENT_CACHE: { - get: (key: string) => yieldThenRun(() => values.get(key) ?? null), - set: (key: string, value: string) => yieldThenRun(() => { values.set(key, value); }), - }, - }); - const [first, second] = await Promise.all([ - claimAgentMaintenanceLock(env, "owner/agent-repo", 7), - claimAgentMaintenanceLock(env, "owner/agent-repo", 7), - ]); - expect([first, second]).toEqual([true, true]); - }); - it("claimAiReviewLock claims when free, denies when held (per-PR+head+mode, not globally), and release frees it again (#confirmed-bug)", async () => { const env = createTestEnv({}); // First claim for this exact (repo, PR, head, mode) succeeds — no prior pass in-flight. @@ -3677,8 +3574,9 @@ describe("queue processors", () => { expect([first, second]).toEqual([true, true]); }); - // claimPrActuationLock (#2135) mirrors claimAgentMaintenanceLock's atomic-claim design exactly — same test - // shapes, same reasoning, a different lock namespace. + // claimPrActuationLock (#2129/#2135) is the ONE shared per-PR actuation lock: maybeRunAgentMaintenance, + // maybeCloseDraftDodgeAttempt, and maybeRecloseDisallowedReopen all claim/release the SAME key so none of the + // three mutating PR paths can race any other (review round 4) — a single namespace, not one lock per path. it("claimPrActuationLock claims when free, denies when held (per-PR), and release frees it again (#2135)", async () => { const env = createTestEnv({}); expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); @@ -3734,9 +3632,8 @@ describe("queue processors", () => { }); it("claimPrActuationLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => { - // Mirrors claimAgentMaintenanceLock's #confirmed-bug fix: a get-then-set pair (even with a re-read) is not - // a real exclusivity guarantee under concurrent load, so a cache without claim() now gets NO exclusivity at - // all rather than a fallback that only looks atomic. + // A get-then-set pair (even with a re-read) is not a real exclusivity guarantee under concurrent load, so a + // cache without claim() now gets NO exclusivity at all rather than a fallback that only looks atomic. const values = new Map(); const env = createTestEnv({ SELFHOST_TRANSIENT_CACHE: { @@ -3798,9 +3695,11 @@ describe("queue processors", () => { vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z")); // Simulate a webhook pass already in-flight for this exact PR — a github-webhook:pr-refresh job's coalesce - // key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the new - // per-PR advisory lock is what makes a second, independently-triggered pass defer instead of racing it. - await env.SELFHOST_TRANSIENT_CACHE?.set("agent-maintenance-lock:owner/agent-repo#7", "1", 60); + // key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the + // shared per-PR actuation lock is what makes a second, independently-triggered pass defer instead of racing + // it. Pre-claims the SAME pr-actuation-lock key the draft-dodge/reopen-reclose paths use (#2129/#2135, + // review round 4) — one shared namespace, not a maintenance-only lock. + await env.SELFHOST_TRANSIENT_CACHE?.set("pr-actuation-lock:owner/agent-repo#7", "1", 60); await processJob(env, { type: "agent-regate-pr", deliveryId: "race-sweep", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 });