diff --git a/src/lib/jobs/apple-health-import-worker.ts b/src/lib/jobs/apple-health-import-worker.ts index becfca88d..f3d224894 100644 --- a/src/lib/jobs/apple-health-import-worker.ts +++ b/src/lib/jobs/apple-health-import-worker.ts @@ -26,7 +26,7 @@ */ import { unlinkSync } from "node:fs"; import { PrismaClient } from "@/generated/prisma/client"; -import { toJson } from "@/lib/db"; +import { buildSessionOptions, getPoolMax, toJson } from "@/lib/db"; import { PrismaPg } from "@prisma/adapter-pg"; import type { Job } from "pg-boss"; @@ -61,8 +61,15 @@ export interface AppleHealthImportPayload { let workerPrismaSingleton: PrismaClient | null = null; 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 }); } diff --git a/src/lib/jobs/reminder/__tests__/shared.test.ts b/src/lib/jobs/reminder/__tests__/shared.test.ts new file mode 100644 index 000000000..feecd51c3 --- /dev/null +++ b/src/lib/jobs/reminder/__tests__/shared.test.ts @@ -0,0 +1,107 @@ +/** + * 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"; + +const adapterCtor = vi.fn(); +const clientCtor = vi.fn(); + +vi.mock("@prisma/adapter-pg", () => ({ + PrismaPg: class { + constructor(...args: unknown[]) { + adapterCtor(...args); + } + }, +})); + +vi.mock("@/generated/prisma/client", () => ({ + PrismaClient: class { + constructor(...args: unknown[]) { + clientCtor(...args); + } + }, +})); + +describe("getWorkerPrisma", () => { + 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; + }); + + 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(); + + 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 05cd9ba58..079ab63f7 100644 --- a/src/lib/jobs/reminder/shared.ts +++ b/src/lib/jobs/reminder/shared.ts @@ -6,6 +6,7 @@ */ import { PrismaClient } from "@/generated/prisma/client"; import { PrismaPg } from "@prisma/adapter-pg"; +import { buildSessionOptions, getPoolMax } from "@/lib/db"; export function parseTimeToMinutes(value: string): number { const [h, m] = value.split(":").map(Number); @@ -36,7 +37,22 @@ let workerPrisma: PrismaClient | null = null; export function getWorkerPrisma(): PrismaClient { if (!workerPrisma) { - const adapter = new PrismaPg({ connectionString: DATABASE_URL }); + // 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;