diff --git a/apps/loopover-ui/src/lib/selfhost-env-reference.ts b/apps/loopover-ui/src/lib/selfhost-env-reference.ts index 9ab6cfcf3..e7c0067ed 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 0ab733c1d..f4cae13f6 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 ca12e5308..b8b26ef21 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 0a38064ac..b2d991c43 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 a514dbbbe..b5ef29fd3 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 3d44101b5..7a00abb84 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 ce148888f..6cb6ebbfa 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 c6902cf50..e8d5a70ca 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 92b5df701..1d07ad11e 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 1b68a46c9..fdc1b84b3 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 3f18af59d..486a69074 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 65681fc41..cff6867b3 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 660441d8e..d62b19dc3 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,51 @@ 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 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); + 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 9fc06b254..635beb4ef 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); + }); +}); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index 6eaa69bf8..a87ec0627 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 c2f9954c0..fff461e12 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 e8e960a0b..d47ecffeb 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 + }); +});