mirror of
https://github.com/MBombeck/HealthLog.git
synced 2026-11-09 21:10:59 +00:00
fix(jobs): apply pool cap + statement timeout to worker Prisma clients
The 60s statement_timeout + pool ceiling applied to the web client in db.ts were never applied to the worker clients. A pathological nightly drain on a heavy tenant could run unbounded, pin a connection, and wedge the shared worker pool. Route both worker clients (reminder handlers and the Apple Health import worker) through the same getPoolMax() + buildSessionOptions() helpers so every worker session is timeout-bounded from its first query.
This commit is contained in:
1 parent
6ad2dc858a
commit
77c73517c0
3 files changed
+132
-2
No files matched your search
@@ -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 });
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user