From c41827e5a6fda2084f770417856367ea42b9a59e Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Thu, 2 Jul 2026 01:38:20 -0700 Subject: [PATCH] fix(selfhost): catch unhandled pump() rejections in sqlite-queue/pg-queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit claimNext()/reclaimExpiredProcessingJobs() run outside processOne()'s own try/finally in both queue drivers, so a raw driver failure (a locked SQLite file, a dropped Postgres connection) propagated out of pump() uncaught. Every void pump() call site (kickOne/kickAll) is fire-and-forget, so this surfaced as an unhandled promise rejection — fatal when SENTRY_DSN is unset, since server.ts only installs the unhandledRejection handler when Sentry is configured. A single transient DB error could silently kill the whole self-host server. Wrap pump()'s body in a try/catch that logs and captures the failure instead of letting it propagate, matching the pattern already used throughout processOne() for job-level errors. --- src/selfhost/pg-queue.ts | 13 ++++++++ src/selfhost/sqlite-queue.ts | 13 ++++++++ test/unit/selfhost-pg-queue.test.ts | 39 +++++++++++++++++++++++ test/unit/selfhost-sqlite-queue.test.ts | 41 +++++++++++++++++++++++++ 4 files changed, 106 insertions(+) diff --git a/src/selfhost/pg-queue.ts b/src/selfhost/pg-queue.ts index d4d00a6924..5dc1773823 100644 --- a/src/selfhost/pg-queue.ts +++ b/src/selfhost/pg-queue.ts @@ -553,6 +553,19 @@ export function createPgQueue( while (await processOne()) { /* drain due jobs */ } + } catch (error) { + // claimNext()/reclaimExpiredProcessingJobs() run OUTSIDE processOne's own try/finally, so a raw pool + // failure (a dropped connection, a lock timeout) lands here. Every `void pump()` call site (kickOne/kickAll) + // is fire-and-forget, so an uncaught rejection here would surface as an unhandled promise rejection — fatal + // when SENTRY_DSN is unset (server.ts only installs the handler when Sentry is configured) (#2498). + console.error( + JSON.stringify({ + level: "error", + event: "selfhost_queue_pump_crashed", + error: errorMessageWithCause(error), + }), + ); + captureError(error, { kind: "queue_pump_crashed" }); } finally { active--; } diff --git a/src/selfhost/sqlite-queue.ts b/src/selfhost/sqlite-queue.ts index 48cd1b35c6..923c8ad8b2 100644 --- a/src/selfhost/sqlite-queue.ts +++ b/src/selfhost/sqlite-queue.ts @@ -496,6 +496,19 @@ export function createSqliteQueue( while (await processOne()) { /* keep draining due jobs */ } + } catch (error) { + // claimNext()/reclaimExpiredProcessingJobs() run OUTSIDE processOne's own try/finally, so a raw driver + // failure (e.g. a transient SQLite error) lands here. Every `void pump()` call site (kickOne/kickAll) is + // fire-and-forget, so an uncaught rejection here would surface as an unhandled promise rejection — fatal + // when SENTRY_DSN is unset (server.ts only installs the handler when Sentry is configured) (#2498). + console.error( + JSON.stringify({ + level: "error", + event: "selfhost_queue_pump_crashed", + error: errorMessageWithCause(error), + }), + ); + captureError(error, { kind: "queue_pump_crashed" }); } finally { active--; } diff --git a/test/unit/selfhost-pg-queue.test.ts b/test/unit/selfhost-pg-queue.test.ts index f08424d3fe..8bd245821f 100644 --- a/test/unit/selfhost-pg-queue.test.ts +++ b/test/unit/selfhost-pg-queue.test.ts @@ -1184,6 +1184,45 @@ describe("createPgQueue (durable #977)", () => { } }); + it("pump absorbs a claimNext() pool failure instead of crashing the process (regression for #2498)", async () => { + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + if (String(sql).includes("RETURNING id, payload, attempts, job_key, priority")) throw new Error("connection terminated unexpectedly"); + return { rows: [], rowCount: 0 }; + }); + const pool = { query: fn } as unknown as Pool; + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + const q = createPgQueue(pool, async () => undefined); + await q.init(); + + await expect(q.drain()).resolves.toBeUndefined(); + + const logged = errorSpy.mock.calls.map(([line]) => String(line)); + expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("connection terminated unexpectedly"))).toBe(true); + }); + + it("pump absorbs a reclaimExpiredProcessingJobs() pool failure instead of crashing the process (regression for #2498)", async () => { + const oldTimeout = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1"; + try { + const fn = vi.fn().mockImplementation(async (sql: unknown) => { + if (String(sql).includes("WHERE status='processing' AND run_after<=$1")) throw new Error("connection terminated unexpectedly"); + return { rows: [], rowCount: 0 }; + }); + const pool = { query: fn } as unknown as Pool; + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + const q = createPgQueue(pool, async () => undefined); + await q.init(); + + await expect(q.drain()).resolves.toBeUndefined(); + + const logged = errorSpy.mock.calls.map(([line]) => String(line)); + expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("connection terminated unexpectedly"))).toBe(true); + } finally { + if (oldTimeout === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = oldTimeout; + } + }); + it("reschedules retryable incomplete review jobs while consuming attempts", async () => { const m = makePool(); m.enqueueJob("1", { type: "agent-regate-pr" }, 0); diff --git a/test/unit/selfhost-sqlite-queue.test.ts b/test/unit/selfhost-sqlite-queue.test.ts index 5790101c58..eb33d0ad2b 100644 --- a/test/unit/selfhost-sqlite-queue.test.ts +++ b/test/unit/selfhost-sqlite-queue.test.ts @@ -1757,6 +1757,47 @@ describe("createSqliteQueue (durable #980)", () => { } }); + it("pump absorbs a claimNext() driver failure instead of crashing the process (regression for #2498)", async () => { + const driver = makeDriver(); + const realQuery = driver.query.bind(driver); + const q = createSqliteQueue(driver, async () => undefined); + // Only claimNextWhere's SELECT starts with this exact column list — spreadDueJobsOnStartup (which already + // ran during construction above) selects a different column set, so this doesn't clobber setup. + vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { + if (sql.includes("SELECT id, payload, attempts, job_key, priority")) throw new Error("database is locked"); + return realQuery(sql, params); + }); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + + await expect(q.drain()).resolves.toBeUndefined(); + + const logged = errorSpy.mock.calls.map(([line]) => String(line)); + expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("database is locked"))).toBe(true); + }); + + it("pump absorbs a reclaimExpiredProcessingJobs() driver failure instead of crashing the process (regression for #2498)", async () => { + const oldTimeout = process.env.QUEUE_PROCESSING_TIMEOUT_MS; + process.env.QUEUE_PROCESSING_TIMEOUT_MS = "1"; + try { + const driver = makeDriver(); + const realQuery = driver.query.bind(driver); + vi.spyOn(driver, "query").mockImplementation((sql: string, params: unknown[]) => { + if (sql.includes("WHERE status='processing' AND run_after<=?")) throw new Error("disk I/O error"); + return realQuery(sql, params); + }); + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => undefined); + const q = createSqliteQueue(driver, async () => undefined); + + await expect(q.drain()).resolves.toBeUndefined(); + + const logged = errorSpy.mock.calls.map(([line]) => String(line)); + expect(logged.some((line) => line.includes("selfhost_queue_pump_crashed") && line.includes("disk I/O error"))).toBe(true); + } finally { + if (oldTimeout === undefined) delete process.env.QUEUE_PROCESSING_TIMEOUT_MS; + else process.env.QUEUE_PROCESSING_TIMEOUT_MS = oldTimeout; + } + }); + it("records 'unknown error' when a consumer throws a non-Error", async () => { const q = createSqliteQueue( makeDriver(),