fix(locks): stop contention killing jobs, and stop three paths freeing a live holder's lock (#9465, #9468)

apps/loopover-ui/src/lib/selfhost-env-reference.ts

@@ -297,6 +297,14 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [
     name: "LOOPOVER_REVIEW_RAG",
     firstReference: "src/selfhost/ai.ts",
   },
+  {
+    name: "LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS",
+    firstReference: "src/server.ts",
+  },
+  {
+    name: "LOOPOVER_SINGLE_INSTANCE",
+    firstReference: "src/selfhost/redis-cache.ts",
+  },
   {
     name: "LOOPOVER_VERSION",
     firstReference: "src/selfhost/otel.ts",
@@ -727,6 +735,8 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [
   "| `LOOPOVER_REPO_CONFIG_DIR` | `src/server.ts` |",
   "| `LOOPOVER_REVIEW_CONTINUOUS` | `src/queue/processors.ts` |",
   "| `LOOPOVER_REVIEW_RAG` | `src/selfhost/ai.ts` |",
+  "| `LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS` | `src/server.ts` |",
+  "| `LOOPOVER_SINGLE_INSTANCE` | `src/selfhost/redis-cache.ts` |",
   "| `LOOPOVER_VERSION` | `src/selfhost/otel.ts` |",
   "| `MAINTENANCE_ADMISSION_DEFER_MS` | `src/selfhost/maintenance-admission.ts` |",
   "| `MAINTENANCE_ADMISSION_DRAIN_AGE_MS` | `src/selfhost/maintenance-admission.ts` |",

src/env.d.ts

@@ -381,6 +381,11 @@ declare global {
      *  for a repo ONLY IF this flag is ON *AND* the repo is in LOOPOVER_REVIEW_REPOS (the per-repo cutover
      *  allowlist) — see review/visual-wire.ts screenshotsAllowed. Default OFF — unset/false captures nothing
      *  (no render, no audit write, no comment change) so the review path is byte-identical to today. */
+    /** #9468: declare this deployment single-instance, permitting the boot-time orphaned-lock flush. */
+    LOOPOVER_SINGLE_INSTANCE?: string;
+    /** #9468: force-release held locks if the shutdown drain has not finished within this many ms. Unset =
+     *  wait for the drain, which never frees a lock whose job is still running. */
+    LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS?: string;
     LOOPOVER_REVIEW_SCREENSHOTS?: string;
     /** Convergence (grounding): when truthy, the AI reviewer prompt is GROUNDED — the PR's finished CI status
      *  + the FULL post-change content of the changed files are appended so a non-frontier model verifies its

src/queue/ai-review-orchestration.ts

@@ -129,7 +129,7 @@ export async function claimAiReviewLock(
   // no-op adapter, actually owns the key) so a shutdown signal can release it immediately instead of only via
   // graceful drain or the full 1800s TTL. A fail-open claim (ownerToken null) has nothing to release.
   if (claim.acquired && claim.ownerToken !== null) {
-    registerHeldLock(key, () => releaseTransientLockIfOwner(env, key, claim.ownerToken));
+    registerHeldLock(key, claim.ownerToken, () => releaseTransientLockIfOwner(env, key, claim.ownerToken));
   }
   return claim;
 }
@@ -148,7 +148,10 @@ export async function releaseAiReviewLock(
   // #8998: this process no longer holds it -- a later shutdown must not attempt to release a key it already
   // gave up (harmless either way, since the compare-and-delete release is idempotent, but there is no reason
   // to carry a stale entry).
-  unregisterHeldLock(key);
+  // #9468: token-scoped, so a pass whose lock was STOLEN from it (#9008) cannot delete the stealer's live
+  // registry entry on its way out. A null token means this pass never really owned the key (fail-open claim),
+  // so there is nothing of its own to unregister either.
+  if (ownerToken !== null) unregisterHeldLock(key, ownerToken);
 }
 
 /**

src/queue/held-lock-registry.ts

@@ -19,21 +19,31 @@
  */
 
 type HeldLockRelease = () => Promise<void>;
+type HeldLockEntry = { ownerToken: string; release: HeldLockRelease };
 
-const heldLocks = new Map<string, HeldLockRelease>();
+const heldLocks = new Map<string, HeldLockEntry>();
 
-/** Record that this process now holds `key`, with `release` as how to give it up. Call the moment a claim
- *  actually succeeds with real ownership (a fail-open "acquired but nothing to release" claim has nothing
- *  worth registering — see the call site's own guard). */
-export function registerHeldLock(key: string, release: HeldLockRelease): void {
-  heldLocks.set(key, release);
+/** Record that this process now holds `key` under `ownerToken`, with `release` as how to give it up. Call the
+ *  moment a claim actually succeeds with real ownership (a fail-open "acquired but nothing to release" claim
+ *  has nothing worth registering — see the call site's own guard). */
+export function registerHeldLock(key: string, ownerToken: string, release: HeldLockRelease): void {
+  heldLocks.set(key, { ownerToken, release });
 }
 
-/** Record that this process no longer holds `key` (its own release already ran). A shutdown racing the same
- *  release is harmless either way: `releaseTransientLockIfOwner`'s compare-and-delete makes a second release
- *  attempt against an already-released key a safe no-op. */
-export function unregisterHeldLock(key: string): void {
-  heldLocks.delete(key);
+/**
+ * Record that this process no longer holds `key` under `ownerToken` (its own release already ran).
+ *
+ * #9468: the unregister is CONDITIONAL on the token, mirroring the store-side compare-and-delete. Keyed by
+ * lock key alone, a later holder's entry could be deleted by an earlier holder's cleanup: a maintainer's
+ * forced re-run STEALS a live lock in the same process (#9008), overwriting the registry entry with the
+ * stealer's release; when the original pass then finished, its `releaseTransientLockIfOwner` correctly no-oped
+ * (wrong token) but its unregister still deleted the STEALER's entry. A SIGTERM during the stealer's
+ * still-running review then skipped that key entirely, stranding it for the full 1800s TTL after a hard kill —
+ * precisely the #8998 incident this registry exists to prevent. The same shape occurs when a fail-open claim
+ * (null token from a Redis blip) later "releases".
+ */
+export function unregisterHeldLock(key: string, ownerToken: string): void {
+  if (heldLocks.get(key)?.ownerToken === ownerToken) heldLocks.delete(key);
 }
 
 /** Best-effort release every lock currently on record, clearing the registry as it goes. Never throws: a
@@ -45,9 +55,9 @@ export async function releaseAllHeldLocksAtShutdown(): Promise<number> {
   const entries = [...heldLocks.entries()];
   heldLocks.clear();
   let attempted = 0;
-  for (const [, release] of entries) {
+  for (const [, entry] of entries) {
     attempted += 1;
-    await release().catch(() => undefined);
+    await entry.release().catch(() => undefined);
   }
   return attempted;
 }

src/queue/retryable.ts

@@ -37,3 +37,30 @@ export function retryableJobDelayMs(error: unknown): number | null {
   if (!isRetryableJobError(error)) return null;
   return error.retryAfterMs;
 }
+
+/** Retry kinds that must NOT consume the job's attempt budget (#9465). */
+const ATTEMPT_FREE_RETRY_KINDS = new Set(["pr_actuation_lock_contended"]);
+
+/**
+ * #9465: is this error a "come back in a moment" condition rather than a failure?
+ *
+ * Lock contention was charged an attempt like any other error, so with a flat 5s retry and maxRetries 5 a job
+ * DIED after roughly 25 seconds of contention -- while the lock it was waiting on is designed to be held for
+ * minutes (PR_ACTUATION_LOCK_TTL_SECONDS is 600, and the section it guards spans an entire publish -> AI review
+ * -> maintain pass). Confirmed in production: the only jobs in the dead-letter queue over a 7-day window were
+ * three actuation-lock contentions, one of which was a `reopen-reclose` -- a policy enforcement with a single
+ * webhook-gated trigger and no reconciler, so that enforcement was lost outright.
+ *
+ * The queue already models this correctly for the other "not our turn yet" condition: a GitHub rate-limit
+ * failure re-pends WITHOUT incrementing attempts (`deferred_by='rate_limit'`). This extends the same treatment
+ * to lock contention, bounded by {@link ATTEMPT_FREE_RETRY_DEADLINE_MS} so a genuinely wedged lock still
+ * surfaces as a dead job rather than looping forever.
+ */
+export function isAttemptFreeRetry(error: unknown): boolean {
+  return isRetryableJobError(error) && ATTEMPT_FREE_RETRY_KINDS.has(error.retryKind);
+}
+
+/** How long a job may keep deferring attempt-free before it starts consuming its budget again. Comfortably
+ *  exceeds the 600s actuation-lock TTL, so a holder that runs to its full TTL never kills a waiter, while a
+ *  lock that is somehow never released still converges to a dead job an operator can see. */
+export const ATTEMPT_FREE_RETRY_DEADLINE_MS = 15 * 60 * 1000;

src/selfhost/audit.ts

@@ -14,6 +14,9 @@ export type AuditEventType =
   | "job_complete"
   | "job_dead"
   | "job_error"
+  // #9465: re-pended without consuming an attempt because a per-PR lock was held by another pass -- distinct
+  // from job_error so "waiting its turn" is never read as a failure in the audit trail or its dashboards.
+  | "job_lock_contended"
   | "job_rate_limited";
 
 export interface AuditEvent {

src/selfhost/metrics.ts

@@ -66,6 +66,7 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [
   ["loopover_jobs_dead_total", { help: "Durable queue jobs moved to dead status.", type: "counter" }],
   ["loopover_jobs_rate_limited_total", { help: "Durable queue jobs rate-limited before processing.", type: "counter" }],
   ["loopover_jobs_rate_limit_deferred_total", { help: "Durable queue jobs deferred by a rate-limit window.", type: "counter" }],
+  ["loopover_jobs_lock_contended_deferred_total", { help: "Durable queue jobs re-pended without consuming an attempt because a per-PR lock was held by another pass (#9465).", type: "counter" }],
   ["loopover_jobs_coalesced_total", { help: "Durable queue jobs coalesced with an existing queued item.", type: "counter" }],
   ["loopover_jobs_recovered_total", { help: "Durable queue jobs recovered from stale in-flight state.", type: "counter" }],
   ["loopover_jobs_maintenance_admission_deferred_total", { help: "Maintenance jobs deferred by admission control.", type: "counter" }],

src/selfhost/pg-queue.ts

@@ -10,7 +10,9 @@ import { withReviewSpan } from "./tracing";
 import { withOtelSpan } from "./otel";
 import { capturePostHogError, withPostHogMonitor } from "./posthog";
 import {
+  ATTEMPT_FREE_RETRY_DEADLINE_MS,
   consumingRetryDelayMs,
+  isAttemptFreeRetry,
   deterministicJitterMs,
   FOREGROUND_QUEUE_PRIORITY_FLOOR,
   errorMessageWithCause,
@@ -1582,6 +1584,31 @@ export function createPgQueue(
         }
         const attempts = Number(job.attempts) + 1;
         const errMsg = errorMessageWithCause(error);
+        // #9465: lock contention is "not our turn yet", not a failure -- charging it an attempt killed jobs
+        // after ~25s of contention against a lock designed to be held for minutes, losing a reopen-reclose
+        // enforcement outright in production. Re-pend WITHOUT consuming the budget, exactly as the sibling
+        // rate-limit path below does, bounded by the job's age so a genuinely wedged lock still converges to a
+        // dead job instead of deferring forever.
+        if (isAttemptFreeRetry(error) && Date.now() - Number(job.created_at) < ATTEMPT_FREE_RETRY_DEADLINE_MS) {
+          const retryDelayMs = consumingRetryDelayMs(error, backoff(attempts));
+          await pool.query(
+            `UPDATE ${TABLE} SET status='pending', run_after=$1, last_error=$2, deferred_by='lock_contended' WHERE id=$3`,
+            [Date.now() + retryDelayMs, errMsg, job.id],
+          );
+          await recordQueueMetric("loopover_jobs_lock_contended_deferred_total");
+          logAudit({
+            event: "job_lock_contended",
+            ts: Date.now(),
+            job_id: job.id,
+            payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
+            latency_ms: Date.now() - claimedAt,
+            attempts: Number(job.attempts),
+            retry_after_ms: retryDelayMs,
+            error: errMsg,
+          }, jobTraceParent);
+          return true;
+        }
         const rateLimitDelayMs = githubRateLimitRetryDelayMs(error);
         if (rateLimitDelayMs !== null) {
           const now = Date.now();

src/selfhost/queue-common.ts

@@ -1,6 +1,8 @@
 import { createHash } from "node:crypto";
 
-import { retryableJobDelayMs } from "../queue/retryable";
+import { ATTEMPT_FREE_RETRY_DEADLINE_MS, isAttemptFreeRetry, retryableJobDelayMs } from "../queue/retryable";
+
+export { ATTEMPT_FREE_RETRY_DEADLINE_MS, isAttemptFreeRetry };
 import {
   LOW_REST_RATE_LIMIT_REMAINING,
   MAINTENANCE_RESERVED_HEADROOM,

src/selfhost/redis-cache.ts

@@ -119,6 +119,23 @@ export const ORPHANED_LOCK_KEY_PATTERNS: readonly string[] = [
  *  self-host boot before the queue starts processing. Never throws: a flush failure just means a lock (or
  *  several) rides out its own TTL as before -- exactly the pre-#9021 behavior -- rather than blocking startup.
  *  Returns the number of keys deleted, for a single boot-time log line. */
+/**
+ * #9468: is this deployment declared single-instance, making a boot-time lock flush safe?
+ *
+ * {@link flushOrphanedLocksAtBoot} deletes EVERY key matching {@link ORPHANED_LOCK_KEY_PATTERNS}, which is only
+ * sound when no sibling process can be holding one. Nothing previously enforced that -- the same Redis is
+ * explicitly shared across replicas for the token cache, and the pg queue backend exists to support more than
+ * one instance -- so a restarting replica would free a live sibling's ai-review/pr-actuation locks and let the
+ * next pass duplicate that review and its actuation, with no TTL expiry involved.
+ *
+ * Defaults to FALSE (no flush) rather than true: the failure mode of skipping is a genuinely orphaned lock
+ * riding out its TTL, which #9008's steal path already recovers on demand; the failure mode of flushing
+ * wrongly is a double close/merge. Operators of a genuinely single-instance deployment opt in.
+ */
+export function isSingleInstanceDeployment(env: Record<string, string | undefined>): boolean {
+  return /^(1|true|yes|on)$/i.test(env["LOOPOVER_SINGLE_INSTANCE"] ?? "");
+}
+
 export async function flushOrphanedLocksAtBoot(redis: Redis): Promise<number> {
   let deleted = 0;
   for (const pattern of ORPHANED_LOCK_KEY_PATTERNS) {

src/selfhost/sqlite-queue.ts

@@ -11,7 +11,9 @@ import { withReviewSpan } from "./tracing";
 import { withOtelSpan } from "./otel";
 import { capturePostHogError, withPostHogMonitor } from "./posthog";
 import {
+  ATTEMPT_FREE_RETRY_DEADLINE_MS,
   consumingRetryDelayMs,
+  isAttemptFreeRetry,
   deterministicJitterMs,
   FOREGROUND_QUEUE_PRIORITY_FLOOR,
   githubRateLimitAdmissionDelayMs,
@@ -1188,6 +1190,29 @@ export function createSqliteQueue(
       } catch (error) {
         const attempts = job.attempts + 1;
         const errMsg = errorMessageWithCause(error);
+        // #9465: parity with the pg backend -- lock contention re-pends WITHOUT consuming an attempt, so a
+        // waiter cannot die after ~25s against a lock legitimately held for minutes. Bounded by job age so a
+        // wedged lock still converges to a dead job.
+        if (isAttemptFreeRetry(error) && Date.now() - job.created_at < ATTEMPT_FREE_RETRY_DEADLINE_MS) {
+          const retryDelayMs = consumingRetryDelayMs(error, backoff(attempts));
+          driver.query(
+            `UPDATE ${TABLE} SET status='pending', run_after=?, last_error=?, deferred_by='lock_contended' WHERE id=?`,
+            [Date.now() + retryDelayMs, errMsg, job.id],
+          );
+          recordQueueMetric(driver, "loopover_jobs_lock_contended_deferred_total");
+          logAudit({
+            event: "job_lock_contended",
+            ts: Date.now(),
+            job_id: job.id,
+            payload_type: extractPayloadType(job.payload),
+            ...payloadContext,
+            latency_ms: Date.now() - claimedAt,
+            attempts: job.attempts,
+            retry_after_ms: retryDelayMs,
+            error: errMsg,
+          }, jobTraceParent);
+          return true;
+        }
         const rateLimitDelayMs = githubRateLimitRetryDelayMs(error);
         if (rateLimitDelayMs !== null) {
           const now = Date.now();