feat(queue): back per-PR transient locks with SubmissionLock DO (#8896)

scripts/gen-selfhost-env-reference.ts

@@ -32,6 +32,7 @@ const INJECTED_BINDING_NAMES = new Set([
   "RATE_LIMITER",
   "REVIEW_AUDIT",
   "SELFHOST_TRANSIENT_CACHE",
+  "SUBMISSION_LOCK",
   "VECTORIZE",
   "WEBHOOKS",
 ]);

src/env.d.ts

@@ -5,6 +5,9 @@ declare global {
     /** Self-host webhook queue binding. Cloudflare no longer binds this because hosted reviews are retired. */
     WEBHOOKS?: Queue;
     RATE_LIMITER?: DurableObjectNamespace;
+    /** Per-key exclusive mutex Durable Object (`SubmissionLock`, #8896). Optional so self-host installs
+     *  without Durable Objects keep the transient-cache lock path in `src/queue/transient-locks.ts`. */
+    SUBMISSION_LOCK?: DurableObjectNamespace;
     AI?: Ai;
     /** Self-host (RAG): a DEDICATED embedding provider, kept SEPARATE from the review chat chain so the reviewer
      *  stays frontier-only (claude-code/codex) while embeddings — which those CLIs cannot produce — route to a
@@ -70,9 +73,6 @@ declare global {
        *  `claim()` (validated at self-host boot). */
       releaseIfValue?(key: string, value: string): Promise<boolean>;
     };
-    /** 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.
-     *  Deliberately NOT declared in this chunk; the review path keeps its current concurrency behavior. */
     PUBLIC_API_ORIGIN?: string;
     PUBLIC_SITE_ORIGIN?: string;
     /** Comma-separated extra origins (each `scheme://host[:port]`) allowed as a post-GitHub-OAuth `returnTo`

src/index.ts

@@ -1,5 +1,6 @@
 import { createApp } from "./api/routes";
 import { RateLimiter } from "./auth/rate-limit";
+import { SubmissionLock } from "./queue/submission-lock";
 import { delayUntil, shouldWaitForGitHubRateLimit, LOW_REST_RATE_LIMIT_REMAINING, MAINTENANCE_RESERVED_HEADROOM } from "./github/rate-limit";
 import { processDlqBatch } from "./queue/dlq";
 import { processJob } from "./queue/processors";
@@ -42,7 +43,7 @@ const REGATE_SWEEP_TRIGGER_TYPES = ["agent-regate-sweep"] as const;
 // queueProcessingTimeoutMs(), which defaults to this sweep's own 30-min cadence) go unnoticed by the next tick.
 const BACKLOG_CONVERGENCE_SWEEP_TRIGGER_TYPES = ["backlog-convergence-sweep"] as const;
 
-export { RateLimiter };
+export { RateLimiter, SubmissionLock };
 
 export default {
   fetch: app.fetch,

src/queue/processors.ts

@@ -2435,14 +2435,14 @@ async function maybeRunAgentMaintenance(
   if (pr.isDraft) return;
   if (!gate) return;
 
-  // Per-PR mutual exclusion (#2129): a webhook re-review and a sweep-driven agent-regate-pr job use different
-  // coalesce-key shapes (jobCoalesceKey never matches one against the other) and QUEUE_CONCURRENCY explicitly
-  // overlaps I/O-bound jobs, so two passes for the SAME PR can both reach this point concurrently, each with its
-  // own independently-timed live CI/mergeable/reviewDecision read. If those reads disagree, both could plan and
-  // execute DIFFERENT actions for the same PR. Claim a short-TTL advisory lock before the plan-and-execute
-  // 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.
+  // Per-PR mutual exclusion (#2129/#8896): a webhook re-review and a sweep-driven agent-regate-pr job use
+  // different coalesce-key shapes (jobCoalesceKey never matches one against the other) and QUEUE_CONCURRENCY
+  // explicitly overlaps I/O-bound jobs, so two passes for the SAME PR can both reach this point concurrently,
+  // each with its own independently-timed live CI/mergeable/reviewDecision read. If those reads disagree, both
+  // could plan and execute DIFFERENT actions for the same PR. Claim a short-TTL advisory lock before the
+  // plan-and-execute 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. Prefers
+  // the SubmissionLock Durable Object when bound; otherwise the transient-cache mutex in transient-locks.ts.
   const actuationLock = await claimPrActuationLock(env, repoFullName, pr.number);
   if (!actuationLock.acquired) return;
   try {

src/queue/submission-lock.ts

@@ -0,0 +1,72 @@
+import { DurableObject } from "cloudflare:workers";
+
+// Per-key exclusive mutex Durable Object (#8896). One instance per lock key (`idFromName(key)`), used by
+// `claimTransientLock` when `env.SUBMISSION_LOCK` is bound. Replaces the cache-only "interim" mutex for hosted
+// Workers while self-host (no DO binding) keeps the Redis/transient-cache path unchanged.
+
+const STORAGE_KEY = "lock";
+
+type LockRecord = {
+  ownerToken: string;
+  expiresAt: number;
+};
+
+type ClaimBody = {
+  ownerToken?: unknown;
+  ttlSeconds?: unknown;
+};
+
+type ReleaseBody = {
+  ownerToken?: unknown;
+};
+
+/**
+ * Strongly-consistent per-key lock. Concurrent claims against the same DO id serialize at the platform input
+ * gate; only the first unexpired claim succeeds until release or TTL.
+ */
+export class SubmissionLock extends DurableObject<Env> {
+  constructor(ctx: DurableObjectState, env: Env) {
+    super(ctx, env);
+  }
+
+  override async fetch(request: Request): Promise<Response> {
+    const action = new URL(request.url).pathname.replace(/^\/+/, "") || "claim";
+    if (action === "claim") return this.handleClaim(request);
+    if (action === "release") return this.handleRelease(request);
+    return Response.json({ error: "unknown_action" }, { status: 404 });
+  }
+
+  private async handleClaim(request: Request): Promise<Response> {
+    const body = (await request.json().catch(() => null)) as ClaimBody | null;
+    const ownerToken = typeof body?.ownerToken === "string" ? body.ownerToken : "";
+    const ttlSeconds = typeof body?.ttlSeconds === "number" ? body.ttlSeconds : NaN;
+    if (!ownerToken || !Number.isFinite(ttlSeconds) || ttlSeconds <= 0) {
+      return Response.json({ error: "invalid_claim" }, { status: 400 });
+    }
+
+    const now = Date.now();
+    const existing = await this.ctx.storage.get<LockRecord>(STORAGE_KEY);
+    if (existing && existing.expiresAt > now && existing.ownerToken !== ownerToken) {
+      return Response.json({ acquired: false });
+    }
+
+    await this.ctx.storage.put(STORAGE_KEY, {
+      ownerToken,
+      expiresAt: now + ttlSeconds * 1000,
+    } satisfies LockRecord);
+    return Response.json({ acquired: true });
+  }
+
+  private async handleRelease(request: Request): Promise<Response> {
+    const body = (await request.json().catch(() => null)) as ReleaseBody | null;
+    const ownerToken = typeof body?.ownerToken === "string" ? body.ownerToken : "";
+    if (!ownerToken) return Response.json({ error: "invalid_release" }, { status: 400 });
+
+    const existing = await this.ctx.storage.get<LockRecord>(STORAGE_KEY);
+    if (!existing || existing.ownerToken !== ownerToken) {
+      return Response.json({ released: false });
+    }
+    await this.ctx.storage.delete(STORAGE_KEY);
+    return Response.json({ released: true });
+  }
+}

src/queue/transient-locks.ts

@@ -1,31 +1,23 @@
-// Best-effort exclusive locking against the self-host transient cache (#4013 step 1 -- extracted from
-// processors.ts, first step of the file's own module-split sequence). Two lock domains are built on the same
-// generic primitive here: the per-PR actuation mutex (below) and the per-(repo, PR, head SHA, mode) AI-review
-// lock, which stays in processors.ts (its own extraction is a later step in the split sequence) and imports
-// claimTransientLock/releaseTransientLockIfOwner/TransientLockClaim back from this module.
+// Best-effort exclusive locking (#4013 step 1 -- extracted from processors.ts). Two lock domains share the
+// same generic primitive: the per-PR actuation mutex and the per-(repo, PR, head SHA, mode) AI-review lock
+// (still wrapped in processors.ts / ai-review-orchestration.ts).
 //
-// ONE shared per-PR actuation mutex (#2129/#2135) for every mutating PR pass: the sweep/webhook-driven
-// maintenance plan-and-execute, the draft-dodge close, and the reopen-reclose. These are three INDEPENDENTLY
-// triggered webhook/sweep paths for the SAME PR (e.g. a `reopened` event and a concurrent `check_suite
-// completed` event, or a sweep tick racing either) that can be dequeued by separate workers at nearly the same
-// time; each would read its own stale-but-still-"current" state, each would pass its own freshness checks, and
-// each could independently fire a mutating call for the same PR. A single lock namespace is deliberate: separate
-// per-path locks (the original design) do not exclude each other, so a maintenance pass and a draft-dodge close
-// could still race — the whole point of this mutex is to make "does something else already own this PR" one
-// question with one answer, not one question per code path (review round 4). This is a lightweight interim
-// mutex (a full per-PR Durable Object / SubmissionLock is a separate, more-involved follow-up — see the TODO in
-// env.d.ts) built on the SAME transient cache used for CI-completion coalescing in processors.ts, claimed
-// ATOMICALLY (see claimTransientLock) so two racing deliveries can never both win the claim — a short TTL,
-// best-effort release. A lock-contended caller fails OPEN (returns false / skips this pass) rather than
-// blocking — the delivery holding the lock is evaluating the SAME PR, and the periodic sweep is the backstop
-// if this specific trigger is dropped. A cache adapter with no claim() primitive gets NO exclusivity at all
-// (every call proceeds) rather than a get-then-set pair that only *looks* atomic — see claimTransientLock's
-// doc comment for why that fallback was removed.
+// ONE shared per-PR actuation mutex (#2129/#2135) for every mutating PR pass: maintenance plan-and-execute,
+// draft-dodge close, and reopen-reclose. Independently triggered webhook/sweep paths for the SAME PR can be
+// dequeued by separate workers at nearly the same time; a single lock namespace makes "does something else
+// already own this PR" one question with one answer.
 //
-// Per-holder ownership tokens + releaseIfValue (atomic compare-and-delete) close the race a shared constant
+// Prefer the SubmissionLock Durable Object when `env.SUBMISSION_LOCK` is bound (#8896) — strongly consistent
+// per-key serialization on hosted Workers. Self-host installs without Durable Objects keep the transient-cache
+// mutex (Redis SET NX via claim()/releaseIfValue). A short TTL, best-effort release. Lock-contended callers
+// fail closed at the call site (PrActuationLockContendedError); missing DO/cache or a thrown claim fails OPEN
+// (returns acquired: true / skips exclusivity) so the lock stays defense-in-depth, never the primary safety
+// gate. A cache adapter with no claim() primitive gets NO exclusivity (every call proceeds) — see
+// claimTransientLock's doc comment.
+//
+// Per-holder ownership tokens + releaseIfValue (or DO 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.
+// delete a later claimer's live lock (#2129/#2135).
 
 import { randomUUID } from "node:crypto";
 import { RetryableJobError } from "./retryable";
@@ -39,30 +31,23 @@ export type TransientLockClaim = {
 };
 
 /**
- * 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
- * the only way to close the race between two concurrent callers each observing an absent key. A plain
- * get-then-set pair CANNOT close that race in general, even with an extra write-then-verify re-read: caller A
- * can write its own token, read it straight back, and return true entirely BEFORE caller B's later write/read
- * also completes and also returns true — both callers "win" (#confirmed-bug). Rather than pretend to serialize
- * via a check that silently fails under exactly the concurrent load this lock exists to guard against, an
- * adapter without claim() gets NO exclusivity from this helper: every caller proceeds. This is honest about the
- * 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
- * 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.
+ * Exclusive claim preferring SubmissionLock when bound (#8896), else the self-host transient cache.
+ * Requires the store's native atomic claim() (Redis SET NX) on the cache path to provide real exclusivity —
+ * a plain get-then-set pair cannot close the race. An adapter without claim() gets NO exclusivity from this
+ * helper: every caller proceeds. A missing cache/DO or a thrown claim() also fails OPEN (acquired: true) —
+ * every lock built on this helper is defense-in-depth 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.
+ * The claimed value is a fresh random token per call (#2129/#2135): release verifies this exact token still
+ * owns the key before deleting it.
  */
 export async function claimTransientLock(
   env: Env,
   key: string,
   ttlSeconds: number,
 ): Promise<TransientLockClaim> {
+  const viaDo = await claimSubmissionLockIfBound(env, key, ttlSeconds);
+  if (viaDo !== null) return viaDo;
+
   const cache = env.SELFHOST_TRANSIENT_CACHE;
   if (!cache?.claim) return { acquired: true, ownerToken: null }; // no atomic primitive — nothing to serialize against.
   // A claim()-only adapter without releaseIfValue would pin locks until TTL after normal work — reject that
@@ -80,9 +65,12 @@ export async function claimTransientLock(
 
 /** 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). */
+ *  on every fail-open claim path (nothing was actually claimed, so nothing to release). Prefers SubmissionLock
+ *  when bound (#8896); otherwise the cache path. */
 export async function releaseTransientLockIfOwner(env: Env, key: string, ownerToken: string | null): Promise<void> {
   if (!ownerToken) return;
+  if (await releaseSubmissionLockIfBound(env, key, ownerToken)) return;
+
   const cache = env.SELFHOST_TRANSIENT_CACHE;
   if (!cache?.releaseIfValue) return;
   try {
@@ -92,6 +80,47 @@ export async function releaseTransientLockIfOwner(env: Env, key: string, ownerTo
   }
 }
 
+/** Returns a claim result when SUBMISSION_LOCK is bound; `null` means "use the cache fallback". */
+async function claimSubmissionLockIfBound(
+  env: Env,
+  key: string,
+  ttlSeconds: number,
+): Promise<TransientLockClaim | null> {
+  const ns = env.SUBMISSION_LOCK;
+  if (!ns) return null;
+  const ownerToken = randomUUID();
+  try {
+    const id = ns.idFromName(key);
+    const response = await ns.get(id).fetch("https://submission-lock/claim", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ ownerToken, ttlSeconds }),
+    });
+    const body = (await response.json().catch(() => null)) as { acquired?: unknown } | null;
+    if (typeof body?.acquired !== "boolean") return { acquired: true, ownerToken: null }; // fail open
+    return { acquired: body.acquired, ownerToken: body.acquired ? ownerToken : null };
+  } catch {
+    return { acquired: true, ownerToken: null }; // fail open — same posture as the cache path
+  }
+}
+
+/** Returns true when the DO path handled the release (binding present); false means fall through to cache. */
+async function releaseSubmissionLockIfBound(env: Env, key: string, ownerToken: string): Promise<boolean> {
+  const ns = env.SUBMISSION_LOCK;
+  if (!ns) return false;
+  try {
+    const id = ns.idFromName(key);
+    await ns.get(id).fetch("https://submission-lock/release", {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify({ ownerToken }),
+    });
+  } catch {
+    // best-effort; the TTL is the backstop if release fails
+  }
+  return true;
+}
+
 const PR_ACTUATION_LOCK_TTL_SECONDS = 600;
 function prActuationLockKey(repoFullName: string, prNumber: number): string {
   return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;

src/selfhost/cf-workers-shim.ts

@@ -1,6 +1,6 @@
-// Minimal stand-in for the `cloudflare:workers` module on the Node self-host runtime. The only import of it
-// in the codebase is `DurableObject` (auth/rate-limit.ts → the RateLimiter DO). That DO is NEVER instantiated
-// on self-host — env.RATE_LIMITER is undefined, so enforceRateLimit returns null before any DO is touched —
+// Minimal stand-in for the `cloudflare:workers` module on the Node self-host runtime. Imports of
+// `DurableObject` (RateLimiter, SubmissionLock) resolve here. Those DOs are NEVER instantiated on self-host —
+// env.RATE_LIMITER / env.SUBMISSION_LOCK are undefined, so callers fall through before any DO is touched —
 // so this base class only needs to make the import + `extends DurableObject` resolve. The self-host esbuild
 // build aliases `cloudflare:workers` to this file (see the Docker build / build:selfhost script).
 export class DurableObject<E = unknown> {