From d8996d518196c901cba27fe23c34378934463568 Mon Sep 17 00:00:00 2001 From: RealDiligent Date: Sat, 4 Jul 2026 16:49:45 +0800 Subject: [PATCH 1/4] fix(selfhost): use ownership tokens for transient PR actuation locks Per-PR actuation and AI-review mutexes claimed Redis keys with a constant value and released via blind del(). A holder running past the TTL could delete a successor's live lock in finally, reopening merge/close races the mutex exists to prevent (#2129/#2135). Store a per-holder UUID at claim time and release with compare-and-delete (releaseIfValue) on the Redis cache adapter. Skip release when fail-open (no cache) or when the adapter lacks compare-and-delete (TTL backstop). Co-authored-by: Cursor --- src/env.d.ts | 3 + src/queue/processors.ts | 101 +++++++++++++++---------- src/selfhost/redis-cache.ts | 10 +++ test/helpers/d1.ts | 7 +- test/unit/ai-review-advisory.test.ts | 2 +- test/unit/queue.test.ts | 81 +++++++++++--------- test/unit/selfhost-redis-cache.test.ts | 17 +++++ 7 files changed, 143 insertions(+), 78 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 124a4ea4b5..1261d551aa 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -34,6 +34,9 @@ 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; + /** Delete `key` only when its current value equals `value` (compare-and-delete). Returns true when the + * key was removed. Optional; lock release skips when absent and relies on TTL instead of blind `del()`. */ + 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 dbc71d3b45..db8899c9d7 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -320,6 +320,7 @@ import { import { isDuplicateClusterWinnerByClaim } from "../signals/duplicate-winner"; import { buildUnifiedReviewDiff } from "../review/review-diff"; import { buildUnifiedCommentBody } from "../review/unified-comment-bridge"; +import { randomUUID } from "node:crypto"; import { isRetryableJobError, RetryableJobError } from "./retryable"; import { screenshotsAllowed } from "../review/visual-wire"; import { isVisualPath } from "../review/visual/paths"; @@ -2143,7 +2144,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 actuationLock = await claimPrActuationLock(env, repoFullName, pr.number); + if (!actuationLock.acquired) return; try { await runAgentMaintenancePlanAndExecute(env, { installationId, @@ -2157,7 +2159,7 @@ async function maybeRunAgentMaintenance( liveFacts: args.liveFacts, }); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } } @@ -3103,23 +3105,24 @@ async function putTransientKey( // 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. +// Ownership tokens + releaseIfValue (compare-and-delete) prevent a stale holder's finally block from +// deleting a successor's live lock after TTL expiry (#2129/#2135). const PR_ACTUATION_LOCK_TTL_SECONDS = 600; function prActuationLockKey(repoFullName: string, prNumber: number): string { return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`; } + +/** Result of claiming a transient-cache mutex. `ownerToken` is null on fail-open paths (no cache / claim error). */ +export type TransientLockClaim = { + acquired: boolean; + ownerToken: string | null; +}; + export async function claimPrActuationLock( env: Env, repoFullName: string, prNumber: number, -): Promise { +): Promise { return claimTransientLock( env, prActuationLockKey(repoFullName, prNumber), @@ -3130,12 +3133,9 @@ export async function releasePrActuationLock( env: Env, repoFullName: string, prNumber: number, + ownerToken: string | null, ): Promise { - try { - await env.SELFHOST_TRANSIENT_CACHE?.del?.(prActuationLockKey(repoFullName, prNumber)); - } catch { - // best-effort - } + await releaseTransientLockIfOwner(env, prActuationLockKey(repoFullName, prNumber), ownerToken); } // A plain thrown Error still reaches the queue's retry path (this call site is deliberately uncaught, same as @@ -3473,12 +3473,32 @@ async function claimTransientLock( env: Env, key: string, ttlSeconds: number, -): Promise { - if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return true; // no atomic primitive — nothing to serialize against. +): Promise { + if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return { acquired: true, ownerToken: null }; + const ownerToken = randomUUID(); + try { + const acquired = await env.SELFHOST_TRANSIENT_CACHE.claim(key, ownerToken, ttlSeconds); + return { acquired, ownerToken: acquired ? ownerToken : null }; + } catch { + return { acquired: true, ownerToken: null }; + } +} + +async function releaseTransientLockIfOwner( + env: Env, + key: string, + ownerToken: string | null, +): Promise { + if (!ownerToken) return; try { - return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", ttlSeconds); + const cache = env.SELFHOST_TRANSIENT_CACHE; + if (cache?.releaseIfValue) { + await cache.releaseIfValue(key, ownerToken); + return; + } + // Without compare-and-delete, skip release and let TTL expire — blind del() would reopen #2129/#2135. } catch { - return true; // fail open — see the doc comment above. + // best-effort; TTL is the backstop if release fails } } @@ -3525,7 +3545,7 @@ export async function claimAiReviewLock( prNumber: number, headSha: string, mode: string, -): Promise { +): Promise { return claimTransientLock( env, aiReviewLockKey(repoFullName, prNumber, headSha, mode), @@ -3540,12 +3560,13 @@ export async function releaseAiReviewLock( prNumber: number, headSha: string, mode: string, + ownerToken: string | null, ): Promise { - try { - await env.SELFHOST_TRANSIENT_CACHE?.del?.(aiReviewLockKey(repoFullName, prNumber, headSha, mode)); - } catch { - // best-effort; the TTL is the backstop if release fails - } + await releaseTransientLockIfOwner( + env, + aiReviewLockKey(repoFullName, prNumber, headSha, mode), + ownerToken, + ); } /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`; @@ -6055,15 +6076,14 @@ 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. - if ( - !(await claimAiReviewLock( - env, - args.repoFullName, - args.pr.number, - args.advisory.headSha, - args.settings.aiReviewMode, - )) - ) { + 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", @@ -6397,6 +6417,7 @@ export async function runAiReviewForAdvisory( args.pr.number, args.advisory.headSha, args.settings.aiReviewMode, + aiReviewLock.ownerToken, ); } } @@ -9624,7 +9645,8 @@ async function maybeCloseDraftDodgeAttempt( pr: PullRequestRecord, settings: RepositorySettings, ): Promise { - if (!(await claimPrActuationLock(env, repoFullName, pr.number))) { + const actuationLock = await claimPrActuationLock(env, repoFullName, pr.number); + if (!actuationLock.acquired) { throw new PrActuationLockContendedError(repoFullName, pr.number, "draft-dodge"); } try { @@ -9637,7 +9659,7 @@ async function maybeCloseDraftDodgeAttempt( settings, ); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } } @@ -9830,7 +9852,8 @@ async function maybeRecloseDisallowedReopen( pr: PullRequestRecord, payload: GitHubWebhookPayload, ): Promise { - if (!(await claimPrActuationLock(env, repoFullName, pr.number))) { + const actuationLock = await claimPrActuationLock(env, repoFullName, pr.number); + if (!actuationLock.acquired) { throw new PrActuationLockContendedError(repoFullName, pr.number, "reopen-reclose"); } try { @@ -9844,7 +9867,7 @@ async function maybeRecloseDisallowedReopen( ); return reclosed ? "reclosed" : "allowed"; } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } } diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index 4b55aeeea0..7d5083e71f 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -23,6 +23,16 @@ export function createRedisCache(redis: Redis) { const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); return result === "OK"; }, + // Atomic compare-and-delete: only the holder whose token still matches may release the key. + async releaseIfValue(key: string, value: string): Promise { + const result = await redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end", + 1, + key, + value, + ); + return result === 1; + }, }; } diff --git a/test/helpers/d1.ts b/test/helpers/d1.ts index 3ca21cd0b2..8284e57042 100644 --- a/test/helpers/d1.ts +++ b/test/helpers/d1.ts @@ -102,7 +102,7 @@ export function createTestEnv(overrides: Partial = {}): Env { async get(key: string) { return transientCache.get(key) ?? null; }, - async set(key: string, value: string) { + async set(key: string, value: string, _ttlSeconds: number) { transientCache.set(key, value); }, async del(key: string) { @@ -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/ai-review-advisory.test.ts b/test/unit/ai-review-advisory.test.ts index f772f39e67..bde0996690 100644 --- a/test/unit/ai-review-advisory.test.ts +++ b/test/unit/ai-review-advisory.test.ts @@ -582,7 +582,7 @@ describe("runAiReviewForAdvisory", () => { // Simulate a webhook pass already in-flight for this exact (repo, PR, head, mode) tuple — the caller under // test (a sweep-shaped pass, say) must defer instead of racing it with a second, independently-decided // LLM call. - expect(await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).toBe(true); + expect((await claimAiReviewLock(env, "acme/widgets", 3, "sha3", "block")).acquired).toBe(true); const result = await runAiReviewForAdvisory(env, { settings: { aiReviewMode: "block" } as RepositorySettings, diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index bd3c8cdf70..b93b25b4d4 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -4610,7 +4610,7 @@ describe("queue processors", () => { // The "first pass" (webhook-shaped) claims the lock for this exact (repo, PR, head, mode) tuple and is still // in-flight when the "second pass" (agent-regate-pr sweep-shaped) below reaches runAiReviewForAdvisory. - expect(await claimAiReviewLock(env, "JSONbored/gittensory", 49, "a49", "block")).toBe(true); + expect((await claimAiReviewLock(env, "JSONbored/gittensory", 49, "a49", "block")).acquired).toBe(true); await expect( processJob(env, { @@ -5390,20 +5390,14 @@ describe("queue processors", () => { 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. - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); - // A second, concurrent pass for the SAME PR at the SAME head and mode (regardless of what triggered it — - // webhook or sweep) is denied while the first is still in-flight — exactly the race this lock exists for. - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(false); - // A DIFFERENT head SHA for the same PR is unaffected — a new commit is a genuinely new review, not a dup. - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha2", "block")).toBe(true); - // A DIFFERENT mode for the same PR+head is also unaffected — advisory vs block are independent lock keys. - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "advisory")).toBe(true); - // A DIFFERENT PR in the same repo is unaffected — the lock is per-PR+head+mode, not repo-wide. - expect(await claimAiReviewLock(env, "owner/agent-repo", 8, "sha1", "block")).toBe(true); - // Release (the finally block's job) frees the (PR, head, mode) tuple — a subsequent pass can claim it again. - await releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); + const first = await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"); + expect(first.acquired).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(false); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha2", "block")).acquired).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "advisory")).acquired).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 8, "sha1", "block")).acquired).toBe(true); + await releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block", first.ownerToken); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); }); it("claimAiReviewLock fails OPEN on a broken transient cache — never itself blocks a real review from running (#confirmed-bug)", async () => { @@ -5414,14 +5408,14 @@ describe("queue processors", () => { del: async () => { throw new Error("cache delete error"); }, }, }); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); - await expect(releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).resolves.toBeUndefined(); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); + await expect(releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block", null)).resolves.toBeUndefined(); }); it("claimAiReviewLock fails OPEN when no transient cache is configured at all — nothing to serialize against (#confirmed-bug)", async () => { const env = createTestEnv({}); delete env.SELFHOST_TRANSIENT_CACHE; - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); }); it("claimAiReviewLock fails OPEN when the atomic claim primitive itself throws (#confirmed-bug)", async () => { @@ -5432,7 +5426,7 @@ describe("queue processors", () => { claim: async () => { throw new Error("redis unavailable"); }, }, }); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); }); it("REGRESSION: claimAiReviewLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME (repo, PR, head, mode) can never both succeed", async () => { @@ -5447,7 +5441,7 @@ describe("queue processors", () => { claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"), claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"), ]); - expect([first, second].filter(Boolean)).toHaveLength(1); + expect([first, second].filter((claim) => claim.acquired)).toHaveLength(1); }); it("REGRESSION: claimAiReviewLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => { @@ -5459,7 +5453,7 @@ describe("queue processors", () => { claim: async () => { calls.push("claim"); return true; }, }, }); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available }); @@ -5478,8 +5472,8 @@ describe("queue processors", () => { set: async (key: string, value: string) => { values.set(key, value); }, }, }); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); - expect(await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); }); it("REGRESSION (#confirmed-bug, review round 2): claimAiReviewLock does not falsely claim exclusivity for two genuinely concurrent callers when the cache has no claim()", async () => { @@ -5499,7 +5493,7 @@ describe("queue processors", () => { claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"), claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"), ]); - expect([first, second]).toEqual([true, true]); + expect([first.acquired, second.acquired]).toEqual([true, true]); }); // claimPrActuationLock (#2129/#2135) is the ONE shared per-PR actuation lock: maybeRunAgentMaintenance, @@ -5507,11 +5501,12 @@ 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(first.acquired).toBe(true); + expect((await claimPrActuationLock(env, "owner/act-repo", 7)).acquired).toBe(false); + expect((await claimPrActuationLock(env, "owner/act-repo", 8)).acquired).toBe(true); + await releasePrActuationLock(env, "owner/act-repo", 7, first.ownerToken); + expect((await claimPrActuationLock(env, "owner/act-repo", 7)).acquired).toBe(true); }); it("claimPrActuationLock fails OPEN on a broken transient cache — never itself blocks actuation (#2135)", async () => { @@ -5522,8 +5517,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)).acquired).toBe(true); + await expect(releasePrActuationLock(env, "owner/act-repo", 7, null)).resolves.toBeUndefined(); }); it("claimPrActuationLock fails OPEN when the atomic claim primitive itself throws (#2135)", async () => { @@ -5534,7 +5529,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)).acquired).toBe(true); }); 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 +5538,7 @@ 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((claim) => claim.acquired)).toHaveLength(1); }); it("REGRESSION (#2135): claimPrActuationLock calls the atomic claim primitive, not a separate get+set pair, when the cache supports it", async () => { @@ -5555,7 +5550,7 @@ 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)).acquired).toBe(true); expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available }); @@ -5569,8 +5564,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)).acquired).toBe(true); + expect((await claimPrActuationLock(env, "owner/act-repo", 7)).acquired).toBe(true); }); 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 +5581,19 @@ describe("queue processors", () => { claimPrActuationLock(env, "owner/act-repo", 7), claimPrActuationLock(env, "owner/act-repo", 7), ]); - expect([first, second]).toEqual([true, true]); + expect([first.acquired, second.acquired]).toEqual([true, true]); + }); + + it("REGRESSION: stale actuation-lock holder releaseIfValue does not delete a successor's live lock", async () => { + const env = createTestEnv({}); + const staleHolder = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(staleHolder.acquired).toBe(true); + expect(staleHolder.ownerToken).toBeTruthy(); + await env.SELFHOST_TRANSIENT_CACHE!.set!("pr-actuation-lock:owner/act-repo#7", "successor-token", 600); + await releasePrActuationLock(env, "owner/act-repo", 7, staleHolder.ownerToken); + expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("pr-actuation-lock:owner/act-repo#7")).toBe("successor-token"); + await releasePrActuationLock(env, "owner/act-repo", 7, "successor-token"); + expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("pr-actuation-lock:owner/act-repo#7")).toBeNull(); }); 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..83e9b2f7c5 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, expected: string) { + if (_store.get(key) === expected) { + _store.delete(key); + return 1; + } + return 0; + }, } as unknown as Redis & { _store: Map }; } @@ -63,4 +70,14 @@ 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 value matches (#2129 ownership release)", async () => { + const r = fakeRedis(); + const cache = createRedisCache(r); + await cache.set("lock", "holder-a", 60); + expect(await cache.releaseIfValue("lock", "holder-b")).toBe(false); + expect(await cache.get("lock")).toBe("holder-a"); + expect(await cache.releaseIfValue("lock", "holder-a")).toBe(true); + expect(await cache.get("lock")).toBeNull(); + }); }); From 1a3e7602b96a01094e18bc8bb358962bd20eb339 Mon Sep 17 00:00:00 2001 From: RealDiligent Date: Sat, 4 Jul 2026 17:04:14 +0800 Subject: [PATCH 2/4] test(selfhost): cover transient lock release edge paths for codecov Add regression tests for caches without releaseIfValue and for releaseIfValue failures so stale-holder protection branches are fully exercised. Co-authored-by: Cursor --- test/unit/queue.test.ts | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index b93b25b4d4..031593ee00 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -5596,6 +5596,41 @@ describe("queue processors", () => { expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("pr-actuation-lock:owner/act-repo#7")).toBeNull(); }); + it("does not blind-del when the cache lacks releaseIfValue (TTL backstop)", async () => { + let deleted = false; + const store = new Map(); + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async (key: string) => store.get(key) ?? null, + set: async (key: string, value: string) => { store.set(key, value); }, + claim: async (key: string, value: string) => { + if (store.has(key)) return false; + store.set(key, value); + return true; + }, + del: async () => { deleted = true; }, + }, + }); + const lock = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(lock.acquired).toBe(true); + await releasePrActuationLock(env, "owner/act-repo", 7, lock.ownerToken); + expect(deleted).toBe(false); + expect(store.get("pr-actuation-lock:owner/act-repo#7")).toBe(lock.ownerToken); + }); + + it("releaseIfValue errors are best-effort (TTL backstop)", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async (_key: string, value: string) => value.length > 0, + releaseIfValue: async () => { throw new Error("redis eval failed"); }, + }, + }); + const lock = await claimPrActuationLock(env, "owner/act-repo", 7); + await expect(releasePrActuationLock(env, "owner/act-repo", 7, lock.ownerToken)).resolves.toBeUndefined(); + }); + it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => { const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); From f00f90132f7ca01d0ac31c54fa0f920d9038bc9c Mon Sep 17 00:00:00 2001 From: RealDiligent Date: Sun, 5 Jul 2026 01:33:29 +0800 Subject: [PATCH 3/4] docs(selfhost): align transient cache claim() comment with fail-open behavior Co-authored-by: Cursor --- src/env.d.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/env.d.ts b/src/env.d.ts index 1261d551aa..3cbe98c932 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -32,7 +32,7 @@ declare global { * already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent * callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the * 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). */ + * type-checks; callers fail open (proceed without exclusivity) when absent — no get-then-set fallback (#2129). */ claim?(key: string, value: string, ttlSeconds: number): Promise; /** Delete `key` only when its current value equals `value` (compare-and-delete). Returns true when the * key was removed. Optional; lock release skips when absent and relies on TTL instead of blind `del()`. */ From 9840e5207aee5f3b0d49aa0b4d48379adc3d6b86 Mon Sep 17 00:00:00 2001 From: RealDiligent Date: Sun, 5 Jul 2026 02:13:03 +0800 Subject: [PATCH 4/4] fix(selfhost): require releaseIfValue for transient lock claim adapters Ownership-token release fixed stale-holder blind del() (#2129), but skipping release when releaseIfValue was absent pinned locks for 600s/1800s after normal work on misconfigured adapters (#3153). - Boot: assertSelfhostTransientCacheOwnershipRelease() in server.ts - Runtime: fail open without calling claim() when releaseIfValue is missing - Tests: stale-holder regressions for both lock namespaces, boot guard, #3153 path Co-authored-by: Cursor --- src/env.d.ts | 6 +++--- src/queue/processors.ts | 16 +++++++++------- src/selfhost/redis-cache.ts | 11 +++++++++++ src/server.ts | 3 ++- test/unit/queue.test.ts | 26 ++++++++++++++++++++------ test/unit/selfhost-redis-cache.test.ts | 11 ++++++++++- 6 files changed, 55 insertions(+), 18 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 3cbe98c932..0aca8203d2 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -31,11 +31,11 @@ declare global { /** Atomic "set only if absent": returns true when this call newly claimed the key, false when it was * already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent * callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the - * check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still - * type-checks; callers fail open (proceed without exclusivity) when absent — no get-then-set fallback (#2129). */ + * check-and-set as one operation. Must be paired with `releaseIfValue` — self-host boot rejects + * `claim()` without it; runtime callers fail open without exclusivity rather than pin locks (#2129). */ claim?(key: string, value: string, ttlSeconds: number): Promise; /** Delete `key` only when its current value equals `value` (compare-and-delete). Returns true when the - * key was removed. Optional; lock release skips when absent and relies on TTL instead of blind `del()`. */ + * key was removed. Required on any adapter that implements `claim()` (validated at self-host boot). */ releaseIfValue?(key: string, value: string): Promise; }; /** TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) is a separate, diff --git a/src/queue/processors.ts b/src/queue/processors.ts index db8899c9d7..214bbdf83a 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -3474,10 +3474,15 @@ async function claimTransientLock( key: string, ttlSeconds: number, ): Promise { - if (!env.SELFHOST_TRANSIENT_CACHE?.claim) return { acquired: true, ownerToken: null }; + const cache = env.SELFHOST_TRANSIENT_CACHE; + if (!cache?.claim) return { acquired: true, ownerToken: null }; + // A claim()-only adapter without releaseIfValue would pin locks until TTL after normal work — reject that + // shape at self-host boot (assertSelfhostTransientCacheOwnershipRelease). At runtime, fail open without + // calling claim() so misconfigured test/custom adapters never acquire an unreleasable lock (#2129/#3153). + if (!cache.releaseIfValue) return { acquired: true, ownerToken: null }; const ownerToken = randomUUID(); try { - const acquired = await env.SELFHOST_TRANSIENT_CACHE.claim(key, ownerToken, ttlSeconds); + const acquired = await cache.claim(key, ownerToken, ttlSeconds); return { acquired, ownerToken: acquired ? ownerToken : null }; } catch { return { acquired: true, ownerToken: null }; @@ -3492,11 +3497,8 @@ async function releaseTransientLockIfOwner( if (!ownerToken) return; try { const cache = env.SELFHOST_TRANSIENT_CACHE; - if (cache?.releaseIfValue) { - await cache.releaseIfValue(key, ownerToken); - return; - } - // Without compare-and-delete, skip release and let TTL expire — blind del() would reopen #2129/#2135. + if (!cache?.releaseIfValue) return; + await cache.releaseIfValue(key, ownerToken); } catch { // best-effort; TTL is the backstop if release fails } diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index 7d5083e71f..71e08f58ca 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -36,4 +36,15 @@ export function createRedisCache(redis: Redis) { }; } +/** Self-host boot guard: `claim()` without ownership-aware release pins actuation locks for minutes. */ +export function assertSelfhostTransientCacheOwnershipRelease( + cache: { claim?(key: string, value: string, ttlSeconds: number): Promise; releaseIfValue?(key: string, value: string): Promise }, +): void { + if (cache.claim && !cache.releaseIfValue) { + throw new Error( + "SELFHOST_TRANSIENT_CACHE.claim requires releaseIfValue for ownership-aware transient locks (#2129)", + ); + } +} + export type RedisCache = ReturnType; diff --git a/src/server.ts b/src/server.ts index 1b4f4e0b71..f0d47ca724 100644 --- a/src/server.ts +++ b/src/server.ts @@ -478,9 +478,10 @@ async function main(): Promise { const { Redis } = await import("ioredis"); const redisClient = new Redis(redisUrl); const { createRedisRateLimiter } = await import("./selfhost/redis-ratelimit"); - const { createRedisCache } = await import("./selfhost/redis-cache"); + const { createRedisCache, assertSelfhostTransientCacheOwnershipRelease } = await import("./selfhost/redis-cache"); const rateLimiter = createRedisRateLimiter(redisClient); const webhookCache = createRedisCache(redisClient); + assertSelfhostTransientCacheOwnershipRelease(webhookCache); // Persist the installation-token cache in Redis so warm GitHub App tokens survive restarts/deploys and are // shared across replicas (the in-isolate Map otherwise re-mints — an Orb round-trip — per replica/cold start). const { createRedisTokenCache } = await import("./selfhost/redis-token-cache"); diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 031593ee00..0c2758d306 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -5451,6 +5451,7 @@ describe("queue processors", () => { get: async () => { calls.push("get"); return null; }, set: async () => { calls.push("set"); }, claim: async () => { calls.push("claim"); return true; }, + releaseIfValue: async () => true, }, }); expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).toBe(true); @@ -5548,6 +5549,7 @@ describe("queue processors", () => { get: async () => { calls.push("get"); return null; }, set: async () => { calls.push("set"); }, claim: async () => { calls.push("claim"); return true; }, + releaseIfValue: async () => true, }, }); expect((await claimPrActuationLock(env, "owner/act-repo", 7)).acquired).toBe(true); @@ -5596,26 +5598,38 @@ describe("queue processors", () => { expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("pr-actuation-lock:owner/act-repo#7")).toBeNull(); }); - it("does not blind-del when the cache lacks releaseIfValue (TTL backstop)", async () => { - let deleted = false; + it("REGRESSION: stale AI-review-lock holder releaseIfValue does not delete a successor's live lock", async () => { + const env = createTestEnv({}); + const staleHolder = await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"); + expect(staleHolder.acquired).toBe(true); + expect(staleHolder.ownerToken).toBeTruthy(); + await env.SELFHOST_TRANSIENT_CACHE!.set!("ai-review-lock:owner/agent-repo#7@sha1:block", "successor-token", 1800); + await releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block", staleHolder.ownerToken); + expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("ai-review-lock:owner/agent-repo#7@sha1:block")).toBe("successor-token"); + await releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block", "successor-token"); + expect(await env.SELFHOST_TRANSIENT_CACHE!.get!("ai-review-lock:owner/agent-repo#7@sha1:block")).toBeNull(); + }); + + it("claimPrActuationLock fails open without exclusivity when claim() is present but releaseIfValue is absent (#3153)", async () => { + let claimed = false; const store = new Map(); const env = createTestEnv({ SELFHOST_TRANSIENT_CACHE: { get: async (key: string) => store.get(key) ?? null, set: async (key: string, value: string) => { store.set(key, value); }, claim: async (key: string, value: string) => { + claimed = true; if (store.has(key)) return false; store.set(key, value); return true; }, - del: async () => { deleted = true; }, }, }); const lock = await claimPrActuationLock(env, "owner/act-repo", 7); expect(lock.acquired).toBe(true); - await releasePrActuationLock(env, "owner/act-repo", 7, lock.ownerToken); - expect(deleted).toBe(false); - expect(store.get("pr-actuation-lock:owner/act-repo#7")).toBe(lock.ownerToken); + expect(lock.ownerToken).toBeNull(); + expect(claimed).toBe(false); + expect(store.size).toBe(0); }); it("releaseIfValue errors are best-effort (TTL backstop)", async () => { diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index 83e9b2f7c5..c4416ac706 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -1,6 +1,6 @@ import type { Redis } from "ioredis"; import { describe, expect, it } from "vitest"; -import { createRedisCache } from "../../src/selfhost/redis-cache"; +import { assertSelfhostTransientCacheOwnershipRelease, createRedisCache } from "../../src/selfhost/redis-cache"; /** Minimal in-memory stand-in for the ioredis methods the cache uses. Emulates real Redis SET NX * semantics (refuse + return null when NX is requested and the key already exists) so a test @@ -80,4 +80,13 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { expect(await cache.releaseIfValue("lock", "holder-a")).toBe(true); expect(await cache.get("lock")).toBeNull(); }); + + it("assertSelfhostTransientCacheOwnershipRelease rejects claim() without releaseIfValue at boot (#3153)", () => { + expect(() => + assertSelfhostTransientCacheOwnershipRelease({ + claim: async () => true, + }), + ).toThrow(/releaseIfValue/); + expect(() => assertSelfhostTransientCacheOwnershipRelease(createRedisCache(fakeRedis()))).not.toThrow(); + }); });