From a20cf1753973f1e172de23ea1051b6d6616cff8a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Marc-Andr=C3=A9=20Bombeck?= Date: Thu, 22 Oct 2026 06:41:17 +0200 Subject: [PATCH] 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 | 6 ++ src/lib/__tests__/db.test.ts | 14 +++ src/lib/db.ts | 23 ++++- src/lib/jobs/__tests__/boss-instance.test.ts | 3 +- src/lib/jobs/boss-instance.ts | 11 ++- .../worker-lifecycle-shutdown.test.ts | 86 +++++++++++++++++++ src/lib/jobs/reminder/worker-lifecycle.ts | 61 +++++++++---- 7 files changed, 178 insertions(+), 26 deletions(-) create mode 100644 src/lib/jobs/reminder/__tests__/worker-lifecycle-shutdown.test.ts diff --git a/docker-compose.yml b/docker-compose.yml index dad7cf80d..aa17c56b1 100644 --- a/docker-compose.yml +++ b/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` diff --git a/src/lib/__tests__/db.test.ts b/src/lib/__tests__/db.test.ts index 251b55e68..e8438f02b 100644 --- a/src/lib/__tests__/db.test.ts +++ b/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); }); diff --git a/src/lib/db.ts b/src/lib/db.ts index 9cfff9d20..b4849c81b 100644 --- a/src/lib/db.ts +++ b/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 { diff --git a/src/lib/jobs/__tests__/boss-instance.test.ts b/src/lib/jobs/__tests__/boss-instance.test.ts index 6b1950999..bcb2676e0 100644 --- a/src/lib/jobs/__tests__/boss-instance.test.ts +++ b/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", }); diff --git a/src/lib/jobs/boss-instance.ts b/src/lib/jobs/boss-instance.ts index c9f056acd..ef3fa2121 100644 --- a/src/lib/jobs/boss-instance.ts +++ b/src/lib/jobs/boss-instance.ts @@ -74,10 +74,13 @@ export function getGlobalBoss(): PgBoss | null { return ((globalThis as Record)[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, diff --git a/src/lib/jobs/reminder/__tests__/worker-lifecycle-shutdown.test.ts b/src/lib/jobs/reminder/__tests__/worker-lifecycle-shutdown.test.ts new file mode 100644 index 000000000..79a57389a --- /dev/null +++ b/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 void>(); + const exits: Array = []; + 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, + 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((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(() => {})) }, + 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); + }); +}); diff --git a/src/lib/jobs/reminder/worker-lifecycle.ts b/src/lib/jobs/reminder/worker-lifecycle.ts index 803e33705..9fcce6f06 100644 --- a/src/lib/jobs/reminder/worker-lifecycle.ts +++ b/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, + proc: Pick = process, +): void { + let drained: Promise | 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((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")); } /**