diff --git a/docker-compose.yml b/docker-compose.yml index 469ab0a13..1987d0311 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -227,7 +227,7 @@ services: test: [ "CMD-SHELL", - "wget --no-verbose --tries=1 --spider http://127.0.0.1:3000/api/version || exit 1", + "wget --no-verbose --tries=1 --spider http://127.0.0.1:3000/api/health || exit 1", ] interval: 30s timeout: 5s @@ -245,7 +245,7 @@ services: # dockerfile: Dockerfile # container_name: healthlog-worker # environment: - # DATABASE_URL: "postgresql://healthlog:${POSTGRES_PASSWORD}@db:5432/healthlog?schema=public" + # DATABASE_URL: "postgresql://healthlog:${POSTGRES_PASSWORD}@db:5432/healthlog?schema=public&connection_limit=${DB_CONNECTION_LIMIT:-20}&pool_timeout=${DB_POOL_TIMEOUT:-20}" # NODE_ENV: production # ENCRYPTION_KEY: "${ENCRYPTION_KEY}" # ENCRYPTION_KEYS: "${ENCRYPTION_KEYS:-}" diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh index 968c88d10..3386a8879 100644 --- a/docker-entrypoint.sh +++ b/docker-entrypoint.sh @@ -9,7 +9,7 @@ set -e # Validating here guarantees a clear, deploy-platform-agnostic # failure message before we touch the database. missing="" -for var in DATABASE_URL ENCRYPTION_KEY API_TOKEN_HMAC_KEY; do +for var in DATABASE_URL API_TOKEN_HMAC_KEY; do eval "value=\$$var" if [ -z "$value" ]; then missing="$missing $var" @@ -21,6 +21,48 @@ if [ -n "$missing" ]; then exit 1 fi +# ENCRYPTION_KEYS supersedes the legacy single key, so validate the same +# fail-closed shape the runtime crypto loader accepts before touching the DB. +if ! node -e ' + const validKey = (value) => { + if (typeof value !== "string") return false; + if (/^[0-9a-fA-F]{64}$/.test(value)) return true; + if (!/^[A-Za-z0-9+/=]+$/.test(value)) return false; + try { return Buffer.from(value, "base64").length === 32; } + catch { return false; } + }; + const rawKeyring = (process.env.ENCRYPTION_KEYS || "").trim(); + if (rawKeyring) { + let keyring; + try { keyring = JSON.parse(rawKeyring); } + catch { throw new Error("ENCRYPTION_KEYS is invalid JSON"); } + if (!keyring || typeof keyring !== "object" || Array.isArray(keyring)) { + throw new Error("ENCRYPTION_KEYS is invalid: expected an object map"); + } + const entries = Object.entries(keyring); + if ( + entries.length === 0 || + entries.some(([id, value]) => + !/^[A-Za-z0-9_-]{1,32}$/.test(id) || !validKey(value) + ) + ) { + throw new Error("ENCRYPTION_KEYS is invalid: every entry must be a valid 32-byte key"); + } + const activeId = + (process.env.ENCRYPTION_ACTIVE_KEY_ID || "").trim() || + (entries.length === 1 ? entries[0][0] : ""); + if (!activeId || !Object.prototype.hasOwnProperty.call(keyring, activeId)) { + throw new Error("ENCRYPTION_ACTIVE_KEY_ID must name an entry in ENCRYPTION_KEYS"); + } + } else if (!validKey((process.env.ENCRYPTION_KEY || "").trim())) { + throw new Error("ENCRYPTION_KEY is missing or invalid"); + } +'; then + echo "HealthLog: ERROR — encryption configuration is invalid." >&2 + echo "HealthLog: set a valid ENCRYPTION_KEY or ENCRYPTION_KEYS with ENCRYPTION_ACTIVE_KEY_ID." >&2 + exit 1 +fi + # ── Wait for PostgreSQL ────────────────────────────────────── MAX_RETRIES=30 RETRY=0 diff --git a/src/__tests__/proxy-worker-health.test.ts b/src/__tests__/proxy-worker-health.test.ts new file mode 100644 index 000000000..de5880d45 --- /dev/null +++ b/src/__tests__/proxy-worker-health.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it, vi } from "vitest"; +import { NextRequest } from "next/server"; + +vi.mock("@/lib/process-type", () => ({ + shouldRunWeb: () => false, +})); + +import { proxy } from "../proxy"; + +function request(pathname: string): NextRequest { + return new NextRequest(`http://localhost${pathname}`); +} + +describe("worker-only process HTTP boundary", () => { + it("allows the process-aware health endpoint through", () => { + const response = proxy(request("/api/health")); + + expect(response.status).toBe(200); + expect(response.headers.get("X-HealthLog-Process-Type")).toBeNull(); + }); + + it("continues to refuse non-health HTTP traffic", () => { + const response = proxy(request("/api/version")); + + expect(response.status).toBe(503); + expect(response.headers.get("X-HealthLog-Process-Type")).toBe("worker"); + }); +}); diff --git a/src/app/api/health/__tests__/route.test.ts b/src/app/api/health/__tests__/route.test.ts new file mode 100644 index 000000000..60d87ca9a --- /dev/null +++ b/src/app/api/health/__tests__/route.test.ts @@ -0,0 +1,119 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +interface WorkerStatusFixture { + running: boolean; + startedAt: string | null; + lastHeartbeat: string | null; + lastReminderCheck: string | null; + lastWithingsSync: string | null; + lastInsightsRun: string | null; + jobsProcessed: number; + errors: number; +} + +const { queryRaw, getWorkerStatus, getGlobalBoss, getSession } = vi.hoisted( + () => ({ + queryRaw: vi.fn(async () => 1), + getWorkerStatus: vi.fn((): WorkerStatusFixture => ({ + running: false, + startedAt: null, + lastHeartbeat: null, + lastReminderCheck: null, + lastWithingsSync: null, + lastInsightsRun: null, + jobsProcessed: 0, + errors: 0, + })), + getGlobalBoss: vi.fn((): object | null => null), + getSession: vi.fn(async () => null), + }), +); + +vi.mock("@/lib/db", () => ({ + prisma: { $queryRaw: queryRaw }, +})); +vi.mock("@/lib/jobs/worker-status", () => ({ getWorkerStatus })); +vi.mock("@/lib/jobs/boss-instance", () => ({ getGlobalBoss })); +vi.mock("@/lib/auth/session", () => ({ getSession })); +vi.mock("@/lib/api-handler", () => ({ + apiHandler: unknown>(fn: T) => fn, +})); +vi.mock("@/lib/logging/context", () => ({ annotate: vi.fn() })); + +import { GET } from "../route"; + +async function health() { + const response = await (GET as unknown as () => Promise)(); + return { response, body: (await response.json()) as { status: string } }; +} + +beforeEach(() => { + vi.clearAllMocks(); + queryRaw.mockResolvedValue(1); + getWorkerStatus.mockReturnValue({ + running: false, + startedAt: null, + lastHeartbeat: null, + lastReminderCheck: null, + lastWithingsSync: null, + lastInsightsRun: null, + jobsProcessed: 0, + errors: 0, + }); + getGlobalBoss.mockReturnValue(null); + getSession.mockResolvedValue(null); +}); + +describe("GET /api/health process topology", () => { + it("web mode requires DB and producer readiness, not an in-process worker", async () => { + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "web"); + getGlobalBoss.mockReturnValue({}); + + const { response, body } = await health(); + + expect(response.status).toBe(200); + expect(body).toEqual({ status: "ok" }); + }); + + it("worker mode requires DB and worker readiness, not web producer readiness", async () => { + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "worker"); + getWorkerStatus.mockReturnValue({ + running: true, + startedAt: "2026-07-20T00:00:00.000Z", + lastHeartbeat: "2026-07-20T00:00:00.000Z", + lastReminderCheck: null, + lastWithingsSync: null, + lastInsightsRun: null, + jobsProcessed: 0, + errors: 0, + }); + + const { response, body } = await health(); + + expect(response.status).toBe(200); + expect(body).toEqual({ status: "ok" }); + }); + + it("all mode requires both producer and worker readiness", async () => { + vi.stubEnv("HEALTHLOG_PROCESS_TYPE", "all"); + getWorkerStatus.mockReturnValue({ + running: true, + startedAt: "2026-07-20T00:00:00.000Z", + lastHeartbeat: "2026-07-20T00:00:00.000Z", + lastReminderCheck: null, + lastWithingsSync: null, + lastInsightsRun: null, + jobsProcessed: 0, + errors: 0, + }); + + const missingProducer = await health(); + expect(missingProducer.response.status).toBe(503); + expect(missingProducer.body).toEqual({ status: "degraded" }); + + getGlobalBoss.mockReturnValue({}); + const ready = await health(); + expect(ready.response.status).toBe(200); + expect(ready.body).toEqual({ status: "ok" }); + }); +}); diff --git a/src/app/api/health/route.ts b/src/app/api/health/route.ts index e3490fb9f..3524d61c0 100644 --- a/src/app/api/health/route.ts +++ b/src/app/api/health/route.ts @@ -1,5 +1,7 @@ import { prisma } from "@/lib/db"; import { getWorkerStatus } from "@/lib/jobs/worker-status"; +import { getGlobalBoss } from "@/lib/jobs/boss-instance"; +import { shouldRunWeb, shouldRunWorker } from "@/lib/process-type"; import { getSession } from "@/lib/auth/session"; import { NextResponse } from "next/server"; import { apiHandler } from "@/lib/api-handler"; @@ -19,7 +21,13 @@ export const GET = apiHandler(async () => { } const worker = getWorkerStatus(); - const status = dbOk && worker.running ? "ok" : "degraded"; + const producerReady = getGlobalBoss() !== null; + const status = + dbOk && + (!shouldRunWeb() || producerReady) && + (!shouldRunWorker() || worker.running) + ? "ok" + : "degraded"; const statusCode = status === "ok" ? 200 : 503; const cacheHeaders = { diff --git a/src/lib/__tests__/db.test.ts b/src/lib/__tests__/db.test.ts index 73f395e39..95d2fac1e 100644 --- a/src/lib/__tests__/db.test.ts +++ b/src/lib/__tests__/db.test.ts @@ -1,65 +1,71 @@ /** - * v1.4.40 W-POOL — pool-config contract for `src/lib/db.ts`. - * - * The v1.4.39 empirical cold-mount trace - * (`.planning/round-v1439-empirical-trace.md` § B2) traced the - * Wave-B/C dashboard-query stall to the Prisma `pg.Pool` falling - * back to the library default ceiling of 10 connections. W-POOL - * pinned the ceiling to 20 via `getPoolMax()`; this test prevents a - * future refactor from silently dropping the override and re-introducing - * the saturation regression on a power-user cold mount. - * - * The test only exercises the `getPoolMax()` env-resolver. Constructing - * a real `PrismaClient` requires a live `DATABASE_URL`, which would - * make this an integration test instead of a unit test — the helper - * isolation is intentional. + * Pool configuration stays unit-tested here so connection ceilings and wait + * timeouts cannot silently fall back to separate library defaults. */ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { buildSessionOptions, getPoolMax, getStatementTimeoutMs } from "../db"; +import { + buildSessionOptions, + getConnectionBudget, + getPgBossPoolMax, + getPoolConnectionTimeoutMs, + getPrismaPoolMax, + getStatementTimeoutMs, +} from "../db"; -describe("getPoolMax", () => { - const originalEnv = process.env.DATABASE_POOL_MAX; +describe("per-process database connection budget", () => { + const originalDatabaseUrl = process.env.DATABASE_URL; + const originalConnectionLimit = process.env.DB_CONNECTION_LIMIT; + const originalPoolTimeout = process.env.DB_POOL_TIMEOUT; + const originalLegacyPoolMax = process.env.DATABASE_POOL_MAX; beforeEach(() => { + process.env.DATABASE_URL = + "postgresql://healthlog:test@db:5432/healthlog?connection_limit=20&pool_timeout=20"; + delete process.env.DB_CONNECTION_LIMIT; + delete process.env.DB_POOL_TIMEOUT; delete process.env.DATABASE_POOL_MAX; }); afterEach(() => { - if (originalEnv === undefined) { - delete process.env.DATABASE_POOL_MAX; - } else { - process.env.DATABASE_POOL_MAX = originalEnv; - } + const restore = (name: string, value: string | undefined) => { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + }; + restore("DATABASE_URL", originalDatabaseUrl); + restore("DB_CONNECTION_LIMIT", originalConnectionLimit); + restore("DB_POOL_TIMEOUT", originalPoolTimeout); + restore("DATABASE_POOL_MAX", originalLegacyPoolMax); }); - it("defaults to 20 when DATABASE_POOL_MAX is unset", () => { - expect(getPoolMax()).toBe(20); + it("splits the URL connection_limit across Prisma and pg-boss", () => { + expect(getConnectionBudget()).toBe(20); + expect(getPrismaPoolMax()).toBe(18); + expect(getPgBossPoolMax()).toBe(2); + expect(getPrismaPoolMax() + getPgBossPoolMax()).toBe( + getConnectionBudget(), + ); }); - it("honours DATABASE_POOL_MAX when it parses to a positive int", () => { + it("uses the existing pool_timeout seconds for both pool constructors", () => { + expect(getPoolConnectionTimeoutMs()).toBe(20_000); + }); + + it("prefers explicit DB_CONNECTION_LIMIT and DB_POOL_TIMEOUT values", () => { + process.env.DB_CONNECTION_LIMIT = "12"; + process.env.DB_POOL_TIMEOUT = "7"; + + expect(getConnectionBudget()).toBe(12); + expect(getPrismaPoolMax()).toBe(10); + expect(getPgBossPoolMax()).toBe(2); + expect(getPoolConnectionTimeoutMs()).toBe(7_000); + }); + + it("keeps DATABASE_POOL_MAX as a backward-compatible explicit override", () => { process.env.DATABASE_POOL_MAX = "30"; - expect(getPoolMax()).toBe(30); - }); - it("falls back to 20 on a malformed DATABASE_POOL_MAX", () => { - process.env.DATABASE_POOL_MAX = "not-a-number"; - expect(getPoolMax()).toBe(20); - }); - - it("falls back to 20 on a non-positive DATABASE_POOL_MAX", () => { - process.env.DATABASE_POOL_MAX = "0"; - expect(getPoolMax()).toBe(20); - process.env.DATABASE_POOL_MAX = "-5"; - expect(getPoolMax()).toBe(20); - }); - - it("never returns the library default of 10 on the happy path", () => { - // The whole point of the W-POOL change. If anybody removes the - // `max:` config from `createPrismaClient` and reverts to the - // library default, this assertion stops the regression before the - // next saturated cold mount. - expect(getPoolMax()).toBeGreaterThanOrEqual(20); + expect(getConnectionBudget()).toBe(30); + expect(getPrismaPoolMax()).toBe(28); }); }); diff --git a/src/lib/__tests__/docker-entrypoint.test.ts b/src/lib/__tests__/docker-entrypoint.test.ts new file mode 100644 index 000000000..25a3c9b29 --- /dev/null +++ b/src/lib/__tests__/docker-entrypoint.test.ts @@ -0,0 +1,99 @@ +import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { spawnSync } from "node:child_process"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; + +const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), "../../.."); +const entrypoint = join(repoRoot, "docker-entrypoint.sh"); +const key = "a".repeat(64); +let fakeBin: string; + +beforeAll(() => { + fakeBin = mkdtempSync(join(tmpdir(), "healthlog-entrypoint-")); + const fakeNode = join(fakeBin, "node"); + writeFileSync( + fakeNode, + `#!/bin/sh +case "$1" in + -e) + case "$2" in + *ENCRYPTION_KEYS*) exec "$REAL_NODE" "$@" ;; + *) exit 0 ;; + esac + ;; + *) exit 0 ;; +esac +`, + ); + chmodSync(fakeNode, 0o755); +}); + +afterAll(() => { + rmSync(fakeBin, { recursive: true, force: true }); +}); + +function runEntrypoint(overrides: Record) { + const env: NodeJS.ProcessEnv = { + PATH: `${fakeBin}:${process.env.PATH ?? ""}`, + REAL_NODE: process.execPath, + NODE_ENV: "test", + DATABASE_URL: "postgresql://healthlog:test@db:5432/healthlog", + API_TOKEN_HMAC_KEY: "test-hmac-key", + ...overrides, + }; + for (const [name, value] of Object.entries(env)) { + if (value === undefined) delete env[name]; + } + return spawnSync("sh", [entrypoint, "true"], { + cwd: repoRoot, + env, + encoding: "utf8", + }); +} + +describe("docker entrypoint crypto configuration", () => { + it("accepts the legacy ENCRYPTION_KEY configuration", () => { + const result = runEntrypoint({ + ENCRYPTION_KEY: key, + ENCRYPTION_KEYS: undefined, + ENCRYPTION_ACTIVE_KEY_ID: undefined, + }); + + expect(result.status, result.stderr).toBe(0); + }, 20_000); + + it("accepts a keyring-only configuration", () => { + const result = runEntrypoint({ + ENCRYPTION_KEY: undefined, + ENCRYPTION_KEYS: JSON.stringify({ v2: key }), + ENCRYPTION_ACTIVE_KEY_ID: undefined, + }); + + expect(result.status, result.stderr).toBe(0); + }, 20_000); + + it("fails closed when ENCRYPTION_KEYS is invalid even if a legacy key exists", () => { + const result = runEntrypoint({ + ENCRYPTION_KEY: key, + ENCRYPTION_KEYS: JSON.stringify({ v2: "not-a-32-byte-key" }), + ENCRYPTION_ACTIVE_KEY_ID: "v2", + }); + + expect(result.status).toBe(1); + expect(result.stderr).toContain("ENCRYPTION_KEYS"); + expect(result.stderr).toContain("invalid"); + }); + + it("fails closed when the active key is absent from the keyring", () => { + const result = runEntrypoint({ + ENCRYPTION_KEY: undefined, + ENCRYPTION_KEYS: JSON.stringify({ v2: key }), + ENCRYPTION_ACTIVE_KEY_ID: "v3", + }); + + expect(result.status).toBe(1); + expect(result.stderr).toContain("ENCRYPTION_ACTIVE_KEY_ID"); + }); +}); diff --git a/src/lib/db.ts b/src/lib/db.ts index 0c3b8472f..7b46e69ad 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -5,34 +5,54 @@ const globalForPrisma = globalThis as unknown as { prisma: PrismaClient | undefined; }; -/** - * v1.4.40 W-POOL — Prisma `pg.Pool` ceiling raised from the library - * default of 10 → 20. - * - * The v1.4.39 empirical cold-mount trace - * (`.planning/round-v1439-empirical-trace.md` § B2) showed thick - * `/api/analytics` holding ≥ 8 of the 10 default pool slots for - * 6.5 s on the maintainer's 347k-row tenant, starving every other Wave-B and - * Wave-C useQuery fan-out for the duration. The production Postgres runs a - * `max_connections` of 200 (raised from the stock 100), so a 20-slot - * Node-side pool sits well under the server's hard cap with plenty of - * headroom for concurrent power-users. Note the web + worker containers - * share this one Postgres, so the effective ceiling covers both pools. - * Complements the W-POOL `p-limit(4)` cap on the analytics fan-out itself — - * the bounded concurrency keeps any single analytics call to ≤ 4 slots, so - * the remaining 16 stay available for every other dashboard query. - * - * Env-overridable via `DATABASE_POOL_MAX` so an operator running - * with a smaller Postgres `max_connections` can dial down without a - * code change. - */ -export function getPoolMax(): number { - const raw = process.env.DATABASE_POOL_MAX; - if (raw) { - const parsed = Number.parseInt(raw, 10); - if (Number.isFinite(parsed) && parsed > 0) return parsed; +const DEFAULT_CONNECTION_BUDGET = 20; +const DEFAULT_POOL_TIMEOUT_SECONDS = 20; +const PG_BOSS_CONNECTIONS = 2; + +function positiveInteger(raw: string | undefined): number | null { + if (!raw) return null; + const parsed = Number.parseInt(raw, 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : null; +} + +function databaseUrlParameter(name: string): string | undefined { + const rawUrl = process.env.DATABASE_URL; + if (!rawUrl) return undefined; + try { + return new URL(rawUrl).searchParams.get(name) ?? undefined; + } catch { + return undefined; } - return 20; +} + +/** + * Total PostgreSQL connections this process may own. Compose already exposes + * this as DB_CONNECTION_LIMIT through DATABASE_URL's `connection_limit`. + * DATABASE_POOL_MAX remains a backward-compatible explicit override. + */ +export function getConnectionBudget(): number { + const configured = + positiveInteger(process.env.DB_CONNECTION_LIMIT) ?? + positiveInteger(process.env.DATABASE_POOL_MAX) ?? + positiveInteger(databaseUrlParameter("connection_limit")) ?? + DEFAULT_CONNECTION_BUDGET; + return Math.max(2, configured); +} + +export function getPgBossPoolMax(): number { + return Math.min(PG_BOSS_CONNECTIONS, getConnectionBudget() - 1); +} + +export function getPrismaPoolMax(): number { + return getConnectionBudget() - getPgBossPoolMax(); +} + +export function getPoolConnectionTimeoutMs(): number { + const seconds = + positiveInteger(process.env.DB_POOL_TIMEOUT) ?? + positiveInteger(databaseUrlParameter("pool_timeout")) ?? + DEFAULT_POOL_TIMEOUT_SECONDS; + return seconds * 1_000; } /** @@ -79,7 +99,8 @@ function createPrismaClient() { const sessionOptions = buildSessionOptions(); const adapter = new PrismaPg({ connectionString: process.env.DATABASE_URL!, - max: getPoolMax(), + max: getPrismaPoolMax(), + connectionTimeoutMillis: getPoolConnectionTimeoutMs(), ...(sessionOptions ? { options: sessionOptions } : {}), }); return new PrismaClient({ adapter }); diff --git a/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts index c6659c456..64367ac29 100644 --- a/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts +++ b/src/lib/jobs/__tests__/apple-health-import-send-policy.test.ts @@ -16,8 +16,6 @@ import { describe, expect, it, vi } from "vitest"; vi.mock("@/lib/db", () => ({ prisma: {}, - buildSessionOptions: () => null, - getPoolMax: () => 5, toJson: (value: unknown) => value, })); diff --git a/src/lib/jobs/__tests__/boss-instance.test.ts b/src/lib/jobs/__tests__/boss-instance.test.ts index 9de31b2ca..1afc12fad 100644 --- a/src/lib/jobs/__tests__/boss-instance.test.ts +++ b/src/lib/jobs/__tests__/boss-instance.test.ts @@ -1,44 +1,118 @@ -import { beforeEach, describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi, type Mock } from "vitest"; -const { start, PgBoss } = vi.hoisted(() => { - const start = vi.fn(async () => undefined); - const stop = vi.fn(async () => undefined); - const on = vi.fn(); - const PgBoss = vi.fn(function MockPgBoss(this: Record) { - this.start = start; - this.stop = stop; - this.on = on; +interface MockBoss { + start: Mock<() => Promise>; + stop: Mock<(options?: { graceful?: boolean }) => Promise>; + on: Mock< + (event: string, handler: (error: Error) => void) => MockBoss + >; + emitError(error: Error): void; +} + +const { PgBoss, instances, startOutcomes } = vi.hoisted(() => { + const instances: MockBoss[] = []; + const startOutcomes: Array<() => Promise> = []; + const PgBoss = vi.fn(function MockPgBoss(this: MockBoss) { + let errorHandler: ((error: Error) => void) | undefined; + this.start = vi.fn(() => + (startOutcomes.shift() ?? (() => Promise.resolve(undefined)))(), + ); + this.stop = vi.fn(async () => undefined); + this.on = vi.fn((event: string, handler: (error: Error) => void) => { + if (event === "error") errorHandler = handler; + return this; + }); + this.emitError = (error: Error) => errorHandler?.(error); + instances.push(this); }); - return { start, PgBoss }; + return { PgBoss, instances, startOutcomes }; }); vi.mock("pg-boss", () => ({ PgBoss })); -import { getGlobalBoss, startGlobalBossProducer } from "../boss-instance"; +import { + createWorkerBoss, + getGlobalBoss, + startGlobalBossProducer, +} from "../boss-instance"; + +const globalState = globalThis as Record; + +beforeEach(() => { + vi.useRealTimers(); + vi.clearAllMocks(); + instances.length = 0; + startOutcomes.length = 0; + delete globalState.__healthlog_pgboss__; + delete globalState.__healthlog_pgboss_start__; + vi.spyOn(console, "error").mockImplementation(() => undefined); +}); describe("web-process pg-boss producer", () => { - beforeEach(() => { - vi.clearAllMocks(); - }); + it("waits for delayed pg-boss schema readiness with capped exponential backoff", async () => { + vi.useFakeTimers(); + const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); + startOutcomes.push( + () => Promise.reject(new Error("schema not ready")), + () => Promise.reject(new Error("schema still not ready")), + () => Promise.reject(new Error("schema still not ready")), + () => Promise.resolve(undefined), + ); - it("retries startup and caps the send-only connection pool", async () => { - start - .mockRejectedValueOnce(new Error("schema not ready")) - .mockResolvedValueOnce(undefined); - - const boss = await startGlobalBossProducer("postgres://healthlog@test/db", { - maxAttempts: 2, - retryDelayMs: 0, + const starting = startGlobalBossProducer("postgres://healthlog@test/db", { + maxAttempts: 4, + retryBaseDelayMs: 10, + retryMaxDelayMs: 20, }); + await vi.runAllTimersAsync(); + const boss = await starting; - expect(PgBoss).toHaveBeenCalledTimes(2); - expect(PgBoss).toHaveBeenCalledWith({ + expect(PgBoss).toHaveBeenCalledTimes(4); + expect(timeoutSpy.mock.calls.map((call) => call[1])).toEqual([10, 20, 20]); + expect(PgBoss).toHaveBeenLastCalledWith({ connectionString: "postgres://healthlog@test/db", max: 2, + connectionTimeoutMillis: 20_000, migrate: false, schedule: false, supervise: false, }); - expect(start).toHaveBeenCalledTimes(2); expect(boss).toBe(getGlobalBoss()); }); + + it("keeps supervised reconnect active after a bounded failed batch", async () => { + vi.useFakeTimers(); + const timeoutSpy = vi.spyOn(globalThis, "setTimeout"); + startOutcomes.push(() => Promise.resolve(undefined)); + await startGlobalBossProducer("postgres://healthlog@test/db", { + maxAttempts: 2, + retryBaseDelayMs: 10, + retryMaxDelayMs: 20, + }); + + startOutcomes.push( + () => Promise.reject(new Error("db unavailable 1")), + () => Promise.reject(new Error("db unavailable 2")), + () => Promise.resolve(undefined), + ); + instances[0].emitError(new Error("connection lost")); + await vi.runAllTimersAsync(); + await Promise.resolve(); + + expect(PgBoss).toHaveBeenCalledTimes(4); + expect(timeoutSpy.mock.calls.map((call) => call[1])).toEqual([10, 20]); + expect(instances[0].stop).toHaveBeenCalledWith({ graceful: false }); + expect(getGlobalBoss()).toBe(instances[3]); + }); +}); + +describe("worker pg-boss pool", () => { + it("uses the pg-boss share of the process connection budget", () => { + createWorkerBoss("postgres://healthlog@test/db"); + + expect(PgBoss).toHaveBeenCalledWith({ + connectionString: "postgres://healthlog@test/db", + max: 2, + connectionTimeoutMillis: 20_000, + }); + }); }); diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index 888f80c29..e599c1e8a 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -25,9 +25,8 @@ * Locks per `.planning/research/v1434-r-1-xml-import.md` §5.1. */ import { unlinkSync } from "node:fs"; -import { PrismaClient } from "@/generated/prisma/client"; -import { buildSessionOptions, getPoolMax, toJson } from "@/lib/db"; -import { PrismaPg } from "@prisma/adapter-pg"; +import type { PrismaClient } from "@/generated/prisma/client"; +import { prisma, toJson } from "@/lib/db"; import type { Job } from "pg-boss"; import { extractExportXml } from "@/lib/import/unzip-export-xml"; @@ -94,20 +93,7 @@ export function _setWorkerPrismaForTests(client: PrismaClient | null): void { } function getWorkerPrisma(): PrismaClient { - if (!workerPrismaSingleton) { - // Inherit the web client's pool ceiling + per-session statement timeouts - // (`src/lib/db.ts`). The timeout is per-statement, not per-job, so the - // long-running bulk import stays safe while a single wedged UPSERT can no - // longer pin a worker connection indefinitely. - const sessionOptions = buildSessionOptions(); - const adapter = new PrismaPg({ - connectionString: process.env.DATABASE_URL!, - max: getPoolMax(), - ...(sessionOptions ? { options: sessionOptions } : {}), - }); - workerPrismaSingleton = new PrismaClient({ adapter }); - } - return workerPrismaSingleton; + return workerPrismaSingleton ?? prisma; } /** diff --git a/src/lib/jobs/boss-instance.ts b/src/lib/jobs/boss-instance.ts index 6925e3cad..7e29c2b95 100644 --- a/src/lib/jobs/boss-instance.ts +++ b/src/lib/jobs/boss-instance.ts @@ -7,15 +7,27 @@ */ import { PgBoss } from "pg-boss"; +import { + getPgBossPoolMax, + getPoolConnectionTimeoutMs, +} from "@/lib/db"; + const BOSS_KEY = "__healthlog_pgboss__" as const; const BOSS_START_KEY = "__healthlog_pgboss_start__" as const; -const PRODUCER_POOL_MAX = 2; const PRODUCER_START_MAX_ATTEMPTS = 5; -const PRODUCER_START_RETRY_DELAY_MS = 1_000; +const PRODUCER_RETRY_BASE_DELAY_MS = 250; +const PRODUCER_RETRY_MAX_DELAY_MS = 2_000; export interface ProducerStartOptions { maxAttempts?: number; - retryDelayMs?: number; + retryBaseDelayMs?: number; + retryMaxDelayMs?: number; +} + +interface ProducerRetryPolicy { + maxAttempts: number; + baseDelayMs: number; + maxDelayMs: number; } function wait(ms: number): Promise { @@ -24,20 +36,122 @@ function wait(ms: number): Promise { return promise; } -export function setGlobalBoss(boss: PgBoss) { - (globalThis as Record)[BOSS_KEY] = boss; +function resolveRetryPolicy(options: ProducerStartOptions): ProducerRetryPolicy { + const maxAttempts = Math.max( + 1, + Math.trunc(options.maxAttempts ?? PRODUCER_START_MAX_ATTEMPTS), + ); + const baseDelayMs = Math.max( + 0, + Math.trunc(options.retryBaseDelayMs ?? PRODUCER_RETRY_BASE_DELAY_MS), + ); + const maxDelayMs = Math.max( + baseDelayMs, + Math.trunc(options.retryMaxDelayMs ?? PRODUCER_RETRY_MAX_DELAY_MS), + ); + return { maxAttempts, baseDelayMs, maxDelayMs }; +} + +function retryDelayMs(failedAttempt: number, policy: ProducerRetryPolicy): number { + return Math.min( + policy.maxDelayMs, + policy.baseDelayMs * 2 ** Math.max(0, failedAttempt - 1), + ); +} + +export function setGlobalBoss(boss: PgBoss | null): void { + const globalState = globalThis as Record; + if (boss) globalState[BOSS_KEY] = boss; + else delete globalState[BOSS_KEY]; } export function getGlobalBoss(): PgBoss | null { return ((globalThis as Record)[BOSS_KEY] as PgBoss) ?? null; } +function databaseOptions(connectionString: string) { + return { + connectionString, + max: getPgBossPoolMax(), + connectionTimeoutMillis: getPoolConnectionTimeoutMs(), + } as const; +} + +export function createWorkerBoss(connectionString: string): PgBoss { + return new PgBoss(databaseOptions(connectionString)); +} + +function producerOptions(connectionString: string) { + return { + ...databaseOptions(connectionString), + migrate: false, + schedule: false, + supervise: false, + } as const; +} + +async function startProducerAttempts( + connectionString: string, + policy: ProducerRetryPolicy, +): Promise { + let lastError: unknown; + for (let attempt = 1; attempt <= policy.maxAttempts; attempt += 1) { + const boss = new PgBoss(producerOptions(connectionString)); + let ready = false; + boss.on("error", (error: unknown) => { + console.error("[pg-boss-producer] error", error); + if (ready) superviseProducerReconnect(boss, connectionString, policy); + }); + try { + await boss.start(); + ready = true; + setGlobalBoss(boss); + return boss; + } catch (error) { + lastError = error; + await boss.stop({ graceful: false }).catch(() => undefined); + if (attempt < policy.maxAttempts) { + await wait(retryDelayMs(attempt, policy)); + } + } + } + throw lastError; +} + +function superviseProducerReconnect( + failedBoss: PgBoss, + connectionString: string, + policy: ProducerRetryPolicy, +): void { + const globalState = globalThis as Record; + if (getGlobalBoss() !== failedBoss || globalState[BOSS_START_KEY]) return; + + setGlobalBoss(null); + const reconnecting = (async () => { + await failedBoss.stop({ graceful: false }).catch(() => undefined); + for (;;) { + try { + return await startProducerAttempts(connectionString, policy); + } catch (error) { + console.error("[pg-boss-producer] reconnect batch exhausted", error); + await wait(policy.maxDelayMs); + } + } + })(); + globalState[BOSS_START_KEY] = reconnecting; + void reconnecting.finally(() => { + if (globalState[BOSS_START_KEY] === reconnecting) { + delete globalState[BOSS_START_KEY]; + } + }); +} + /** * Start the web process's send-only pg-boss connection exactly once. * - * `PgBoss.start()` is still required to open/check the database connection. - * Disabling migration, schedules, and supervision keeps this process from - * performing worker duties; `send()` remains available through the manager. + * `PgBoss.start()` is required to confirm the worker-owned schema exists. + * Migration, schedules, and pg-boss maintenance stay disabled in web mode; + * this module only supervises the producer connection itself. */ export async function startGlobalBossProducer( connectionString: string | null | undefined, @@ -50,50 +164,20 @@ export async function startGlobalBossProducer( if (existing) return existing; const starting = globalState[BOSS_START_KEY] as - Promise | undefined; + | Promise + | undefined; if (starting) return starting; - const maxAttempts = Math.max( - 1, - Math.trunc(options.maxAttempts ?? PRODUCER_START_MAX_ATTEMPTS), + const startPromise = startProducerAttempts( + connectionString, + resolveRetryPolicy(options), ); - const retryDelayMs = Math.max( - 0, - options.retryDelayMs ?? PRODUCER_START_RETRY_DELAY_MS, - ); - const startPromise = (async () => { - let lastError: unknown; - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { - const boss = new PgBoss({ - connectionString, - max: PRODUCER_POOL_MAX, - migrate: false, - schedule: false, - supervise: false, - }); - boss.on("error", (error: unknown) => { - console.error("[pg-boss-producer] error", error); - }); - try { - await boss.start(); - setGlobalBoss(boss); - return boss; - } catch (error) { - lastError = error; - await boss.stop({ graceful: false }).catch(() => {}); - if (attempt < maxAttempts) { - await wait(retryDelayMs); - } - } - } - throw lastError; - })(); - globalState[BOSS_START_KEY] = startPromise; try { return await startPromise; - } catch (error) { - delete globalState[BOSS_START_KEY]; - throw error; + } finally { + if (globalState[BOSS_START_KEY] === startPromise) { + delete globalState[BOSS_START_KEY]; + } } } diff --git a/src/lib/jobs/reminder-worker.ts b/src/lib/jobs/reminder-worker.ts index d3ccefc3a..e61f62348 100644 --- a/src/lib/jobs/reminder-worker.ts +++ b/src/lib/jobs/reminder-worker.ts @@ -14,9 +14,8 @@ * Usage: Run as a standalone process or call startReminderWorker() from a * custom server setup. In dev, use: npx tsx src/lib/jobs/reminder-worker.ts */ -import { PgBoss } from "pg-boss"; import { markWorkerStarted, recordError } from "@/lib/jobs/worker-status"; -import { setGlobalBoss } from "@/lib/jobs/boss-instance"; +import { createWorkerBoss, setGlobalBoss } from "@/lib/jobs/boss-instance"; import { assertSubsystemEnabled } from "@/lib/process-type"; import { DATABASE_URL, workerLog } from "./reminder/shared"; import { @@ -50,7 +49,7 @@ export async function startReminderWorker() { return; } - const boss = new PgBoss(DATABASE_URL); + const boss = createWorkerBoss(DATABASE_URL); boss.on("error", (error: unknown) => { workerLog("error", "boss emitted error", error); diff --git a/src/lib/jobs/reminder/__tests__/shared.test.ts b/src/lib/jobs/reminder/__tests__/shared.test.ts index feecd51c3..bb6b11ed1 100644 --- a/src/lib/jobs/reminder/__tests__/shared.test.ts +++ b/src/lib/jobs/reminder/__tests__/shared.test.ts @@ -1,15 +1,4 @@ -/** - * F-DB-1 — the worker Prisma client must inherit the same pool ceiling + - * session timeouts as the web client (`src/lib/db.ts`). - * - * Before this fix the worker adapter was built with a bare - * `{ connectionString }`, so a pathological nightly drain on a heavy tenant - * could run unbounded, pin a connection indefinitely, and wedge the shared - * worker pool. This test pins the contract: `getWorkerPrisma()` constructs its - * `PrismaPg` adapter with `max` from `getPoolMax()` and the `options` startup - * string from `buildSessionOptions()`. - */ -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; const adapterCtor = vi.fn(); const clientCtor = vi.fn(); @@ -23,6 +12,7 @@ vi.mock("@prisma/adapter-pg", () => ({ })); vi.mock("@/generated/prisma/client", () => ({ + Prisma: {}, PrismaClient: class { constructor(...args: unknown[]) { clientCtor(...args); @@ -30,78 +20,24 @@ vi.mock("@/generated/prisma/client", () => ({ }, })); -describe("getWorkerPrisma", () => { +describe("worker Prisma connection ownership", () => { beforeEach(() => { vi.resetModules(); adapterCtor.mockClear(); clientCtor.mockClear(); - process.env.DATABASE_URL = "postgres://user:pass@localhost:5432/db"; - delete process.env.DATABASE_POOL_MAX; - delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; + process.env.DATABASE_URL = + "postgres://user:pass@localhost:5432/db?connection_limit=20&pool_timeout=20"; }); - afterEach(() => { - delete process.env.DATABASE_POOL_MAX; - delete process.env.DATABASE_STATEMENT_TIMEOUT_MS; - }); - - it("builds the adapter with a pool cap and session timeouts", async () => { - const { getWorkerPrisma } = await import("../shared"); - // Importing ../shared pulls in src/lib/db.ts, which constructs the web - // client's adapter at module load. Clear so we count only the worker's. - adapterCtor.mockClear(); - clientCtor.mockClear(); - getWorkerPrisma(); + it("reuses the process Prisma client instead of opening a second pool", async () => { + const [{ prisma }, { getWorkerPrisma }] = await Promise.all([ + import("@/lib/db"), + import("../shared"), + ]); + expect(getWorkerPrisma()).toBe(prisma); + expect(getWorkerPrisma()).toBe(prisma); expect(adapterCtor).toHaveBeenCalledTimes(1); - const config = adapterCtor.mock.calls[0][0] as { - connectionString: string; - max: number; - options?: string; - }; - expect(config.connectionString).toBe( - "postgres://user:pass@localhost:5432/db", - ); - // Default pool ceiling — must not fall back to the pg library default of 10. - expect(config.max).toBe(20); - // Both session timeouts applied at connection establishment. - expect(config.options).toContain("-c statement_timeout=60000"); - expect(config.options).toContain( - "-c idle_in_transaction_session_timeout=60000", - ); - }); - - it("honours the shared env knobs", async () => { - process.env.DATABASE_POOL_MAX = "12"; - process.env.DATABASE_STATEMENT_TIMEOUT_MS = "30000"; - const { getWorkerPrisma } = await import("../shared"); - adapterCtor.mockClear(); - getWorkerPrisma(); - - const config = adapterCtor.mock.calls[0][0] as { - max: number; - options?: string; - }; - expect(config.max).toBe(12); - expect(config.options).toContain("-c statement_timeout=30000"); - }); - - it("omits the options string when the timeout is disabled", async () => { - process.env.DATABASE_STATEMENT_TIMEOUT_MS = "0"; - const { getWorkerPrisma } = await import("../shared"); - adapterCtor.mockClear(); - getWorkerPrisma(); - - const config = adapterCtor.mock.calls[0][0] as { options?: string }; - expect(config.options).toBeUndefined(); - }); - - it("reuses a single client across calls", async () => { - const { getWorkerPrisma } = await import("../shared"); - clientCtor.mockClear(); - const a = getWorkerPrisma(); - const b = getWorkerPrisma(); - expect(a).toBe(b); expect(clientCtor).toHaveBeenCalledTimes(1); }); }); diff --git a/src/lib/jobs/reminder/shared.ts b/src/lib/jobs/reminder/shared.ts index 079ab63f7..5a14af029 100644 --- a/src/lib/jobs/reminder/shared.ts +++ b/src/lib/jobs/reminder/shared.ts @@ -4,9 +4,8 @@ * Extracted from reminder-worker.ts, which owns the queue names, cron * schedules, and boss.work registrations. */ -import { PrismaClient } from "@/generated/prisma/client"; -import { PrismaPg } from "@prisma/adapter-pg"; -import { buildSessionOptions, getPoolMax } from "@/lib/db"; +import type { PrismaClient } from "@/generated/prisma/client"; +import { prisma } from "@/lib/db"; export function parseTimeToMinutes(value: string): number { const [h, m] = value.split(":").map(Number); @@ -31,31 +30,10 @@ export const DATABASE_URL = process.env.DATABASE_URL!; // immediate semantics. export const BOOT_BACKFILL_STAGGER_SECONDS = 30; -// Reuse a single PrismaClient across all job handlers to avoid connection pool exhaustion - -let workerPrisma: PrismaClient | null = null; - +// Every queue handler shares the process-level client and its single bounded +// pg.Pool. Separate worker clients multiplied the configured connection limit. export function getWorkerPrisma(): PrismaClient { - if (!workerPrisma) { - // Apply the same pool ceiling + session timeouts the web client uses - // (`src/lib/db.ts`). Without them the worker client ran unbounded: a - // pathological nightly drain on a heavy tenant could pin a connection - // indefinitely and wedge the shared worker pool. `getPoolMax()` caps the - // slot count; `buildSessionOptions()` applies `statement_timeout` + - // `idle_in_transaction_session_timeout` at connection establishment so - // every worker session is timeout-bounded from its first query. Both share - // the same `DATABASE_POOL_MAX` / `DATABASE_STATEMENT_TIMEOUT_MS` env knobs - // as the web tier, and the web + worker containers share one Postgres so a - // single ceiling covers both pools. - const sessionOptions = buildSessionOptions(); - const adapter = new PrismaPg({ - connectionString: DATABASE_URL, - max: getPoolMax(), - ...(sessionOptions ? { options: sessionOptions } : {}), - }); - workerPrisma = new PrismaClient({ adapter }); - } - return workerPrisma; + return prisma; } /** diff --git a/src/proxy.ts b/src/proxy.ts index f8712927d..f42321058 100644 --- a/src/proxy.ts +++ b/src/proxy.ts @@ -194,7 +194,7 @@ export function proxy(request: NextRequest) { // Worker-only container: refuse HTTP traffic with a clear hint instead of // serving requests that would either crash (no DB pool ready for writes from // the worker) or duplicate work that the dedicated web container is doing. - if (!shouldRunWeb()) { + if (!shouldRunWeb() && pathname !== "/api/health") { return NextResponse.json( { data: null,