From 3eacc8cbe5d56616846235daa4d9dd8346b41534 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:56:38 -0700 Subject: [PATCH 1/2] fix(locks): stop contention killing jobs, and stop three paths freeing a live holder's lock (#9465, #9468) Lock contention was charged an attempt like any other error, so with a flat 5s retry and maxRetries 5 a waiter died after roughly 25 seconds -- against a lock designed to be held for minutes (PR_ACTUATION_LOCK_TTL_SECONDS is 600s, spanning a whole publish -> AI review -> maintain pass). Production confirmed it: the only dead-lettered jobs in a 7-day window were three actuation-lock contentions, and one was a reopen-reclose, whose single webhook-gated trigger and absent reconciler meant that one-shot enforcement was lost outright (PR #9450). Both backends now re-pend without consuming the budget -- the same treatment the sibling GitHub rate-limit path already gets -- bounded by job age so a genuinely wedged lock still converges to a dead job an operator can see, under a distinct job_lock_contended audit event so "waiting its turn" is never read as a failure. Three paths freed locks whose owners were still running. The boot flush deletes every key matching the orphaned-lock patterns, which is only sound on a single instance -- but nothing enforced that, while the same Redis is explicitly shared across replicas and the pg backend exists to support more than one; a restarting sibling therefore freed a live replica's in-flight ai-review and pr-actuation locks, letting the next pass duplicate that review and its actuation with no TTL expiry involved. It is now opt-in via LOOPOVER_SINGLE_INSTANCE, defaulting to skip because the cost of skipping is a lock riding out its TTL (which #9008's steal path already recovers) while the cost of flushing wrongly is a double close or merge. Shutdown released every held lock BEFORE the drain, but the drain deliberately lets in-flight work finish -- so a completed drain left jobs running with their locks already freed, and a sibling or the new container in an overlapped deploy could claim one and duplicate the actuation. The drain now runs first and each job releases its own lock through its own finally; the proactive bulk release remains for the short-grace case it was written for, but only fires when the drain has not finished within LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS. The held-lock registry was keyed by lock key alone, so an earlier holder's cleanup could evict a later one's live entry: a forced re-run steals a lock in-process, and the original pass's unregister then deleted the stealer's entry, so a SIGTERM skipped that key and it stranded for the full TTL after a hard kill -- exactly the shape the registry exists to prevent. Register and unregister are now token-scoped, mirroring the store-side compare-and-delete. --- .../src/lib/selfhost-env-reference.ts | 10 +++ src/env.d.ts | 5 ++ src/queue/ai-review-orchestration.ts | 7 +- src/queue/held-lock-registry.ts | 36 +++++---- src/queue/retryable.ts | 27 +++++++ src/selfhost/audit.ts | 3 + src/selfhost/metrics.ts | 1 + src/selfhost/pg-queue.ts | 27 +++++++ src/selfhost/queue-common.ts | 4 +- src/selfhost/redis-cache.ts | 17 +++++ src/selfhost/sqlite-queue.ts | 25 +++++++ src/server.ts | 72 ++++++++++++++---- test/unit/held-lock-registry.test.ts | 50 ++++++++++--- test/unit/retryable.test.ts | 73 +++++++++++++------ 14 files changed, 295 insertions(+), 62 deletions(-) diff --git a/apps/loopover-ui/src/lib/selfhost-env-reference.ts b/apps/loopover-ui/src/lib/selfhost-env-reference.ts index 9ab6cfcf33..e7c0067ed3 100644 --- a/apps/loopover-ui/src/lib/selfhost-env-reference.ts +++ b/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` |", diff --git a/src/env.d.ts b/src/env.d.ts index 0ab733c1d2..f4cae13f68 100644 --- a/src/env.d.ts +++ b/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 diff --git a/src/queue/ai-review-orchestration.ts b/src/queue/ai-review-orchestration.ts index ca12e53088..b8b26ef213 100644 --- a/src/queue/ai-review-orchestration.ts +++ b/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); } /** diff --git a/src/queue/held-lock-registry.ts b/src/queue/held-lock-registry.ts index 0a38064acd..b2d991c43b 100644 --- a/src/queue/held-lock-registry.ts +++ b/src/queue/held-lock-registry.ts @@ -19,21 +19,31 @@ */ type HeldLockRelease = () => Promise; +type HeldLockEntry = { ownerToken: string; release: HeldLockRelease }; -const heldLocks = new Map(); +const heldLocks = new Map(); -/** 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 { 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; } diff --git a/src/queue/retryable.ts b/src/queue/retryable.ts index a514dbbbed..b5ef29fd3e 100644 --- a/src/queue/retryable.ts +++ b/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; diff --git a/src/selfhost/audit.ts b/src/selfhost/audit.ts index 3d44101b54..7a00abb848 100644 --- a/src/selfhost/audit.ts +++ b/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 { diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index ce148888fd..6cb6ebbfa4 100644 --- a/src/selfhost/metrics.ts +++ b/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" }], diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index c6902cf509..e8d5a70ca1 100644 --- a/src/selfhost/pg-queue.ts +++ b/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(); diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index 92b5df7010..1d07ad11ee 100644 --- a/src/selfhost/queue-common.ts +++ b/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, diff --git a/src/selfhost/redis-cache.ts b/src/selfhost/redis-cache.ts index 1b68a46c9a..fdc1b84b30 100644 --- a/src/selfhost/redis-cache.ts +++ b/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): boolean { + return /^(1|true|yes|on)$/i.test(env["LOOPOVER_SINGLE_INSTANCE"] ?? ""); +} + export async function flushOrphanedLocksAtBoot(redis: Redis): Promise { let deleted = 0; for (const pattern of ORPHANED_LOCK_KEY_PATTERNS) { diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 3f18af59da..486a69074b 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/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(); diff --git a/src/server.ts b/src/server.ts index 65681fc415..cff6867b3a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -731,19 +731,37 @@ async function main(): Promise { const { Redis } = await import("ioredis"); const redisClient = new Redis(redisUrl); const { createRedisRateLimiter } = await import("./selfhost/redis-ratelimit"); - const { createRedisCache, assertSelfhostTransientCacheOwnershipRelease, flushOrphanedLocksAtBoot, isWebhookDeliveryDuplicate, rememberWebhookDelivery } = await import("./selfhost/redis-cache"); + const { createRedisCache, assertSelfhostTransientCacheOwnershipRelease, flushOrphanedLocksAtBoot, isSingleInstanceDeployment, isWebhookDeliveryDuplicate, rememberWebhookDelivery } = await import("./selfhost/redis-cache"); const rateLimiter = createRedisRateLimiter(redisClient); const webhookCache = createRedisCache(redisClient); assertSelfhostTransientCacheOwnershipRelease(webhookCache); // #9021: every Redis-backed lock (pr-actuation-lock, ai-review-lock, contributor-cap-wake/-lock) survives a - // container restart with its TTL intact. On this single-instance deployment any lock present at boot is - // provably orphaned -- the process that claimed it is gone -- so flush them before the queue starts - // processing rather than let each strand real work for up to its own TTL (30 min for ai-review-lock, #8998). - // Best-effort: a flush failure just falls back to the pre-#9021 behavior (each lock rides out its TTL). - const flushedOrphanedLocks = await flushOrphanedLocksAtBoot(redisClient); - if (flushedOrphanedLocks > 0) { + // container restart with its TTL intact. On a SINGLE-INSTANCE deployment any lock present at boot is provably + // orphaned -- the process that claimed it is gone -- so flushing them before the queue starts beats letting + // each strand real work for up to its own TTL (30 min for ai-review-lock, #8998). + // + // #9468: that reasoning holds ONLY for a single instance, and nothing used to enforce it. Redis here is + // explicitly shared across replicas (the token cache below says so in as many words), and the pg queue + // backend exists precisely to support more than one. With a sibling running, this flush DELETES ITS LIVE + // LOCKS: replica B restarting (OOM, crash-loop, scale-out, or a start-before-stop rolling deploy) frees + // replica A's in-flight ai-review-lock and pr-actuation-lock, and the next pass claims them fresh and + // duplicates the review and the actuation -- with no TTL expiry needed at all. A crash-looping replica + // becomes a periodic fleet-wide lock wipe. Opt in explicitly instead; TTL plus the #9008 steal path already + // recover a genuinely orphaned lock without this, just more slowly. + // `env` is not assigned yet this early in boot -- this whole block reads process.env directly (see REDIS_URL above). + if (isSingleInstanceDeployment(process.env)) { + const flushedOrphanedLocks = await flushOrphanedLocksAtBoot(redisClient); + if (flushedOrphanedLocks > 0) { + console.log( + JSON.stringify({ event: "selfhost_orphaned_locks_flushed", count: flushedOrphanedLocks }), + ); + } + } else { console.log( - JSON.stringify({ event: "selfhost_orphaned_locks_flushed", count: flushedOrphanedLocks }), + JSON.stringify({ + event: "selfhost_orphaned_lock_flush_skipped", + reason: "LOOPOVER_SINGLE_INSTANCE is not enabled; a shared-Redis flush would delete a sibling replica's live locks", + }), ); } // Persist the installation-token cache in Redis so warm GitHub App tokens survive restarts/deploys and are @@ -1416,16 +1434,38 @@ async function main(): Promise { if (shuttingDown) return; shuttingDown = true; console.log(JSON.stringify({ event: "selfhost_shutdown", signal })); - // #8998: release every transient lock THIS process currently holds FIRST, before anything that might take - // real time (the queue drain below, telemetry flush). A container orchestrator's SIGKILL grace period is - // often shorter than an in-flight AI-review call legitimately runs, so waiting for the graceful drain to - // release a lock via its own finally block is not always fast enough -- doing it proactively here means the - // lock is gone the instant SIGTERM/SIGINT arrives, regardless of whether the rest of shutdown gets to finish. - const releasedLocks = await releaseAllHeldLocksAtShutdown(); - if (releasedLocks > 0) console.log(JSON.stringify({ event: "selfhost_shutdown_locks_released", count: releasedLocks })); clearInterval(cron); server.close(); - await backend.shutdown(); + // #8998 released every held lock HERE, before the drain -- so that a SIGKILL landing mid-drain could not + // strand an ai-review-lock for its full 1800s TTL. #9468: that ordering also frees the locks of jobs that + // are STILL RUNNING, because `backend.shutdown()` deliberately lets in-flight work finish (#9007). In the + // case where the drain does complete, a sibling replica -- or the new container in an overlapped deploy -- + // could claim the freed lock at t0+e and duplicate the very actuation the lock exists to serialize. The + // token-checked release protects the new claim from corruption; it does not stop the double-run. + // + // So: drain FIRST and let each job release its own lock through its own finally block, which is both + // correct and precise. The proactive bulk release stays as a last resort for the short-grace case #8998 + // was written for, but it now only fires when the drain has NOT finished in time -- i.e. when a hard kill + // is genuinely imminent and a stranded lock is the worse outcome. Opt in with + // LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS; unset means "wait for the drain", which is right wherever the + // orchestrator's grace period comfortably exceeds a review (this deployment's stop_grace_period is 300s). + const forceReleaseAfterMs = Number(process.env["LOOPOVER_SHUTDOWN_LOCK_RELEASE_AFTER_MS"] ?? ""); + const drainPromise = backend.shutdown(); + const drainedInTime = + Number.isFinite(forceReleaseAfterMs) && forceReleaseAfterMs > 0 + ? await Promise.race([ + drainPromise.then(() => true), + new Promise((resolve) => setTimeout(() => resolve(false), forceReleaseAfterMs)), + ]) + : await drainPromise.then(() => true); + // After a completed drain this finds an empty registry (every job unregistered its own lock on the way + // out), so the count is 0 and nothing is force-released -- the log line only appears in the cut-short case. + const releasedLocks = await releaseAllHeldLocksAtShutdown(); + if (releasedLocks > 0) { + console.log( + JSON.stringify({ event: "selfhost_shutdown_locks_released", count: releasedLocks, drainedInTime }), + ); + } /* v8 ignore next -- graceful process signal path is not imported in unit tests; shutdown helper is covered. */ await shutdownOpenTelemetry(); await shutdownPostHog(); diff --git a/test/unit/held-lock-registry.test.ts b/test/unit/held-lock-registry.test.ts index 660441d8e8..64b25adb69 100644 --- a/test/unit/held-lock-registry.test.ts +++ b/test/unit/held-lock-registry.test.ts @@ -11,15 +11,15 @@ import { createTestEnv } from "../helpers/d1"; describe("held-lock-registry (#8998)", () => { it("releases every registered lock and reports how many it attempted", async () => { const released: string[] = []; - registerHeldLock("lock:a", async () => void released.push("a")); - registerHeldLock("lock:b", async () => void released.push("b")); + registerHeldLock("lock:a", "tok-a", async () => void released.push("a")); + registerHeldLock("lock:b", "tok-b", async () => void released.push("b")); expect(await releaseAllHeldLocksAtShutdown()).toBe(2); expect(released.sort()).toEqual(["a", "b"]); }); it("clears the registry as it goes, so a second shutdown call has nothing left to release", async () => { - registerHeldLock("lock:c", async () => undefined); + registerHeldLock("lock:c", "tok-c", async () => undefined); await releaseAllHeldLocksAtShutdown(); expect(await releaseAllHeldLocksAtShutdown()).toBe(0); @@ -27,8 +27,8 @@ describe("held-lock-registry (#8998)", () => { it("unregistering removes a lock before shutdown ever needs to touch it — the normal, non-crash path", async () => { const release = vi.fn(async () => undefined); - registerHeldLock("lock:d", release); - unregisterHeldLock("lock:d"); + registerHeldLock("lock:d", "tok-d", release); + unregisterHeldLock("lock:d", "tok-d"); expect(await releaseAllHeldLocksAtShutdown()).toBe(0); expect(release).not.toHaveBeenCalled(); @@ -36,10 +36,10 @@ describe("held-lock-registry (#8998)", () => { it("keeps releasing the rest when one release throws — one bad lock must not strand the others", async () => { const released: string[] = []; - registerHeldLock("lock:e", async () => { + registerHeldLock("lock:e", "tok-e", async () => { throw new Error("cache unreachable"); }); - registerHeldLock("lock:f", async () => void released.push("f")); + registerHeldLock("lock:f", "tok-f", async () => void released.push("f")); expect(await releaseAllHeldLocksAtShutdown()).toBe(2); expect(released).toEqual(["f"]); @@ -48,8 +48,8 @@ describe("held-lock-registry (#8998)", () => { it("re-registering the same key replaces the prior release rather than accumulating both", async () => { const first = vi.fn(async () => undefined); const second = vi.fn(async () => undefined); - registerHeldLock("lock:g", first); - registerHeldLock("lock:g", second); + registerHeldLock("lock:g", "tok-g1", first); + registerHeldLock("lock:g", "tok-g2", second); expect(heldLockCountForTest()).toBe(1); await releaseAllHeldLocksAtShutdown(); @@ -60,6 +60,38 @@ describe("held-lock-registry (#8998)", () => { it("does nothing and reports zero when the registry is empty", async () => { expect(await releaseAllHeldLocksAtShutdown()).toBe(0); }); + + // #9468 regression: keyed by lock key alone, an EARLIER holder's cleanup could delete a LATER holder's live + // entry. A maintainer's forced re-run steals a live lock in the same process (#9008), overwriting the entry + // with the stealer's release; when the original pass then finished, its store-side release 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, stranding it for the full TTL after a hard kill -- exactly the + // #8998 shape this registry exists to prevent. + it("#9468 a stale holder's unregister does NOT evict the live holder that stole the key", async () => { + const stolenRelease = vi.fn(async () => undefined); + registerHeldLock("lock:stolen", "tok-original", async () => undefined); + registerHeldLock("lock:stolen", "tok-stealer", stolenRelease); // the steal (#9008) + + unregisterHeldLock("lock:stolen", "tok-original"); // the original pass finishing afterwards + + expect(heldLockCountForTest()).toBe(1); + await releaseAllHeldLocksAtShutdown(); + expect(stolenRelease).toHaveBeenCalledTimes(1); // the live holder is still released at shutdown + }); + + it("#9468 unregistering with a non-matching token is a no-op, not a silent eviction", async () => { + const before = heldLockCountForTest(); + registerHeldLock("lock:h", "tok-real", async () => undefined); + unregisterHeldLock("lock:h", "tok-someone-else"); + expect(heldLockCountForTest()).toBe(before + 1); + }); + + it("#9468 unregistering with the matching token still removes it (the normal path is unchanged)", async () => { + const before = heldLockCountForTest(); + registerHeldLock("lock:i", "tok-i", async () => undefined); + unregisterHeldLock("lock:i", "tok-i"); + expect(heldLockCountForTest()).toBe(before); + }); }); // The actual integration this registry exists for: claimAiReviewLock/releaseAiReviewLock are the canonical diff --git a/test/unit/retryable.test.ts b/test/unit/retryable.test.ts index 9fc06b2541..635beb4efa 100644 --- a/test/unit/retryable.test.ts +++ b/test/unit/retryable.test.ts @@ -1,21 +1,52 @@ -import { describe, expect, it } from "vitest"; -import { RetryableJobError, isRetryableJobError, retryableJobDelayMs } from "../../src/queue/retryable"; - -describe("RetryableJobError", () => { - it("clamps retryAfterMs between 1s and 1h with a 5m default", () => { - expect(new RetryableJobError("retry", { retryKind: "rate_limit" }).retryAfterMs).toBe(5 * 60 * 1000); - expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: Number.NaN }).retryAfterMs).toBe(5 * 60 * 1000); - expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 0 }).retryAfterMs).toBe(1_000); - expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 500 }).retryAfterMs).toBe(1_000); - expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 90_000 }).retryAfterMs).toBe(90_000); - expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 9_999_999 }).retryAfterMs).toBe(60 * 60 * 1000); - }); - - it("identifies retryable errors for queue delay helpers", () => { - const err = new RetryableJobError("backoff", { retryKind: "github", retryAfterMs: 2_000 }); - expect(isRetryableJobError(err)).toBe(true); - expect(retryableJobDelayMs(err)).toBe(2_000); - expect(isRetryableJobError(new Error("nope"))).toBe(false); - expect(retryableJobDelayMs(new Error("nope"))).toBeNull(); - }); -}); +import { describe, expect, it } from "vitest"; +import { ATTEMPT_FREE_RETRY_DEADLINE_MS, RetryableJobError, isAttemptFreeRetry, isRetryableJobError, retryableJobDelayMs } from "../../src/queue/retryable"; + +describe("RetryableJobError", () => { + it("clamps retryAfterMs between 1s and 1h with a 5m default", () => { + expect(new RetryableJobError("retry", { retryKind: "rate_limit" }).retryAfterMs).toBe(5 * 60 * 1000); + expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: Number.NaN }).retryAfterMs).toBe(5 * 60 * 1000); + expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 0 }).retryAfterMs).toBe(1_000); + expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 500 }).retryAfterMs).toBe(1_000); + expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 90_000 }).retryAfterMs).toBe(90_000); + expect(new RetryableJobError("retry", { retryKind: "x", retryAfterMs: 9_999_999 }).retryAfterMs).toBe(60 * 60 * 1000); + }); + + it("identifies retryable errors for queue delay helpers", () => { + const err = new RetryableJobError("backoff", { retryKind: "github", retryAfterMs: 2_000 }); + expect(isRetryableJobError(err)).toBe(true); + expect(retryableJobDelayMs(err)).toBe(2_000); + expect(isRetryableJobError(new Error("nope"))).toBe(false); + expect(retryableJobDelayMs(new Error("nope"))).toBeNull(); + }); +}); + +// #9465 regression: lock contention was charged an attempt like any other error, so with a flat 5s retry and +// maxRetries 5 a waiter DIED after ~25s -- against a lock designed to be held for minutes (the actuation lock's +// TTL is 600s and it spans a whole publish -> AI review -> maintain pass). Production confirmed it: the only +// dead-lettered jobs in a 7-day window were three actuation-lock contentions, one a `reopen-reclose`, whose +// single webhook-gated trigger and absent reconciler meant that enforcement was lost outright. +describe("attempt-free retry classification (#9465)", () => { + it("treats per-PR actuation-lock contention as attempt-free", () => { + expect(isAttemptFreeRetry(new RetryableJobError("contended", { retryKind: "pr_actuation_lock_contended" }))).toBe(true); + }); + + it("does NOT extend attempt-free treatment to other retryable kinds", () => { + // Deliberately narrow: a genuine failure must still consume its budget and converge to a dead job. + expect(isAttemptFreeRetry(new RetryableJobError("later", { retryKind: "rate_limit" }))).toBe(false); + expect(isAttemptFreeRetry(new RetryableJobError("later", { retryKind: "some_other_kind" }))).toBe(false); + }); + + it("is false for a plain error and for a non-error value", () => { + expect(isAttemptFreeRetry(new Error("boom"))).toBe(false); + expect(isAttemptFreeRetry("boom")).toBe(false); + expect(isAttemptFreeRetry(undefined)).toBe(false); + }); + + it("bounds attempt-free deferral well beyond the actuation lock's own TTL, so a holder never kills a waiter", () => { + // PR_ACTUATION_LOCK_TTL_SECONDS is 600 (src/queue/transient-locks.ts). The deadline must exceed it, or a + // holder running to its full TTL would still dead-letter the job waiting behind it -- the original bug. + expect(ATTEMPT_FREE_RETRY_DEADLINE_MS).toBeGreaterThan(600 * 1000); + // ...but remain finite, so a lock that is somehow never released still surfaces as a dead job. + expect(Number.isFinite(ATTEMPT_FREE_RETRY_DEADLINE_MS)).toBe(true); + }); +}); From 5e0887795dcb4456dfa4fd35dc4027fe03f49b69 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 18:27:53 -0700 Subject: [PATCH 2/2] test(locks): cover attempt-free deferral on both backends and the #9468 lifecycle arms The attempt-free deferral block -- the core of #9465 -- had no coverage in either backend. Adds a regression per backend (re-pends without consuming an attempt, tagged deferred_by='lock_contended') plus two invariants: a genuine failure still consumes its budget and converges to dead, and past the age deadline a wedged lock does too rather than deferring forever. Also covers the single-instance boot-flush gate (defaulting to false is the fail-safe direction) and the null-token release arm, which must never evict a live holder. makePool's enqueueJob gained an optional createdAt so the age-based deadline is testable; it defaults to now, so every existing caller is unaffected. --- test/unit/held-lock-registry.test.ts | 13 +++++ test/unit/selfhost-pg-queue.test.ts | 47 +++++++++++++++-- test/unit/selfhost-redis-cache.test.ts | 21 ++++++++ test/unit/selfhost-sqlite-queue.test.ts | 67 +++++++++++++++++++++++++ 4 files changed, 145 insertions(+), 3 deletions(-) diff --git a/test/unit/held-lock-registry.test.ts b/test/unit/held-lock-registry.test.ts index 64b25adb69..d62b19dc36 100644 --- a/test/unit/held-lock-registry.test.ts +++ b/test/unit/held-lock-registry.test.ts @@ -86,6 +86,19 @@ describe("held-lock-registry (#8998)", () => { expect(heldLockCountForTest()).toBe(before + 1); }); + it("#9468 a fail-open release (null token) never unregisters anyone — there was nothing to own", async () => { + // claimAiReviewLock only registers a REAL claim, so a null-token release has no entry of its own; the + // guard at the call site must not let it evict whatever IS registered for that key. + const env = createTestEnv(); + const before = heldLockCountForTest(); + const claim = await claimAiReviewLock(env, "acme/widgets", 11, "sha11", "block"); + expect(heldLockCountForTest()).toBe(before + 1); + await releaseAiReviewLock(env, "acme/widgets", 11, "sha11", "block", null); + expect(heldLockCountForTest()).toBe(before + 1); // still registered — a null token releases nothing + await releaseAiReviewLock(env, "acme/widgets", 11, "sha11", "block", claim.ownerToken); + expect(heldLockCountForTest()).toBe(before); + }); + it("#9468 unregistering with the matching token still removes it (the normal path is unchanged)", async () => { const before = heldLockCountForTest(); registerHeldLock("lock:i", "tok-i", async () => undefined); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 6eaa69bf8d..a87ec06271 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -3,6 +3,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { Pool, QueryResult } from "pg"; import { createPgQueue, setPgRetryPoolQueryDelayMsForTest } from "../../src/selfhost/pg-queue"; +import { PrActuationLockContendedError } from "../../src/queue/transient-locks"; +import { ATTEMPT_FREE_RETRY_DEADLINE_MS } from "../../src/queue/retryable"; import { queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; import { DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS } from "../../src/selfhost/queue-fairness"; import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; @@ -69,7 +71,8 @@ interface MockPool { fn: MockFn; enqueueResult(r: Partial): void; /** Pre-load a job to be returned by the next RETURNING claim query. */ - enqueueJob(id: string, payload: object, attempts?: number, jobKey?: string | null): void; + /** #9465: `createdAt` backs the attempt-free-deferral deadline, which is evaluated against job age. */ + enqueueJob(id: string, payload: object, attempts?: number, jobKey?: string | null, createdAt?: number): void; setDeferUpdateRowCount(rowCount: number): void; /** Queues per-call rowCounts for the "AND status='dead'" revive UPDATE, one entry consumed per call in order * (default 1 when the queue is empty) — lets a test simulate an overlapping reviver already winning the race @@ -268,8 +271,8 @@ function makePool(): MockPool { pool: { query: fn } as unknown as Pool, fn: fn as unknown as MockFn, enqueueResult(r) { results.push(r); }, - enqueueJob(id, payload, attempts = 0, jobKey = null) { - results.push({ rows: [{ id, payload: JSON.stringify(payload), attempts, job_key: jobKey }], rowCount: 1 }); + enqueueJob(id, payload, attempts = 0, jobKey = null, createdAt = Date.now()) { + results.push({ rows: [{ id, payload: JSON.stringify(payload), attempts, job_key: jobKey, created_at: createdAt }], rowCount: 1 }); }, setDeferUpdateRowCount(rowCount) { deferUpdateRowCount = rowCount; @@ -1099,6 +1102,44 @@ describe("createPgQueue (durable #977)", () => { expect(calls.some((sql) => sql.includes("attempts=$1, run_after=$2"))).toBe(false); }); + // #9465: parity with the sqlite backend. Lock contention is "not our turn yet", not a failure -- charging it + // an attempt killed waiters after ~25s against a lock designed to be held for minutes, losing a reopen-reclose + // enforcement outright in production. + it("REGRESSION (#9465): lock contention re-pends WITHOUT consuming an attempt, and is tagged distinctly", async () => { + const m = makePool(); + m.enqueueJob("1", { type: "agent-regate-pr" }); + const consume = vi.fn().mockImplementation(async () => { + throw new PrActuationLockContendedError("acme/widgets", 7, "public-surface-publish"); + }); + const q = createPgQueue(m.pool, consume); + await q.init(); + await q.drain(); + + const calls = (m.fn as unknown as ReturnType).mock.calls.map((c: unknown[]) => String(c[0])); + // Re-pended under its own provenance tag, distinguishable from a rate-limit defer in the audit trail. + expect(calls.some((sql) => sql.includes("deferred_by='lock_contended'"))).toBe(true); + // Crucially: NOT the attempts-consuming retry UPDATE, and NOT a dead-letter. + expect(calls.some((sql) => sql.includes("attempts=$1, run_after=$2"))).toBe(false); + expect(calls.some((sql) => sql.includes("status='dead'"))).toBe(false); + }); + + it("INVARIANT (#9465): past the attempt-free deadline a wedged lock converges to the normal retry path", async () => { + // The deferral is bounded by job age so a lock that is somehow never released still surfaces as a dead job + // an operator can see, rather than deferring forever. + const m = makePool(); + m.enqueueJob("1", { type: "agent-regate-pr" }, 0, null, Date.now() - (ATTEMPT_FREE_RETRY_DEADLINE_MS + 60_000)); + const consume = vi.fn().mockImplementation(async () => { + throw new PrActuationLockContendedError("acme/widgets", 7, "public-surface-publish"); + }); + const q = createPgQueue(m.pool, consume); + await q.init(); + await q.drain(); + + const calls = (m.fn as unknown as ReturnType).mock.calls.map((c: unknown[]) => String(c[0])); + expect(calls.some((sql) => sql.includes("deferred_by='lock_contended'"))).toBe(false); + expect(calls.some((sql) => sql.includes("attempts=$1, run_after=$2") || sql.includes("status='dead'"))).toBe(true); + }); + it("regression: a GENERIC network error from consume() (e.g. GitHub API, not Postgres) still goes through normal retry, not silent reclaim", async () => { // ECONNRESET/ECONNREFUSED/EPIPE are NOT unique to Postgres -- consume() can throw them from its own // unrelated network calls. Only unambiguous Postgres SQLSTATE codes should trigger the reclaim path here. diff --git a/test/unit/selfhost-redis-cache.test.ts b/test/unit/selfhost-redis-cache.test.ts index c2f9954c0e..fff461e129 100644 --- a/test/unit/selfhost-redis-cache.test.ts +++ b/test/unit/selfhost-redis-cache.test.ts @@ -4,6 +4,7 @@ import { afterEach, describe, expect, it } from "vitest"; import { assertSelfhostTransientCacheOwnershipRelease, createRedisCache, + isSingleInstanceDeployment, flushOrphanedLocksAtBoot, isWebhookDeliveryDuplicate, ORPHANED_LOCK_KEY_PATTERNS, @@ -220,3 +221,23 @@ describe("flushOrphanedLocksAtBoot (#9021)", () => { expect(redis._store.has("contributor-cap-lock:owner/repo:alice")).toBe(false); }); }); + +// #9468: the boot-time orphaned-lock flush deletes EVERY key matching the lock patterns, which is only sound +// when no sibling process can be holding one. Nothing enforced that -- the same Redis is shared across replicas +// for the token cache, and the pg queue backend exists to support more than one instance -- so a restarting +// replica freed a live sibling's in-flight ai-review and pr-actuation locks and let the next pass duplicate +// that review and its actuation, with no TTL expiry involved. +describe("isSingleInstanceDeployment (#9468)", () => { + it.each([["1"], ["true"], ["TRUE"], ["yes"], ["on"]])("treats %s as an explicit opt-in", (value) => { + expect(isSingleInstanceDeployment({ LOOPOVER_SINGLE_INSTANCE: value })).toBe(true); + }); + + it.each([[undefined], [""], ["0"], ["false"], ["no"], ["off"], ["maybe"]])( + "defaults to false for %s — the flush must be opted into, never assumed", + (value) => { + // Fail-safe direction: skipping the flush costs a genuinely orphaned lock riding out its TTL (which + // #9008's steal path already recovers on demand); flushing wrongly costs a double close or merge. + expect(isSingleInstanceDeployment(value === undefined ? {} : { LOOPOVER_SINGLE_INSTANCE: value })).toBe(false); + }, + ); +}); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index e8e960a0ba..d47ecffeba 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -2,6 +2,8 @@ import { DatabaseSync } from "node:sqlite"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; import { createSqliteQueue } from "../../src/selfhost/sqlite-queue"; +import { PrActuationLockContendedError } from "../../src/queue/transient-locks"; +import { ATTEMPT_FREE_RETRY_DEADLINE_MS } from "../../src/queue/retryable"; import { jobCoalesceKey, queueSnapshotFromBinding } from "../../src/selfhost/queue-common"; import { DEFAULT_FOREGROUND_LANE_MAX_STARVE_AGE_MS } from "../../src/selfhost/queue-fairness"; import { renderMetrics, resetMetrics } from "../../src/selfhost/metrics"; @@ -4475,3 +4477,68 @@ describe("createSqliteQueue (durable #980)", () => { }); }); }); + +// #9465: lock contention is "not our turn yet", not a failure. It used to be charged an attempt like any other +// error, so with a flat 5s retry and maxRetries 5 a waiter DIED after roughly 25 seconds -- against a lock +// designed to be held for minutes (PR_ACTUATION_LOCK_TTL_SECONDS is 600, spanning a whole publish -> AI review +// -> maintain pass). Confirmed in production: the only dead-lettered jobs in a 7-day window were three +// actuation-lock contentions, one a reopen-reclose whose single webhook-gated trigger and absent reconciler +// meant that one-shot enforcement was lost outright. +describe("attempt-free deferral on lock contention (#9465)", () => { + beforeEach(() => { vi.spyOn(process.stdout, "write").mockImplementation(() => true); }); + afterEach(() => { vi.useRealTimers(); resetMetrics(); vi.restoreAllMocks(); }); + + const contend = () => { + throw new PrActuationLockContendedError("acme/widgets", 7, "public-surface-publish"); + }; + + it("REGRESSION: re-pends WITHOUT consuming an attempt, so a waiter cannot die while the lock is legitimately held", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => contend(), { maxRetries: 2 }); + await q.binding.send(msg("agent-regate-pr")); + + // Far more contended passes than maxRetries would ever allow. Each retry schedules a future run_after, so + // clear it between drains to actually re-process rather than silently no-op. + for (let i = 0; i < 8; i += 1) { + driver.query("UPDATE _selfhost_jobs SET run_after = 0", []); + await q.drain(); + } + + const { rows } = driver.query("SELECT status, attempts, deferred_by FROM _selfhost_jobs LIMIT 1", []); + const row = (rows as Array<{ status: string; attempts: number; deferred_by: string | null }>)[0]; + expect(row?.status).toBe("pending"); // NOT dead + expect(row?.attempts).toBe(0); // budget untouched + expect(row?.deferred_by).toBe("lock_contended"); // distinguishable from a rate-limit defer + }); + + it("INVARIANT: a genuine failure still consumes its budget and converges to dead — the deferral is narrow", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => { + throw new Error("a real failure"); + }, { maxRetries: 2 }); + await q.binding.send(msg("agent-regate-pr")); + for (let i = 0; i < 5; i += 1) { + driver.query("UPDATE _selfhost_jobs SET run_after = 0", []); + await q.drain(); + } + + const { rows } = driver.query("SELECT status FROM _selfhost_jobs LIMIT 1", []); + expect((rows as Array<{ status: string }>)[0]?.status).toBe("dead"); + }); + + it("INVARIANT: past the attempt-free deadline a wedged lock still converges to dead, rather than deferring forever", async () => { + const driver = makeDriver(); + const q = createSqliteQueue(driver, async () => contend(), { maxRetries: 1 }); + await q.binding.send(msg("agent-regate-pr")); + // Age the job beyond ATTEMPT_FREE_RETRY_DEADLINE_MS so the deferral no longer applies. + driver.query("UPDATE _selfhost_jobs SET created_at = ?", [Date.now() - (ATTEMPT_FREE_RETRY_DEADLINE_MS + 60_000)]); + + for (let i = 0; i < 4; i += 1) { + driver.query("UPDATE _selfhost_jobs SET run_after = 0", []); + await q.drain(); + } + + const { rows } = driver.query("SELECT status FROM _selfhost_jobs LIMIT 1", []); + expect((rows as Array<{ status: string }>)[0]?.status).toBe("dead"); // an operator can see it + }); +});