fix(jobs): let running jobs finish on stop, and give the worker four pg-boss sockets

Next's signal handler closed the server and called process.exit, which
won the race against boss.stop, so a running job was cut off and stayed
active until its expiry. The shutdown handler now holds process.exit until
the boss has stopped, at most 30 s plus a margin, and compose gives the
container 45 s to stop.

A process that runs the worker gets four pg-boss connections instead of
two for its pollers, cron, supervision and job writes; a web-only producer
keeps two. Both come out of the same per-process budget.

docker-compose.yml

@@ -508,6 +508,11 @@ services:
     # support report and Admin > System status say so. On a busy instance,
     # raise it with APP_MEMORY_LIMIT in .env; do not lower it.
     mem_limit: "${APP_MEMORY_LIMIT:-1536m}"
+    # On `docker stop` (a redeploy) the app stops taking requests, lets the
+    # requests in flight finish and gives running jobs up to 30 s to end
+    # before it exits. Docker's default of 10 s cut that short and left
+    # jobs marked active until their expiry, holding exclusive queues.
+    stop_grace_period: 45s
     healthcheck:
       test:
         [
@@ -564,6 +569,7 @@ services:
   #       max-size: "10m"
   #       max-file: "3"
   #   mem_limit: "${WORKER_MEMORY_LIMIT:-1536m}"
+  #   stop_grace_period: 45s
 
   db:
     # Pinned to the concrete minor instead of the floating `16-alpine`

src/lib/__tests__/db.test.ts

@@ -21,8 +21,10 @@ describe("per-process database connection budget", () => {
   const originalConnectionLimit = process.env.DB_CONNECTION_LIMIT;
   const originalPoolTimeout = process.env.DB_POOL_TIMEOUT;
   const originalLegacyPoolMax = process.env.DATABASE_POOL_MAX;
+  const originalProcessType = process.env.HEALTHLOG_PROCESS_TYPE;
 
   beforeEach(() => {
+    process.env.HEALTHLOG_PROCESS_TYPE = "web";
     process.env.DATABASE_URL =
       "postgresql://healthlog:test@db:5432/healthlog?connection_limit=20&pool_timeout=20";
     delete process.env.DB_CONNECTION_LIMIT;
@@ -39,6 +41,7 @@ describe("per-process database connection budget", () => {
     restore("DB_CONNECTION_LIMIT", originalConnectionLimit);
     restore("DB_POOL_TIMEOUT", originalPoolTimeout);
     restore("DATABASE_POOL_MAX", originalLegacyPoolMax);
+    restore("HEALTHLOG_PROCESS_TYPE", originalProcessType);
   });
 
   it("splits the URL connection_limit across Prisma and pg-boss", () => {
@@ -48,6 +51,17 @@ describe("per-process database connection budget", () => {
     expect(getPrismaPoolMax() + getPgBossPoolMax()).toBe(getConnectionBudget());
   });
 
+  it("gives a process that runs the worker four pg-boss connections", () => {
+    for (const role of ["all", "worker"]) {
+      process.env.HEALTHLOG_PROCESS_TYPE = role;
+      expect(getPgBossPoolMax()).toBe(4);
+      expect(getPrismaPoolMax()).toBe(16);
+    }
+    process.env.DB_CONNECTION_LIMIT = "3";
+    expect(getPgBossPoolMax()).toBe(2);
+    expect(getPrismaPoolMax()).toBe(1);
+  });
+
   it("uses the existing pool_timeout seconds for both pool constructors", () => {
     expect(getPoolConnectionTimeoutMs()).toBe(20_000);
   });

src/lib/db.ts

@@ -6,6 +6,7 @@ import {
   sessionOptionsDisabled,
   TIME_ZONE_OPTION,
 } from "@/lib/db-session";
+import { shouldRunWorker } from "@/lib/process-type";
 
 export {
   pinTimeZoneOnConnect,
@@ -21,7 +22,17 @@ const globalForPrisma = globalThis as unknown as {
 
 const DEFAULT_CONNECTION_BUDGET = 20;
 const DEFAULT_POOL_TIMEOUT_SECONDS = 20;
-const PG_BOSS_CONNECTIONS = 2;
+/**
+ * pg-boss connections. A web-only process only sends jobs, and two sockets
+ * carry that. A process that runs the worker polls: about a hundred bindings
+ * (two dozen of them every two seconds, several with two to five slots), the
+ * cron timekeeper, supervision, and every job's own start and completion
+ * writes, plus the web sends when web and worker share the process. On two
+ * sockets those queued behind each other; four keep a fetch from waiting on
+ * a completion. The connections come out of the same per-process budget.
+ */
+const PG_BOSS_PRODUCER_CONNECTIONS = 2;
+const PG_BOSS_WORKER_CONNECTIONS = 4;
 
 function positiveInteger(raw: string | undefined): number | null {
   if (!raw) return null;
@@ -53,8 +64,14 @@ export function getConnectionBudget(): number {
   return Math.max(2, configured);
 }
 
-export function getPgBossPoolMax(): number {
-  return Math.min(PG_BOSS_CONNECTIONS, getConnectionBudget() - 1);
+export function getPgBossPoolMax(
+  role: "worker" | "producer" = shouldRunWorker() ? "worker" : "producer",
+): number {
+  const wanted =
+    role === "worker"
+      ? PG_BOSS_WORKER_CONNECTIONS
+      : PG_BOSS_PRODUCER_CONNECTIONS;
+  return Math.min(wanted, getConnectionBudget() - 1);
 }
 
 export function getPrismaPoolMax(): number {

src/lib/jobs/__tests__/boss-instance.test.ts

@@ -108,9 +108,10 @@ describe("worker pg-boss pool", () => {
   it("uses the pg-boss share of the process connection budget", () => {
     createWorkerBoss("postgres://healthlog@test/db");
 
+    // A worker polls; a web-only producer keeps two (see `db.ts`).
     expect(PgBoss).toHaveBeenCalledWith({
       connectionString: "postgres://healthlog@test/db",
-      max: 2,
+      max: 4,
       connectionTimeoutMillis: 20_000,
       options: "-c TimeZone=UTC",
     });

src/lib/jobs/boss-instance.ts

@@ -74,10 +74,13 @@ export function getGlobalBoss(): PgBoss | null {
   return ((globalThis as Record<string, unknown>)[BOSS_KEY] as PgBoss) ?? null;
 }
 
-function databaseOptions(connectionString: string) {
+function databaseOptions(
+  connectionString: string,
+  role: "worker" | "producer",
+) {
   return {
     connectionString,
-    max: getPgBossPoolMax(),
+    max: getPgBossPoolMax(role),
     connectionTimeoutMillis: getPoolConnectionTimeoutMs(),
     // pg-boss keeps its own tables in `timestamptz`, so its queue math does
     // not depend on the zone; the pin keeps every HealthLog session alike.
@@ -86,12 +89,12 @@ function databaseOptions(connectionString: string) {
 }
 
 export function createWorkerBoss(connectionString: string): PgBoss {
-  return new PgBoss(databaseOptions(connectionString));
+  return new PgBoss(databaseOptions(connectionString, "worker"));
 }
 
 function producerOptions(connectionString: string) {
   return {
-    ...databaseOptions(connectionString),
+    ...databaseOptions(connectionString, "producer"),
     migrate: false,
     schedule: false,
     supervise: false,

src/lib/jobs/reminder/__tests__/worker-lifecycle-shutdown.test.ts

@@ -0,0 +1,86 @@
+import { describe, expect, it, vi } from "vitest";
+
+vi.mock("@/lib/jobs/apple-health-import-worker", () => ({
+  reconcileOrphanImportJobs: vi.fn(),
+}));
+vi.mock("@/lib/jobs/integration-status-null-probe", () => ({
+  probeIntegrationStatusNullBuckets: vi.fn(),
+}));
+vi.mock("@/lib/logging/background", () => ({ withBackgroundEvent: vi.fn() }));
+vi.mock("../shared", () => ({ getWorkerPrisma: vi.fn(), workerLog: vi.fn() }));
+
+import {
+  registerShutdownHandlers,
+  SHUTDOWN_JOB_TIMEOUT_MS,
+} from "../worker-lifecycle";
+
+function fakeProcess() {
+  const listeners = new Map<string, () => void>();
+  const exits: Array<number | string | null | undefined> = [];
+  const proc = {
+    once: (event: string, fn: () => void) => {
+      listeners.set(event, fn);
+      return proc;
+    },
+    exit: (code?: number | string | null) => {
+      exits.push(code);
+    },
+  };
+  return {
+    proc: proc as unknown as Pick<NodeJS.Process, "once" | "exit">,
+    signal: (name: string) => listeners.get(name)?.(),
+    exits,
+  };
+}
+
+describe("graceful shutdown", () => {
+  it("holds the process exit until running jobs have finished", async () => {
+    let finishJobs!: () => void;
+    const stopped = new Promise<void>((r) => (finishJobs = r));
+    const stop = vi.fn(() => stopped);
+    const fake = fakeProcess();
+    registerShutdownHandlers({ stop }, fake.proc);
+
+    fake.signal("SIGTERM");
+    expect(stop).toHaveBeenCalledWith({
+      graceful: true,
+      timeout: SHUTDOWN_JOB_TIMEOUT_MS,
+    });
+    // Next's handler closes the server and exits; the exit waits.
+    fake.proc.exit(143);
+    await new Promise((r) => setTimeout(r, 10));
+    expect(fake.exits).toEqual([]);
+
+    finishJobs();
+    await new Promise((r) => setTimeout(r, 0));
+    expect(fake.exits).toEqual([143]);
+  });
+
+  it("exits at the deadline when a job does not finish", async () => {
+    vi.useFakeTimers();
+    try {
+      const fake = fakeProcess();
+      registerShutdownHandlers(
+        { stop: vi.fn(() => new Promise<void>(() => {})) },
+        fake.proc,
+      );
+      fake.signal("SIGTERM");
+      fake.proc.exit(143);
+      await vi.advanceTimersByTimeAsync(SHUTDOWN_JOB_TIMEOUT_MS);
+      expect(fake.exits).toEqual([]);
+      await vi.advanceTimersByTimeAsync(5_000);
+      expect(fake.exits).toEqual([143]);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("drains once, whatever signals follow", () => {
+    const stop = vi.fn(() => Promise.resolve());
+    const fake = fakeProcess();
+    registerShutdownHandlers({ stop }, fake.proc);
+    fake.signal("SIGTERM");
+    fake.signal("SIGINT");
+    expect(stop).toHaveBeenCalledTimes(1);
+  });
+});

src/lib/jobs/reminder/worker-lifecycle.ts

@@ -16,30 +16,55 @@ import { withBackgroundEvent } from "@/lib/logging/background";
 
 import { getWorkerPrisma, workerLog } from "./shared";
 
+/** How long running jobs get to finish on a stop signal. */
+export const SHUTDOWN_JOB_TIMEOUT_MS = 30_000;
+
 /**
  * Graceful shutdown: drain in-flight jobs on SIGTERM/SIGINT (sent by Docker
  * Compose `docker stop`, Kubernetes pod termination, Coolify redeploys).
- * Without this, pending handlers were force-killed and could either be lost
- * or replayed on next start. The listeners register once — re-entering the
- * worker boot (e.g. on hot-reload in dev) is a no-op for the handlers because
- * they capture `boss` by closure and the first signal stops everything.
+ *
+ * Next's own signal handler closes the HTTP server (finishing the requests
+ * in flight) and then calls `process.exit`. That exit used to win the race
+ * against `boss.stop`, so a running job was cut off mid-way and stayed
+ * `active` until its expiry, which kept an exclusive queue blocked. The
+ * handler therefore holds `process.exit` until the boss has stopped: both
+ * drains run side by side, the process leaves when the slower one is done,
+ * and never later than {@link SHUTDOWN_JOB_TIMEOUT_MS} plus a margin, so a
+ * stuck handler cannot block a deploy. Compose gives the container 45 s
+ * (`stop_grace_period`) for this.
+ *
+ * The listeners register once; re-entering the worker boot (hot reload in
+ * dev) is a no-op for them because the first signal stops everything.
  */
-export function registerShutdownHandlers(boss: PgBoss): void {
-  let shutdownInProgress = false;
-  const onSignal = async (signal: "SIGTERM" | "SIGINT") => {
-    if (shutdownInProgress) return;
-    shutdownInProgress = true;
+export function registerShutdownHandlers(
+  boss: Pick<PgBoss, "stop">,
+  proc: Pick<NodeJS.Process, "once" | "exit"> = process,
+): void {
+  let drained: Promise<void> | null = null;
+  const onSignal = (signal: "SIGTERM" | "SIGINT") => {
+    if (drained) return;
     workerLog("error", `received ${signal}, draining boss`);
-    try {
-      // graceful=true waits for active handlers to finish, then closes the
-      // pg connection pool. timeout cap so a stuck handler can't block deploys.
-      await boss.stop({ graceful: true, timeout: 30_000 });
-    } catch (err) {
-      workerLog("error", "boss.stop failed during shutdown", err);
-    }
+    // graceful=true waits for active handlers to finish, then closes the
+    // pg connection pool. The timeout caps it.
+    drained = boss
+      .stop({ graceful: true, timeout: SHUTDOWN_JOB_TIMEOUT_MS })
+      .catch((err: unknown) => {
+        workerLog("error", "boss.stop failed during shutdown", err);
+      });
+    const exit = proc.exit.bind(proc);
+    const deadline = new Promise<void>((resolve) =>
+      setTimeout(resolve, SHUTDOWN_JOB_TIMEOUT_MS + 5_000).unref(),
+    );
+    let exiting = false;
+    proc.exit = ((code?: number | string | null) => {
+      if (exiting) return undefined as never;
+      exiting = true;
+      void Promise.race([drained, deadline]).finally(() => exit(code));
+      return undefined as never;
+    }) as typeof proc.exit;
   };
-  process.once("SIGTERM", () => void onSignal("SIGTERM"));
-  process.once("SIGINT", () => void onSignal("SIGINT"));
+  proc.once("SIGTERM", () => onSignal("SIGTERM"));
+  proc.once("SIGINT", () => onSignal("SIGINT"));
 }
 
 /**