fix(runtime): bound the shutdown drain, serialize boot migrations, and make capture failures visible (#9485, #9486, #9487)

stop() waited on in-flight work with NO deadline, so a redeploy blocked on a multi-minute AI
review until the orchestrator SIGKILLed the process -- and because the killed job's lease had
been heartbeated right up to the kill, reclaimExpiredProcessingJobs (which only recovers rows
older than 30 minutes) left it stalled for another 20-30. Against a 1-5 minute latency target
with automated redeploys, that was the largest single source of a silently stalled review. The
wait is now bounded, and on expiry this process re-pends its OWN in-flight rows -- safe because
activeJobIds is process-local -- turning a SIGKILL race into an immediate retry. Both backends.

Boot migrations took no cross-instance lock. Two instances booting together meant one applied a
migration atomically while the other's identical transaction failed "already exists", which
runSelfHostMigrations treats as DRIFT and retries statement-by-statement -- re-executing any
table-rebuild INSERT ... SELECT or UPDATE backfill against the already-migrated schema, then
dying on the ledger INSERT's unique violation (a message matching neither tolerated shape) and
crash-looping. #9027 made a single migration atomic against a crash; this makes the whole run
atomic against a concurrent boot, via a Postgres session advisory lock. A session lock is held
by its connection, so it cannot be taken through the pooled adapter -- hence the adapter->pool
registry rather than a new statement.

Visual capture failures were logged but never counted, so a browserless outage was invisible in
Prometheus while it degraded every screenshot to a dash cell -- and since the screenshot gate
treats absent evidence as a close signal, that outage could close legitimate visual PRs before
anyone noticed.

Author: JSONbored

src/review/visual/shot.ts

@@ -18,6 +18,7 @@
 // account API token. Returns null on any failure so callers degrade gracefully (the cell becomes a dash).
 import puppeteer from "@cloudflare/puppeteer";
 import { hostIsPrivateOrLocal, isSafeHttpUrl } from "../content-lane/safe-url";
+import { incr } from "../../selfhost/metrics";
 
 export type Viewport = { width: number; height: number };
 /** A `prefers-color-scheme` value the renderer can emulate before capture (#3678). */
@@ -362,10 +363,15 @@ export async function captureShot(env: Env, url: string, viewport: Viewport = VI
     // normal review pages without letting attacker-controlled document height or PNG size drive unbounded
     // Chromium raster work on the public screenshot route.
     const shot = await captureBoundedFullPageShot(page, viewport);
+    incr("loopover_visual_capture_total", { result: "ok" });
     return { png: shot, authWalled: false };
   } catch (error) {
     // Log before degrading to null — otherwise a networkidle0 timeout, a binding quota error, or a render
     // crash is indistinguishable from "no page" and the cell silently blanks.
+    // #9487: also COUNTED. The log alone is invisible to Prometheus, so a browserless outage silently degraded
+    // every screenshot to a dash cell with nothing to alert on -- and, because the screenshot-table gate treats
+    // absent evidence as a close signal, that outage could close legitimate visual PRs before anyone noticed.
+    incr("loopover_visual_capture_total", { result: "error" });
     console.log(JSON.stringify({ event: "render_screenshot_error", mode: "binding", url, message: String(error).slice(0, 200) }));
     return { png: null, authWalled: false };
   } finally {

src/selfhost/metrics.ts

@@ -58,6 +58,7 @@ export const DEFAULT_METRIC_META: readonly (readonly [string, MetricMeta])[] = [
   ["loopover_config_dir_empty_acknowledged", { help: "1 when LOOPOVER_REPO_CONFIG_DIR is unset, has entries, or is acknowledged; 0 when it's configured but the mounted directory is empty.", type: "gauge" }],
   ["loopover_http_requests_total", { help: "HTTP app requests by response status class.", type: "counter" }],
   ["loopover_http_request_duration_seconds", { help: "HTTP app request duration in seconds.", type: "histogram" }],
+  ["loopover_visual_capture_total", { help: "Visual capture attempts by result -- a browserless outage is otherwise invisible while it silently degrades screenshots to dash cells, and the screenshot gate treats absent evidence as a close signal (#9487).", type: "counter" }],
   ["loopover_webhook_dedup_total", { help: "Webhook deliveries deduplicated before enqueue.", type: "counter" }],
   ["loopover_webhook_enqueue_total", { help: "Webhook enqueue outcomes by event and action.", type: "counter" }],
   ["loopover_jobs_enqueued_total", { help: "Durable queue jobs enqueued.", type: "counter" }],

src/selfhost/pg-adapter.ts

@@ -62,6 +62,52 @@ class PgStatement implements SelfHostD1PreparedStatement {
   }
 }
 
+/**
+ * #9486: adapter -> pool registry, so a caller that needs a DEDICATED connection (not a pooled statement) can
+ * find it. A Postgres session advisory lock is held by the connection that took it, so it cannot be acquired
+ * through `prepare()`/`batch()` -- those hand back a pooled client per call. Kept as a WeakMap rather than a
+ * property on the adapter so the D1Database surface stays exactly the shape every other backend implements.
+ */
+const adapterPools = new WeakMap<D1Database, Pool>();
+
+/** The pool backing `db`, when it is a Postgres adapter; undefined for SQLite/D1. */
+export function pgPoolForAdapter(db: D1Database): Pool | undefined {
+  return adapterPools.get(db);
+}
+
+/**
+ * #9486: run `fn` while holding a Postgres session advisory lock, so two instances booting concurrently
+ * cannot apply migrations at the same time.
+ *
+ * Without it, instance A applies a migration atomically while instance B's identical transaction fails
+ * "already exists" -- which `runSelfHostMigrations` treats as DRIFT and retries statement-by-statement, so a
+ * table-rebuild `INSERT ... SELECT` or an UPDATE backfill RE-EXECUTES against the already-migrated schema.
+ * B then dies on the ledger INSERT's unique violation, whose message matches neither "duplicate column" nor
+ * "already exists", and crash-loops a cycle. #9027 made a single migration atomic against a crash; this makes
+ * the whole run atomic against a concurrent boot.
+ *
+ * Falls through to running `fn` directly on a non-Postgres backend (SQLite is single-process by construction).
+ */
+export async function withPgMigrationLock<T>(db: D1Database, fn: () => Promise<T>): Promise<T> {
+  const pool = adapterPools.get(db);
+  if (!pool) return fn();
+  const client = await pool.connect();
+  try {
+    // A stable, arbitrary key derived from the purpose; any instance using this same constant serializes.
+    await client.query("SELECT pg_advisory_lock($1)", [MIGRATION_ADVISORY_LOCK_KEY]);
+    try {
+      return await fn();
+    } finally {
+      await client.query("SELECT pg_advisory_unlock($1)", [MIGRATION_ADVISORY_LOCK_KEY]).catch(() => undefined);
+    }
+  } finally {
+    client.release();
+  }
+}
+
+/** Arbitrary but stable: every instance must use the SAME key for the lock to serialize them. */
+export const MIGRATION_ADVISORY_LOCK_KEY = 8_140_9486;
+
 export function createPgAdapter(pool: Pool): D1Database {
   const adapter: SelfHostD1Database = {
     prepare: (sql: string) => new PgStatement(pool, sql),
@@ -116,7 +162,9 @@ export function createPgAdapter(pool: Pool): D1Database {
       return new ArrayBuffer(0); // unused; present for D1 surface completeness
     },
   };
-  return adapter as unknown as D1Database;
+  const built = adapter as unknown as D1Database;
+  adapterPools.set(built, pool);
+  return built;
 }
 
 // #2543: github_rate_limit_observations receives one INSERT per outbound GitHub API response and is pruned in

src/selfhost/pg-queue.ts

@@ -12,6 +12,7 @@ import { capturePostHogError, withPostHogMonitor } from "./posthog";
 import {
   ATTEMPT_FREE_RETRY_DEADLINE_MS,
   consumingRetryDelayMs,
+  shutdownDrainDeadlineMs,
   isAttemptFreeRetry,
   deterministicJitterMs,
   FOREGROUND_QUEUE_PRIORITY_FLOOR,
@@ -1822,7 +1823,39 @@ export function createPgQueue(
       if (timer) clearTimeout(timer);
       if (deadLetterReviveTimer) clearInterval(deadLetterReviveTimer);
       if (foregroundLivenessTimer) clearInterval(foregroundLivenessTimer);
-      while (active > 0) await new Promise((r) => setTimeout(r, 10));
+      // #9485: the drain wait is BOUNDED. It used to be `while (active > 0)` with no deadline, so a redeploy
+      // (redeploy-companion recreates the container) waited on an in-flight multi-minute AI review until the
+      // orchestrator's grace period expired and SIGKILLed the process. The killed job's lease had been
+      // heartbeated right up to the kill, so reclaimExpiredProcessingJobs -- which only recovers rows older
+      // than processingTimeoutMs (30 min by default) -- did not pick it up for another 20-30 minutes. Against a
+      // 1-5 minute latency target with AUTOMATED redeploys, that is the single largest source of "a review
+      // silently stalled for half an hour".
+      //
+      // On expiry we re-pend THIS PROCESS'S OWN in-flight rows before returning. That is safe precisely because
+      // activeJobIds is process-local: these are rows we claimed and are about to abandon, so releasing them
+      // converts a SIGKILL race into an immediate retry by whoever comes up next, instead of a lease-expiry
+      // wait. Jobs that finish normally inside the deadline are unaffected.
+      const deadlineMs = shutdownDrainDeadlineMs();
+      const waitUntil = Date.now() + deadlineMs;
+      while (active > 0 && Date.now() < waitUntil) await new Promise((r) => setTimeout(r, 10));
+      if (active > 0) {
+        const abandoned = [...activeJobIds];
+        console.warn(
+          JSON.stringify({
+            level: "warn",
+            event: "selfhost_queue_shutdown_drain_deadline",
+            active,
+            abandoned: abandoned.length,
+            deadline_ms: deadlineMs,
+            message: "drain deadline reached; re-pending this process's in-flight jobs so they retry immediately instead of waiting out their lease",
+          }),
+        );
+        if (abandoned.length > 0) {
+          await pool
+            .query(`UPDATE ${TABLE} SET status='pending', run_after=$1 WHERE id = ANY($2::bigint[]) AND status='processing'`, [Date.now(), abandoned])
+            .catch(() => undefined); // best-effort: a failure just falls back to the pre-#9485 lease-expiry path
+        }
+      }
     },
     async drain() {
       while (active > 0) await new Promise((r) => setTimeout(r, 5));

src/selfhost/queue-common.ts

@@ -22,6 +22,7 @@ const DEFAULT_STARTUP_JITTER_MS = 3 * 60_000;
 const DEFAULT_RECOVERY_JITTER_MS = 60_000;
 const DEFAULT_SCHEDULED_ENQUEUE_JITTER_MS = 5 * 60_000;
 const DEFAULT_STARTUP_JITTER_MIN_JOBS = 8;
+const DEFAULT_SHUTDOWN_DRAIN_DEADLINE_MS = 120_000;
 const DEFAULT_PROCESSING_TIMEOUT_MS = 30 * 60_000;
 const DEFAULT_BACKGROUND_CONCURRENCY = 4;
 // Dead-letter auto-retry (#audit-rate-headroom): a job that exhausted its normal retry budget and landed in
@@ -725,6 +726,22 @@ export function queueProcessingTimeoutMs(): number {
   );
 }
 
+/**
+ * #9485: how long {@link stop} waits for in-flight work before re-pending it and returning.
+ *
+ * The wait used to be unbounded, so a redeploy blocked on a multi-minute AI review until the orchestrator's
+ * grace period expired and SIGKILLed the process -- and because the killed job's lease had been heartbeated
+ * right up to the kill, `reclaimExpiredProcessingJobs` (which only recovers rows older than
+ * QUEUE_PROCESSING_TIMEOUT_MS, 30 min by default) left it stalled for another 20-30 minutes.
+ *
+ * The default is deliberately generous: long enough that an ordinary review finishes and releases its own
+ * lock and lease cleanly, short enough to be well inside a typical `stop_grace_period` (this deployment's is
+ * 300s) so the re-pend happens on OUR terms rather than as a SIGKILL race.
+ */
+export function shutdownDrainDeadlineMs(): number {
+  return envDurationMs("QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS", DEFAULT_SHUTDOWN_DRAIN_DEADLINE_MS);
+}
+
 export function queueStartupJitterMinJobs(): number {
   return parsePositiveIntEnv("QUEUE_STARTUP_JITTER_MIN_JOBS", { min: 0, fallback: DEFAULT_STARTUP_JITTER_MIN_JOBS });
 }

src/selfhost/sqlite-queue.ts

@@ -13,6 +13,7 @@ import { capturePostHogError, withPostHogMonitor } from "./posthog";
 import {
   ATTEMPT_FREE_RETRY_DEADLINE_MS,
   consumingRetryDelayMs,
+  shutdownDrainDeadlineMs,
   isAttemptFreeRetry,
   deterministicJitterMs,
   FOREGROUND_QUEUE_PRIORITY_FLOOR,
@@ -1432,7 +1433,33 @@ export function createSqliteQueue(
       if (timer) clearTimeout(timer);
       if (deadLetterReviveTimer) clearInterval(deadLetterReviveTimer);
       if (foregroundLivenessTimer) clearInterval(foregroundLivenessTimer);
-      while (active > 0) await new Promise((r) => setTimeout(r, 10)); // let in-flight pumps finish
+      // #9485: bounded, matching the pg backend. An unbounded wait meant a redeploy blocked on a multi-minute
+      // AI review until SIGKILL, and the killed job then sat unclaimed until its lease expired. On expiry we
+      // re-pend THIS PROCESS'S OWN in-flight rows (activeJobIds is process-local, so these are rows we claimed
+      // and are abandoning) so they retry immediately instead of waiting out the lease.
+      const deadlineMs = shutdownDrainDeadlineMs();
+      const waitUntil = Date.now() + deadlineMs;
+      while (active > 0 && Date.now() < waitUntil) await new Promise((r) => setTimeout(r, 10));
+      if (active > 0) {
+        const abandoned = [...activeJobIds];
+        console.warn(
+          JSON.stringify({
+            level: "warn",
+            event: "selfhost_queue_shutdown_drain_deadline",
+            active,
+            abandoned: abandoned.length,
+            deadline_ms: deadlineMs,
+            message: "drain deadline reached; re-pending this process's in-flight jobs so they retry immediately instead of waiting out their lease",
+          }),
+        );
+        for (const id of abandoned) {
+          try {
+            driver.query(`UPDATE ${TABLE} SET status='pending', run_after=? WHERE id=? AND status='processing'`, [Date.now(), id]);
+          } catch {
+            // best-effort: a failure just falls back to the pre-#9485 lease-expiry path
+          }
+        }
+      }
     },
     async drain() {
       // send() fire-and-forgets a pump; wait for any in-flight pumps to settle, then drain to completion.

src/server.ts

@@ -551,9 +551,11 @@ async function main(): Promise<void> {
       }),
     );
 
-  const applied = await runSelfHostMigrations(
-    backend.db,
-    process.env.MIGRATIONS_DIR ?? "migrations",
+  // #9486: serialize the whole migration run across instances. On Postgres this takes a session advisory
+  // lock; on SQLite (single-process by construction) it runs straight through.
+  const { withPgMigrationLock } = await import("./selfhost/pg-adapter");
+  const applied = await withPgMigrationLock(backend.db, () =>
+    runSelfHostMigrations(backend.db, process.env.MIGRATIONS_DIR ?? "migrations"),
   );
   console.log(
     JSON.stringify({ event: "selfhost_migrations_applied", count: applied }),

test/unit/selfhost-migrate.test.ts

@@ -7,6 +7,8 @@ import { describe, expect, it } from "vitest";
 import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter";
 import { runSelfHostMigrations } from "../../src/selfhost/migrate";
 import { normalizePostgresValue } from "../../scripts/migrate-selfhost-sqlite-to-postgres";
+import type { Pool } from "pg";
+import { createPgAdapter, MIGRATION_ADVISORY_LOCK_KEY, withPgMigrationLock } from "../../src/selfhost/pg-adapter";
 
 const sha256 = (sql: string) => createHash("sha256").update(sql, "utf8").digest("hex");
 
@@ -169,3 +171,73 @@ describe("SQLite-to-Postgres migrator helpers", () => {
     expect(normalizePostgresValue(42)).toBe(42);
   });
 });
+
+// #9486: two instances booting concurrently could both apply migrations. Instance A applies one atomically;
+// B's identical transaction fails "already exists", which runSelfHostMigrations treats as DRIFT and retries
+// statement-by-statement -- so a table-rebuild INSERT ... SELECT or an UPDATE backfill RE-EXECUTES against the
+// already-migrated schema. B then dies on the ledger INSERT's unique violation, whose message matches neither
+// "duplicate column" nor "already exists", and crash-loops a cycle. #9027 made a single migration atomic
+// against a crash; this makes the whole RUN atomic against a concurrent boot.
+describe("withPgMigrationLock (#9486)", () => {
+  it("runs straight through on a non-Postgres backend (SQLite is single-process by construction)", async () => {
+    // Any db that was not built by createPgAdapter has no registered pool, so there is no lock to take.
+    const notPg = {} as unknown as D1Database;
+    let ran = false;
+    const result = await withPgMigrationLock(notPg, async () => {
+      ran = true;
+      return 42;
+    });
+    expect(ran).toBe(true);
+    expect(result).toBe(42);
+  });
+
+  it("takes and releases a session advisory lock on the Postgres adapter, around the work", async () => {
+    const queries: string[] = [];
+    const client = {
+      query: async (sql: string) => {
+        queries.push(sql);
+        return { rows: [], rowCount: 0 };
+      },
+      release: () => undefined,
+    };
+    const pool = { connect: async () => client } as unknown as Pool;
+    const db = createPgAdapter(pool);
+
+    const order: string[] = [];
+    const out = await withPgMigrationLock(db, async () => {
+      order.push("work");
+      return "done";
+    });
+
+    expect(out).toBe("done");
+    // Locked BEFORE the work and unlocked after -- an unlock-before-work ordering would serialize nothing.
+    expect(queries[0]).toContain("pg_advisory_lock");
+    expect(queries.at(-1)).toContain("pg_advisory_unlock");
+    expect(order).toEqual(["work"]);
+  });
+
+  it("INVARIANT: releases the lock and the client even when the work THROWS", async () => {
+    // A migration failure must not strand the advisory lock -- every later boot would block on it forever.
+    const queries: string[] = [];
+    let released = false;
+    const client = {
+      query: async (sql: string) => {
+        queries.push(sql);
+        return { rows: [], rowCount: 0 };
+      },
+      release: () => { released = true; },
+    };
+    const pool = { connect: async () => client } as unknown as Pool;
+    const db = createPgAdapter(pool);
+
+    await expect(withPgMigrationLock(db, async () => { throw new Error("migration blew up"); })).rejects.toThrow("migration blew up");
+
+    expect(queries.some((q) => q.includes("pg_advisory_unlock"))).toBe(true);
+    expect(released).toBe(true);
+  });
+
+  it("uses one stable key, so separate instances actually serialize against each other", () => {
+    expect(typeof MIGRATION_ADVISORY_LOCK_KEY).toBe("number");
+    expect(Number.isInteger(MIGRATION_ADVISORY_LOCK_KEY)).toBe(true);
+  });
+});

test/unit/selfhost-sqlite-queue.test.ts

@@ -4542,3 +4542,51 @@ describe("attempt-free deferral on lock contention (#9465)", () => {
     expect((rows as Array<{ status: string }>)[0]?.status).toBe("dead"); // an operator can see it
   });
 });
+
+// #9485: stop() used to wait `while (active > 0)` with NO deadline, so a redeploy blocked on an in-flight
+// multi-minute AI review until the orchestrator's grace period expired and SIGKILLed the process. Because the
+// killed job's lease had been heartbeated right up to the kill, reclaimExpiredProcessingJobs -- which only
+// recovers rows older than QUEUE_PROCESSING_TIMEOUT_MS (30 min by default) -- left it stalled for another
+// 20-30 minutes. Against a 1-5 minute latency target with AUTOMATED redeploys, that was the single largest
+// source of "a review silently stalled for half an hour".
+describe("bounded shutdown drain (#9485)", () => {
+  beforeEach(() => { vi.spyOn(process.stdout, "write").mockImplementation(() => true); });
+  afterEach(() => { vi.useRealTimers(); resetMetrics(); vi.restoreAllMocks(); delete process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"]; });
+
+  it("REGRESSION: stop() returns rather than hanging on a job that outlives the deadline", async () => {
+    process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"] = "50";
+    const driver = makeDriver();
+    let release!: () => void;
+    const blocked = new Promise<void>((resolve) => { release = resolve; });
+    const q = createSqliteQueue(driver, async () => { await blocked; });
+    await q.binding.send(msg("agent-regate-pr"));
+    void q.drain();
+    await new Promise((r) => setTimeout(r, 20)); // let the job be claimed
+
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    await q.stop(); // must not hang on the still-running job
+
+    const logged = warn.mock.calls.map((c) => String(c[0])).join("\n");
+    expect(logged).toContain("selfhost_queue_shutdown_drain_deadline");
+    // The abandoned row is re-pended, so the next process retries it immediately instead of waiting out the
+    // lease -- that recovery delay is the whole defect.
+    const { rows } = driver.query("SELECT status FROM _selfhost_jobs LIMIT 1", []);
+    expect((rows as Array<{ status: string }>)[0]?.status).toBe("pending");
+    release();
+  });
+
+  it("INVARIANT: a job that finishes inside the deadline drains normally and is NOT re-pended", async () => {
+    process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"] = "5000";
+    const driver = makeDriver();
+    const q = createSqliteQueue(driver, async () => undefined);
+    await q.binding.send(msg("agent-regate-pr"));
+    await q.drain();
+
+    const warn = vi.spyOn(console, "warn").mockImplementation(() => undefined);
+    await q.stop();
+
+    expect(warn.mock.calls.map((c) => String(c[0])).join("\n")).not.toContain("shutdown_drain_deadline");
+    const { rows } = driver.query("SELECT count(*) AS n FROM _selfhost_jobs", []);
+    expect((rows as Array<{ n: number }>)[0]?.n).toBe(0); // completed and deleted, not re-pended
+  });
+});