fix(queue): make claimPrActuationLock atomic, add a real concurrency test

Review findings on this PR:
- claimPrActuationLock was a non-atomic getTransientKey-then-putTransientKey
  pair, so two genuinely concurrent deliveries for the same PR could both
  observe an absent key and both proceed — defeating the exact race this
  mutex exists to close. The sibling claimAgentMaintenanceLock (#2129,
  #2368) already solved this: env.SELFHOST_TRANSIENT_CACHE.claim performs
  the check-and-set as one atomic operation (Redis SET NX server-side),
  with a documented fallback to the old get/set pair for a cache adapter
  that hasn't implemented claim yet. Mirrored that exact pattern here.
- The existing lock tests only pre-seeded the key before the call started,
  proving the contended branch but not the actual race. Added a Promise.all
  test that fires two draft-dodge deliveries for the SAME PR with neither
  pre-claiming anything, asserting exactly one PATCH and one completed
  audit row. Verified this test is meaningful by temporarily reverting to
  the non-atomic implementation and confirming it fails (2 PATCH calls),
  then restoring the fix and confirming it passes.
- Exported claimPrActuationLock/releasePrActuationLock (matching the
  already-exported claimAgentMaintenanceLock/releaseAgentMaintenanceLock)
  and mirrored that sibling's full direct-unit-test suite — fail-open on a
  broken cache, fail-open when claim() itself throws, atomic-claim-used
  verification, and the no-claim-method fallback — closing the branch
  coverage gap the new code left in the fallback/catch paths.

Author: JSONbored

src/queue/processors.ts

@@ -2309,24 +2309,41 @@ async function putTransientKey(
 // 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: a short-TTL claim, 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.
+// cache used for CI-completion coalescing above, claimed ATOMICALLY (see claimPrActuationLock) 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.
 const PR_ACTUATION_LOCK_TTL_SECONDS = 60;
 function prActuationLockKey(repoFullName: string, prNumber: number): string {
   return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;
 }
-async function claimPrActuationLock(
+export async function claimPrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,
 ): Promise<boolean> {
   const key = prActuationLockKey(repoFullName, prNumber);
+  // Atomic claim (#2129, mirroring claimAgentMaintenanceLock): a get-then-set pair has a window between the
+  // read and the write where two concurrent deliveries for the SAME PR can both observe an absent key and both
+  // claim it, defeating this mutex entirely. env.SELFHOST_TRANSIENT_CACHE.claim performs the check-and-set as
+  // one operation (Redis SET NX server-side), closing that window. Falls back to the non-atomic get/set pair
+  // only for a cache adapter that hasn't implemented claim yet — strictly no worse than this function's prior
+  // behavior.
+  if (env.SELFHOST_TRANSIENT_CACHE?.claim) {
+    try {
+      return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", PR_ACTUATION_LOCK_TTL_SECONDS);
+    } catch {
+      return true; // fail open — see the doc comment above.
+    }
+  }
+  // getTransientKey/putTransientKey already fail open internally (a missing cache or a thrown read/write error
+  // both resolve rather than throw), so this never needs its own try/catch — a cache fault surfaces here as
+  // "no lock held", which correctly falls through to claiming it.
   if (await getTransientKey(env, key)) return false;
   await putTransientKey(env, key, "1", PR_ACTUATION_LOCK_TTL_SECONDS);
   return true;
 }
-async function releasePrActuationLock(
+export async function releasePrActuationLock(
   env: Env,
   repoFullName: string,
   prNumber: number,

test/unit/queue.test.ts

@@ -44,7 +44,7 @@ import {
   upsertRepositoryFromGitHub,
   putCachedAiReview,
 } from "../../src/db/repositories";
-import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock } from "../../src/queue/processors";
+import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, claimPrActuationLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock, 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";
@@ -3453,6 +3453,74 @@ describe("queue processors", () => {
     expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false);
   });
 
+  // claimPrActuationLock (#2135) mirrors claimAgentMaintenanceLock's atomic-claim design exactly — same test
+  // shapes, same reasoning, a different lock namespace.
+  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);
+  });
+
+  it("claimPrActuationLock fails OPEN on a broken transient cache — never itself blocks actuation (#2135)", 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 claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    await expect(releasePrActuationLock(env, "owner/act-repo", 7)).resolves.toBeUndefined();
+  });
+
+  it("claimPrActuationLock fails OPEN when the atomic claim primitive itself throws (#2135)", async () => {
+    const env = createTestEnv({
+      SELFHOST_TRANSIENT_CACHE: {
+        get: async () => null,
+        set: async () => undefined,
+        claim: async () => { throw new Error("redis unavailable"); },
+      },
+    });
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).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 () => {
+    const env = createTestEnv({});
+    const [first, second] = await Promise.all([
+      claimPrActuationLock(env, "owner/act-repo", 7),
+      claimPrActuationLock(env, "owner/act-repo", 7),
+    ]);
+    expect([first, second].filter(Boolean)).toHaveLength(1);
+  });
+
+  it("REGRESSION (#2135): claimPrActuationLock 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 claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available
+  });
+
+  it("claimPrActuationLock falls back to the get/set pair and still denies a held key when the cache has no claim() (#2135)", async () => {
+    const values = new Map<string, string>();
+    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 claimPrActuationLock(env, "owner/act-repo", 7)).toBe(true);
+    expect(await claimPrActuationLock(env, "owner/act-repo", 7)).toBe(false);
+  });
+
   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: [] } });
@@ -11892,6 +11960,37 @@ describe("converted_to_draft gate-close (draft-dodge prevention)", () => {
     expect(audit?.n).toBe(0); // no decision recorded either way — the in-flight delivery owns this pass
   });
 
+  it("REGRESSION: exactly ONE of two genuinely concurrent draft-dodge deliveries for the SAME PR wins the actuation lock (#2135)", async () => {
+    // Unlike the lock-contended test above (which pre-seeds the key before the call even starts), this fires
+    // two deliveries together via Promise.all with NEITHER pre-claiming anything — exercising the actual
+    // check-and-set race claimPrActuationLock must arbitrate, not just "the key was already there". A
+    // get-then-set (non-atomic) implementation lets both deliveries observe an absent key and both proceed,
+    // which this test would catch as more than one PATCH / more than one completed audit row.
+    const calls: string[] = [];
+    vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = input.toString();
+      calls.push(`${init?.method ?? "GET"} ${url}`);
+      if (url.includes("/access_tokens")) return Response.json({ token: "t" });
+      if (url.endsWith("/issues/42/comments")) return Response.json({ id: 1 }, { status: 201 });
+      if (url.endsWith("/pulls/42")) return Response.json({ state: "closed" });
+      return new Response("not found", { status: 404 });
+    });
+
+    const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: generateRsaPrivateKeyPem(), GITHUB_APP_SLUG: "gittensory" });
+    await setupRepo(env);
+    await recordGateBlockOutcome(env, { repoFullName: "JSONbored/gittensory", pullNumber: 42, headSha: "abc123", blockerCodes: ["missing_linked_issue"] });
+
+    await Promise.all([
+      processJob(env, { type: "github-webhook", deliveryId: "draft-dodge-race-a", eventName: "pull_request", payload: draftPayload("contributor") }),
+      processJob(env, { type: "github-webhook", deliveryId: "draft-dodge-race-b", eventName: "pull_request", payload: draftPayload("contributor") }),
+    ]);
+
+    const patchCalls = calls.filter((c) => c.includes("PATCH") && c.includes("/pulls/42"));
+    expect(patchCalls).toHaveLength(1); // exactly one delivery won the race and closed the PR
+    const audit = await env.DB.prepare("select count(*) as n from audit_events where event_type = ? and outcome = 'completed'").bind("github_app.draft_dodge_closed").first<{ n: number }>();
+    expect(audit?.n).toBe(1); // exactly one completed close recorded — not two (the race), not zero
+  });
+
   it("no-ops when no prior gate failure exists for the PR", async () => {
     const calls: string[] = [];
     vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => {