From 2972dfc969638248cb53bb75282038de9da779b5 Mon Sep 17 00:00:00 2001 From: joaovictor91123 Date: Sat, 4 Jul 2026 14:12:58 +0400 Subject: [PATCH] fix(selfhost): use ownership tokens for transient PR actuation locks --- src/env.d.ts | 5 ++ src/queue/processors.ts | 107 ++++++++++++++---------- src/selfhost/redis-cache.ts | 12 +++ test/helpers/d1.ts | 7 ++ test/unit/ai-review-advisory.test.ts | 2 +- test/unit/queue.test.ts | 108 ++++++++++++++++++------- test/unit/selfhost-redis-cache.test.ts | 24 ++++++ 7 files changed, 193 insertions(+), 72 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 124a4ea4b5..3917c05f95 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -34,6 +34,11 @@ 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 compare-and-delete: deletes `key` only when its current value equals `value`, returning whether + * it was removed. Lets a lock holder release its OWN claim without risking a stale post-TTL release + * deleting a different holder's live claim on the same key. Optional; a cache without it skips release + * entirely and relies on the TTL to reclaim the key (#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 2812d02a11..801d0a0432 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -315,6 +315,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"; @@ -2059,7 +2060,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, @@ -2073,7 +2075,7 @@ async function maybeRunAgentMaintenance( liveFacts: args.liveFacts, }); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } } @@ -3005,14 +3007,10 @@ 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. +// Per-holder ownership tokens + releaseIfValue (atomic compare-and-delete) close the race a shared constant +// lock value used to leave open: a holder that ran past the TTL can never have its stale `finally` release +// delete a later claimer's live lock (#2129/#2135) — release only succeeds when the caller's own token still +// matches what's stored. const PR_ACTUATION_LOCK_TTL_SECONDS = 600; function prActuationLockKey(repoFullName: string, prNumber: number): string { return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`; @@ -3021,7 +3019,7 @@ export async function claimPrActuationLock( env: Env, repoFullName: string, prNumber: number, -): Promise { +): Promise { return claimTransientLock( env, prActuationLockKey(repoFullName, prNumber), @@ -3032,12 +3030,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 @@ -3356,6 +3351,14 @@ async function ciHeadShaResolutionCoalesced( ); } +/** Result of a transient-lock claim attempt. `ownerToken` is the random value THIS call wrote when it actually + * acquired the lock, or null on every fail-open path (no cache, no atomic claim() primitive, a thrown claim(), + * or a lost race) — there is nothing for a null-token caller to release later. */ +export type TransientLockClaim = { + acquired: boolean; + ownerToken: string | null; +}; + /** * Best-effort exclusive claim against the self-host transient cache, shared by every per-PR/per-review advisory * lock below. Requires the store's native atomic claim() (Redis SET NX) to provide any real exclusivity — it is @@ -3368,19 +3371,42 @@ async function ciHeadShaResolutionCoalesced( * limitation rather than a false guarantee, and costs nothing in practice — self-host's Redis-backed cache (the * only cache adapter this codebase ships) always implements claim(), so this is a documented limitation for a * hypothetical future adapter, not a live gap. A missing cache or a thrown claim() also fails OPEN (returns - * true) — every lock built on this helper is defense-in-depth, never the primary safety gate, and must never - * itself block real work from running. + * acquired: true) — every lock built on this helper is defense-in-depth, never the primary safety gate, and + * must never itself block real work from running. + * + * The claimed value is a fresh random token per call, not a shared constant (#2129/#2135): release then + * verifies this exact token still owns the key (see releaseTransientLockIfOwner) before deleting it, so a + * holder that runs past its TTL can never have its stale `finally` release delete a DIFFERENT, live holder's + * claim on the same key — the race this mutex exists to close in the first place. */ 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 }; // no atomic primitive — nothing to serialize against. + const ownerToken = randomUUID(); try { - return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", ttlSeconds); + const acquired = await env.SELFHOST_TRANSIENT_CACHE.claim(key, ownerToken, ttlSeconds); + return { acquired, ownerToken: acquired ? ownerToken : null }; } catch { - return true; // fail open — see the doc comment above. + return { acquired: true, ownerToken: null }; // fail open — see the doc comment above. + } +} + +/** Releases a transient lock ONLY when `ownerToken` still matches the stored value (atomic compare-and-delete), + * so a stale holder can never delete a different, live holder's claim on the same key. `ownerToken` is null + * on every fail-open claim path (nothing was actually claimed, so nothing to release). A cache with no + * releaseIfValue() skips release entirely and relies on the TTL to reclaim the key, rather than falling back + * to a blind del() that would reopen the exact race the token scheme exists to close. */ +async function releaseTransientLockIfOwner(env: Env, key: string, ownerToken: string | null): Promise { + if (!ownerToken) return; + const cache = env.SELFHOST_TRANSIENT_CACHE; + if (!cache?.releaseIfValue) return; + try { + await cache.releaseIfValue(key, ownerToken); + } catch { + // best-effort; the TTL is the backstop if release fails } } @@ -3427,7 +3453,7 @@ export async function claimAiReviewLock( prNumber: number, headSha: string, mode: string, -): Promise { +): Promise { return claimTransientLock( env, aiReviewLockKey(repoFullName, prNumber, headSha, mode), @@ -3442,12 +3468,9 @@ 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`; @@ -5892,15 +5915,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", @@ -6234,6 +6256,7 @@ export async function runAiReviewForAdvisory( args.pr.number, args.advisory.headSha, args.settings.aiReviewMode, + aiReviewLock.ownerToken, ); } } @@ -9462,7 +9485,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 { @@ -9475,7 +9499,7 @@ async function maybeCloseDraftDodgeAttempt( settings, ); } finally { - await releasePrActuationLock(env, repoFullName, pr.number); + await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } } @@ -9668,7 +9692,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 { @@ -9682,7 +9707,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..e607c0cb57 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -23,6 +23,18 @@ export function createRedisCache(redis: Redis) { const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); return result === "OK"; }, + // Compare-and-delete: the read and the delete must be one atomic server-side step (a Lua eval), or a + // holder's own release could race a NEW claimant's write between a separate GET and DEL and delete the + // wrong holder's key -- the exact race per-holder ownership tokens exist to close. + 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..4281d880e1 100644 --- a/test/helpers/d1.ts +++ b/test/helpers/d1.ts @@ -117,6 +117,13 @@ export function createTestEnv(overrides: Partial = {}): Env { transientCache.set(key, value); return true; }, + // Mirrors createRedisCache's atomic compare-and-delete (#2129): only deletes when the stored value still + // equals the caller's own token, so a stale holder's release can never delete a different, live claim. + 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 43bf215c12..69d1b9f009 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -4180,7 +4180,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, { @@ -4961,19 +4961,20 @@ 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); + const first = await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block"); + expect(first.acquired).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); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block")).acquired).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); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha2", "block")).acquired).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); + expect((await claimAiReviewLock(env, "owner/agent-repo", 7, "sha1", "advisory")).acquired).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); + expect((await claimAiReviewLock(env, "owner/agent-repo", 8, "sha1", "block")).acquired).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); + 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 () => { @@ -4984,14 +4985,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 () => { @@ -5002,7 +5003,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 () => { @@ -5017,7 +5018,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 () => { @@ -5029,7 +5030,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 }); @@ -5048,8 +5049,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 () => { @@ -5069,7 +5070,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, @@ -5077,11 +5078,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 () => { @@ -5092,8 +5094,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 () => { @@ -5104,7 +5106,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 () => { @@ -5113,7 +5115,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 () => { @@ -5125,7 +5127,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 }); @@ -5139,8 +5141,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 () => { @@ -5156,7 +5158,53 @@ 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 (#2129/#2135): a stale actuation-lock holder's release does not delete a successor's live lock", async () => { + // The exact race the ownership-token scheme exists to close: holder A's claim TTL lapses (or its finally + // block simply runs late), a NEW holder B claims the same key in the meantime, and then A's release finally + // runs. A blind del() would delete B's still-live lock; releaseIfValue only deletes when the caller's OWN + // token still matches what's stored, so A's late release is a safe no-op against B's key. + const env = createTestEnv({}); + const staleHolder = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(staleHolder.acquired).toBe(true); + expect(staleHolder.ownerToken).toBeTruthy(); + // Simulate B's claim landing in the same key slot after A's token would have expired. + 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"); + // B's own release, with the matching token, does free the key. + 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("releaseAiReviewLock and releasePrActuationLock are no-ops when ownerToken is null (nothing was actually claimed)", async () => { + const env = createTestEnv({}); + const calls: string[] = []; + env.SELFHOST_TRANSIENT_CACHE = { + get: async () => null, + set: async () => undefined, + releaseIfValue: async () => { calls.push("releaseIfValue"); return true; }, + }; + await releasePrActuationLock(env, "owner/act-repo", 7, null); + await releaseAiReviewLock(env, "owner/agent-repo", 7, "sha1", "block", null); + expect(calls).toEqual([]); // a null token means nothing was claimed, so release must never touch the cache + }); + + it("release skips the cache entirely (relies on TTL) when the cache has no releaseIfValue()", async () => { + const calls: string[] = []; + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async (key: string) => { calls.push(`claim:${key}`); return true; }, + }, + }); + const claim = await claimPrActuationLock(env, "owner/act-repo", 7); + expect(claim.acquired).toBe(true); + await releasePrActuationLock(env, "owner/act-repo", 7, claim.ownerToken); + expect(calls).toEqual(["claim:pr-actuation-lock:owner/act-repo#7"]); // release never called releaseIfValue — it doesn't exist on this cache }); 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..34e4f9fb1d 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -21,6 +21,12 @@ function fakeRedis(): Redis & { _store: Map } { _store.delete(k); return 1; }, + // Emulates the Lua eval releaseIfValue runs: delete k only when its stored value equals the expected arg. + async eval(_script: string, _numkeys: number, k: string, expected: string) { + if (_store.get(k) !== expected) return 0; + _store.delete(k); + return 1; + }, } as unknown as Redis & { _store: Map }; } @@ -63,4 +69,22 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { const cache = createRedisCache(brokenRedis); await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused"); }); + + it("releaseIfValue deletes the key only when the stored value matches the caller's own token (#2129)", async () => { + const r = fakeRedis(); + const cache = createRedisCache(r); + await cache.set("lock", "holder-a", 60); + // A stale/different holder's token does not match — the live key is left untouched. + expect(await cache.releaseIfValue("lock", "holder-b")).toBe(false); + expect(await cache.get("lock")).toBe("holder-a"); + // The genuine owner's token matches — the key is removed. + expect(await cache.releaseIfValue("lock", "holder-a")).toBe(true); + expect(await cache.get("lock")).toBeNull(); + }); + + it("releaseIfValue propagates a Redis error to the caller (releaseTransientLockIfOwner treats this as best-effort)", async () => { + const brokenRedis = { async eval() { throw new Error("connection refused"); } } as unknown as Redis; + const cache = createRedisCache(brokenRedis); + await expect(cache.releaseIfValue("lock", "1")).rejects.toThrow("connection refused"); + }); });