From 1883573d580dfe7d38e7c3c6b3cc5305291eafff Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 1 Jul 2026 05:56:32 -0700 Subject: [PATCH 1/2] fix(queue): serialize concurrent webhook/sweep passes for the same PR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit check_run/check_suite completion, pull_request synchronize, and the scheduled agent-regate-pr sweep all key their job coalescing differently (github-webhook:ci-completed:..., github-webhook:pr-refresh:..., and agent-regate-pr:... respectively), so they never dedup against each other. Combined with QUEUE_CONCURRENCY explicitly overlapping I/O-bound jobs, two of these can be claimed and processed concurrently for the same PR, each computing its own independently-timed live CI/mergeable/ reviewDecision snapshot — a narrow window where disagreeing reads could race to plan and execute different actions for the same PR. Add a short-TTL per-PR advisory lock (agent-maintenance-lock:{repo}#{pr} via the existing SELFHOST_TRANSIENT_CACHE, the same mechanism the CI-completion re-review coalesce window already uses) around the plan-and-execute critical section of maybeRunAgentMaintenance. A pass that can't claim the lock defers cleanly; the next webhook/sweep tick is the backstop. The critical section is extracted into runAgentMaintenancePlanAndExecute so the try/finally wrapping it doesn't force-reindent the whole existing block. Both the direct webhook path and the sweep path (via reReviewStoredPullRequest) converge on the same maybeRunAgentMaintenance call, so wrapping it there covers both without touching either caller. Lightweight stand-in for the per-PR SubmissionLock Durable Object already noted as a longer-term TODO in env.d.ts. --- src/queue/processors.ts | 86 +++++++++++++++++++++++++++++++++++++++++ test/unit/queue.test.ts | 75 ++++++++++++++++++++++++++++++++++- 2 files changed, 160 insertions(+), 1 deletion(-) diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 7f32dfc58b..9d6cda1e42 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -1571,6 +1571,48 @@ 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. + if (!(await claimAgentMaintenanceLock(env, repoFullName, pr.number))) return; + try { + await runAgentMaintenancePlanAndExecute(env, { + installationId, + repoFullName, + repo: args.repo, + pr, + settings, + otherOpenPullRequests, + gate, + liveFacts: args.liveFacts, + }); + } finally { + await releaseAgentMaintenanceLock(env, repoFullName, pr.number); + } +} + +/** The plan-and-execute critical section of {@link maybeRunAgentMaintenance}, extracted so the caller's + * per-PR lock (#2129) wraps it in a try/finally without reindenting this whole block. */ +async function runAgentMaintenancePlanAndExecute( + env: Env, + args: { + installationId: number; + repoFullName: string; + repo: Awaited>; + pr: PullRequestRecord; + settings: RepositorySettings; + otherOpenPullRequests: PullRequestRecord[]; + gate: ReturnType; + liveFacts: LiveGithubFacts; + }, +): Promise { + const { installationId, repoFullName, pr, settings, otherOpenPullRequests, gate } = args; + // Convergence safety: feed the planner the PR's changed paths + the repo's hard-guardrail globs so guarded // paths force manual review, and flag owner-authored PRs so they are never auto-closed (standing rule). // FIX B: resolve files via the shared resolver so an EMPTY stored list (the maintenance ran before the @@ -2184,6 +2226,50 @@ async function ciHeadShaResolutionCoalesced( ); } +// 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 { + // getTransientKey/putTransientKey already fail open internally (a missing cache or a thrown read/write error + // both resolve rather than throw), so this never needs its own try/catch — a cache fault surfaces here as + // "no lock held", which correctly falls through to claiming it. + const key = agentMaintenanceLockKey(repoFullName, prNumber); + if (await getTransientKey(env, key)) return false; // another pass is already in-flight for this PR + await putTransientKey(env, key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS); + return true; +} + +/** 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 { + try { + await env.SELFHOST_TRANSIENT_CACHE?.del?.( + agentMaintenanceLockKey(repoFullName, prNumber), + ); + } catch { + // best-effort; the TTL is the backstop if release fails + } +} + /** Read the CI head SHA off a `check_suite`/`check_run` `completed` payload (the event node carries `head_sha`; * `check_run` also nests it under `check_suite.head_sha`). Returns "" when absent. The payload type doesn't model * these events, so we narrow off `Record` the same way the `pull_requests[]` read does. */ diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index ec69054c93..be705aa673 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -44,7 +44,7 @@ import { upsertRepositoryFromGitHub, putCachedAiReview, } from "../../src/db/repositories"; -import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, contributorEvidenceBatchSize, processJob } from "../../src/queue/processors"; +import { agentMaintenanceHeadMatchesGate, changedPathsForGuardrail, claimAgentMaintenanceLock, contributorEvidenceBatchSize, processJob, releaseAgentMaintenanceLock } from "../../src/queue/processors"; import { upsertRepoFocusManifest } from "../../src/signals/focus-manifest-loader"; import { normalizeRegistryPayload } from "../../src/registry/normalize"; import { persistRegistrySnapshot } from "../../src/registry/sync"; @@ -2324,6 +2324,79 @@ 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("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => { + const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); + await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); + await upsertRepositoryFromGitHub(env, { name: "agent-repo", full_name: "owner/agent-repo", private: false, owner: { login: "owner" } }, 9001); + await upsertRepositorySettings(env, { repoFullName: "owner/agent-repo", autonomy: { merge: "auto" }, aiReviewMode: "off", gatePack: "oss-anti-slop", gateCheckMode: "enabled", checkRunMode: "off", commentMode: "off", publicSurface: "off" }); + await upsertPullRequestFromGitHub(env, "owner/agent-repo", { number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1" }); + let mergeCalls = 0; + vi.stubGlobal("fetch", async (input: RequestInfo | URL, init?: RequestInit) => { + const url = input.toString(); + if (url.includes("/access_tokens")) return Response.json({ token: "installation-token" }); + if (url.includes("/pulls/7/files")) return Response.json([{ filename: "src/a.ts", status: "modified", additions: 1, deletions: 0, changes: 1, patch: "@@\n+export const ok = true;" }]); + if (url.includes("/pulls/7/merge")) { + mergeCalls += 1; + return new Response(null, { status: 204 }); + } + if (url.includes("/pulls/7/reviews") && init?.method === "POST") return Response.json({ id: 1 }); + if (url.includes("/pulls/7/reviews")) return Response.json([]); + // Only the bare PR resource (no sub-path) — the more specific checks above already claimed + // /pulls/7/files, /pulls/7/merge, and /pulls/7/reviews. + if (/\/pulls\/7(\?|$)/.test(url)) return Response.json({ number: 7, title: "Clean PR", state: "open", user: { login: "contributor" }, head: { sha: "a7" }, labels: [], body: "Closes #1", mergeable_state: "clean" }); + if (url.includes("/commits/a7/check-runs")) return Response.json({ total_count: 0, check_runs: [] }); + if (url.includes("/commits/a7/status")) return Response.json({ state: "success", statuses: [] }); + if (url.includes("/commits/a7/check-suites")) return Response.json({ check_suites: [] }); + if (url.includes("/issues/1")) return Response.json({ number: 1, title: "Issue", state: "open", labels: [], user: { login: "reporter" } }); + if (url.includes("/branches/")) return Response.json({ protected: false, protection: { required_status_checks: { contexts: [] } } }); + if (url.includes(".gittensory.yml")) return new Response("Not Found", { status: 404 }); + if (url.endsWith("/check-runs") && init?.method === "POST") return Response.json({ id: 1 }); + if (url === "https://api.gittensor.io/miners") return Response.json([]); + if (url.endsWith("/graphql")) return Response.json({ data: {} }); + return Response.json({}); + }); + 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); + + await processJob(env, { type: "agent-regate-pr", deliveryId: "race-sweep", repoFullName: "owner/agent-repo", prNumber: 7, installationId: 9001 }); + + // The held lock made this pass skip its plan-and-execute critical section entirely — no mutation attempted. + expect(mergeCalls).toBe(0); + const actionAudits = await env.DB.prepare("select count(*) as n from audit_events where event_type like 'agent.action.%'").first<{ n: number }>(); + expect(actionAudits?.n).toBe(0); + }); + it("the sweep stamps the marker INLINE when the repo has no installation (audit-only, still converges) (#audit-sweep-fanout)", async () => { const sent: import("../../src/types").JobMessage[] = []; const env = createTestEnv({ JOBS: { async send(m: import("../../src/types").JobMessage) { sent.push(m); } } as unknown as Queue }); From 5a40e513728e4ea1df79a917598477bb924d2a2b Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 1 Jul 2026 13:52:56 -0700 Subject: [PATCH 2/2] fix(queue): make the per-PR agent-maintenance lock atomic claimAgentMaintenanceLock performed getTransientKey then putTransientKey as two separate operations, so two concurrent passes for the same PR (e.g. a webhook and the sweep, whose job-coalesce keys never match each other) could both observe an absent key, both write it, and both proceed to execute conflicting maintenance actions -- defeating the serializer the lock exists to provide. Add an atomic claim(key, value, ttlSeconds) primitive to the SELFHOST_TRANSIENT_CACHE interface, backed in production by Redis's SET ... EX ttl NX (a single server-side check-and-set, so no window exists where two callers can both see "absent"). Wire claimAgentMaintenanceLock to use it, falling back to the prior get/set pair only for a cache adapter that hasn't implemented claim yet -- strictly no worse than the previous behavior, never a regression. Add a genuine concurrent-race regression test (races two claims for the same PR via Promise.all and asserts exactly one wins), plus dedicated coverage for the new claim() primitive itself and the fallback path. --- src/env.d.ts | 6 +++ src/queue/processors.ts | 14 ++++++- src/selfhost/redis-cache.ts | 7 ++++ test/helpers/d1.ts | 9 +++++ test/unit/queue.test.ts | 53 ++++++++++++++++++++++++++ test/unit/selfhost-redis-cache.test.ts | 27 ++++++++++++- 6 files changed, 113 insertions(+), 3 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 7053788fe7..9169af8c3b 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -28,6 +28,12 @@ declare global { get(key: string): Promise; set(key: string, value: string, ttlSeconds: number): Promise; del?(key: string): Promise; + /** Atomic "set only if absent": returns true when this call newly claimed the key, false when it was + * already held by someone else. Unlike a get-then-set pair, there is no window where two concurrent + * callers can both observe an absent key and both claim it — the store (e.g. Redis SET NX) performs the + * check-and-set as one operation. Optional so a cache adapter that hasn't implemented it yet still + * type-checks; callers fall back to the non-atomic get/set pair when absent (#2129). */ + claim?(key: string, value: string, ttlSeconds: number): 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. diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 65f0f36f9a..09c5b961a8 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -2369,10 +2369,22 @@ export async function claimAgentMaintenanceLock( repoFullName: string, prNumber: number, ): Promise { + const key = agentMaintenanceLockKey(repoFullName, prNumber); + // Atomic claim (#2129): a get-then-set pair has a window between the read and the write where two concurrent + // passes for the SAME PR can both observe an absent key and both claim it, defeating the serializer entirely. + // env.SELFHOST_TRANSIENT_CACHE.claim performs the check-and-set as one operation (Redis SET NX server-side), + // closing that window. Falls back to the non-atomic get/set pair only for a cache adapter that hasn't + // implemented claim yet — strictly no worse than this function's prior behavior. + if (env.SELFHOST_TRANSIENT_CACHE?.claim) { + try { + return await env.SELFHOST_TRANSIENT_CACHE.claim(key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS); + } catch { + return true; // fail open — see the doc comment above. + } + } // getTransientKey/putTransientKey already fail open internally (a missing cache or a thrown read/write error // both resolve rather than throw), so this never needs its own try/catch — a cache fault surfaces here as // "no lock held", which correctly falls through to claiming it. - const key = agentMaintenanceLockKey(repoFullName, prNumber); if (await getTransientKey(env, key)) return false; // another pass is already in-flight for this PR await putTransientKey(env, key, "1", AGENT_MAINTENANCE_LOCK_TTL_SECONDS); return true; diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index d5d6b13d37..b38297ad8d 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -16,6 +16,13 @@ export function createRedisCache(redis: Redis) { async del(key: string): Promise { await redis.del(key); }, + // Redis performs the existence check and the write as a single atomic command server-side (SET ... NX), so + // two concurrent callers racing on the same key can never both receive "OK" -- unlike a get-then-set pair, + // which has a window between the read and the write where both callers can observe an absent key. + async claim(key: string, value: string, ttlSeconds: number): Promise { + const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); + return result === "OK"; + }, }; } diff --git a/test/helpers/d1.ts b/test/helpers/d1.ts index 324244c8c4..984840531b 100644 --- a/test/helpers/d1.ts +++ b/test/helpers/d1.ts @@ -89,6 +89,15 @@ export function createTestEnv(overrides: Partial = {}): Env { async del(key: string) { transientCache.delete(key); }, + // Mirrors createRedisCache's atomic claim (#2129): the check-and-set below has no `await` between the + // `has` read and the `set` write, so it completes synchronously within one microtask — a concurrent + // caller can never observe the key as absent partway through another caller's claim, matching Redis's + // SET NX server-side atomicity. + async claim(key: string, value: string) { + if (transientCache.has(key)) return false; + transientCache.set(key, value); + return true; + }, }, // Per-repo review allowlist: default to the test repos so flag-ON wiring tests activate the // gated review features. Override to "" to assert the dormant (no-repo) default. diff --git a/test/unit/queue.test.ts b/test/unit/queue.test.ts index 1260634942..0dc379929c 100644 --- a/test/unit/queue.test.ts +++ b/test/unit/queue.test.ts @@ -2949,6 +2949,59 @@ describe("queue processors", () => { 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 falls back to the get/set pair and still denies a held key when the cache has no claim() (#2368)", async () => { + // A cache adapter that hasn't implemented the atomic claim() primitive yet must still deny a second claim — + // just via the older, non-atomic get-then-set pair (documented as strictly no worse than this function's + // prior behavior), not by skipping the check entirely. + const values = new Map(); + 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(false); + }); + it("INVARIANT (#2129 per-PR lock): a maintenance pass defers when another pass already holds the PR's lock", async () => { const env = createTestEnv({ GITHUB_APP_PRIVATE_KEY: await generatePrivateKeyPem() }); await upsertInstallation(env, { action: "created", installation: { id: 9001, account: { login: "owner", id: 1, type: "Organization" }, target_type: "Organization", repository_selection: "selected", permissions: {}, events: [] } }); diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index d350afb5cf..b4bf3d5afd 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -2,7 +2,9 @@ import type { Redis } from "ioredis"; import { describe, expect, it } from "vitest"; import { checkAndMarkDelivery, createRedisCache } from "../../src/selfhost/redis-cache"; -/** Minimal in-memory stand-in for the ioredis methods the cache uses. */ +/** Minimal in-memory stand-in for the ioredis methods the cache uses. Emulates real Redis SET NX + * semantics (refuse + return null when NX is requested and the key already exists) so a test + * using this fake actually exercises the atomicity claim() depends on, not just a plain overwrite. */ function fakeRedis(): Redis & { _store: Map } { const _store = new Map(); return { @@ -10,7 +12,8 @@ function fakeRedis(): Redis & { _store: Map } { async get(k: string) { return _store.get(k) ?? null; }, - async set(k: string, v: string, _ex: "EX", _ttl: number) { + async set(k: string, v: string, _ex: "EX", _ttl: number, nx?: "NX") { + if (nx === "NX" && _store.has(k)) return null; _store.set(k, v); return "OK"; }, @@ -40,6 +43,26 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { await cache.del("k"); expect(await cache.get("k")).toBeNull(); }); + + it("claim atomically sets an absent key and returns true (#2129)", async () => { + const cache = createRedisCache(fakeRedis()); + expect(await cache.claim("lock", "1", 60)).toBe(true); + expect(await cache.get("lock")).toBe("1"); + }); + + it("claim refuses and returns false when the key is already held, without overwriting it (#2129)", async () => { + const r = fakeRedis(); + const cache = createRedisCache(r); + await cache.set("lock", "holder-A", 60); + expect(await cache.claim("lock", "holder-B", 60)).toBe(false); + expect(await cache.get("lock")).toBe("holder-A"); // the second claimant never overwrote the first + }); + + it("claim propagates a Redis error to the caller (claimAgentMaintenanceLock is responsible for failing open)", async () => { + const brokenRedis = { async set() { throw new Error("connection refused"); } } as unknown as Redis; + const cache = createRedisCache(brokenRedis); + await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused"); + }); }); describe("checkAndMarkDelivery (#1216 webhook idempotency)", () => {