From 1089e12fcbeb11e3bc882c07129ca428b06f27e9 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 19:25:41 -0700 Subject: [PATCH 1/3] 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. --- src/review/visual/shot.ts | 6 +++ src/selfhost/metrics.ts | 1 + src/selfhost/pg-adapter.ts | 50 ++++++++++++++++- src/selfhost/pg-queue.ts | 35 +++++++++++- src/selfhost/queue-common.ts | 17 ++++++ src/selfhost/sqlite-queue.ts | 29 +++++++++- src/server.ts | 8 +-- test/unit/selfhost-migrate.test.ts | 72 +++++++++++++++++++++++++ test/unit/selfhost-sqlite-queue.test.ts | 48 +++++++++++++++++ 9 files changed, 260 insertions(+), 6 deletions(-) diff --git a/src/review/visual/shot.ts b/src/review/visual/shot.ts index 1696112687..20c457a069 100644 --- a/src/review/visual/shot.ts +++ b/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 { diff --git a/src/selfhost/metrics.ts b/src/selfhost/metrics.ts index d505d14e79..c60d1d5c5a 100644 --- a/src/selfhost/metrics.ts +++ b/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" }], diff --git a/src/selfhost/pg-adapter.ts b/src/selfhost/pg-adapter.ts index 5651453940..b804f387e8 100644 --- a/src/selfhost/pg-adapter.ts +++ b/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(); + +/** 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(db: D1Database, fn: () => Promise): Promise { + 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 diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index e8d5a70ca1..75f0084412 100644 --- a/src/selfhost/pg-queue.ts +++ b/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)); diff --git a/src/selfhost/queue-common.ts b/src/selfhost/queue-common.ts index 1d07ad11ee..d5acfc62b7 100644 --- a/src/selfhost/queue-common.ts +++ b/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 }); } diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 486a69074b..6b1bf7eafa 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/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. diff --git a/src/server.ts b/src/server.ts index cff6867b3a..2ecb7563cd 100644 --- a/src/server.ts +++ b/src/server.ts @@ -551,9 +551,11 @@ async function main(): Promise { }), ); - 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 }), diff --git a/test/unit/selfhost-migrate.test.ts b/test/unit/selfhost-migrate.test.ts index d25868b6bb..434dada67a 100644 --- a/test/unit/selfhost-migrate.test.ts +++ b/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); + }); +}); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index d47ecffeba..96a6ad265b 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/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((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 + }); +}); From 2d85835c0aa8d27f6234370c5961ebead51bf259 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 19:30:36 -0700 Subject: [PATCH 2/3] 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). --- src/selfhost/pg-adapter.ts | 2 ++ src/selfhost/pg-queue.ts | 5 ++- test/unit/selfhost-migrate.test.ts | 11 ++++++- test/unit/selfhost-pg-queue.test.ts | 47 +++++++++++++++++++++++++++++ test/unit/visual-shot.test.ts | 16 ++++++++++ 5 files changed, 79 insertions(+), 2 deletions(-) diff --git a/src/selfhost/pg-adapter.ts b/src/selfhost/pg-adapter.ts index b804f387e8..253eb6b800 100644 --- a/src/selfhost/pg-adapter.ts +++ b/src/selfhost/pg-adapter.ts @@ -98,6 +98,8 @@ export async function withPgMigrationLock(db: D1Database, fn: () => Promise undefined); } } finally { diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index 75f0084412..44fe3b9fbc 100644 --- a/src/selfhost/pg-queue.ts +++ b/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); } } }, diff --git a/test/unit/selfhost-migrate.test.ts b/test/unit/selfhost-migrate.test.ts index 434dada67a..bce267c9bc 100644 --- a/test/unit/selfhost-migrate.test.ts +++ b/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); diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index a87ec06271..e833ff5c55 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/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((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).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).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" }); diff --git a/test/unit/visual-shot.test.ts b/test/unit/visual-shot.test.ts index 807438deea..b6151644aa 100644 --- a/test/unit/visual-shot.test.ts +++ b/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); + }); + +}); From e55717e3858878a86eef11a04553a6bba3f741de Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:07:07 -0700 Subject: [PATCH 3/3] chore(selfhost): regenerate the env reference for the new env reads --- apps/loopover-ui/src/lib/selfhost-env-reference.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/apps/loopover-ui/src/lib/selfhost-env-reference.ts b/apps/loopover-ui/src/lib/selfhost-env-reference.ts index e7c0067ed3..001a325a55 100644 --- a/apps/loopover-ui/src/lib/selfhost-env-reference.ts +++ b/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` |",