From 00f5f973fd41302fc5f190f57b425e3eff4d54da Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 18:09:18 -0700 Subject: [PATCH 1/2] fix(locks): renew a held lock while its work runs, so a slow pass cannot be claimed out from under (#9467) The transient-lock TTLs were sized when the actuation lock covered a short plan-and-execute section. #9013 moved the claim to BEFORE maybePublishPrPublicSurface, so the 600s actuation lock now spans the whole publish -> AI review -> maintain unit -- and the AI review alone can exceed it (3 attempts x up to 600s per attempt per model, with an operator override clamped at 30 minutes PER ATTEMPT). When the TTL lapsed mid-work the holder was never told: a second worker claimed the same PR and both actuated it, producing a duplicate close plus a duplicate explanation comment and duplicate decision records, or a losing pass's placeholder republishing over a real verdict -- the exact thrash #9013 existed to eliminate. The AI-review lock has the same shape: its 1800s TTL is exactly one model's max-effort retry budget, so a second reviewer or a per-repo timeout override runs past it and the two passes' ai_review_cache upserts race. Renewal is compare-and-extend, mirroring the compare-and-delete release and for the same reason: a holder whose key already lapsed and was re-claimed must NOT extend the new owner's lock. It learns it lost instead, so a caller can abort before mutating anything. Fails open throughout, matching every other operation in this module: an adapter without renewIfValue, a fail-open claim with no token, or a throwing renewal all leave the lock on its original fixed TTL -- never worse than before, and never a reason to block real work. A transient renewal error is deliberately NOT treated as losing the lock, since that would abort work that is still legitimately holding it. The end-to-end tests drive claim -> heartbeat -> competing claim against a cache that genuinely expires keys, so the invariant asserted is the one that matters -- a competing pass stays refused while the holder works -- and the paired regression shows the same lock lapsing without the heartbeat. --- src/env.d.ts | 6 + src/queue/ai-review-orchestration.ts | 34 ++++ src/queue/processors.ts | 25 +++ src/queue/transient-locks.ts | 68 ++++++- src/selfhost/redis-cache.ts | 14 ++ test/unit/selfhost-redis-cache.test.ts | 54 +++++- test/unit/transient-locks.test.ts | 244 +++++++++++++++++++++++++ 7 files changed, 439 insertions(+), 6 deletions(-) diff --git a/src/env.d.ts b/src/env.d.ts index 0ab733c1d2..a54a6b95f2 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -72,6 +72,12 @@ declare global { * deleting a different holder's live claim on the same key. Required on any adapter that implements * `claim()` (validated at self-host boot). */ releaseIfValue?(key: string, value: string): Promise; + /** Atomic compare-and-extend: resets `key`'s TTL only when its current value equals `value`, returning + * whether it was extended (#9467). The compare is what makes a renewal safe — a holder whose lock has + * already expired and been re-claimed by someone else must NOT extend the new holder's key, and must be + * able to learn that it lost ownership. Optional: an adapter without it simply gets no renewal, and the + * lock behaves exactly as it did before (fixed TTL). */ + renewIfValue?(key: string, value: string, ttlSeconds: number): Promise; }; PUBLIC_API_ORIGIN?: string; PUBLIC_SITE_ORIGIN?: string; diff --git a/src/queue/ai-review-orchestration.ts b/src/queue/ai-review-orchestration.ts index ca12e53088..129f56f0d1 100644 --- a/src/queue/ai-review-orchestration.ts +++ b/src/queue/ai-review-orchestration.ts @@ -17,6 +17,8 @@ import { claimTransientLock, releaseTransientLockIfOwner, + startLockHeartbeat, + type LockHeartbeat, type TransientLockClaim, } from "./transient-locks"; import { buildPullRequestAdvisory } from "../rules/advisory"; @@ -134,6 +136,38 @@ export async function claimAiReviewLock( return claim; } +/** + * #9467: keep an acquired AI-review lock alive for as long as the review is actually running. + * + * The 1800s TTL is documented as "a crash-safety backstop, not a throughput bound" -- but nothing bounded the + * work below it. At max effort a SINGLE model's retry budget is exactly 3 x 600s = 1800s, so any second + * reviewer, fallback chain, or per-repo claudeTimeoutMs override (clamped at 30 min PER ATTEMPT, #8458) runs + * past it. A second pass then claimed the same key, fired a duplicate LLM call, and the two passes' + * ai_review_cache upserts raced -- last writer wins, so two contradictory verdicts could alternate across + * passes at an unchanged head. + * + * The caller stops it in the same finally that releases the lock. See startLockHeartbeat for the fail-open + * posture: no compare-and-extend support, a fail-open claim, or a throwing renewal all leave the lock on its + * original fixed TTL rather than blocking work. + */ +export function startAiReviewLockHeartbeat( + env: Env, + repoFullName: string, + prNumber: number, + headSha: string, + mode: string, + ownerToken: string | null, + options?: { onLost?: () => void }, +): LockHeartbeat { + return startLockHeartbeat( + env, + aiReviewLockKey(repoFullName, prNumber, headSha, mode), + ownerToken, + AI_REVIEW_LOCK_TTL_SECONDS, + options, + ); +} + /** Best-effort release, called from a finally block so the lock frees promptly instead of waiting out the TTL. */ export async function releaseAiReviewLock( env: Env, diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 166891cce8..c84aed08c3 100644 --- a/src/queue/processors.ts +++ b/src/queue/processors.ts @@ -362,6 +362,8 @@ import { // unchanged -- those tests are deeply interspersed with unrelated ones in that file family, 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 { PR_ACTUATION_LOCK_TTL_SECONDS, prActuationLockKeyForHeartbeat, startLockHeartbeat } from "./transient-locks"; +import { startAiReviewLockHeartbeat } from "./ai-review-orchestration"; // #4013 step 2: same shim shape for generateSignalSnapshots -- imported here for processJob's own internal // call below, and re-exported so src/api/routes.ts and test/unit/queue-trends.test.ts's existing // `import { generateSignalSnapshots } from "../../src/queue/processors"` keeps working unchanged. @@ -4227,6 +4229,16 @@ export async function reReviewStoredPullRequest( }).catch(() => undefined); throw new PrActuationLockContendedError(repoFullName, pr.number, "public-surface-publish"); } + // #9467: this lock now spans the WHOLE publish -> AI review -> maintain unit (#9013 moved the claim here), + // and the AI review alone can outlive the 600s TTL. Renew it while the work runs so a slow-but-healthy pass + // cannot have its lock claimed out from under it mid-flight. Compare-and-extend, so if this pass has already + // lost the key it learns that instead of extending the new owner's lock. + const actuationHeartbeat = startLockHeartbeat( + env, + prActuationLockKeyForHeartbeat(repoFullName, pr.number), + actuationLock.ownerToken, + PR_ACTUATION_LOCK_TTL_SECONDS, + ); let gate: ReturnType | undefined; try { gate = await withReviewPipelineSpan( @@ -4312,6 +4324,7 @@ export async function reReviewStoredPullRequest( ); }); } finally { + actuationHeartbeat.stop(); await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken); } return true; @@ -11127,9 +11140,21 @@ async function maybePublishPrPublicSurface( }).catch(() => undefined); aiReview = aiReviewLockContendedResult(advisory); } else { + // #9467: an LLM call can outlive the 1800s TTL (a single model's max-effort retry budget is exactly + // 3 x 600s, before a second reviewer or a per-repo timeout override). Renew while the review runs so a + // slow-but-healthy pass cannot have its lock claimed out from under it and pay for a duplicate call. + const aiReviewHeartbeat = startAiReviewLockHeartbeat( + env, + repoFullName, + pr.number, + aiReviewHeadSha, + settings.aiReviewMode, + aiReviewLock.ownerToken, + ); try { await aiReviewCacheReadDecideAndRun(aiReviewLock); } finally { + aiReviewHeartbeat.stop(); await releaseAiReviewLock( env, repoFullName, diff --git a/src/queue/transient-locks.ts b/src/queue/transient-locks.ts index 06ac669900..2a8f4d6d4f 100644 --- a/src/queue/transient-locks.ts +++ b/src/queue/transient-locks.ts @@ -136,7 +136,73 @@ async function releaseSubmissionLockIfBound(env: Env, key: string, ownerToken: s return true; } -const PR_ACTUATION_LOCK_TTL_SECONDS = 600; +/** + * #9467: keep a held lock alive for as long as its owner is actually working, instead of betting that the work + * finishes inside a fixed TTL. + * + * The TTLs were sized when the actuation lock covered only a short plan-and-execute section. #9013 moved the + * claim to BEFORE `maybePublishPrPublicSurface`, so the 600s actuation lock now wraps the entire publish -> AI + * review -> maintain unit -- and the AI review alone can legitimately exceed it (3 attempts x up to 600s per + * attempt, per model, with an operator override clamped at 1,800,000 ms per attempt). When the TTL lapsed + * mid-work the holder was never told: a second worker claimed the same PR and both actuated it, producing a + * duplicate close plus a duplicate explanation comment, or a losing pass's placeholder republished over a real + * verdict -- the exact thrash #9013 existed to eliminate. + * + * The queue proved this shape for its own processing lease in #9023; this is the same idea for transient locks. + * Renewal is compare-and-extend, so a holder that has ALREADY lost the key (TTL lapsed and someone re-claimed, + * or a maintainer stole it via #9008) cannot extend the new owner's lock -- it learns it lost instead, via + * `onLost`, and its caller aborts before mutating anything. + * + * Fails OPEN, exactly like every other operation in this module: an adapter without `renewIfValue`, or a + * throwing renewal, simply leaves the lock on its original fixed TTL -- never worse than the pre-#9467 + * behavior, and never a reason to block real work. + */ +export type LockHeartbeat = { stop: () => void }; + +export function startLockHeartbeat( + env: Env, + key: string, + ownerToken: string | null, + ttlSeconds: number, + options?: { onLost?: () => void; intervalMsOverride?: number }, +): LockHeartbeat { + const cache = env.SELFHOST_TRANSIENT_CACHE; + // Nothing to renew: a fail-open claim owns no key, and an adapter without compare-and-extend cannot renew + // safely (a blind re-set would extend whoever holds it now, which is the bug this exists to prevent). + if (!ownerToken || !cache?.renewIfValue) return { stop: () => undefined }; + // A third of the TTL gives two chances to recover from a transient renewal failure before the lock lapses. + const intervalMs = options?.intervalMsOverride ?? Math.max(1_000, Math.floor((ttlSeconds * 1000) / 3)); + let stopped = false; + const timer = setInterval(() => { + void (async () => { + if (stopped) return; + try { + const stillOurs = await cache.renewIfValue!(key, ownerToken, ttlSeconds); + if (!stillOurs && !stopped) { + stopped = true; + clearInterval(timer); + options?.onLost?.(); + } + } catch { + // Fail open: a transient renewal error must not itself abort the work. The next tick retries, and if + // renewals keep failing the lock simply lapses on its original TTL -- the pre-#9467 behavior. + } + })(); + }, intervalMs); + // Never hold the process open for a heartbeat (Node only; the Workers runtime has no unref). + (timer as unknown as { unref?: () => void }).unref?.(); + return { + stop: () => { + stopped = true; + clearInterval(timer); + }, + }; +} + +export const PR_ACTUATION_LOCK_TTL_SECONDS = 600; +export function prActuationLockKeyForHeartbeat(repoFullName: string, prNumber: number): string { + return prActuationLockKey(repoFullName, prNumber); +} function prActuationLockKey(repoFullName: string, prNumber: number): string { return `pr-actuation-lock:${repoFullName.toLowerCase()}#${prNumber}`; } diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index 1b68a46c9a..ca9e273944 100644 --- a/src/selfhost/redis-cache.ts +++ b/src/selfhost/redis-cache.ts @@ -71,6 +71,20 @@ export function createRedisCache(redis: Redis) { const result = await redis.set(key, value, "EX", ttlSeconds, "NX"); return result === "OK"; }, + // Compare-and-extend (#9467): same atomicity requirement as releaseIfValue below, for the same reason -- + // a GET followed by a separate EXPIRE could extend a key that a NEW claimant wrote in between, handing the + // stale holder's renewal to the live holder's lock. Returning false is how a holder discovers it no longer + // owns the key (its TTL lapsed and someone else claimed it, or a maintainer stole it, #9008). + async renewIfValue(key: string, value: string, ttlSeconds: number): Promise { + const result = await redis.eval( + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('expire', KEYS[1], ARGV[2]) else return 0 end", + 1, + key, + value, + String(ttlSeconds), + ); + return result === 1; + }, // Compare-and-delete: the read and the delete must be one atomic server-side step (a Lua eval), or a // holder's own release could race a NEW claimant's write between a separate GET and DEL and delete the // wrong holder's key -- the exact race per-holder ownership tokens exist to close. diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index c2f9954c0e..f60838b396 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -15,29 +15,39 @@ import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; /** 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 } { +function fakeRedis(): Redis & { _store: Map; _ttls: Map } { const _store = new Map(); + const _ttls = new Map(); return { _store, + _ttls, async get(k: string) { return _store.get(k) ?? null; }, - async set(k: string, v: string, _ex: "EX", _ttl: number, nx?: "NX") { + async set(k: string, v: string, _ex: "EX", ttl: number, nx?: "NX") { if (nx === "NX" && _store.has(k)) return null; _store.set(k, v); + _ttls.set(k, ttl); return "OK"; }, async del(k: string) { _store.delete(k); return 1; }, - // Emulates the Lua eval releaseIfValue runs: delete k only when its stored value equals the expected arg. - async eval(_script: string, _numkeys: number, k: string, expected: string) { + // Emulates the two compare-and-act Lua evals this module runs, distinguished by the command they call: + // releaseIfValue deletes on match; renewIfValue (#9467) resets the TTL on match. Both must be no-ops on a + // mismatch -- that is the whole ownership guarantee. + async eval(script: string, _numkeys: number, k: string, expected: string, ttl?: string) { if (_store.get(k) !== expected) return 0; + if (script.includes("expire")) { + _ttls.set(k, Number(ttl)); + return 1; + } _store.delete(k); + _ttls.delete(k); return 1; }, - } as unknown as Redis & { _store: Map }; + } as unknown as Redis & { _store: Map; _ttls: Map }; } describe("createRedisCache (#1216 webhook dedup cache)", () => { @@ -80,6 +90,40 @@ describe("createRedisCache (#1216 webhook dedup cache)", () => { await expect(cache.claim("lock", "1", 60)).rejects.toThrow("connection refused"); }); + // #9467: renewal must be compare-and-extend for the same reason release is compare-and-delete -- a GET + // followed by a separate EXPIRE could extend a key a NEW claimant wrote in between, handing the stale + // holder's renewal to the live holder's lock. + it("renewIfValue extends the TTL only when the stored value matches the caller's own token (#9467)", async () => { + const redis = fakeRedis(); + const cache = createRedisCache(redis); + await cache.claim!("lock", "holder-a", 600); + expect(redis._ttls.get("lock")).toBe(600); + + // The owner renews: TTL is reset, ownership unchanged. + expect(await cache.renewIfValue!("lock", "holder-a", 900)).toBe(true); + expect(redis._ttls.get("lock")).toBe(900); + expect(redis._store.get("lock")).toBe("holder-a"); + + // A DIFFERENT holder's renewal must not touch it -- this is the double-actuation guard. + expect(await cache.renewIfValue!("lock", "holder-b", 5)).toBe(false); + expect(redis._ttls.get("lock")).toBe(900); + }); + + it("renewIfValue returns false for a key that no longer exists, so a stale holder learns it lost (#9467)", async () => { + const redis = fakeRedis(); + const cache = createRedisCache(redis); + expect(await cache.renewIfValue!("never-claimed", "holder-a", 600)).toBe(false); + }); + + it("renewIfValue does not resurrect a released lock (#9467)", async () => { + const redis = fakeRedis(); + const cache = createRedisCache(redis); + await cache.claim!("lock", "holder-a", 600); + await cache.releaseIfValue!("lock", "holder-a"); + expect(await cache.renewIfValue!("lock", "holder-a", 600)).toBe(false); + expect(redis._store.has("lock")).toBe(false); + }); + it("releaseIfValue deletes the key only when the stored value matches the caller's own token (#2129)", async () => { const r = fakeRedis(); const cache = createRedisCache(r); diff --git a/test/unit/transient-locks.test.ts b/test/unit/transient-locks.test.ts index f60e0ea744..6a11d73865 100644 --- a/test/unit/transient-locks.test.ts +++ b/test/unit/transient-locks.test.ts @@ -8,6 +8,7 @@ import { releaseContributorCapLock, releasePrActuationLock, releaseTransientLockIfOwner, + startLockHeartbeat, } from "../../src/queue/transient-locks"; import { createTestEnv } from "../helpers/d1"; @@ -597,3 +598,246 @@ describe("releaseTransientLockIfOwner — no-op when there's no releaseIfValue p await expect(releaseTransientLockIfOwner(env, "key", "token")).resolves.toBeUndefined(); }); }); + +// #9467: the TTLs were sized when the actuation lock covered a short plan-and-execute section. #9013 moved the +// claim BEFORE maybePublishPrPublicSurface, so the 600s lock now wraps the whole publish -> AI review -> +// maintain unit -- and the AI review alone can exceed it (3 attempts x up to 600s, per model). When the TTL +// lapsed mid-work the holder was never told, so a second worker claimed the same PR and both actuated it. +describe("startLockHeartbeat (#9467)", () => { + /** Minimal transient-cache stand-in with a real compare-and-extend, so ownership is genuinely modelled. */ + function heartbeatCache(initial: Record = {}) { + const store = new Map(Object.entries(initial)); + const renewals: string[] = []; + return { + renewals, + store, + cache: { + async get(k: string) { + return store.get(k) ?? null; + }, + async set(k: string, v: string) { + store.set(k, v); + }, + async claim(k: string, v: string) { + if (store.has(k)) return false; + store.set(k, v); + return true; + }, + async releaseIfValue(k: string, v: string) { + if (store.get(k) !== v) return false; + store.delete(k); + return true; + }, + async renewIfValue(k: string, v: string) { + renewals.push(k); + return store.get(k) === v; + }, + }, + }; + } + + afterEach(() => vi.useRealTimers()); + + it("INVARIANT: a holder still working past its TTL keeps the lock — the whole point", async () => { + vi.useFakeTimers(); + const { cache, renewals } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", "tok", 600, { intervalMsOverride: 1_000 }); + + // Simulate work running for well past the nominal TTL. + for (let i = 0; i < 5; i += 1) { + await vi.advanceTimersByTimeAsync(1_000); + } + beat.stop(); + expect(renewals.length).toBeGreaterThanOrEqual(5); // renewed repeatedly rather than lapsing + }); + + it("INVARIANT: stopping halts renewal, so a completed job never extends a key it released", async () => { + vi.useFakeTimers(); + const { cache, renewals } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", "tok", 600, { intervalMsOverride: 1_000 }); + await vi.advanceTimersByTimeAsync(1_000); + const afterFirst = renewals.length; + beat.stop(); + await vi.advanceTimersByTimeAsync(10_000); + expect(renewals.length).toBe(afterFirst); // no further renewals after stop + }); + + it("REGRESSION: a holder that LOST the key is told, and stops renewing — it must not extend the new owner's lock", async () => { + vi.useFakeTimers(); + const { cache, store } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + let lost = false; + startLockHeartbeat(env, "lock:k", "tok", 600, { intervalMsOverride: 1_000, onLost: () => void (lost = true) }); + + store.set("lock:k", "someone-elses-token"); // the TTL lapsed and another worker claimed it (or #9008 stole it) + await vi.advanceTimersByTimeAsync(1_000); + + expect(lost).toBe(true); + expect(store.get("lock:k")).toBe("someone-elses-token"); // the new owner's claim is untouched + }); + + it("REGRESSION: a lost key stops the heartbeat permanently (no further renewals after onLost)", async () => { + vi.useFakeTimers(); + const { cache, store, renewals } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + startLockHeartbeat(env, "lock:k", "tok", 600, { intervalMsOverride: 1_000 }); + store.set("lock:k", "other"); + await vi.advanceTimersByTimeAsync(1_000); + const afterLoss = renewals.length; + await vi.advanceTimersByTimeAsync(5_000); + expect(renewals.length).toBe(afterLoss); + }); + + it("fails OPEN for a fail-open claim (null token) — nothing to renew, and no crash", async () => { + const { cache, renewals } = heartbeatCache(); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", null, 600); + beat.stop(); + expect(renewals).toEqual([]); + }); + + it("fails OPEN on an adapter without compare-and-extend — a blind re-set would extend the WRONG holder", async () => { + const env = { SELFHOST_TRANSIENT_CACHE: { async get() { return null; }, async set() {} } } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", "tok", 600); + expect(() => beat.stop()).not.toThrow(); + }); + + it("fails OPEN with no cache bound at all", async () => { + const beat = startLockHeartbeat({} as unknown as Env, "lock:k", "tok", 600); + expect(() => beat.stop()).not.toThrow(); + }); + + it("fails OPEN when renewal THROWS, and keeps trying on later ticks", async () => { + vi.useFakeTimers(); + let calls = 0; + const cache = { + async get() { return null; }, + async set() {}, + async renewIfValue() { + calls += 1; + throw new Error("redis unreachable"); + }, + }; + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + let lost = false; + const beat = startLockHeartbeat(env, "lock:k", "tok", 600, { intervalMsOverride: 1_000, onLost: () => void (lost = true) }); + await vi.advanceTimersByTimeAsync(3_000); + beat.stop(); + expect(calls).toBeGreaterThanOrEqual(3); // retried rather than giving up + expect(lost).toBe(false); // a transient error is NOT "you lost the lock" -- that would abort real work + }); + + it("derives a sub-TTL interval by default, leaving room to recover from a failed renewal", () => { + vi.useFakeTimers(); + const { cache, renewals } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", "tok", 600); // → 200s interval, i.e. 2 chances before lapse + vi.advanceTimersByTime(199_000); + expect(renewals.length).toBe(0); + beat.stop(); + }); + + it("clamps a tiny TTL to a floor so it cannot spin the event loop", () => { + vi.useFakeTimers(); + const { cache, renewals } = heartbeatCache({ "lock:k": "tok" }); + const env = { SELFHOST_TRANSIENT_CACHE: cache } as unknown as Env; + const beat = startLockHeartbeat(env, "lock:k", "tok", 1); // 1s/3 → floored to 1000ms + vi.advanceTimersByTime(999); + expect(renewals.length).toBe(0); + beat.stop(); + }); +}); + +// #9467 end-to-end: the property that actually matters is not "renewIfValue was called" but "a competing pass +// is still refused while the holder works". This drives claim -> heartbeat -> competing claim against a cache +// that genuinely EXPIRES keys, so without the heartbeat the second claim would succeed and both passes would +// actuate the same PR. +describe("lock heartbeat end-to-end (#9467)", () => { + /** Transient cache with real TTL expiry, so a lapse is observable rather than simulated. */ + function expiringCache() { + const store = new Map(); + const alive = (k: string) => { + const e = store.get(k); + if (!e) return null; + if (e.expiresAtMs <= Date.now()) { + store.delete(k); + return null; + } + return e; + }; + return { + async get(k: string) { + return alive(k)?.value ?? null; + }, + async set(k: string, v: string, ttl: number) { + store.set(k, { value: v, expiresAtMs: Date.now() + ttl * 1000 }); + }, + async claim(k: string, v: string, ttl: number) { + if (alive(k)) return false; + store.set(k, { value: v, expiresAtMs: Date.now() + ttl * 1000 }); + return true; + }, + async releaseIfValue(k: string, v: string) { + if (alive(k)?.value !== v) return false; + store.delete(k); + return true; + }, + async renewIfValue(k: string, v: string, ttl: number) { + const e = alive(k); + if (e?.value !== v) return false; + e.expiresAtMs = Date.now() + ttl * 1000; + return true; + }, + }; + } + + afterEach(() => vi.useRealTimers()); + + it("INVARIANT: a competing claim stays refused for the whole time the holder is still working", async () => { + vi.useFakeTimers(); + const env = { SELFHOST_TRANSIENT_CACHE: expiringCache() } as unknown as Env; + const held = await claimTransientLock(env, "lock:pr", 3); // deliberately tiny TTL + expect(held.acquired).toBe(true); + + const beat = startLockHeartbeat(env, "lock:pr", held.ownerToken, 3, { intervalMsOverride: 1_000 }); + // Work for 5x the TTL. + for (let i = 0; i < 15; i += 1) { + await vi.advanceTimersByTimeAsync(1_000); + expect((await claimTransientLock(env, "lock:pr", 3)).acquired).toBe(false); + } + beat.stop(); + }); + + it("REGRESSION: WITHOUT a heartbeat the same lock lapses and a second pass claims it (the bug)", async () => { + vi.useFakeTimers(); + const env = { SELFHOST_TRANSIENT_CACHE: expiringCache() } as unknown as Env; + expect((await claimTransientLock(env, "lock:pr", 3)).acquired).toBe(true); + await vi.advanceTimersByTimeAsync(4_000); // past the TTL, holder still "working" + // This is exactly the double-actuation window #9467 closes. + expect((await claimTransientLock(env, "lock:pr", 3)).acquired).toBe(true); + }); + + it("INVARIANT: once the holder stops, the lock lapses normally and the next pass can claim it", async () => { + vi.useFakeTimers(); + const env = { SELFHOST_TRANSIENT_CACHE: expiringCache() } as unknown as Env; + const held = await claimTransientLock(env, "lock:pr", 3); + const beat = startLockHeartbeat(env, "lock:pr", held.ownerToken, 3, { intervalMsOverride: 1_000 }); + await vi.advanceTimersByTimeAsync(2_000); + beat.stop(); + await vi.advanceTimersByTimeAsync(4_000); // no longer renewed + expect((await claimTransientLock(env, "lock:pr", 3)).acquired).toBe(true); + }); + + it("INVARIANT: releasing while the heartbeat ran still frees the lock immediately", async () => { + vi.useFakeTimers(); + const env = { SELFHOST_TRANSIENT_CACHE: expiringCache() } as unknown as Env; + const held = await claimTransientLock(env, "lock:pr", 60); + const beat = startLockHeartbeat(env, "lock:pr", held.ownerToken, 60, { intervalMsOverride: 1_000 }); + await vi.advanceTimersByTimeAsync(2_000); + beat.stop(); + await releaseTransientLockIfOwner(env, "lock:pr", held.ownerToken); + expect((await claimTransientLock(env, "lock:pr", 60)).acquired).toBe(true); + }); +}); From 4d9736e321ca8657a95287ece0a2731a21215d47 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 18:41:44 -0700 Subject: [PATCH 2/2] refactor(locks): drop an unreachable pre-await stopped guard in the heartbeat (#9467) stop() clears the interval, so a callback can never START after it -- the guard before the first await was dead. The check that matters is the one after the renewal await, where stop() can genuinely have landed mid-call, and that stays. --- src/queue/transient-locks.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/queue/transient-locks.ts b/src/queue/transient-locks.ts index 2a8f4d6d4f..f29678e468 100644 --- a/src/queue/transient-locks.ts +++ b/src/queue/transient-locks.ts @@ -175,7 +175,8 @@ export function startLockHeartbeat( let stopped = false; const timer = setInterval(() => { void (async () => { - if (stopped) return; + // No pre-await `stopped` guard: stop() clears the interval, so a callback can never START after it. + // The check that matters is the one AFTER the await below, where stop() CAN have landed mid-renewal. try { const stillOurs = await cache.renewIfValue!(key, ownerToken, ttlSeconds); if (!stillOurs && !stopped) {