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

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

@@ -585,6 +585,10 @@ export const SELFHOST_ENV_REFERENCE_ROWS: SelfHostEnvReferenceRow[] = [
     name: "QUEUE_RECOVERY_JITTER_MS",
     firstReference: "src/selfhost/queue-common.ts",
   },
+  {
+    name: "QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS",
+    firstReference: "src/selfhost/queue-common.ts",
+  },
   {
     name: "QUEUE_STARTUP_JITTER_MIN_JOBS",
     firstReference: "src/selfhost/queue-common.ts",
@@ -807,6 +811,7 @@ export const SELFHOST_ENV_REFERENCE_MARKDOWN = [
   "| `QUEUE_PROCESSING_TIMEOUT_MS` | `src/selfhost/queue-common.ts` |",
   "| `QUEUE_RATE_LIMIT_JITTER_MS` | `src/selfhost/queue-common.ts` |",
   "| `QUEUE_RECOVERY_JITTER_MS` | `src/selfhost/queue-common.ts` |",
+  "| `QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS` | `src/selfhost/queue-common.ts` |",
   "| `QUEUE_STARTUP_JITTER_MIN_JOBS` | `src/selfhost/queue-common.ts` |",
   "| `QUEUE_STARTUP_JITTER_MS` | `src/selfhost/queue-common.ts` |",
   "| `REDEPLOY_COMPANION_SOCKET_PATH` | `src/server.ts` |",

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,54 @@ 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 {
+      /* v8 ignore next -- best-effort: if the unlock query itself fails the session is already broken, and
+         releasing the client below drops the lock with it. */
+      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 +164,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,42 @@ 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",
+          }),
+        );
+        /* v8 ignore next 5 -- the empty-set arm needs `active > 0` with no claimed ids (the counters can only
+           diverge transiently), and the .catch is best-effort: either way it degrades to the pre-#9485
+           lease-expiry path, which is exactly the behaviour this block improves on rather than depends on. */
+        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);
+        }
+      }
     },
     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, pgPoolForAdapter, withPgMigrationLock } from "../../src/selfhost/pg-adapter";
 
 const sha256 = (sql: string) => createHash("sha256").update(sql, "utf8").digest("hex");
 
@@ -169,3 +171,82 @@ 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("registers the pool so a caller needing a DEDICATED connection can find it", () => {
+    // A session advisory lock is held by its connection, so it cannot be taken through the pooled adapter --
+    // hence the registry rather than a new statement on the D1 surface.
+    const pool = { connect: async () => ({ query: async () => ({ rows: [], rowCount: 0 }), release: () => undefined }) } as unknown as Pool;
+    const db = createPgAdapter(pool);
+    expect(pgPoolForAdapter(db)).toBe(pool);
+    expect(pgPoolForAdapter({} as unknown as D1Database)).toBeUndefined();
+  });
+
+  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);
+  });
+});