fix(selfhost): catch unhandled pump() rejections in sqlite-queue/pg-queue

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--;
     }

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--;
     }

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);

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(),