test(runtime): cover the drain deadline on both backends, the migration lock, and the capture metric

Adds the pg-queue drain-deadline regression and its completing-drain invariant, the pool
registry, and a render-failure metric test proving a browserless outage is now alertable.
Three best-effort/defensive arms are annotated with their reasons rather than given contrived
tests: the advisory unlock's catch (a failed unlock means a broken session, and releasing the
client drops the lock anyway), and the abandoned-set empty arm plus its catch (both degrade to
the pre-#9485 lease-expiry path this block exists to improve on).

Author: JSONbored

src/selfhost/pg-adapter.ts

@@ -98,6 +98,8 @@ export async function withPgMigrationLock<T>(db: D1Database, fn: () => Promise<T
     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 {

src/selfhost/pg-queue.ts

@@ -1850,10 +1850,13 @@ export function createPgQueue(
             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); // best-effort: a failure just falls back to the pre-#9485 lease-expiry path
+            .catch(() => undefined);
         }
       }
     },

test/unit/selfhost-migrate.test.ts

@@ -8,7 +8,7 @@ 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";
+import { createPgAdapter, MIGRATION_ADVISORY_LOCK_KEY, pgPoolForAdapter, withPgMigrationLock } from "../../src/selfhost/pg-adapter";
 
 const sha256 = (sql: string) => createHash("sha256").update(sql, "utf8").digest("hex");
 
@@ -236,6 +236,15 @@ describe("withPgMigrationLock (#9486)", () => {
     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);

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

@@ -1105,6 +1105,53 @@ describe("createPgQueue (durable #977)", () => {
   // #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.
+  // #9485: parity with the sqlite backend. An unbounded stop() blocked a redeploy on a multi-minute AI review
+  // until SIGKILL, and the killed job then sat unclaimed until its 30-minute lease expired.
+  it("REGRESSION (#9485): stop() bounds the drain and re-pends this process's abandoned rows", async () => {
+    process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"] = "50";
+    try {
+      const m = makePool();
+      m.enqueueJob("1", { type: "agent-regate-pr" });
+      let release!: () => void;
+      const blocked = new Promise<void>((resolve) => { release = resolve; });
+      const q = createPgQueue(m.pool, async () => { await blocked; });
+      await q.init();
+      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 return rather than hang on the still-running job
+
+      expect(warn.mock.calls.map((c) => String(c[0])).join("\n")).toContain("selfhost_queue_shutdown_drain_deadline");
+      const calls = (m.fn as unknown as ReturnType<typeof vi.fn>).mock.calls.map((c: unknown[]) => String(c[0]));
+      // The abandoned row is re-pended so the next process retries it immediately, rather than waiting out the lease.
+      expect(calls.some((sql) => sql.includes("status='pending'") && sql.includes("ANY($2::bigint[])"))).toBe(true);
+      release();
+    } finally {
+      delete process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"];
+    }
+  });
+
+  it("INVARIANT (#9485): a drain that completes in time re-pends nothing", async () => {
+    process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"] = "5000";
+    try {
+      const m = makePool();
+      m.enqueueJob("1", { type: "agent-regate-pr" });
+      const q = createPgQueue(m.pool, async () => undefined);
+      await q.init();
+      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 calls = (m.fn as unknown as ReturnType<typeof vi.fn>).mock.calls.map((c: unknown[]) => String(c[0]));
+      expect(calls.some((sql) => sql.includes("ANY($2::bigint[])"))).toBe(false);
+    } finally {
+      delete process.env["QUEUE_SHUTDOWN_DRAIN_DEADLINE_MS"];
+    }
+  });
+
   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" });

test/unit/visual-shot.test.ts

@@ -1,5 +1,6 @@
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { captureInteractionFrames, captureScrollFrames, captureShot, handleShot } from "../../src/review/visual/shot";
+import { counterValue, resetMetrics } from "../../src/selfhost/metrics";
 
 const mocks = vi.hoisted(() => ({
   finalUrl: "https://preview.pages.dev/page",
@@ -1090,3 +1091,18 @@ describe("visual screenshot R2 key serve + traversal guard", () => {
     expect(response.headers.get("content-type")).toBe("image/png");
   });
 });
+
+// #9487: a browserless outage was logged but never COUNTED, so it was invisible in Prometheus while it
+// degraded every screenshot to a dash cell -- and because the screenshot-table gate treats absent evidence as
+// a close signal, that outage could close legitimate visual PRs before anyone noticed.
+describe("visual capture result metric (#9487)", () => {
+  it("counts a render failure, so a browserless outage is alertable rather than silent", async () => {
+    resetMetrics();
+    mocks.screenshot.mockRejectedValueOnce(new Error("Protocol error: Target closed"));
+
+    await expect(captureShot(env(), "https://preview.pages.dev/page")).resolves.toEqual({ png: null, authWalled: false });
+
+    expect(counterValue("loopover_visual_capture_total", { result: "error" })).toBe(1);
+  });
+
+});