Skip to content

refactor(queue): extract transient-lock primitives into their own module #4157

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JSONbored merged 1 commit into from
Jul 8, 2026
+181 −122
Merged

refactor(queue): extract transient-lock primitives into their own module #4157

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 13 additions & 122 deletions src/queue/processors.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -355,8 +355,20 @@ import { isDuplicateClusterWinnerByClaim, resolveDuplicateClusterWinnerNumber }
import { buildUnifiedReviewDiff, totalAddedLineCount } from "../review/review-diff";
import { estimateReviewEffort } from "../review/review-effort";
import { buildUnifiedCommentBody } from "../review/unified-comment-bridge";
import { randomUUID } from "node:crypto";
import { isRetryableJobError, RetryableJobError } from "./retryable";
import {
claimPrActuationLock,
claimTransientLock,
PrActuationLockContendedError,
releasePrActuationLock,
releaseTransientLockIfOwner,
type TransientLockClaim,
} from "./transient-locks";
// #4013 step 1: temporary re-export shim so test/unit/queue.test.ts's existing
// `import { claimPrActuationLock, releasePrActuationLock } from "../../src/queue/processors"` keeps working
// unchanged -- those tests are deeply interspersed with unrelated ones in that file, not in a cleanly
// extractable describe block, so relocating them is deliberately deferred rather than forced into this PR.
export { claimPrActuationLock, releasePrActuationLock } from "./transient-locks";
import { screenshotsAllowed } from "../review/visual-wire";
import { isVisualPath } from "../review/visual/paths";
import { buildCapture, fetchShotContentBlock, hasSuccessfulBotCapture, resolveVisualRoutes, type CaptureRoute } from "../review/visual/capture";
Expand Down Expand Up @@ -3678,66 +3690,6 @@ async function putTransientKey(
}
}

// 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 above, 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.
//
// 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}`;
}
export async function claimPrActuationLock(
env: Env,
repoFullName: string,
prNumber: number,
): Promise<TransientLockClaim> {
return claimTransientLock(
env,
prActuationLockKey(repoFullName, prNumber),
PR_ACTUATION_LOCK_TTL_SECONDS,
);
}
export async function releasePrActuationLock(
env: Env,
repoFullName: string,
prNumber: number,
ownerToken: string | null,
): Promise<void> {
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
// maybeRecloseDisallowedReopen's other error paths), but it only gets the queue's generic default backoff — far
// slower than the near-instant window a per-PR actuation lock is actually held for. Extending RetryableJobError
// gives lock contention its own fast, deterministic retry plus a distinct retryKind for observability, without
// changing the uncaught-and-propagate shape either call site already relies on (#2135/#2447).
class PrActuationLockContendedError extends RetryableJobError {
constructor(repoFullName: string, prNumber: number, policy: string) {
super(`pr actuation lock contended for ${repoFullName}#${prNumber} during ${policy}`, {
retryAfterMs: 5_000,
retryKind: "pr_actuation_lock_contended",
});
this.name = "PrActuationLockContendedError";
}
}

/**
* True when CI for this PR+headSha has been pending past `capMs`. Stamps the first-seen time in a transient
Expand Down Expand Up @@ -4040,67 +3992,6 @@ 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
* 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.
*
* 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<TransientLockClaim> {
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
// shape at self-host boot (assertSelfhostTransientCacheOwnershipRelease). At runtime, fail open without
// calling claim() so misconfigured test/custom adapters never acquire an unreleasable lock (#2129/#3153).
if (!cache.releaseIfValue) return { acquired: true, ownerToken: null };
const ownerToken = randomUUID();
try {
const acquired = await cache.claim(key, ownerToken, ttlSeconds);
return { acquired, ownerToken: acquired ? ownerToken : null };
} catch {
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). */
async function releaseTransientLockIfOwner(env: Env, key: string, ownerToken: string | null): Promise<void> {
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
}
}

// Per-(repo, PR, head SHA) advisory lock around runAiReviewForAdvisory's expensive grounding/RAG/enrichment/LLM
// section (#confirmed-bug: a webhook pass and an agent-regate-pr sweep pass can independently reach this same
Expand Down
132 changes: 132 additions & 0 deletions src/queue/transient-locks.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
// 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.
//
// 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.
//
// 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.

import { randomUUID } from "node:crypto";
import { RetryableJobError } from "./retryable";

/** 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
* 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.
*
* 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.
*/
export async function claimTransientLock(
env: Env,
key: string,
ttlSeconds: number,
): Promise<TransientLockClaim> {
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
// shape at self-host boot (assertSelfhostTransientCacheOwnershipRelease). At runtime, fail open without
// calling claim() so misconfigured test/custom adapters never acquire an unreleasable lock (#2129/#3153).
if (!cache.releaseIfValue) return { acquired: true, ownerToken: null };
const ownerToken = randomUUID();
try {
const acquired = await cache.claim(key, ownerToken, ttlSeconds);
return { acquired, ownerToken: acquired ? ownerToken : null };
} catch {
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). */
export async function releaseTransientLockIfOwner(env: Env, key: string, ownerToken: string | null): Promise<void> {
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
}
}

const PR_ACTUATION_LOCK_TTL_SECONDS = 600;
function prActuationLockKey(repoFullName: string, prNumber: number): string {
return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;
}
export async function claimPrActuationLock(
env: Env,
repoFullName: string,
prNumber: number,
): Promise<TransientLockClaim> {
return claimTransientLock(
env,
prActuationLockKey(repoFullName, prNumber),
PR_ACTUATION_LOCK_TTL_SECONDS,
);
}
export async function releasePrActuationLock(
env: Env,
repoFullName: string,
prNumber: number,
ownerToken: string | null,
): Promise<void> {
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
// maybeRecloseDisallowedReopen's other error paths), but it only gets the queue's generic default backoff — far
// slower than the near-instant window a per-PR actuation lock is actually held for. Extending RetryableJobError
// gives lock contention its own fast, deterministic retry plus a distinct retryKind for observability, without
// changing the uncaught-and-propagate shape either call site already relies on (#2135/#2447).
export class PrActuationLockContendedError extends RetryableJobError {
constructor(repoFullName: string, prNumber: number, policy: string) {
super(`pr actuation lock contended for ${repoFullName}#${prNumber} during ${policy}`, {
retryAfterMs: 5_000,
retryKind: "pr_actuation_lock_contended",
});
this.name = "PrActuationLockContendedError";
}
}
36 changes: 36 additions & 0 deletions test/unit/transient-locks.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
import { describe, expect, it } from "vitest";
import { claimTransientLock, 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.

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 () => {
const env = createTestEnv({
SELFHOST_TRANSIENT_CACHE: {
get: async () => null,
set: async () => undefined,
claim: async () => {
throw new Error("redis unavailable");
},
releaseIfValue: async () => true,
},
});
const result = await claimTransientLock(env, "some-lock-key", 600);
expect(result).toEqual({ acquired: true, ownerToken: null });
});
});

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;
await expect(releaseTransientLockIfOwner(env, "some-lock-key", "a-real-token")).resolves.toBeUndefined();
});
});