fix(locks): renew a held lock while its work runs, so a slow pass cannot be claimed out from under (#9467) (#9507)

* 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.

* 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.

Author: JSONbored

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<boolean>;
+      /** 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<boolean>;
     };
     PUBLIC_API_ORIGIN?: string;
     PUBLIC_SITE_ORIGIN?: string;

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,

src/queue/processors.ts

@@ -363,6 +363,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.
@@ -4236,6 +4238,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<typeof evaluateGateCheck> | undefined;
   try {
     gate = await withReviewPipelineSpan(
@@ -4321,6 +4333,7 @@ export async function reReviewStoredPullRequest(
       );
     });
   } finally {
+    actuationHeartbeat.stop();
     await releasePrActuationLock(env, repoFullName, pr.number, actuationLock.ownerToken);
   }
   return true;
@@ -11155,9 +11168,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,

src/queue/transient-locks.ts

@@ -136,7 +136,74 @@ 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 () => {
+      // 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) {
+          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}`;
 }

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<boolean> {
+      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.

test/unit/selfhost-redis-cache.test.ts

@@ -16,29 +16,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<string, string> } {
+function fakeRedis(): Redis & { _store: Map<string, string>; _ttls: Map<string, number> } {
   const _store = new Map<string, string>();
+  const _ttls = new Map<string, number>();
   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<string, string> };
+  } as unknown as Redis & { _store: Map<string, string>; _ttls: Map<string, number> };
 }
 
 describe("createRedisCache (#1216 webhook dedup cache)", () => {
@@ -81,6 +91,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);