From 252a3fbc605e6c5607da258c0fc27f26fae22ac7 Mon Sep 17 00:00:00 2001 From: andriypolandki <=> Date: Sat, 4 Jul 2026 03:45:35 -0700 Subject: [PATCH] fix(queue): owner-verified release for per-PR actuation lock --- src/env.d.ts | 4 ++ src/queue/processors.ts | 84 +++++++++++++++++--------- src/selfhost/redis-cache.ts | 11 ++++ test/helpers/d1.ts | 5 ++ test/unit/queue.test.ts | 80 +++++++++++++++++++----- test/unit/selfhost-redis-cache.test.ts | 21 +++++++ 6 files changed, 163 insertions(+), 42 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 124a4ea4b5..9e349abc37 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -34,6 +34,10 @@ declare global { * check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still * type-checks; callers fall back to the non-atomic get/set pair when absent (#2129). */ claim?(key: string, value: string, ttlSeconds: number): Promise; + /** Atomic "delete only if value matches": returns true when this call removed the key, false when the + * key was absent or held by a different owner. Used by per-holder transient locks so a stale holder's + * finally release cannot delete a later claimer's live lock after TTL expiry (#2129). */ + releaseIfValue?(key: string, value: string): Promise; }; /** TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) is a separate, * more-involved sub-task — it needs the ported DO class + its own migration tag, not just a binding here. diff --git a/src/queue/processors.ts b/src/queue/processors.ts index bc2cc0610a..d724d60464 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -2143,7 +2143,8 @@ 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 claimPrActuationLock(env, repoFullName, pr.number))) return; + const lockToken = await claimPrActuationLock(env, repoFullName, pr.number); + if (lockToken === null) return; try { await runAgentMaintenancePlanAndExecute(env, { installationId, @@ -2157,7 +2158,7 @@ async function maybeRunAgentMaintenance( liveFacts: args.liveFacts, }); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, lockToken); } } @@ -3094,32 +3095,31 @@ async function putTransientKey( // 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. +// env.d.ts) built on the SAME transient cache used for CI-completion coalescing above, claimed ATOMICALLY with a +// per-holder token (see claimTransientLockWithOwnerToken) so two racing deliveries can never both win the claim — +// a short TTL, owner-verified release. A lock-contended caller skips this pass (returns null) 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 as fail-open) 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 -// first holder's stale `finally` release, reopening the exact race this mutex exists to close. A per-holder -// 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, or a maintenance pass's plan-and-execute) should never legitimately -// run anywhere near this long. +// Per-PR mutual exclusion (#2129/#2135): 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. Each claim stores a per-holder +// ownership token; release is compare-and-delete so a holder that ran past the TTL cannot delete a later +// claimer's live lock in its stale `finally` block. const PR_ACTUATION_LOCK_TTL_SECONDS = 600; function prActuationLockKey(repoFullName: string, prNumber: number): string { return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`; } +/** Non-empty string = claimed (release with this token). Empty string = fail-open (no lock held). Null = contended. */ +export type PrActuationLockClaim = string | null; export async function claimPrActuationLock( env: Env, repoFullName: string, prNumber: number, -): Promise { - return claimTransientLock( +): Promise { + return claimTransientLockWithOwnerToken( env, prActuationLockKey(repoFullName, prNumber), PR_ACTUATION_LOCK_TTL_SECONDS, @@ -3129,12 +3129,9 @@ export async function releasePrActuationLock( env: Env, repoFullName: string, prNumber: number, + token: PrActuationLockClaim, ): Promise { - try { - await env.SELFHOST_TRANSIENT_CACHE?.del?.(prActuationLockKey(repoFullName, prNumber)); - } catch { - // best-effort - } + await releaseTransientLockIfOwner(env, prActuationLockKey(repoFullName, prNumber), token); } // A plain thrown Error still reaches the queue's retry path (this call site is deliberately uncaught, same as @@ -3481,6 +3478,35 @@ async function claimTransientLock( } } +/** Like {@link claimTransientLock}, but stores a per-holder ownership token for compare-and-delete release. */ +async function claimTransientLockWithOwnerToken( + env: Env, + key: string, + ttlSeconds: number, +): Promise { + if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return ""; // fail-open — no lock to release. + try { + const token = crypto.randomUUID(); + return (await env.SELFHOST_TRANSIENT_CACHE.claim(key, token, ttlSeconds)) ? token : null; + } catch { + return ""; // fail open — see claimTransientLock's doc comment above. + } +} + +/** Best-effort owner-verified release. Empty token = fail-open claim, no-op. Without releaseIfValue, leave the + * key to expire via TTL rather than risk deleting a later claimer's live lock. */ +async function releaseTransientLockIfOwner(env: Env, key: string, token: PrActuationLockClaim): Promise { + if (!token) return; + try { + if (env.SELFHOST_TRANSIENT_CACHE?.releaseIfValue) { + await env.SELFHOST_TRANSIENT_CACHE.releaseIfValue(key, token); + return; + } + } catch { + // best-effort + } +} + // 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 @@ -9624,7 +9650,8 @@ async function maybeCloseDraftDodgeAttempt( pr: PullRequestRecord, settings: RepositorySettings, ): Promise { - if (!(await claimPrActuationLock(env, repoFullName, pr.number))) { + const lockToken = await claimPrActuationLock(env, repoFullName, pr.number); + if (lockToken === null) { throw new PrActuationLockContendedError(repoFullName, pr.number, "draft-dodge"); } try { @@ -9637,7 +9664,7 @@ async function maybeCloseDraftDodgeAttempt( settings, ); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, lockToken); } } @@ -9830,7 +9857,8 @@ async function maybeRecloseDisallowedReopen( pr: PullRequestRecord, payload: GitHubWebhookPayload, ): Promise { - if (!(await claimPrActuationLock(env, repoFullName, pr.number))) { + const lockToken = await claimPrActuationLock(env, repoFullName, pr.number); + if (lockToken === null) { throw new PrActuationLockContendedError(repoFullName, pr.number, "reopen-reclose"); } try { @@ -9844,7 +9872,7 @@ async function maybeRecloseDisallowedReopen( ); return reclosed ? "reclosed" : "allowed"; } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, lockToken); } } diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index 4b55aeeea0..1b8a0c665d 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -23,6 +23,17 @@ export function createRedisCache(redis: Redis) { const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); return result === "OK"; }, + async releaseIfValue(key: string, value: string): Promise { + const script = ` + if redis.call("get", KEYS[1]) == ARGV[1] then + return redis.call("del", KEYS[1]) + else + return 0 + end + `; + const result = await redis.eval(script, 1, key, value); + return Number(result) === 1; + }, }; } diff --git a/test/helpers/d1.ts b/test/helpers/d1.ts index 3ca21cd0b2..fbff28d7b5 100644 --- a/test/helpers/d1.ts +++ b/test/helpers/d1.ts @@ -117,6 +117,11 @@ export function createTestEnv(overrides: Partial = {}): Env { transientCache.set(key, value); return true; }, + async releaseIfValue(key: string, value: string) { + if (transientCache.get(key) !== value) return false; + transientCache.delete(key); + return true; + }, }, // Per-repo review allowlist: default to the test repos so flag-ON wiring tests activate the // gated review features. Override to "" to assert the dormant (no-repo) default. diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index bd3c8cdf70..fa1317a45c 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -5507,11 +5507,13 @@ describe("queue processors", () => { // 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); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(false); - expect(await claimPrActuationLock(env, "owner/act-repo", 8)).toBe(true); - await releasePrActuationLock(env, "owner/act-repo", 7); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); + const first = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(typeof first).toBe("string"); + expect(first).not.toBe(""); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBeNull(); + expect(await claimPrActuationLock(env, "owner/act-repo", 8)).not.toBeNull(); + await releasePrActuationLock(env, "owner/act-repo", 7, first); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull(); }); it("claimPrActuationLock fails OPEN on a broken transient cache — never itself blocks actuation (#2135)", async () => { @@ -5522,8 +5524,8 @@ describe("queue processors", () => { del: async () => { throw new Error("cache delete error"); }, }, }); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); - await expect(releasePrActuationLock(env, "owner/act-repo", 7)).resolves.toBeUndefined(); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(""); + await expect(releasePrActuationLock(env, "owner/act-repo", 7, "")).resolves.toBeUndefined(); }); it("claimPrActuationLock fails OPEN when the atomic claim primitive itself throws (#2135)", async () => { @@ -5534,7 +5536,7 @@ describe("queue processors", () => { claim: async () => { throw new Error("redis unavailable"); }, }, }); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(""); }); it("REGRESSION (#2135): claimPrActuationLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => { @@ -5543,7 +5545,57 @@ describe("queue processors", () => { claimPrActuationLock(env, "owner/act-repo", 7), claimPrActuationLock(env, "owner/act-repo", 7), ]); - expect([first, second].filter(Boolean)).toHaveLength(1); + expect([first, second].filter((token) => token !== null && token !== "")).toHaveLength(1); + }); + + it("REGRESSION: a stale holder's releaseIfValue does not delete a later claimer's live lock", async () => { + const env = createTestEnv({}); + const staleToken = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(staleToken).not.toBeNull(); + expect(staleToken).not.toBe(""); + const liveToken = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(liveToken).toBeNull(); + await env.SELFHOST_TRANSIENT_CACHE?.set?.("pr-actuation-lock:owner/act-repo#7", "later-holder-token", 60); + await releasePrActuationLock(env, "owner/act-repo", 7, staleToken); + expect(await env.SELFHOST_TRANSIENT_CACHE?.get("pr-actuation-lock:owner/act-repo#7")).toBe("later-holder-token"); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBeNull(); + await releasePrActuationLock(env, "owner/act-repo", 7, "later-holder-token"); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull(); + }); + + it("releasePrActuationLock leaves the key in place when the cache has claim() but no releaseIfValue()", async () => { + 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); }, + claim: async (key: string, value: string) => { + if (values.has(key)) return false; + values.set(key, value); + return true; + }, + }, + }); + const token = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(token).not.toBe(""); + await releasePrActuationLock(env, "owner/act-repo", 7, token); + expect(values.get("pr-actuation-lock:owner/act-repo#7")).toBe(token); + }); + + it("releasePrActuationLock swallows releaseIfValue errors (best-effort)", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => "holder", + set: async () => undefined, + claim: async (_key: string, value: string) => { + return value.length > 0; + }, + releaseIfValue: async () => { + throw new Error("redis down"); + }, + }, + }); + await expect(releasePrActuationLock(env, "owner/act-repo", 7, "holder")).resolves.toBeUndefined(); }); it("REGRESSION (#2135): claimPrActuationLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => { @@ -5555,11 +5607,11 @@ describe("queue processors", () => { claim: async () => { calls.push("claim"); return true; }, }, }); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).not.toBeNull(); expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available }); - it("claimPrActuationLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => { + it("claimPrActuationLock returns empty string unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => { // 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(); @@ -5569,8 +5621,8 @@ describe("queue processors", () => { set: async (key: string, value: string) => { values.set(key, value); }, }, }); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); - expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(""); + expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(""); }); it("REGRESSION (#2135, review round 2): claimPrActuationLock does not falsely claim exclusivity for two genuinely concurrent callers when the cache has no claim()", async () => { @@ -5586,7 +5638,7 @@ describe("queue processors", () => { claimPrActuationLock(env, "owner/act-repo", 7), claimPrActuationLock(env, "owner/act-repo", 7), ]); - expect([first, second]).toEqual([true, true]); + expect([first, second]).toEqual(["", ""]); }); it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => { diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index f12499dcf7..e58c226445 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -21,6 +21,13 @@ function fakeRedis(): Redis & { _store: Map } { _store.delete(k); return 1; }, + async eval(_script: string, _numKeys: number, key: string, value: string) { + if (_store.get(key) === value) { + _store.delete(key); + return 1; + } + return 0; + }, } as unknown as Redis & { _store: Map }; } @@ -63,4 +70,18 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { const cache = createRedisCache(brokenRedis); await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused"); }); + + it("releaseIfValue deletes only when the stored owner matches (#2129)", async () => { + const cache = createRedisCache(fakeRedis()); + await cache.set("lock", "holder-A", 60); + expect(await cache.releaseIfValue("lock", "holder-A")).toBe(true); + expect(await cache.get("lock")).toBeNull(); + }); + + it("releaseIfValue refuses when a different owner holds the key (#2129)", async () => { + const cache = createRedisCache(fakeRedis()); + await cache.set("lock", "holder-B", 60); + expect(await cache.releaseIfValue("lock", "holder-A")).toBe(false); + expect(await cache.get("lock")).toBe("holder-B"); + }); });