From 81487edbf50af01d250fc7e3c0dbcab3f4aa2ff5 Mon Sep 17 00:00:00 2001 From: madisonsc52-del Date: Mon, 20 Jul 2026 22:48:06 +0100 Subject: [PATCH] #254 Graceful Shutdown Logic Is Duplicated Across All Three Services FIXED --- services/api-gateway/src/index.ts | 27 +-- services/fx-engine/src/index.ts | 25 +-- services/indexer/src/index.ts | 20 +- services/settlement-engine/src/index.ts | 81 ++------ shared/validation/index.ts | 1 + shared/validation/package.json | 2 +- shared/validation/shutdown.test.ts | 190 +++++++++++++++++++ shared/validation/shutdown.ts | 239 ++++++++++++++++++++++++ 8 files changed, 467 insertions(+), 118 deletions(-) create mode 100644 shared/validation/shutdown.test.ts create mode 100644 shared/validation/shutdown.ts diff --git a/services/api-gateway/src/index.ts b/services/api-gateway/src/index.ts index 7ef68fe..346bbf1 100644 --- a/services/api-gateway/src/index.ts +++ b/services/api-gateway/src/index.ts @@ -28,7 +28,7 @@ import fastifyJwt from '@fastify/jwt'; import rateLimit from '@fastify/rate-limit'; import crypto from 'crypto'; import { z } from 'zod'; -import { validateEnv, getPrismaLogLevels, setupPrismaQueryLogging, buildPrismaConnectionUrl, connectWithRetry, registerRequestId, createLoggerOptions, registerTracing } from '@bettapay/validation'; +import { validateEnv, getPrismaLogLevels, setupPrismaQueryLogging, buildPrismaConnectionUrl, connectWithRetry, registerRequestId, createLoggerOptions, registerTracing, registerGracefulShutdown } from '@bettapay/validation'; import { createFxClient } from './clients/fx-client.js'; import { createIndexerClient } from './clients/indexer-client.js'; import { @@ -944,27 +944,10 @@ fastify.get('/api/quote', async (request, reply) => { return proxyFxUpstream(request, reply, path); }); -// Graceful shutdown -let shuttingDown = false; - -async function shutdown(signal: string) { - if (shuttingDown) return; - shuttingDown = true; - - fastify.log.info(`Received ${signal}, shutting down gracefully...`); - - try { - await fastify.close(); - await prisma.$disconnect(); - process.exit(0); - } catch (err) { - fastify.log.error(err, 'Error during shutdown'); - process.exit(1); - } -} - -process.on('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); +// Graceful shutdown — delegated to the shared helper in @bettapay/validation. +// It closes the HTTP server first, then disconnects Prisma, exiting 0 on +// success or 1 on failure. A 30s force-exit timeout guards against a hang. +registerGracefulShutdown({ fastify, prisma }); const start = async () => { try { diff --git a/services/fx-engine/src/index.ts b/services/fx-engine/src/index.ts index ea701fb..79e06bf 100644 --- a/services/fx-engine/src/index.ts +++ b/services/fx-engine/src/index.ts @@ -30,6 +30,7 @@ import { CurrencyCode, buildFxEngineHealthResponse, readServiceVersion, + registerGracefulShutdown, } from '@bettapay/validation'; const env = validateEnv(process.env); @@ -167,7 +168,6 @@ const fastify = Fastify({ registerRequestId(fastify); redis = new Redis(env.REDIS_URL, { enableOfflineQueue: false }); redis.on('error', (err) => fastify.log.warn({ err: err.message }, 'Redis error in fx-engine')); -fastify.addHook('onClose', async () => { await redis.quit().catch(() => {}); }); fastify.register(cors, { origin: env.ALLOWED_ORIGINS, @@ -472,25 +472,10 @@ fastify.post<{ Body: VerifyQuoteRouteBody }>( // ── Start ────────────────────────────────────────────────────────────────── -let shuttingDown = false; - -async function shutdown(signal: string) { - if (shuttingDown) return; - shuttingDown = true; - - fastify.log.info(`Received ${signal}, shutting down gracefully...`); - - try { - await fastify.close(); - process.exit(0); - } catch (err) { - fastify.log.error(err, 'Error during shutdown'); - process.exit(1); - } -} - -process.on('SIGTERM', () => shutdown('SIGTERM')); -process.on('SIGINT', () => shutdown('SIGINT')); +// Graceful shutdown — delegated to the shared helper in @bettapay/validation. +// The shared helper closes the HTTP server, then quits Redis, and now enforces +// a 30s force-exit timeout (previously the FX engine had none). +registerGracefulShutdown({ fastify, redis }); const start = async () => { try { diff --git a/services/indexer/src/index.ts b/services/indexer/src/index.ts index 5a6dacc..969ef80 100644 --- a/services/indexer/src/index.ts +++ b/services/indexer/src/index.ts @@ -42,6 +42,7 @@ import { buildIndexerHealthResponse, readServiceVersion, createAuditLogger, + registerGracefulShutdown, } from '@bettapay/validation'; import type { EventType } from '@bettapay/validation'; @@ -108,8 +109,6 @@ const webhookWorker = createWebhookWorker('indexer-webhooks', connectionParams, }, }); -}); - const redisHealth = new Redis(env.REDIS_URL, { enableOfflineQueue: false }); redisHealth.on('error', (err) => fastify.log.warn({ err: err.message }, '[Indexer] Redis health client error')); fastify.addHook('onClose', async () => { @@ -543,12 +542,17 @@ const start = async () => { } }; -process.on('SIGTERM', async () => { - await prisma.$disconnect(); - await webhookQueue.close(); - await webhookWorker.close(); - await fastify.close(); - process.exit(0); +// Graceful shutdown — delegated to the shared helper in @bettapay/validation. +// It closes resources in the canonical order (server → worker → queue → prisma) +// and now also wires up SIGINT, which the previous inline handler omitted. +// The Redis health client is released via the server's onClose hook. +registerGracefulShutdown({ + fastify, + prisma, + bullmq: { + worker: webhookWorker, + queues: [webhookQueue], + }, }); if (process.env.NODE_ENV !== 'test') { diff --git a/services/settlement-engine/src/index.ts b/services/settlement-engine/src/index.ts index 2392b96..ce8cbfc 100644 --- a/services/settlement-engine/src/index.ts +++ b/services/settlement-engine/src/index.ts @@ -49,6 +49,7 @@ import { registerTracing, buildSettlementEngineHealthResponse, readServiceVersion, + registerGracefulShutdown, } from "@bettapay/validation"; import type { PaginatedResponse, ApiResponse } from '@bettapay/shared-types'; @@ -600,73 +601,19 @@ fastify.post<{ Body: z.infer }>( // ============================================================================ // GRACEFUL SHUTDOWN // ============================================================================ - -let isShuttingDown = false; - -async function gracefulShutdown(signal: string): Promise { - // Prevent multiple shutdown attempts - if (isShuttingDown) { - fastify.log.warn({ signal }, 'Shutdown already in progress, ignoring duplicate signal'); - return; - } - - isShuttingDown = true; - fastify.log.info({ signal }, 'Received shutdown signal, starting graceful shutdown'); - - // Set a timeout to force exit if shutdown hangs - const forceExitTimeout = setTimeout(() => { - fastify.log.error('Graceful shutdown timed out after 30 seconds, forcing exit'); - process.exit(1); - }, 30000); - - try { - // 1. Close Fastify server (stops accepting new connections) - fastify.log.info('Closing Fastify server...'); - await fastify.close(); - fastify.log.info('Fastify server closed'); - - // 2. Close BullMQ worker (drain and close gracefully) - fastify.log.info('Closing BullMQ worker...'); - await worker.close(); - fastify.log.info('BullMQ worker closed'); - - // 3. Close BullMQ queues - fastify.log.info('Closing BullMQ queues...'); - await settlementQueue.close(); - await settlementDLQ.close(); - await webhookWorker.close(); - await webhookQueue.close(); - fastify.log.info('BullMQ queues closed'); - - // 4. Close Redis connection - fastify.log.info('Closing Redis connection...'); - await redis.quit(); - fastify.log.info('Redis connection closed'); - - // 5. Disconnect Prisma - fastify.log.info('Disconnecting Prisma...'); - await prisma.$disconnect(); - fastify.log.info('Prisma disconnected'); - - // Clear the force exit timeout - clearTimeout(forceExitTimeout); - - fastify.log.info({ signal }, 'Graceful shutdown completed successfully'); - process.exit(0); - } catch (error) { - fastify.log.error({ error, signal }, 'Error during graceful shutdown'); - clearTimeout(forceExitTimeout); - process.exit(1); - } -} - -// Register shutdown handlers for SIGTERM and SIGINT -process.on('SIGTERM', () => { - void gracefulShutdown('SIGTERM'); -}); - -process.on('SIGINT', () => { - void gracefulShutdown('SIGINT'); +// +// Delegated to the shared helper in @bettapay/validation. It enforces the +// canonical close order (server → workers → queues → redis → prisma) and a +// 30s force-exit timeout, replacing the previous hand-rolled implementation. + +registerGracefulShutdown({ + fastify, + prisma, + redis, + bullmq: { + worker: [worker, webhookWorker], + queues: [settlementQueue, settlementDLQ, webhookQueue], + }, }); // ============================================================================ diff --git a/shared/validation/index.ts b/shared/validation/index.ts index 962f5e6..7df944f 100644 --- a/shared/validation/index.ts +++ b/shared/validation/index.ts @@ -16,6 +16,7 @@ export * from './envAwareSchema.js'; export * from './webhookSchema.js'; export * from './health.js'; export * from './audit.js'; +export * from './shutdown.js'; import "dotenv/config"; export function genReqId(req: FastifyRequest | IncomingMessage): string { diff --git a/shared/validation/package.json b/shared/validation/package.json index faa2656..886b589 100644 --- a/shared/validation/package.json +++ b/shared/validation/package.json @@ -13,7 +13,7 @@ "build": "tsc", "type-check": "tsc --noEmit", "pretest": "pnpm -F @bettapay/stellar-utils build", - "test": "cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm health.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm cors.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm prisma.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm genReqId.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm plugins.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm tracing.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm logger.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm schemas.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm envAwareSchema.test.ts" + "test": "cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm health.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm cors.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm prisma.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm genReqId.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm plugins.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm tracing.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm logger.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm schemas.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm envAwareSchema.test.ts && cross-env TS_NODE_TRANSPILE_ONLY=true node --loader ts-node/esm shutdown.test.ts" }, "dependencies": { "@bettapay/stellar-utils": "workspace:^", diff --git a/shared/validation/shutdown.test.ts b/shared/validation/shutdown.test.ts new file mode 100644 index 0000000..7220f35 --- /dev/null +++ b/shared/validation/shutdown.test.ts @@ -0,0 +1,190 @@ +import test, { beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert'; +import { + gracefulShutdown, + registerGracefulShutdown, + DEFAULT_SHUTDOWN_TIMEOUT_MS, +} from './shutdown.js'; + +function silentLogger() { + return { info() {}, error() {}, warn() {} }; +} + +// `process.exit` is real during the test run, but we replace it with a spy so +// the shutdown helper can call it without terminating the test process. +let mockExit: { code: number | undefined; calls: number[] }; +let originalExit: typeof process.exit; + +beforeEach(() => { + mockExit = { code: undefined, calls: [] }; + originalExit = process.exit; + (process as unknown as { exit: (code?: number) => never }).exit = ((code?: number) => { + mockExit.code = code ?? 0; + mockExit.calls.push(code ?? 0); + return undefined as never; + }) as (code?: number) => never; +}); + +afterEach(() => { + (process as unknown as { exit: typeof originalExit }).exit = originalExit; +}); + +test('closes resources in the canonical order (server → workers → queues → redis → prisma)', async () => { + const order: string[] = []; + const fastify = { close: async () => { order.push('fastify'); } }; + const worker = { close: async () => { order.push('worker'); } }; + const queue = { close: async () => { order.push('queue'); } }; + const redis = { quit: async () => { order.push('redis'); } }; + const prisma = { $disconnect: async () => { order.push('prisma'); } }; + + await gracefulShutdown('SIGTERM', { + fastify, + prisma, + redis, + bullmq: { worker, queues: [queue] }, + logger: silentLogger(), + }); + + assert.deepStrictEqual(order, ['fastify', 'worker', 'queue', 'redis', 'prisma']); + assert.strictEqual(mockExit.code, 0, 'process.exit(0) should be called on success'); +}); + +test('closes every worker and queue when passed as arrays', async () => { + const closed: string[] = []; + const w1 = { close: async () => { closed.push('w1'); } }; + const w2 = { close: async () => { closed.push('w2'); } }; + const q1 = { close: async () => { closed.push('q1'); } }; + const q2 = { close: async () => { closed.push('q2'); } }; + const fastify = { close: async () => { closed.push('fastify'); } }; + + await gracefulShutdown('SIGINT', { + fastify, + bullmq: { worker: [w1, w2], queues: [q1, q2] }, + logger: silentLogger(), + }); + + assert.deepStrictEqual(closed, ['fastify', 'w1', 'w2', 'q1', 'q2']); + assert.strictEqual(mockExit.code, 0); +}); + +test('accepts a single worker / queue (non-array) value', async () => { + const closed: string[] = []; + const worker = { close: async () => { closed.push('worker'); } }; + const queue = { close: async () => { closed.push('queue'); } }; + const fastify = { close: async () => { closed.push('fastify'); } }; + + await gracefulShutdown('SIGTERM', { + fastify, + bullmq: { worker, queues: queue }, + logger: silentLogger(), + }); + + assert.deepStrictEqual(closed, ['fastify', 'worker', 'queue']); + assert.strictEqual(mockExit.code, 0); +}); + +test('attempts to close every resource even when one fails, then exits 1', async () => { + const order: string[] = []; + const fastify = { close: async () => { order.push('fastify'); } }; + const worker = { + close: async () => { + order.push('worker'); + throw new Error('worker boom'); + }, + }; + const queue = { close: async () => { order.push('queue'); } }; + const redis = { quit: async () => { order.push('redis'); } }; + const prisma = { $disconnect: async () => { order.push('prisma'); } }; + + await gracefulShutdown('SIGTERM', { + fastify, + prisma, + redis, + bullmq: { worker, queues: [queue] }, + logger: silentLogger(), + }); + + // every resource was still given a chance to close + assert.deepStrictEqual(order, ['fastify', 'worker', 'queue', 'redis', 'prisma']); + assert.strictEqual(mockExit.code, 1, 'process.exit(1) should be called when a resource fails'); +}); + +test('triggers process.exit(1) after the timeout when a close never settles', () => { + const originalSetTimeout = globalThis.setTimeout; + const originalClearTimeout = globalThis.clearTimeout; + const timeoutCallbacks: { cb: () => void; ms: number }[] = []; + (globalThis as unknown as { setTimeout: typeof setTimeout }).setTimeout = (( + cb: () => void, + ms: number, + ) => { + timeoutCallbacks.push({ cb, ms }); + return 0 as unknown as ReturnType; + }) as typeof setTimeout; + (globalThis as unknown as { clearTimeout: typeof clearTimeout }).clearTimeout = (() => {}) as typeof clearTimeout; + + const fastify = { close: () => new Promise(() => { /* never settles */ }) }; + const started = gracefulShutdown('SIGTERM', { fastify, logger: silentLogger() }); + + assert.strictEqual(timeoutCallbacks.length, 1, 'a force-exit timer should be registered'); + assert.strictEqual( + timeoutCallbacks[0].ms, + DEFAULT_SHUTDOWN_TIMEOUT_MS, + 'default timeout should be 30s', + ); + + // simulate the force-exit timer firing + timeoutCallbacks[0].cb(); + assert.strictEqual(mockExit.code, 1, 'forced exit should use code 1'); + + // restore timers and avoid an unhandled rejection from the orphaned promise + (globalThis as unknown as { setTimeout: typeof originalSetTimeout }).setTimeout = originalSetTimeout; + (globalThis as unknown as { clearTimeout: typeof originalClearTimeout }).clearTimeout = + originalClearTimeout; + started.catch(() => {}); +}); + +test('honours a custom timeoutMs for the force-exit timer', () => { + const originalSetTimeout = globalThis.setTimeout; + const originalClearTimeout = globalThis.clearTimeout; + const timeoutCallbacks: { cb: () => void; ms: number }[] = []; + (globalThis as unknown as { setTimeout: typeof setTimeout }).setTimeout = (( + cb: () => void, + ms: number, + ) => { + timeoutCallbacks.push({ cb, ms }); + return 0 as unknown as ReturnType; + }) as typeof setTimeout; + (globalThis as unknown as { clearTimeout: typeof clearTimeout }).clearTimeout = (() => {}) as typeof clearTimeout; + + const fastify = { close: () => new Promise(() => { /* never settles */ }) }; + const started = gracefulShutdown('SIGTERM', { + fastify, + timeoutMs: 5000, + logger: silentLogger(), + }); + + assert.strictEqual(timeoutCallbacks[0].ms, 5000); + timeoutCallbacks[0].cb(); + assert.strictEqual(mockExit.code, 1); + + (globalThis as unknown as { setTimeout: typeof originalSetTimeout }).setTimeout = originalSetTimeout; + (globalThis as unknown as { clearTimeout: typeof originalClearTimeout }).clearTimeout = + originalClearTimeout; + started.catch(() => {}); +}); + +test('registerGracefulShutdown registers SIGTERM/SIGINT and ignores every signal after the first', async () => { + let closeCalls = 0; + const fastify = { close: async () => { closeCalls += 1; } }; + const deregister = registerGracefulShutdown({ fastify, logger: silentLogger() }); + + // Fire several signals of differing types; only the first must trigger a shutdown. + process.emit('SIGTERM'); + process.emit('SIGINT'); + process.emit('SIGTERM'); + await new Promise((resolve) => setImmediate(resolve)); + + assert.strictEqual(closeCalls, 1, 'only the first signal should start a shutdown'); + + deregister(); +}); diff --git a/shared/validation/shutdown.ts b/shared/validation/shutdown.ts new file mode 100644 index 0000000..dff980b --- /dev/null +++ b/shared/validation/shutdown.ts @@ -0,0 +1,239 @@ +/** + * @bettapay/validation — shared graceful shutdown helper + * + * Every BettaPay service previously hand-rolled its own shutdown routine + * (API Gateway, FX Engine, Settlement Engine, Indexer) with subtle, + * inconsistent behaviour: only the settlement engine enforced a force-exit + * timeout, the indexer never closed Redis/Prisma in the right order, and the + * gateway never closed its BullMQ/Redis resources at all. + * + * This module centralises that logic. A single `gracefulShutdown(signal, + * options)` function closes every managed resource in a deterministic order + * and guarantees that a hung close cannot block process termination forever + * (a configurable force-exit timeout fires `process.exit(1)`). + * + * Resource close order (mirrors the project's operational requirements): + * server (fastify) → BullMQ workers → BullMQ queues → Redis → Prisma + * + * `Promise.allSettled` is used *within* each phase so that a failure to close + * one worker/queue still lets the others attempt to close, and so the overall + * shutdown is reported as failed only after every resource has been given a + * chance to release. + * + * The helper is intentionally dependency-free: it relies only on structural + * typing (an object exposing the expected `close` / `$disconnect` / `quit` + * methods) so it can be reused by services without forcing a hard dependency + * on fastify / bullmq / ioredis / @prisma/client inside `@bettapay/validation`. + */ + +/** A logger compatible with `fastify.log` (and the default console fallback). */ +export interface ShutdownLogger { + info: (...args: unknown[]) => void; + error: (...args: unknown[]) => void; + warn: (...args: unknown[]) => void; +} + +/** Structural view of a Fastify server. */ +export interface ShutdownFastify { + close: (options?: { timeout?: number }) => Promise; + log?: ShutdownLogger; +} + +/** Structural view of a Prisma client. */ +export interface ShutdownPrisma { + $disconnect: () => Promise; +} + +/** Structural view of an ioredis (or compatible) client. */ +export interface ShutdownRedis { + quit: () => Promise; +} + +/** Structural view of a BullMQ Worker. */ +export interface ShutdownWorker { + close: () => Promise; +} + +/** Structural view of a BullMQ Queue. */ +export interface ShutdownQueue { + close: () => Promise; +} + +export interface GracefulShutdownBullMq { + /** A single worker or an array of workers. All are closed in array order. */ + worker?: ShutdownWorker | ShutdownWorker[]; + /** A single queue or an array of queues. All are closed in array order. */ + queues?: ShutdownQueue | ShutdownQueue[]; +} + +export interface GracefulShutdownOptions { + /** The HTTP server. Closed first so no new connections are accepted. */ + fastify: ShutdownFastify; + /** Optional Prisma client, disconnected last. */ + prisma?: ShutdownPrisma; + /** Optional Redis client, quit after BullMQ resources. */ + redis?: ShutdownRedis; + /** Optional BullMQ workers/queues. */ + bullmq?: GracefulShutdownBullMq; + /** + * Force-exit timeout in milliseconds. If the shutdown sequence has not + * finished within this window a `process.exit(1)` is forced. + * Defaults to 30000 (30s), matching the settlement engine pattern. + */ + timeoutMs?: number; + /** + * Optional logger. When omitted a console-based logger tagged `[shutdown]` + * is used so the helper is safe to call in any context. + */ + logger?: ShutdownLogger; +} + +/** Default force-exit timeout — 30s, matching the settlement engine. */ +export const DEFAULT_SHUTDOWN_TIMEOUT_MS = 30_000; + +function toArray(value: T | T[] | undefined): T[] { + if (value === undefined) return []; + return Array.isArray(value) ? value : [value]; +} + +function formatMessage(msg: unknown): string { + if (typeof msg === 'string') return msg; + if (msg instanceof Error) return msg.message; + try { + return JSON.stringify(msg); + } catch { + return String(msg); + } +} + +function createDefaultLogger(): ShutdownLogger { + return { + info: (...args: unknown[]) => console.info('[shutdown]', ...args.map(formatMessage)), + error: (...args: unknown[]) => console.error('[shutdown]', ...args.map(formatMessage)), + warn: (...args: unknown[]) => console.warn('[shutdown]', ...args.map(formatMessage)), + }; +} + +/** + * Perform a graceful shutdown for the given signal. + * + * Resources are released in the canonical order (server → workers → queues → + * redis → prisma). Each phase uses `Promise.allSettled` so a single failing + * resource never prevents the others from being released. Once every managed + * resource has been given a chance to close the process exits: + * + * - `process.exit(0)` when every resource closed successfully, or + * - `process.exit(1)` when at least one resource failed to close, or when an + * unexpected error occurred. + * + * A force-exit timer (default 30s) guarantees the process cannot hang forever + * if a resource close never settles; it fires `process.exit(1)`. + */ +export async function gracefulShutdown( + signal: string, + options: GracefulShutdownOptions, +): Promise { + const logger = options.logger ?? createDefaultLogger(); + const timeoutMs = options.timeoutMs ?? DEFAULT_SHUTDOWN_TIMEOUT_MS; + + let settled = false; + + const forceExit = setTimeout(() => { + if (settled) return; + logger.error(`Graceful shutdown timed out after ${timeoutMs}ms, forcing exit`); + process.exit(1); + }, timeoutMs); + + try { + logger.info(`Received ${signal}, shutting down gracefully...`); + + // Phase 1 — server: stop accepting new connections. + const serverResults = await Promise.allSettled([options.fastify.close()]); + + // Phase 2 — BullMQ workers (close all, even if some fail). + const workers = toArray(options.bullmq?.worker); + const workerResults = await Promise.allSettled(workers.map((w) => w.close())); + + // Phase 3 — BullMQ queues. + const queues = toArray(options.bullmq?.queues); + const queueResults = await Promise.allSettled(queues.map((q) => q.close())); + + // Phase 4 — Redis. + const redisResults: PromiseSettledResult[] = options.redis + ? await Promise.allSettled([options.redis.quit()]) + : []; + + // Phase 5 — Prisma. + const prismaResults: PromiseSettledResult[] = options.prisma + ? await Promise.allSettled([options.prisma.$disconnect()]) + : []; + + const results = [ + ...serverResults, + ...workerResults, + ...queueResults, + ...redisResults, + ...prismaResults, + ]; + + const failures = results.filter( + (r): r is PromiseRejectedResult => r.status === 'rejected', + ); + + clearTimeout(forceExit); + + if (failures.length > 0) { + for (const failure of failures) { + logger.error(failure.reason, 'Resource failed to close during shutdown'); + } + logger.error({ signal }, 'Graceful shutdown completed with errors'); + settled = true; + process.exit(1); + return; + } + + logger.info({ signal }, 'Graceful shutdown completed successfully'); + settled = true; + process.exit(0); + } catch (error) { + clearTimeout(forceExit); + logger.error(error, `Error during graceful shutdown (${signal})`); + settled = true; + process.exit(1); + } +} + +/** + * Wire the shared graceful-shutdown routine into the current process. + * + * Registers `SIGTERM` and `SIGINT` handlers that call `gracefulShutdown` with + * the supplied options. A re-entrancy guard ensures a second signal (while a + * shutdown is already in flight) is ignored, mirroring the settlement + * engine's original behaviour. + * + * @returns a `deregister` function that removes the installed signal handlers. + */ +export function registerGracefulShutdown( + options: GracefulShutdownOptions, +): () => void { + let shuttingDown = false; + + const handler = (signal: string): void => { + if (shuttingDown) { + return; + } + shuttingDown = true; + void gracefulShutdown(signal, options); + }; + + const onSigterm = (): void => handler('SIGTERM'); + const onSigint = (): void => handler('SIGINT'); + + process.on('SIGTERM', onSigterm); + process.on('SIGINT', onSigint); + + return () => { + process.removeListener('SIGTERM', onSigterm); + process.removeListener('SIGINT', onSigint); + }; +}