From 4b08e32026440859c2fe0df04ec4b26617087c2e Mon Sep 17 00:00:00 2001 From: Andriy Polanski Date: Sun, 26 Jul 2026 13:25:04 +0000 Subject: [PATCH] feat(queue): back per-PR transient locks with SubmissionLock DO (#8896) Prefer the SubmissionLock Durable Object when bound for strongly consistent per-key mutexes; keep the cache-based path for self-host without DOs. Co-authored-by: Cursor --- scripts/gen-selfhost-env-reference.ts | 1 + src/env.d.ts | 6 +- src/index.ts | 3 +- src/queue/processors.ts | 16 +- src/queue/submission-lock.ts | 72 +++++ src/queue/transient-locks.ts | 115 ++++--- src/selfhost/cf-workers-shim.ts | 6 +- test/unit/transient-locks.test.ts | 437 +++++++++++++++++++++++++- worker-configuration.d.ts | 5 +- wrangler.jsonc | 14 +- wrangler.vitest.jsonc | 8 + 11 files changed, 610 insertions(+), 73 deletions(-) create mode 100644 src/queue/submission-lock.ts diff --git a/scripts/gen-selfhost-env-reference.ts b/scripts/gen-selfhost-env-reference.ts index 47053238e1..33952151f4 100644 --- a/scripts/gen-selfhost-env-reference.ts +++ b/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", ]); diff --git a/src/env.d.ts b/src/env.d.ts index 0ec5ed9585..64ca181301 100644 --- a/src/env.d.ts +++ b/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; }; - /** 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` diff --git a/src/index.ts b/src/index.ts index 04e44231d2..86ac703b9e 100644 --- a/src/index.ts +++ b/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, diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 818cd6671f..b399a035b0 100644 --- a/src/queue/processors.ts +++ b/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 { diff --git a/src/queue/submission-lock.ts b/src/queue/submission-lock.ts new file mode 100644 index 0000000000..ae218f6871 --- /dev/null +++ b/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 { + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env); + } + + override async fetch(request: Request): Promise { + 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 { + 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(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 { + 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(STORAGE_KEY); + if (!existing || existing.ownerToken !== ownerToken) { + return Response.json({ released: false }); + } + await this.ctx.storage.delete(STORAGE_KEY); + return Response.json({ released: true }); + } +} diff --git a/src/queue/transient-locks.ts b/src/queue/transient-locks.ts index a8f012cc6c..837af1833c 100644 --- a/src/queue/transient-locks.ts +++ b/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 { + 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 { 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 { + 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 { + 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}`; diff --git a/src/selfhost/cf-workers-shim.ts b/src/selfhost/cf-workers-shim.ts index 4307219150..8ced977f95 100644 --- a/src/selfhost/cf-workers-shim.ts +++ b/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 { diff --git a/test/unit/transient-locks.test.ts b/test/unit/transient-locks.test.ts index 12193dec28..e69bf09764 100644 --- a/test/unit/transient-locks.test.ts +++ b/test/unit/transient-locks.test.ts @@ -1,14 +1,382 @@ -import { describe, expect, it } from "vitest"; -import { claimTransientLock, releaseTransientLockIfOwner } from "../../src/queue/transient-locks"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { SubmissionLock } from "../../src/queue/submission-lock"; +import { + claimContributorCapLock, + claimPrActuationLock, + claimTransientLock, + PrActuationLockContendedError, + releaseContributorCapLock, + releasePrActuationLock, + releaseTransientLockIfOwner, +} from "../../src/queue/transient-locks"; import { createTestEnv } from "../helpers/d1"; -// #4013 step 1: claimPrActuationLock/releasePrActuationLock/claimAiReviewLock/releaseAiReviewLock's own -// extensive existing coverage (test/unit/queue.test.ts, unmoved -- see the re-export shim in processors.ts) -// already exercises claimTransientLock/releaseTransientLockIfOwner indirectly through every domain wrapper. -// This file closes the ONE gap that extraction exposed: every existing "claim() throws" test's mock cache -// omits releaseIfValue, so it hits claimTransientLock's EARLIER `!cache.releaseIfValue` fail-open branch and -// never actually reaches the try/catch around cache.claim() itself -- a pre-existing gap invisible before -// because it was diluted inside processors.ts's aggregate coverage, not something this extraction introduced. +// #8896: SubmissionLock DO + claimTransientLock preference / cache fallback. + +function memoryDurableObjectState() { + const storage = new Map(); + return { + storage: { + async get(key: string) { + return storage.get(key); + }, + async put(key: string, value: unknown) { + storage.set(key, value); + }, + async delete(key: string) { + return storage.delete(key); + }, + }, + }; +} + +function lockFromState(state = memoryDurableObjectState()) { + return new SubmissionLock(state as unknown as DurableObjectState, {} as Env); +} + +function claimRequest(ownerToken: string, ttlSeconds = 60) { + return new Request("https://submission-lock/claim", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ownerToken, ttlSeconds }), + }); +} + +function releaseRequest(ownerToken: string) { + return new Request("https://submission-lock/release", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ ownerToken }), + }); +} + +/** + * Namespace stub that routes each lock key to one in-memory SubmissionLock and serializes concurrent + * fetches per id — the same input-gate serialization real Durable Objects provide. + */ +function submissionLockNamespace(env: Env = {} as Env) { + const states = new Map>(); + const tails = new Map>(); + let claimCalls = 0; + let releaseCalls = 0; + + return { + get claimCalls() { + return claimCalls; + }, + get releaseCalls() { + return releaseCalls; + }, + idFromName(name: string) { + return name; + }, + get(id: string) { + let state = states.get(id); + if (!state) { + state = memoryDurableObjectState(); + states.set(id, state); + } + return { + async fetch(input: string, init?: RequestInit) { + const prev = tails.get(id) ?? Promise.resolve(); + let releaseGate!: () => void; + const gate = new Promise((resolve) => { + releaseGate = resolve; + }); + tails.set( + id, + prev.then(() => gate), + ); + await prev; + try { + const url = typeof input === "string" ? input : String(input); + if (url.includes("/claim")) claimCalls += 1; + if (url.includes("/release")) releaseCalls += 1; + return await new SubmissionLock(state as unknown as DurableObjectState, env).fetch( + new Request(input, init), + ); + } finally { + releaseGate(); + } + }, + }; + }, + }; +} + +describe("SubmissionLock Durable Object (#8896)", () => { + afterEach(() => { + vi.useRealTimers(); + }); + + it("grants the first claim and rejects a second concurrent claim against the same key", async () => { + const ns = submissionLockNamespace(); + const env = createTestEnv({ + SUBMISSION_LOCK: ns as unknown as DurableObjectNamespace, + }); + delete env.SELFHOST_TRANSIENT_CACHE; + + const [first, second] = await Promise.all([ + claimTransientLock(env, "pr-actuation-lock:acme/widgets#1", 600), + claimTransientLock(env, "pr-actuation-lock:acme/widgets#1", 600), + ]); + + const acquired = [first, second].filter((claim) => claim.acquired); + const denied = [first, second].filter((claim) => !claim.acquired); + expect(acquired).toHaveLength(1); + expect(denied).toHaveLength(1); + expect(acquired[0]?.ownerToken).toEqual(expect.any(String)); + expect(denied[0]?.ownerToken).toBeNull(); + expect(ns.claimCalls).toBe(2); + }); + + it("releases only when the owner token still matches, then allows a new claim", async () => { + const lock = lockFromState(); + const first = await lock.fetch(claimRequest("token-a")); + expect(first.status).toBe(200); + await expect(first.json()).resolves.toEqual({ acquired: true }); + + const foreignRelease = await lock.fetch(releaseRequest("token-b")); + await expect(foreignRelease.json()).resolves.toEqual({ released: false }); + + const stillHeld = await lock.fetch(claimRequest("token-b")); + await expect(stillHeld.json()).resolves.toEqual({ acquired: false }); + + const released = await lock.fetch(releaseRequest("token-a")); + await expect(released.json()).resolves.toEqual({ released: true }); + + const reclaim = await lock.fetch(claimRequest("token-b")); + await expect(reclaim.json()).resolves.toEqual({ acquired: true }); + }); + + it("allows a new claim after the TTL expires", async () => { + const lock = lockFromState(); + const now = vi.spyOn(Date, "now"); + now.mockReturnValue(1_000); + await expect(lock.fetch(claimRequest("token-a", 1)).then((r) => r.json())).resolves.toEqual({ + acquired: true, + }); + + now.mockReturnValue(1_500); + await expect(lock.fetch(claimRequest("token-b", 1)).then((r) => r.json())).resolves.toEqual({ + acquired: false, + }); + + now.mockReturnValue(2_001); + await expect(lock.fetch(claimRequest("token-b", 1)).then((r) => r.json())).resolves.toEqual({ + acquired: true, + }); + }); + + it("rejects malformed claim/release bodies and unknown actions", async () => { + const lock = lockFromState(); + await expect(lock.fetch(claimRequest("", 60))).resolves.toMatchObject({ status: 400 }); + await expect( + lock.fetch( + new Request("https://submission-lock/claim", { + method: "POST", + body: JSON.stringify({ ownerToken: "x", ttlSeconds: 0 }), + }), + ), + ).resolves.toMatchObject({ status: 400 }); + // Non-string ownerToken / non-number ttlSeconds → empty / NaN → invalid_claim. + await expect( + lock.fetch( + new Request("https://submission-lock/claim", { + method: "POST", + body: JSON.stringify({ ownerToken: 12, ttlSeconds: "60" }), + }), + ), + ).resolves.toMatchObject({ status: 400 }); + await expect(lock.fetch(new Request("https://submission-lock/claim", { method: "POST", body: "{" }))).resolves + .toMatchObject({ status: 400 }); + await expect(lock.fetch(releaseRequest(""))).resolves.toMatchObject({ status: 400 }); + // Non-string release token hits the `typeof … === "string"` false arm. + await expect( + lock.fetch( + new Request("https://submission-lock/release", { + method: "POST", + body: JSON.stringify({ ownerToken: false }), + }), + ), + ).resolves.toMatchObject({ status: 400 }); + await expect(lock.fetch(new Request("https://submission-lock/nope", { method: "POST", body: "{}" }))).resolves + .toMatchObject({ status: 404 }); + // Empty pathname after strip → default action "claim". + await expect( + lock.fetch( + new Request("https://submission-lock/", { + method: "POST", + body: JSON.stringify({ ownerToken: "token-default", ttlSeconds: 30 }), + }), + ).then((r) => r.json()), + ).resolves.toEqual({ acquired: true }); + }); + + it("reports released:false when nothing is held", async () => { + const lock = lockFromState(); + await expect(lock.fetch(releaseRequest("orphan")).then((r) => r.json())).resolves.toEqual({ + released: false, + }); + }); + + it("treats a same-token re-claim as a refresh while the lock is still held", async () => { + // existing.ownerToken === ownerToken arm: holder may refresh TTL without losing the claim. + const lock = lockFromState(); + await expect(lock.fetch(claimRequest("token-a", 60)).then((r) => r.json())).resolves.toEqual({ + acquired: true, + }); + await expect(lock.fetch(claimRequest("token-a", 120)).then((r) => r.json())).resolves.toEqual({ + acquired: true, + }); + }); +}); + +describe("claimTransientLock / releaseTransientLockIfOwner — cache fallback without SUBMISSION_LOCK (#8896)", () => { + it("still uses the cache claim()/releaseIfValue path when the DO binding is absent", async () => { + const claims: Array<{ key: string; value: string; ttl: number }> = []; + const releases: Array<{ key: string; value: string }> = []; + const held = new Map(); + + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async (key, value, ttlSeconds) => { + claims.push({ key, value, ttl: ttlSeconds }); + if (held.has(key)) return false; + held.set(key, value); + return true; + }, + releaseIfValue: async (key, value) => { + releases.push({ key, value }); + if (held.get(key) !== value) return false; + held.delete(key); + return true; + }, + }, + }); + delete env.SUBMISSION_LOCK; + + const first = await claimTransientLock(env, "cache-lock-key", 30); + const second = await claimTransientLock(env, "cache-lock-key", 30); + expect(first.acquired).toBe(true); + expect(first.ownerToken).toEqual(expect.any(String)); + expect(second.acquired).toBe(false); + expect(second.ownerToken).toBeNull(); + expect(claims).toHaveLength(2); + + await releaseTransientLockIfOwner(env, "cache-lock-key", first.ownerToken); + expect(releases).toEqual([{ key: "cache-lock-key", value: first.ownerToken }]); + + const reclaim = await claimTransientLock(env, "cache-lock-key", 30); + expect(reclaim.acquired).toBe(true); + }); + + it("fails open on DO transport errors without consulting the cache when the binding is present", async () => { + let cacheClaimed = false; + const env = createTestEnv({ + SUBMISSION_LOCK: { + idFromName: () => "id", + get: () => ({ + fetch: async () => { + throw new Error("do unavailable"); + }, + }), + } as unknown as DurableObjectNamespace, + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async () => { + cacheClaimed = true; + return true; + }, + releaseIfValue: async () => true, + }, + }); + + const result = await claimTransientLock(env, "key", 10); + expect(result).toEqual({ acquired: true, ownerToken: null }); + expect(cacheClaimed).toBe(false); + }); + + it("fails open when the DO returns a non-boolean acquired payload", async () => { + const env = createTestEnv({ + SUBMISSION_LOCK: { + idFromName: () => "id", + get: () => ({ + fetch: async () => Response.json({ acquired: "yes" }), + }), + } as unknown as DurableObjectNamespace, + }); + delete env.SELFHOST_TRANSIENT_CACHE; + await expect(claimTransientLock(env, "key", 10)).resolves.toEqual({ + acquired: true, + ownerToken: null, + }); + }); + + it("releases via the DO when bound and ignores a null owner token", async () => { + const ns = submissionLockNamespace(); + const env = createTestEnv({ + SUBMISSION_LOCK: ns as unknown as DurableObjectNamespace, + }); + delete env.SELFHOST_TRANSIENT_CACHE; + + const claim = await claimTransientLock(env, "release-key", 60); + expect(claim.acquired).toBe(true); + await releaseTransientLockIfOwner(env, "release-key", null); + expect(ns.releaseCalls).toBe(0); + + await releaseTransientLockIfOwner(env, "release-key", claim.ownerToken); + expect(ns.releaseCalls).toBe(1); + + const reclaim = await claimTransientLock(env, "release-key", 60); + expect(reclaim.acquired).toBe(true); + }); + + it("swallows DO release failures (TTL is the backstop)", async () => { + const env = createTestEnv({ + SUBMISSION_LOCK: { + idFromName: () => "id", + get: () => ({ + fetch: async () => { + throw new Error("release failed"); + }, + }), + } as unknown as DurableObjectNamespace, + }); + await expect(releaseTransientLockIfOwner(env, "key", "token")).resolves.toBeUndefined(); + }); +}); + +describe("domain wrappers + PrActuationLockContendedError (#8896)", () => { + it("routes claim/release for PR actuation and contributor-cap through the same lock helpers", async () => { + const ns = submissionLockNamespace(); + const env = createTestEnv({ + SUBMISSION_LOCK: ns as unknown as DurableObjectNamespace, + }); + delete env.SELFHOST_TRANSIENT_CACHE; + + const pr = await claimPrActuationLock(env, "Acme/Widgets", 7); + expect(pr.acquired).toBe(true); + await releasePrActuationLock(env, "Acme/Widgets", 7, pr.ownerToken); + + const cap = await claimContributorCapLock(env, "Acme/Widgets", "Alice"); + expect(cap.acquired).toBe(true); + await releaseContributorCapLock(env, "Acme/Widgets", "Alice", cap.ownerToken); + }); + + it("builds a fast-retry contended error with a distinct retryKind", () => { + const error = new PrActuationLockContendedError("acme/widgets", 3, "maintenance"); + expect(error).toBeInstanceOf(PrActuationLockContendedError); + expect(error.name).toBe("PrActuationLockContendedError"); + expect(error.message).toContain("acme/widgets#3"); + expect(error.message).toContain("maintenance"); + expect(error.retryAfterMs).toBe(5_000); + expect(error.retryKind).toBe("pr_actuation_lock_contended"); + }); +}); describe("claimTransientLock — the catch(cache.claim() throws) fail-open branch (#4013 step 1 gap close)", () => { it("fails OPEN when cache.claim() itself throws, even with releaseIfValue present (reaches the try/catch, not the earlier releaseIfValue guard)", async () => { @@ -22,15 +390,66 @@ describe("claimTransientLock — the catch(cache.claim() throws) fail-open branc releaseIfValue: async () => true, }, }); + delete env.SUBMISSION_LOCK; const result = await claimTransientLock(env, "some-lock-key", 600); expect(result).toEqual({ acquired: true, ownerToken: null }); }); + + it("fails OPEN when the cache has no claim() primitive", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + }, + }); + delete env.SUBMISSION_LOCK; + await expect(claimTransientLock(env, "key", 10)).resolves.toEqual({ + acquired: true, + ownerToken: null, + }); + }); + + it("fails OPEN when claim() exists without releaseIfValue (unreleasable lock shape)", async () => { + let claimed = false; + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async () => { + claimed = true; + return true; + }, + }, + }); + delete env.SUBMISSION_LOCK; + await expect(claimTransientLock(env, "key", 10)).resolves.toEqual({ + acquired: true, + ownerToken: null, + }); + expect(claimed).toBe(false); + }); }); describe("releaseTransientLockIfOwner — no-op when there's no releaseIfValue primitive to release against (#4013 step 1 gap close)", () => { it("no-ops (never throws) when SELFHOST_TRANSIENT_CACHE isn't configured at all, given a real owner token", async () => { const env = createTestEnv({}); delete env.SELFHOST_TRANSIENT_CACHE; + delete env.SUBMISSION_LOCK; await expect(releaseTransientLockIfOwner(env, "some-lock-key", "a-real-token")).resolves.toBeUndefined(); }); + + it("swallows cache releaseIfValue failures (TTL is the backstop)", async () => { + const env = createTestEnv({ + SELFHOST_TRANSIENT_CACHE: { + get: async () => null, + set: async () => undefined, + claim: async () => true, + releaseIfValue: async () => { + throw new Error("redis release failed"); + }, + }, + }); + delete env.SUBMISSION_LOCK; + await expect(releaseTransientLockIfOwner(env, "key", "token")).resolves.toBeUndefined(); + }); }); diff --git a/worker-configuration.d.ts b/worker-configuration.d.ts index 0d35a11ba8..94c3666d0c 100644 --- a/worker-configuration.d.ts +++ b/worker-configuration.d.ts @@ -1,5 +1,5 @@ /* eslint-disable */ -// Generated by Wrangler by running `wrangler types` (hash: 3fbc9150a68bb0e93fbf6bf660ccf637) +// Generated by Wrangler by running `wrangler types` (hash: 12ab7fcab3ef47d0e7c428770fc3c36a) // Runtime types generated with workerd@1.20260701.1 2026-05-28 nodejs_compat interface __BaseEnv_Env { REVIEW_AUDIT: R2Bucket; @@ -56,11 +56,12 @@ interface __BaseEnv_Env { LOOPOVER_OPEN_PR_FILE_COLLISION: "true"; LOOPOVER_SKIP_AUTOMATION_BOT_PRS: "true"; RATE_LIMITER: DurableObjectNamespace; + SUBMISSION_LOCK: DurableObjectNamespace; } declare namespace Cloudflare { interface GlobalProps { mainModule: typeof import("./src/index"); - durableNamespaces: "RateLimiter"; + durableNamespaces: "RateLimiter" | "SubmissionLock"; } interface Env extends __BaseEnv_Env {} } diff --git a/wrangler.jsonc b/wrangler.jsonc index 5649b3d75d..809b046a6e 100644 --- a/wrangler.jsonc +++ b/wrangler.jsonc @@ -286,16 +286,18 @@ "migrations_dir": "migrations", }, ], - // TODO (convergence follow-up): a per-PR LOCK Durable Object (`SubmissionLock` mutex) so concurrent - // webhook deliveries for the same PR serialize. That is a separate, more-involved sub-task — it needs a DO - // class (ported) + its own `migrations` tag (`new_sqlite_classes: ["SubmissionLock"]`) + a `durable_objects` - // binding here. NOT ported in this chunk; the review path keeps its current concurrency behavior until then. + // Per-PR / per-lock-key exclusive mutex (#8896). claimTransientLock prefers SUBMISSION_LOCK when bound; + // self-host (no DO) keeps the Redis/transient-cache path. "durable_objects": { "bindings": [ { "name": "RATE_LIMITER", "class_name": "RateLimiter", }, + { + "name": "SUBMISSION_LOCK", + "class_name": "SubmissionLock", + }, ], }, "migrations": [ @@ -303,6 +305,10 @@ "tag": "v1-rate-limiter", "new_sqlite_classes": ["RateLimiter"], }, + { + "tag": "v2-submission-lock", + "new_sqlite_classes": ["SubmissionLock"], + }, ], // Queue rename (#4768): the producer writes to loopover-jobs. The old gittensory-jobs/-dlq drain-window // consumers were removed once `wrangler queues info gittensory-jobs` confirmed 0 producers and the queue diff --git a/wrangler.vitest.jsonc b/wrangler.vitest.jsonc index f2b3bfb736..38ac21c829 100644 --- a/wrangler.vitest.jsonc +++ b/wrangler.vitest.jsonc @@ -18,6 +18,10 @@ { "name": "RATE_LIMITER", "class_name": "RateLimiter" + }, + { + "name": "SUBMISSION_LOCK", + "class_name": "SubmissionLock" } ] }, @@ -25,6 +29,10 @@ { "tag": "v1-rate-limiter", "new_sqlite_classes": ["RateLimiter"] + }, + { + "tag": "v2-submission-lock", + "new_sqlite_classes": ["SubmissionLock"] } ] }