Skip to content

fix(queue): unify the per-PR actuation lock across maintenance and draft-dodge/reopen-reclose #2447

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 2, 2026
Merged

fix(queue): unify the per-PR actuation lock across maintenance and draft-dodge/reopen-reclose #2447

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
84 changes: 26 additions & 58 deletions src/queue/processors.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1704,7 +1704,7 @@ 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 claimAgentMaintenanceLock(env, repoFullName, pr.number))) return;
if (!(await claimPrActuationLock(env, repoFullName, pr.number))) return;
try {
await runAgentMaintenancePlanAndExecute(env, {
installationId,
Expand All @@ -1718,7 +1718,7 @@ async function maybeRunAgentMaintenance(
liveFacts: args.liveFacts,
});
} finally {
await releaseAgentMaintenanceLock(env, repoFullName, pr.number);
await releasePrActuationLock(env, repoFullName, pr.number);
}
}

Expand Down Expand Up @@ -2305,25 +2305,32 @@ async function putTransientKey(
}
}

// Per-PR actuation mutex (#2135). Two DIFFERENT webhook deliveries for the same PR (e.g. a `reopened` event and
// a concurrent `check_suite completed` event) can be dequeued by separate workers at nearly the same time; both
// 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, 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: 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.
//
// 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) should never legitimately run anywhere near this long.
// (a handful of sequential GitHub API calls, or a maintenance pass's plan-and-execute) should never legitimately
// run anywhere near this long.
const PR_ACTUATION_LOCK_TTL_SECONDS = 600;
function prActuationLockKey(repoFullName: string, prNumber: number): string {
return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`;
Expand Down Expand Up @@ -2520,53 +2527,14 @@ async function claimTransientLock(
}
}

// Per-PR advisory lock around maybeRunAgentMaintenance's plan-and-execute critical section (#2129). The TTL is a
// crash-safety backstop only — the normal path releases explicitly in a finally block within a few seconds — so
// it is sized well above any realistic pass duration (matches CI_COALESCE_WINDOW_SECONDS, an already-vetted
// value for a comparable-scale operation in this file), not to bound throughput.
const AGENT_MAINTENANCE_LOCK_TTL_SECONDS = 60;

function agentMaintenanceLockKey(repoFullName: string, prNumber: number): string {
return `agent-maintenance-lock:${repoFullName.toLowerCase()}#${prNumber}`;
}

/**
* Claim the per-PR advisory lock. Returns false when another pass already holds it (caller must skip this pass
* — the next webhook/sweep tick is the backstop). A missing cache or cache hiccup fails OPEN (returns true —
* the lock is a defense-in-depth serializer, not the primary safety gate, and must never itself block actuation).
*/
export async function claimAgentMaintenanceLock(
env: Env,
repoFullName: string,
prNumber: number,
): Promise<boolean> {
return claimTransientLock(
env,
agentMaintenanceLockKey(repoFullName, prNumber),
AGENT_MAINTENANCE_LOCK_TTL_SECONDS,
);
}

/** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */
export async function releaseAgentMaintenanceLock(
env: Env,
repoFullName: string,
prNumber: number,
): Promise<void> {
try {
await env.SELFHOST_TRANSIENT_CACHE?.del?.(
agentMaintenanceLockKey(repoFullName, prNumber),
);
} 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
// code for the SAME PR at the SAME head SHA, both miss the cache, and both fire a real LLM call — which can
// return DIFFERENT verdicts). The TTL is a crash-safety backstop only (see AI_REVIEW_LOCK_TTL_SECONDS below), not
// a throughput bound — same philosophy as AGENT_MAINTENANCE_LOCK_TTL_SECONDS (#2129/#2368).
// a throughput bound — same philosophy as PR_ACTUATION_LOCK_TTL_SECONDS (#2129/#2368). Deliberately its OWN lock
// namespace, not the shared pr-actuation-lock above: this guards an expensive read-and-cache (dedup a redundant
// LLM call for the identical head+mode), not a GitHub-mutating actuation, so it has different scoping (keyed by
// head SHA + mode, not just PR) and a much longer TTL (an LLM call legitimately runs far longer than a close).
const AI_REVIEW_LOCK_TTL_SECONDS = 1_800; // 30 minutes — see justification below.

function aiReviewLockKey(repoFullName: string, prNumber: number, headSha: string, mode: string): string {
Expand Down Expand Up @@ -4642,7 +4610,7 @@ export async function runAiReviewForAdvisory(
}))
)
return undefined;
// Per-(repo, PR, head SHA, mode) advisory lock (#confirmed-bug, mirrors #2129/#2368's claimAgentMaintenanceLock):
// Per-(repo, PR, head SHA, mode) advisory lock (#confirmed-bug, mirrors #2129/#2368's claimPrActuationLock):
// a webhook pass and an agent-regate-pr sweep pass can independently reach this point for the SAME PR at the
// SAME head, both miss the cache (neither has written yet), and both fire a real, wasteful LLM call that can
// return different verdicts. Claim before the expensive section below; a pass that loses the race returns the
Expand Down
123 changes: 11 additions & 112 deletions test/unit/queue.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ import {
upsertRepositoryFromGitHub,
putCachedAiReview,
} from "../../src/db/repositories";
import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock, releaseAiReviewLock, releasePrActuationLock } from "../../src/queue/processors";
import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAiReviewLock, claimPrActuationLock, contributorEvidenceBatchSize, processJob, releaseAiReviewLock, 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";
Expand Down Expand Up @@ -3460,109 +3460,6 @@ describe("queue processors", () => {
expect(denied?.n).toBe(1);
});

it("claimAgentMaintenanceLock claims when free, denies when held (per-PR, not per-repo), and release frees it again (#2129)", async () => {
const env = createTestEnv({});
// First claim for this PR succeeds — no prior pass in-flight.
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
// A second, concurrent pass for the SAME PR (regardless of what triggered it — webhook or sweep) is denied
// while the first is still in-flight — exactly the race #2129 describes, since job-coalesce keys never
// match across trigger shapes but this lock is keyed purely on repo+PR.
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(false);
// A DIFFERENT PR in the same repo is unaffected — the lock is per-PR, not a repo-wide serializer.
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 8)).toBe(true);
// Release (the finally block's job) frees the PR — a subsequent pass can claim it again.
await releaseAgentMaintenanceLock(env, "owner/agent-repo", 7);
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
});

it("claimAgentMaintenanceLock fails OPEN on a broken transient cache — never itself blocks actuation (#2129)", 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 claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
await expect(releaseAgentMaintenanceLock(env, "owner/agent-repo", 7)).resolves.toBeUndefined();
});

it("claimAgentMaintenanceLock fails OPEN when the atomic claim primitive itself throws (#2368)", async () => {
const env = createTestEnv({
SELFHOST_TRANSIENT_CACHE: {
get: async () => null,
set: async () => undefined,
claim: async () => { throw new Error("redis unavailable"); },
},
});
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
});

it("REGRESSION (#2368): claimAgentMaintenanceLock uses an atomic check-and-set, so two genuinely concurrent claims for the SAME PR can never both succeed", async () => {
// #2368: a get-then-set pair has a window between the read and the write where two concurrent callers can
// both observe an absent key and both claim it — exactly what the per-PR lock exists to prevent. This test
// races two claims for the same PR via Promise.all (both kick off before either resolves) against the
// default test cache's claim(), which mirrors createRedisCache's atomic SET NX: the check-and-set happens
// with no `await` boundary in between, so it is impossible for both callers to see "unclaimed".
const env = createTestEnv({});
const [first, second] = await Promise.all([
claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
]);
expect([first, second].filter(Boolean)).toHaveLength(1);
});

it("REGRESSION (#2368): claimAgentMaintenanceLock 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 claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
expect(calls).toEqual(["claim"]); // never falls through to the racy get/set pair when claim is available
});

it("claimAgentMaintenanceLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#confirmed-bug, review round 2)", async () => {
// A prior version of this helper fell back to a get-then-set pair (even with an extra write-then-verify
// re-read) when claim() wasn't available. That is NOT a real exclusivity guarantee: 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". Rather than pretend to serialize via a check that
// silently fails under exactly the concurrent load this lock exists to guard against, a cache without
// claim() now gets NO exclusivity at all -- every call proceeds, sequential or concurrent, even for a key a
// previous call already "set" via get/set.
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 claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
expect(await claimAgentMaintenanceLock(env, "owner/agent-repo", 7)).toBe(true);
});

it("REGRESSION (#confirmed-bug, review round 2): claimAgentMaintenanceLock does not falsely deny — and does not falsely claim exclusivity for — two genuinely concurrent callers when the cache has no claim()", async () => {
// Documents the corrected, honest contract under the exact interleaving the gate flagged: with no atomic
// claim() primitive, BOTH concurrent callers proceed (true), because this helper no longer attempts a
// get/set-based serialization that can't actually provide exclusivity.
const values = new Map<string, string>();
const yieldThenRun = <T,>(fn: () => T): Promise<T> => new Promise((resolve) => queueMicrotask(() => resolve(fn())));
const env = createTestEnv({
SELFHOST_TRANSIENT_CACHE: {
get: (key: string) => yieldThenRun(() => values.get(key) ?? null),
set: (key: string, value: string) => yieldThenRun(() => { values.set(key, value); }),
},
});
const [first, second] = await Promise.all([
claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
claimAgentMaintenanceLock(env, "owner/agent-repo", 7),
]);
expect([first, second]).toEqual([true, true]);
});

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.
Expand Down Expand Up @@ -3677,8 +3574,9 @@ describe("queue processors", () => {
expect([first, second]).toEqual([true, true]);
});

// claimPrActuationLock (#2135) mirrors claimAgentMaintenanceLock's atomic-claim design exactly — same test
// shapes, same reasoning, a different lock namespace.
// claimPrActuationLock (#2129/#2135) is the ONE shared per-PR actuation lock: maybeRunAgentMaintenance,
// maybeCloseDraftDodgeAttempt, and maybeRecloseDisallowedReopen all claim/release the SAME key so none of the
// 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);
Expand Down Expand Up @@ -3734,9 +3632,8 @@ describe("queue processors", () => {
});

it("claimPrActuationLock returns true unconditionally when the cache has no claim() — no false exclusivity guarantee (#2135, review round 2)", async () => {
// Mirrors claimAgentMaintenanceLock's #confirmed-bug fix: a get-then-set pair (even with a re-read) is not
// a real exclusivity guarantee under concurrent load, so a cache without claim() now gets NO exclusivity at
// all rather than a fallback that only looks atomic.
// A get-then-set pair (even with a re-read) is not a real exclusivity guarantee under concurrent load, so a
// cache without claim() now gets NO exclusivity at all rather than a fallback that only looks atomic.
const values = new Map<string, string>();
const env = createTestEnv({
SELFHOST_TRANSIENT_CACHE: {
Expand Down Expand Up @@ -3798,9 +3695,11 @@ describe("queue processors", () => {
vi.setSystemTime(new Date("2026-05-28T02:00:00.000Z"));

// Simulate a webhook pass already in-flight for this exact PR — a github-webhook:pr-refresh job's coalesce
// key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the new
// per-PR advisory lock is what makes a second, independently-triggered pass defer instead of racing it.
await env.SELFHOST_TRANSIENT_CACHE?.set("agent-maintenance-lock:owner/agent-repo#7", "1", 60);
// key never matches agent-regate-pr's, so the two would never dedup against each other pre-#2129; the
// shared per-PR actuation lock is what makes a second, independently-triggered pass defer instead of racing
// it. Pre-claims the SAME pr-actuation-lock key the draft-dodge/reopen-reclose paths use (#2129/#2135,
// review round 4) — one shared namespace, not a maintenance-only lock.
await env.SELFHOST_TRANSIENT_CACHE?.set("pr-actuation-lock:owner/agent-repo#7", "1", 60);

await processJob(env, { type: "agent-regate-pr", deliveryId: "race-sweep", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 });

Expand Down