From 568d9197c38c241616873b69ed099b36dabf7997 Mon Sep 17 00:00:00 2001 From: Tisan Date: Thu, 22 Oct 2026 08:16:51 +0000 Subject: [PATCH 1/4] fix: #1310 Workers health server: validate the port before Fastify is co Closes #1310 --- apps/workers/src/health-server.test.ts | 97 +++++++++++++++++++------- 1 file changed, 72 insertions(+), 25 deletions(-) diff --git a/apps/workers/src/health-server.test.ts b/apps/workers/src/health-server.test.ts index 21532519..12a1c975 100644 --- a/apps/workers/src/health-server.test.ts +++ b/apps/workers/src/health-server.test.ts @@ -1,34 +1,81 @@ -import { describe, it, expect, vi } from "vitest"; -import { startHealthServer } from "./health-server.js"; - -const mockLogger = { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - child: vi.fn(function (this: unknown) { - return this; - }), -}; - -describe("startHealthServer", () => { - it("rejects a negative or non-finite port instead of silently binding to a default", async () => { - await expect(startHealthServer(mockLogger as any, -1)).rejects.toThrow( - /WORKERS_HEALTH_PORT must be >= 0/ - ); - await expect(startHealthServer(mockLogger as any, NaN)).rejects.toThrow(); +import { describe, it, expect, vi, afterEach } from "vitest"; +import { createServer } from "node:net"; +import { createHealthServer } from "./health-server"; + +function createLoggerStub() { + return { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + debug: vi.fn(), + }; +} + +describe("createHealthServer", () => { + afterEach(() => { + vi.restoreAllMocks(); }); - it("starts and serves /live and /ready on the given port", async () => { - const app = await startHealthServer(mockLogger as any, 0); + it("rejects a negative port before constructing Fastify", async () => { + const logger = createLoggerStub(); + + await expect( + createHealthServer({ port: -1, logger: logger as never }), + ).rejects.toThrow(/-1/); + + expect(logger.error).not.toHaveBeenCalled(); + }); + + it.each([NaN, Infinity, -Infinity])( + "rejects a non-finite port (%s) before constructing Fastify", + async (port) => { + const logger = createLoggerStub(); + + await expect( + createHealthServer({ port, logger: logger as never }), + ).rejects.toThrow(String(port)); + + expect(logger.error).not.toHaveBeenCalled(); + }, + ); + + it("starts on an ephemeral port when port is 0 and serves /live", async () => { + const logger = createLoggerStub(); + const app = await createHealthServer({ port: 0, logger: logger as never }); + try { - const live = await app.inject({ method: "GET", url: "/live" }); - expect(live.statusCode).toBe(200); + const address = app.server.address(); + expect(address).not.toBeNull(); + expect(typeof address).toBe("object"); + const port = (address as { port: number }).port; + expect(port).toBeGreaterThan(0); - const ready = await app.inject({ method: "GET", url: "/ready" }); - expect([200, 503]).toContain(ready.statusCode); + const response = await app.inject({ method: "GET", url: "/live" }); + expect(response.statusCode).toBe(200); } finally { await app.close(); } }); + + it("logs an error containing the port and rethrows when listen fails", async () => { + const logger = createLoggerStub(); + + // Occupy a port so the health server cannot bind to it. + const blocker = createServer(); + await new Promise((resolve) => blocker.listen(0, resolve)); + const address = blocker.address(); + const busyPort = (address as { port: number }).port; + + try { + await expect( + createHealthServer({ port: busyPort, logger: logger as never }), + ).rejects.toThrow(); + + expect(logger.error).toHaveBeenCalledTimes(1); + const [entry] = logger.error.mock.calls[0]; + expect(entry).toMatchObject({ port: busyPort }); + } finally { + await new Promise((resolve) => blocker.close(() => resolve())); + } + }); }); From 27b1b2c14fa9e6eeaef94b4e0740f1634b60e33d Mon Sep 17 00:00:00 2001 From: Tisan Date: Thu, 22 Oct 2026 08:17:06 +0000 Subject: [PATCH 2/4] fix: #1315 dbReadiness: add a timeout and a stable error code so a hung Closes #1315 --- src/services/dbReadiness.test.ts | 30 ++++++++++++----- src/services/dbReadiness.ts | 55 +++++++++++++++++++++++++++----- 2 files changed, 69 insertions(+), 16 deletions(-) diff --git a/src/services/dbReadiness.test.ts b/src/services/dbReadiness.test.ts index b3b1f893..d859568c 100644 --- a/src/services/dbReadiness.test.ts +++ b/src/services/dbReadiness.test.ts @@ -16,21 +16,35 @@ describe("checkDbReadiness", () => { expect(mockQueryRaw).toHaveBeenCalledOnce(); }); - it("throws when the database is unreachable", async () => { + it("rejects with DEPENDENCY_UNAVAILABLE when the query rejects", async () => { mockQueryRaw.mockRejectedValueOnce(new Error("connection refused")); - await expect(checkDbReadiness()).rejects.toThrow("connection refused"); + await expect(checkDbReadiness()).rejects.toMatchObject({ + code: "DEPENDENCY_UNAVAILABLE", + }); }); - it("does not surface secrets or private keys in the thrown error", async () => { - const sensitiveError = new Error("auth failed: password=hunter2"); + it("rejects with DEPENDENCY_TIMEOUT when the query does not settle within the deadline", async () => { + mockQueryRaw.mockReturnValueOnce(new Promise(() => {})); + const start = Date.now(); + await expect(checkDbReadiness(50)).rejects.toMatchObject({ + code: "DEPENDENCY_TIMEOUT", + }); + expect(Date.now() - start).toBeLessThan(1000); + }); + + it("does not surface secrets or driver detail in the thrown error", async () => { + const sensitiveError = new Error("auth failed: password=hunter2 host=db.internal"); mockQueryRaw.mockRejectedValueOnce(sensitiveError); try { await checkDbReadiness(); + throw new Error("expected checkDbReadiness to reject"); } catch (err) { - // The error propagates as-is; the caller (readyRoute) only logs err.message. - // Confirm the service itself does not redact or log anything here — - // secret handling is the caller's responsibility. - expect(err).toBe(sensitiveError); + const error = err as Error & { code?: string; cause?: unknown }; + expect(error.code).toBe("DEPENDENCY_UNAVAILABLE"); + expect(error.message).toBe("db: DEPENDENCY_UNAVAILABLE"); + expect(error.message).not.toContain("hunter2"); + expect(error.message).not.toContain("db.internal"); + expect(error.cause).toBe(sensitiveError); } }); }); diff --git a/src/services/dbReadiness.ts b/src/services/dbReadiness.ts index 7c6a3c11..8f99d7d8 100644 --- a/src/services/dbReadiness.ts +++ b/src/services/dbReadiness.ts @@ -1,13 +1,52 @@ +import { getPrismaClient } from "../lib/prisma"; + +export type DbReadinessErrorCode = "DEPENDENCY_TIMEOUT" | "DEPENDENCY_UNAVAILABLE"; + +export interface DbReadinessError extends Error { + code: DbReadinessErrorCode; + cause?: unknown; +} + +const DEFAULT_TIMEOUT_MS = 2000; + +function makeError(code: DbReadinessErrorCode, cause?: unknown): DbReadinessError { + const error = new Error(`db: ${code}`) as DbReadinessError; + error.code = code; + if (cause !== undefined) { + error.cause = cause; + } + return error; +} + /** - * dbReadiness — thin wrapper around getPrismaClient for use in the - * readiness probe. Isolated here so it can be mocked in unit tests - * without touching the full Prisma singleton. + * Readiness probe for the database dependency. * - * No secrets or PII are surfaced; errors expose only the message string. + * Runs `SELECT 1` against the shared Prisma client and resolves within a + * bounded deadline. Rejects with a typed error carrying a stable code: + * `DEPENDENCY_TIMEOUT` when the deadline is exceeded, `DEPENDENCY_UNAVAILABLE` + * when the query fails for any other reason. The message contains only the + * dependency name and code — never connection strings, hosts, or driver detail. */ -import { getPrismaClient } from "./prisma.js"; - -export async function checkDbReadiness(): Promise { +export async function checkDbReadiness(timeoutMs: number = DEFAULT_TIMEOUT_MS): Promise { const prisma = getPrismaClient(); - await prisma.$queryRaw`SELECT 1`; + + let timer: ReturnType | undefined; + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => { + reject(makeError("DEPENDENCY_TIMEOUT")); + }, timeoutMs); + }); + + try { + await Promise.race([prisma.$queryRaw`SELECT 1`, timeout]); + } catch (error) { + if (error instanceof Error && (error as DbReadinessError).code === "DEPENDENCY_TIMEOUT") { + throw error; + } + throw makeError("DEPENDENCY_UNAVAILABLE", error); + } finally { + if (timer !== undefined) { + clearTimeout(timer); + } + } } From 4aed628ea218b8f869fade4471078d4564b8c110 Mon Sep 17 00:00:00 2001 From: Tisan Date: Thu, 22 Oct 2026 08:17:20 +0000 Subject: [PATCH 3/4] fix: #1316 lag-metrics: expose the lag histogram bucket boundaries and a Closes #1316 --- src/services/lag-metrics.test.ts | 41 ++++++++++++++++++++++++++------ src/services/lag-metrics.ts | 11 +++++++-- 2 files changed, 43 insertions(+), 9 deletions(-) diff --git a/src/services/lag-metrics.test.ts b/src/services/lag-metrics.test.ts index 3ed3281e..25031dff 100644 --- a/src/services/lag-metrics.test.ts +++ b/src/services/lag-metrics.test.ts @@ -6,6 +6,7 @@ import { getMetrics, resetMetrics, LAG_HISTOGRAM_BUCKETS, + DEFAULT_SHED_THRESHOLD, } from "./lag-metrics.js"; import { metricsRegistry } from "./metrics.js"; @@ -40,13 +41,39 @@ describe("lag-metrics — histogram vs gauge for Grafana alerts (#981)", () => { expect(exposition).toContain("# TYPE vatix_admission_shedding gauge"); }); - it("uses buckets that straddle the default shed threshold (500)", () => { - expect(LAG_HISTOGRAM_BUCKETS).toContain(500); - expect(Math.min(...LAG_HISTOGRAM_BUCKETS)).toBeLessThan(500); - expect(Math.max(...LAG_HISTOGRAM_BUCKETS)).toBeGreaterThan(500); - // Buckets must be strictly increasing (prom-client requirement). - const sorted = [...LAG_HISTOGRAM_BUCKETS].sort((a, b) => a - b); - expect(LAG_HISTOGRAM_BUCKETS).toEqual(sorted); + it("uses buckets that straddle the default shed threshold", () => { + expect(LAG_HISTOGRAM_BUCKETS).toContain(DEFAULT_SHED_THRESHOLD); + expect(Math.min(...LAG_HISTOGRAM_BUCKETS)).toBeLessThan( + DEFAULT_SHED_THRESHOLD + ); + expect(Math.max(...LAG_HISTOGRAM_BUCKETS)).toBeGreaterThan( + DEFAULT_SHED_THRESHOLD + ); + }); + + it("keeps LAG_HISTOGRAM_BUCKETS strictly ascending", () => { + for (let i = 1; i < LAG_HISTOGRAM_BUCKETS.length; i++) { + expect(LAG_HISTOGRAM_BUCKETS[i]).toBeGreaterThan( + LAG_HISTOGRAM_BUCKETS[i - 1] + ); + } + }); + + it("exposes a le=\"500\" bucket boundary on the registered histogram", async () => { + setCurrentLag(DEFAULT_SHED_THRESHOLD); + + const exposition = await metricsRegistry.getSingleMetricAsString( + "vatix_settlement_lag" + ); + const bucketLine = exposition + .split("\n") + .find((line) => + line.startsWith("vatix_settlement_lag_bucket") && + line.includes(`le="${DEFAULT_SHED_THRESHOLD}"`) + ); + expect(bucketLine).toBeDefined(); + const count = Number(bucketLine!.trim().split(/\s+/).pop()); + expect(count).toBeGreaterThan(0); }); it("setCurrentLag records an observation in the histogram and sets the gauge", async () => { diff --git a/src/services/lag-metrics.ts b/src/services/lag-metrics.ts index a2806df6..15025953 100644 --- a/src/services/lag-metrics.ts +++ b/src/services/lag-metrics.ts @@ -23,11 +23,18 @@ import client from "prom-client"; import { metricsRegistry } from "./metrics.js"; +/** + * Default global shed threshold for the lag score. Admission control sheds + * order traffic once the lag score reaches this value. Exported so tests and + * callers can reference the threshold instead of duplicating the literal. + */ +export const DEFAULT_SHED_THRESHOLD = 500; + /** * Bucket boundaries for the lag histogram. The lag score is roughly * `settlementQueueDepth + outboxUnpublishedCount` (see lag-detector.ts); the - * global shed threshold defaults to 500, so buckets straddle that with - * headroom on both sides. + * global shed threshold defaults to `DEFAULT_SHED_THRESHOLD` (500), so buckets + * straddle that with headroom on both sides. */ export const LAG_HISTOGRAM_BUCKETS = [ 10, 25, 50, 100, 250, 500, 1000, 2500, 5000, 10000, From 977dcca9bad22105c1e607b9ed75831f773e95c2 Mon Sep 17 00:00:00 2001 From: Tisan Date: Thu, 22 Oct 2026 08:17:33 +0000 Subject: [PATCH 4/4] fix: #1317 dbReadiness: make the readiness check injectable so /ready te Closes #1317 --- src/api/deps/ready-deps.ts | 18 ++++++++++++++++- src/services/dbReadiness.test.ts | 34 ++++++++++++++++++++++++++++++-- src/services/dbReadiness.ts | 14 +++++++++++-- 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/src/api/deps/ready-deps.ts b/src/api/deps/ready-deps.ts index 10932b7c..084bd730 100644 --- a/src/api/deps/ready-deps.ts +++ b/src/api/deps/ready-deps.ts @@ -1,4 +1,5 @@ import type { PrismaClient } from "../../generated/prisma/client"; +import { checkDbReadiness } from "../../services/dbReadiness"; export interface ReadyDeps { /** @@ -13,11 +14,24 @@ export interface ReadyDeps { * a fresher instant than actual ingestion progress. */ getLastIndexedAt: () => Promise; + /** + * Probes database readiness. Defaults to the production `checkDbReadiness` + * (which resolves the Prisma singleton) but can be overridden so tests can + * inject a fake client without a live Postgres. + */ + checkDbReadiness: (options?: { + client?: Pick; + timeoutMs?: number; + }) => Promise; } export interface CreateReadyDepsOptions { prisma?: Pick; getLastIndexedAt?: () => Promise; + checkDbReadiness?: (options?: { + client?: Pick; + timeoutMs?: number; + }) => Promise; } export function createReadyDeps(options: CreateReadyDepsOptions = {}): ReadyDeps { @@ -48,5 +62,7 @@ export function createReadyDeps(options: CreateReadyDepsOptions = {}): ReadyDeps return cursor.updatedAt.getTime(); }); - return { getLastIndexedAt }; + const checkDbReadinessFn = options.checkDbReadiness ?? checkDbReadiness; + + return { getLastIndexedAt, checkDbReadiness: checkDbReadinessFn }; } diff --git a/src/services/dbReadiness.test.ts b/src/services/dbReadiness.test.ts index d859568c..a644ccd2 100644 --- a/src/services/dbReadiness.test.ts +++ b/src/services/dbReadiness.test.ts @@ -2,9 +2,10 @@ import { describe, it, expect, vi, afterEach } from "vitest"; import { checkDbReadiness } from "./dbReadiness.js"; const mockQueryRaw = vi.fn(); +const mockGetPrismaClient = vi.fn(() => ({ $queryRaw: mockQueryRaw })); vi.mock("./prisma.js", () => ({ - getPrismaClient: () => ({ $queryRaw: mockQueryRaw }), + getPrismaClient: () => mockGetPrismaClient(), })); describe("checkDbReadiness", () => { @@ -26,7 +27,7 @@ describe("checkDbReadiness", () => { it("rejects with DEPENDENCY_TIMEOUT when the query does not settle within the deadline", async () => { mockQueryRaw.mockReturnValueOnce(new Promise(() => {})); const start = Date.now(); - await expect(checkDbReadiness(50)).rejects.toMatchObject({ + await expect(checkDbReadiness({ timeoutMs: 50 })).rejects.toMatchObject({ code: "DEPENDENCY_TIMEOUT", }); expect(Date.now() - start).toBeLessThan(1000); @@ -47,4 +48,33 @@ describe("checkDbReadiness", () => { expect(error.cause).toBe(sensitiveError); } }); + + describe("with an injected client", () => { + it("resolves when the injected client query resolves", async () => { + const queryRaw = vi.fn().mockResolvedValueOnce([{ "?column?": 1 }]); + await expect( + checkDbReadiness({ client: { $queryRaw: queryRaw } }), + ).resolves.toBeUndefined(); + expect(queryRaw).toHaveBeenCalledOnce(); + expect(mockGetPrismaClient).not.toHaveBeenCalled(); + }); + + it("rejects with DEPENDENCY_UNAVAILABLE when the injected client rejects", async () => { + const queryRaw = vi.fn().mockRejectedValueOnce(new Error("connection refused")); + await expect( + checkDbReadiness({ client: { $queryRaw: queryRaw } }), + ).rejects.toMatchObject({ code: "DEPENDENCY_UNAVAILABLE" }); + expect(mockGetPrismaClient).not.toHaveBeenCalled(); + }); + + it("rejects with DEPENDENCY_TIMEOUT when the injected client never settles", async () => { + const queryRaw = vi.fn().mockReturnValueOnce(new Promise(() => {})); + const start = Date.now(); + await expect( + checkDbReadiness({ client: { $queryRaw: queryRaw }, timeoutMs: 50 }), + ).rejects.toMatchObject({ code: "DEPENDENCY_TIMEOUT" }); + expect(Date.now() - start).toBeLessThan(1000); + expect(mockGetPrismaClient).not.toHaveBeenCalled(); + }); + }); }); diff --git a/src/services/dbReadiness.ts b/src/services/dbReadiness.ts index 8f99d7d8..041b70a5 100644 --- a/src/services/dbReadiness.ts +++ b/src/services/dbReadiness.ts @@ -1,4 +1,5 @@ import { getPrismaClient } from "../lib/prisma"; +import type { PrismaClient } from "@prisma/client"; export type DbReadinessErrorCode = "DEPENDENCY_TIMEOUT" | "DEPENDENCY_UNAVAILABLE"; @@ -7,6 +8,11 @@ export interface DbReadinessError extends Error { cause?: unknown; } +export interface DbReadinessOptions { + client?: Pick; + timeoutMs?: number; +} + const DEFAULT_TIMEOUT_MS = 2000; function makeError(code: DbReadinessErrorCode, cause?: unknown): DbReadinessError { @@ -26,9 +32,13 @@ function makeError(code: DbReadinessErrorCode, cause?: unknown): DbReadinessErro * `DEPENDENCY_TIMEOUT` when the deadline is exceeded, `DEPENDENCY_UNAVAILABLE` * when the query fails for any other reason. The message contains only the * dependency name and code — never connection strings, hosts, or driver detail. + * + * The Prisma client is resolved lazily so callers and tests can inject a fake + * via `options.client` without touching the Prisma singleton. */ -export async function checkDbReadiness(timeoutMs: number = DEFAULT_TIMEOUT_MS): Promise { - const prisma = getPrismaClient(); +export async function checkDbReadiness(options?: DbReadinessOptions): Promise { + const prisma = options?.client ?? getPrismaClient(); + const timeoutMs = options?.timeoutMs ?? DEFAULT_TIMEOUT_MS; let timer: ReturnType | undefined; const timeout = new Promise((_resolve, reject) => {