diff --git a/backend/src/controllers/healthController.ts b/backend/src/controllers/healthController.ts index 2f11c9b8..709946ed 100644 --- a/backend/src/controllers/healthController.ts +++ b/backend/src/controllers/healthController.ts @@ -13,6 +13,22 @@ export const redis: Redis | null = config.REDIS_URL }) : null; +export async function closeHealthDependencies(): Promise { + const closers: Promise[] = [pool.end()]; + if (redis) { + closers.push(redis.quit()); + } + + const results = await Promise.allSettled(closers); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason); + + if (failures.length > 0) { + throw new AggregateError(failures, 'Failed to close health-check dependencies'); + } +} + export interface DependencyStatus { status: 'connected' | 'disconnected' | 'not_configured' | 'unknown'; error?: string; diff --git a/backend/src/index.ts b/backend/src/index.ts index c883ae89..72f8d506 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -3,6 +3,10 @@ import { createServer } from 'node:http'; import app from './app.js'; import logger from './utils/logger.js'; import config from './config/index.js'; +import pool from './config/database.js'; +import { closeHealthDependencies } from './controllers/healthController.js'; +import { closeRateLimitRedis } from './services/rateLimitService.js'; +import { createGracefulShutdown } from './services/gracefulShutdown.js'; import { initializeSocket } from './services/socketService.js'; import { scheduleExecutor } from './services/scheduleExecutor.js'; import { contractEventIndexer } from './services/contractEventIndexer.js'; @@ -16,15 +20,36 @@ dotenv.config(); const server = createServer(app); // Part-49 job handles — assigned on server start, cleaned up on shutdown -let usageSnapshotJob: { stop(): void }; -let integrityCheckJob: { stop(): void }; +let usageSnapshotJob: { stop(): Promise } | undefined; +let integrityCheckJob: { stop(): Promise } | undefined; +let auditCacheCleanup: ReturnType | undefined; +let idempotencyCleanup: ReturnType | undefined; +const activeCleanupRuns = new Set>(); +let isShuttingDown = false; + +function trackCleanupRun(run: Promise): void { + activeCleanupRuns.add(run); + void run.then( + () => { + activeCleanupRuns.delete(run); + }, + () => { + activeCleanupRuns.delete(run); + } + ); +} // Initialize Socket.IO -initializeSocket(server); +const socketServer = initializeSocket(server); const PORT = config.port || process.env.PORT || 4000; server.listen(PORT, () => { + if (isShuttingDown) { + server.close(); + return; + } + logger.info(`Server running on port ${PORT}`); logger.info(`Environment: ${config.nodeEnv}`); logger.info(`Health check: http://localhost:${PORT}/health`); @@ -48,64 +73,93 @@ server.listen(PORT, () => { logger.info('Part-49 jobs scheduled (usage snapshots + audit integrity)'); // Part 45 — cleanup expired audit cache every hour - setInterval( - async () => { - try { - const deleted = await auditAnalyticsService.cleanupExpiredCache(); - if (deleted > 0) { - logger.info(`Cleaned up ${deleted} expired audit cache entries`); + auditCacheCleanup = setInterval( + () => { + const run = (async () => { + try { + const deleted = await auditAnalyticsService.cleanupExpiredCache(); + if (deleted > 0) { + logger.info(`Cleaned up ${deleted} expired audit cache entries`); + } + } catch (error) { + logger.error('Failed to cleanup audit cache', { error }); } - } catch (error) { - logger.error('Failed to cleanup audit cache', { error }); - } + })(); + trackCleanupRun(run); }, 60 * 60 * 1000 ); // Every hour logger.info('Part-45 audit cache cleanup scheduled'); // Idempotency key cleanup — every hour, remove expired keys - setInterval( - async () => { - try { - await cleanupExpiredIdempotencyKeys(); - } catch (error) { - logger.error('Failed to cleanup expired idempotency keys', { error }); - } + idempotencyCleanup = setInterval( + () => { + const run = (async () => { + try { + await cleanupExpiredIdempotencyKeys(); + } catch (error) { + logger.error('Failed to cleanup expired idempotency keys', { error }); + } + })(); + trackCleanupRun(run); }, 60 * 60 * 1000 ); logger.info('Idempotency key cleanup scheduled'); }); -// Graceful shutdown handling -const shutdown = () => { - logger.info('Shutting down gracefully...'); - - // Stop the schedule executor - scheduleExecutor.stop(); - - liquidityAlertChecker.stop(); - - // Stop Part-49 cron jobs - usageSnapshotJob?.stop(); - integrityCheckJob?.stop(); - - // Stop the contract event indexer - contractEventIndexer.stop(); - - // Close the server - server.close(() => { - logger.info('Server closed'); - process.exit(0); - }); +// Stop accepting HTTP before stopping future background work. The shared +// shutdown service drains HTTP for up to 30 seconds before closing dependencies. +const gracefulShutdown = createGracefulShutdown({ + server, + logger, + stopBackgroundWork: async () => { + if (auditCacheCleanup) clearInterval(auditCacheCleanup); + if (idempotencyCleanup) clearInterval(idempotencyCleanup); + + const stops = [ + // HTTP server.close() does not close upgraded WebSocket connections. + // Close this engine's transports so clients can reconnect to other pods. + () => { socketServer.engine.close(); }, + () => scheduleExecutor.stop(), + () => liquidityAlertChecker.stop(), + () => usageSnapshotJob?.stop(), + () => integrityCheckJob?.stop(), + () => contractEventIndexer.stop(), + ]; + const results = await Promise.allSettled([ + ...stops.map((stop) => Promise.resolve().then(stop)), + ...activeCleanupRuns, + ]); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason); + if (failures.length > 0) { + throw new AggregateError(failures, 'Failed to stop one or more background jobs'); + } + }, + closeDependencies: async () => { + const results = await Promise.allSettled([ + Promise.resolve().then(() => pool.end()), + Promise.resolve().then(closeHealthDependencies), + Promise.resolve().then(closeRateLimitRedis), + ]); + const failures = results + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map((result) => result.reason); + if (failures.length > 0) { + throw new AggregateError(failures, 'Failed to close one or more backend dependencies'); + } + }, +}); - // Force shutdown after 10 seconds - setTimeout(() => { - logger.error('Forced shutdown after timeout'); +const shutdown = (signal: NodeJS.Signals): void => { + isShuttingDown = true; + void gracefulShutdown(signal).catch((error) => { + logger.error('Graceful shutdown failed', { error }); process.exit(1); - }, 10000); + }); }; -// Listen for termination signals -process.on('SIGTERM', shutdown); -process.on('SIGINT', shutdown); +process.on('SIGTERM', () => shutdown('SIGTERM')); +process.on('SIGINT', () => shutdown('SIGINT')); diff --git a/backend/src/jobs/part49Jobs.ts b/backend/src/jobs/part49Jobs.ts index 648920d3..746d94f4 100644 --- a/backend/src/jobs/part49Jobs.ts +++ b/backend/src/jobs/part49Jobs.ts @@ -26,40 +26,80 @@ const ADVISORY_LOCK_INTEGRITY_CHECK = 84_901_002; */ export interface ScheduledJob { - stop(): void; + stop(): Promise; +} + +function trackRun(activeRuns: Set>, run: Promise): void { + activeRuns.add(run); + void run.then( + () => { + activeRuns.delete(run); + }, + () => { + activeRuns.delete(run); + } + ); +} + +async function stopAndDrain( + task: ScheduledTask, + activeRuns: Set>, + label: string +): Promise { + task.stop(); + const runs = [...activeRuns]; + if (runs.length > 0) { + await Promise.allSettled(runs); + } + logger.info(`${label} cron stopped and active runs drained`); } export function scheduleDailyUsageSnapshots(): ScheduledJob { + const activeRuns = new Set>(); + // Run once immediately on startup (non-leader-elected — safe idempotent catch-up) - void runDailyUsageSnapshots(); + trackRun(activeRuns, runDailyUsageSnapshots()); const task: ScheduledTask = cron.schedule('0 0 * * *', () => { - void runWithAdvisoryLock(ADVISORY_LOCK_USAGE_SNAPSHOT, 'daily-usage-snapshot', runDailyUsageSnapshots); + trackRun( + activeRuns, + runWithAdvisoryLock( + ADVISORY_LOCK_USAGE_SNAPSHOT, + 'daily-usage-snapshot', + runDailyUsageSnapshots + ) + ); }, { timezone: 'UTC' }); logger.info('Daily usage snapshot cron scheduled (midnight UTC, leader-elected)'); return { - stop() { - task.stop(); - logger.info('Daily usage snapshot cron stopped'); + async stop() { + await stopAndDrain(task, activeRuns, 'Daily usage snapshot'); }, }; } export function scheduleNightlyIntegrityCheck(): ScheduledJob { - void runNightlyIntegrityCheck(); + const activeRuns = new Set>(); + trackRun(activeRuns, runNightlyIntegrityCheck()); const task: ScheduledTask = cron.schedule('0 0 * * *', () => { - void runWithAdvisoryLock(ADVISORY_LOCK_INTEGRITY_CHECK, 'nightly-integrity-check', runNightlyIntegrityCheck); + trackRun( + activeRuns, + runWithAdvisoryLock( + ADVISORY_LOCK_INTEGRITY_CHECK, + 'nightly-integrity-check', + runNightlyIntegrityCheck + ) + ); }, { timezone: 'UTC' }); logger.info('Nightly integrity check cron scheduled (midnight UTC, leader-elected)'); return { - stop() { - task.stop(); - logger.info('Nightly integrity check cron stopped'); + async stop() { + await stopAndDrain(task, activeRuns, 'Nightly integrity check'); }, }; } diff --git a/backend/src/services/__tests__/gracefulShutdown.test.ts b/backend/src/services/__tests__/gracefulShutdown.test.ts new file mode 100644 index 00000000..f3539b39 --- /dev/null +++ b/backend/src/services/__tests__/gracefulShutdown.test.ts @@ -0,0 +1,160 @@ +import type { Server } from 'node:http'; +import { createGracefulShutdown } from '../gracefulShutdown'; + +describe('createGracefulShutdown', () => { + it('drains HTTP before dependency cleanup and exits cleanly', async () => { + let onClose: (() => void) | undefined; + const server = { + close: jest.fn((callback: () => void) => { + onClose = callback; + return server; + }), + closeAllConnections: jest.fn(), + } as unknown as Server; + const stopBackgroundWork = jest.fn(); + const closeDependencies = jest.fn().mockResolvedValue(undefined); + const exit = jest.fn(); + const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn() }; + + const shutdown = createGracefulShutdown({ + server, + logger, + stopBackgroundWork, + closeDependencies, + exit, + }); + + const pending = shutdown('SIGTERM'); + expect(server.close).toHaveBeenCalledTimes(1); + expect(stopBackgroundWork).toHaveBeenCalledTimes(1); + expect(closeDependencies).not.toHaveBeenCalled(); + + onClose?.(); + await pending; + + expect(closeDependencies).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(0); + }); + + it('uses the 30 second drain bound before cleanup', async () => { + jest.useFakeTimers(); + const server = { + close: jest.fn(() => server), + closeAllConnections: jest.fn(), + } as unknown as Server; + const closeDependencies = jest.fn().mockResolvedValue(undefined); + const exit = jest.fn(); + const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn() }; + + const pending = createGracefulShutdown({ + server, + logger, + stopBackgroundWork: jest.fn(), + closeDependencies, + exit, + })('SIGINT'); + + await Promise.resolve(); + jest.advanceTimersByTime(30_000); + await pending; + + expect(server.closeAllConnections).toHaveBeenCalledTimes(1); + expect(closeDependencies).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(0); + jest.useRealTimers(); + }); + + it('does not close dependencies under failed background work and still exits zero', async () => { + let onClose: (() => void) | undefined; + const server = { + close: jest.fn((callback: () => void) => { + onClose = callback; + return server; + }), + closeAllConnections: jest.fn(), + } as unknown as Server; + const stopBackgroundWork = jest.fn().mockRejectedValue(new Error('stop failed')); + const closeDependencies = jest.fn().mockResolvedValue(undefined); + const exit = jest.fn(); + const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn() }; + + const pending = createGracefulShutdown({ + server, + logger, + stopBackgroundWork, + closeDependencies, + exit, + })('SIGTERM'); + + onClose?.(); + await pending; + + expect(logger.error).toHaveBeenCalledWith( + 'Background work shutdown failed', + expect.objectContaining({ error: expect.any(Error) }) + ); + expect(logger.error).toHaveBeenCalledWith( + 'Runtime dependency cleanup skipped because background work did not stop cleanly' + ); + expect(logger.error).toHaveBeenCalledWith( + 'Graceful shutdown completed with cleanup failures' + ); + expect(closeDependencies).not.toHaveBeenCalled(); + expect(exit).toHaveBeenCalledWith(0); + }); + + it('closes dependencies and finishes handled SIGTERM even when server.close throws', async () => { + const server = { + close: jest.fn(() => { throw new Error('not listening'); }), + closeAllConnections: jest.fn(), + } as unknown as Server; + const closeDependencies = jest.fn().mockResolvedValue(undefined); + const stopBackgroundWork = jest.fn(); + const exit = jest.fn(); + const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn() }; + + await createGracefulShutdown({ + server, logger, stopBackgroundWork, closeDependencies, exit, + })('SIGTERM'); + + expect(server.closeAllConnections).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + 'HTTP server close threw synchronously', + expect.objectContaining({ error: expect.any(Error) }), + ); + expect(stopBackgroundWork).toHaveBeenCalledTimes(1); + expect(closeDependencies).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(0); + }); + + it('keeps dependency cleanup reachable after timed-out force-close throws', async () => { + jest.useFakeTimers(); + try { + const server = { + close: jest.fn(() => server), + closeAllConnections: jest.fn(() => { + throw new Error('transport already disconnected'); + }), + } as unknown as Server; + const closeDependencies = jest.fn().mockResolvedValue(undefined); + const exit = jest.fn(); + const logger = { info: jest.fn(), warn: jest.fn(), error: jest.fn() }; + const pending = createGracefulShutdown({ + server, logger, stopBackgroundWork: jest.fn(), + closeDependencies, exit, timeoutMs: 30_000, + })('SIGINT'); + await Promise.resolve(); + jest.advanceTimersByTime(30_000); + await pending; + expect(logger.error).toHaveBeenCalledWith( + 'Unable to force-close timed-out HTTP connections', + expect.objectContaining({ error: expect.any(Error) }), + ); + expect(closeDependencies).toHaveBeenCalledTimes(1); + expect(exit).toHaveBeenCalledWith(0); + } finally { + jest.useRealTimers(); + } + }); + +}); diff --git a/backend/src/services/contractEventIndexer.ts b/backend/src/services/contractEventIndexer.ts index 7e66c746..765b4860 100644 --- a/backend/src/services/contractEventIndexer.ts +++ b/backend/src/services/contractEventIndexer.ts @@ -6,6 +6,7 @@ import * as StellarSdk from '@stellar/stellar-sdk'; export class ContractEventIndexer { private isRunning = false; private intervalId: NodeJS.Timeout | null = null; + private readonly activeRuns = new Set>(); private readonly POLL_INTERVAL_MS = 10000; // Poll every 10 seconds private readonly BATCH_SIZE = 100; private readonly CONTRACTS_TO_INDEX = [ @@ -16,6 +17,18 @@ export class ContractEventIndexer { private readonly RPC_URL = process.env.SOROBAN_RPC_URL || 'https://soroban-testnet.stellar.org'; + private trackRun(run: Promise): void { + this.activeRuns.add(run); + void run.then( + () => { + this.activeRuns.delete(run); + }, + () => { + this.activeRuns.delete(run); + } + ); + } + /** * Initialize the indexer and start polling */ @@ -35,12 +48,18 @@ export class ContractEventIndexer { this.isRunning = true; - // Run immediately on startup - await this.pollAndIndexEvents(); + // Run immediately on startup and retain the promise so stop() can drain + // a shutdown that arrives while the initial poll is still using the DB. + const initialRun = this.pollAndIndexEvents(); + this.trackRun(initialRun); + await initialRun; + + // Shutdown may have stopped the indexer while the initial poll was pending. + if (!this.isRunning) return; // Then poll at regular intervals - this.intervalId = setInterval(async () => { - await this.pollAndIndexEvents(); + this.intervalId = setInterval(() => { + this.trackRun(this.pollAndIndexEvents()); }, this.POLL_INTERVAL_MS); console.log(`[ContractEventIndexer] Started polling every ${this.POLL_INTERVAL_MS}ms`); @@ -49,13 +68,19 @@ export class ContractEventIndexer { /** * Stop the indexer gracefully */ - stop(): void { + async stop(): Promise { if (this.intervalId) { clearInterval(this.intervalId); this.intervalId = null; } this.isRunning = false; - console.log('[ContractEventIndexer] Stopped'); + + const activeRuns = [...this.activeRuns]; + if (activeRuns.length > 0) { + await Promise.allSettled(activeRuns); + } + + console.log('[ContractEventIndexer] Stopped and active polls drained'); } /** diff --git a/backend/src/services/forecasting/liquidityAlertChecker.ts b/backend/src/services/forecasting/liquidityAlertChecker.ts index 600a84ef..b29fc0e2 100644 --- a/backend/src/services/forecasting/liquidityAlertChecker.ts +++ b/backend/src/services/forecasting/liquidityAlertChecker.ts @@ -18,24 +18,46 @@ interface LiquiditySettings { export class LiquidityAlertChecker { private cronJob: ScheduledTask | null = null; + private readonly activeRuns = new Set>(); initialize(): void { - this.cronJob = cron.schedule('0 * * * *', async () => { - try { - await this.checkAllOrganizations(); - } catch (error) { - console.error('[LiquidityAlertChecker] Error executing check:', error); - } + this.cronJob = cron.schedule('0 * * * *', () => { + const run = (async () => { + try { + await this.checkAllOrganizations(); + } catch (error) { + console.error('[LiquidityAlertChecker] Error executing check:', error); + } + })(); + this.trackRun(run); }); console.log('[LiquidityAlertChecker] Cron job initialized - running hourly'); } - stop(): void { + private trackRun(run: Promise): void { + this.activeRuns.add(run); + void run.then( + () => { + this.activeRuns.delete(run); + }, + () => { + this.activeRuns.delete(run); + } + ); + } + + async stop(): Promise { if (this.cronJob) { this.cronJob.stop(); - console.log('[LiquidityAlertChecker] Cron job stopped'); } + + const activeRuns = [...this.activeRuns]; + if (activeRuns.length > 0) { + await Promise.allSettled(activeRuns); + } + + console.log('[LiquidityAlertChecker] Cron job stopped and active runs drained'); } async checkAllOrganizations(): Promise { diff --git a/backend/src/services/gracefulShutdown.ts b/backend/src/services/gracefulShutdown.ts new file mode 100644 index 00000000..b9dd9e7e --- /dev/null +++ b/backend/src/services/gracefulShutdown.ts @@ -0,0 +1,159 @@ +import type { Server } from 'node:http'; + +export interface ShutdownLogger { + info(message: string): unknown; + warn(message: string): unknown; + error(message: string, meta?: unknown): unknown; +} + +export interface GracefulShutdownOptions { + server: Server; + logger: ShutdownLogger; + stopBackgroundWork: () => void | Promise; + closeDependencies: () => Promise; + timeoutMs?: number; + cleanupTimeoutMs?: number; + exit?: (code: number) => void; +} + +async function drainHttpServer( + server: Server, + logger: ShutdownLogger, + timeoutMs: number +): Promise { + await new Promise((resolve) => { + let settled = false; + let timeout: NodeJS.Timeout | undefined; + + const finish = () => { + if (settled) return; + settled = true; + if (timeout) clearTimeout(timeout); + resolve(); + }; + + try { + server.close((error) => { + if (error && (error as NodeJS.ErrnoException).code !== 'ERR_SERVER_NOT_RUNNING') { + logger.error('HTTP server close failed', { error }); + } + finish(); + }); + } catch (error) { + // A closing/listen-racing server or injected transport can fail before + // registering the callback. Do not skip dependency shutdown or turn a + // deliberately handled SIGTERM into an unhandled rejection. + logger.error('HTTP server close threw synchronously', { error }); + try { + server.closeAllConnections(); + } catch (forceError) { + logger.error('Unable to force-close HTTP connections', { error: forceError }); + } + finish(); + } + + if (!settled) { + timeout = setTimeout(() => { + logger.warn(`HTTP drain exceeded ${timeoutMs}ms; closing remaining connections`); + try { + server.closeAllConnections(); + } catch (error) { + // Exceptions inside a timer otherwise escape the drain Promise and + // crash the process before PostgreSQL/Redis cleanup can run. + logger.error('Unable to force-close timed-out HTTP connections', { error }); + } + finish(); + }, timeoutMs); + } + }); +} + +async function runBoundedCleanup( + label: string, + cleanup: () => void | Promise, + timeoutMs: number, + logger: ShutdownLogger +): Promise { + let timeout: NodeJS.Timeout | undefined; + + try { + await Promise.race([ + Promise.resolve().then(cleanup), + new Promise((_, reject) => { + timeout = setTimeout( + () => reject(new Error(`${label} exceeded ${timeoutMs}ms`)), + timeoutMs + ); + }), + ]); + return true; + } catch (error) { + logger.error(`${label} failed`, { error }); + return false; + } finally { + if (timeout) clearTimeout(timeout); + } +} + +export function createGracefulShutdown({ + server, + logger, + stopBackgroundWork, + closeDependencies, + timeoutMs = 30_000, + cleanupTimeoutMs = 10_000, + exit = (code) => process.exit(code), +}: GracefulShutdownOptions): (signal: NodeJS.Signals) => Promise { + let shutdownPromise: Promise | null = null; + + return (signal: NodeJS.Signals): Promise => { + if (shutdownPromise) return shutdownPromise; + + shutdownPromise = (async () => { + logger.info(`Received ${signal}; starting graceful shutdown`); + + // server.close() runs immediately inside drainHttpServer, so new HTTP + // connections are refused before background cleanup begins. + const httpDrain = drainHttpServer(server, logger, timeoutMs); + + const backgroundStopped = await runBoundedCleanup( + 'Background work shutdown', + stopBackgroundWork, + cleanupTimeoutMs, + logger + ); + + await httpDrain; + + let dependenciesClosed = false; + if (backgroundStopped) { + dependenciesClosed = await runBoundedCleanup( + 'Runtime dependency cleanup', + closeDependencies, + cleanupTimeoutMs, + logger + ); + } else { + // A timed-out/rejected background drain may still be using PostgreSQL + // or Redis. Do not tear those dependencies out from under in-flight + // work; process exit will reclaim them after the failed drain is logged. + logger.error( + 'Runtime dependency cleanup skipped because background work did not stop cleanly' + ); + } + + const clean = backgroundStopped && dependenciesClosed; + if (clean) { + logger.info('Graceful shutdown complete'); + } else { + logger.error('Graceful shutdown completed with cleanup failures'); + } + // SIGTERM/SIGINT is an intentionally handled shutdown path. Cleanup + // failures remain visible above, but the issue contract requires an + // orderly signal-driven shutdown to terminate with status zero. + exit(0); + })(); + + return shutdownPromise; + }; +} diff --git a/backend/src/services/rateLimitService.ts b/backend/src/services/rateLimitService.ts index f90773e7..4cd5fb64 100644 --- a/backend/src/services/rateLimitService.ts +++ b/backend/src/services/rateLimitService.ts @@ -261,3 +261,7 @@ export class RateLimitService { } export const rateLimitService = new RateLimitService(); + +export async function closeRateLimitRedis(): Promise { + await RedisClient.disconnect(); +} diff --git a/backend/src/services/scheduleExecutor.ts b/backend/src/services/scheduleExecutor.ts index f34c242a..543929bb 100644 --- a/backend/src/services/scheduleExecutor.ts +++ b/backend/src/services/scheduleExecutor.ts @@ -10,6 +10,7 @@ import os from 'node:os'; export class ScheduleExecutor { private cronJob: ScheduledTask | null = null; private readonly podId: string; + private readonly activeRuns = new Set>(); constructor() { this.podId = `${os.hostname()}-${process.pid}`; @@ -21,26 +22,48 @@ export class ScheduleExecutor { */ initialize(): void { // Cron expression: run every minute - this.cronJob = cron.schedule('* * * * *', async () => { - try { - console.log('[ScheduleExecutor] Running scheduled task check...'); - await this.processDueSchedules(); - } catch (error) { - console.error('[ScheduleExecutor] Error in cron job execution:', error); - } + this.cronJob = cron.schedule('* * * * *', () => { + const run = (async () => { + try { + console.log('[ScheduleExecutor] Running scheduled task check...'); + await this.processDueSchedules(); + } catch (error) { + console.error('[ScheduleExecutor] Error in cron job execution:', error); + } + })(); + this.trackRun(run); }); console.log('[ScheduleExecutor] Cron job initialized - running every minute'); } + private trackRun(run: Promise): void { + this.activeRuns.add(run); + void run.then( + () => { + this.activeRuns.delete(run); + }, + () => { + this.activeRuns.delete(run); + } + ); + } + /** - * Stop the cron job (for graceful shutdown) + * Stop future cron ticks and drain callbacks that are already using + * shared runtime dependencies. */ - stop(): void { + async stop(): Promise { if (this.cronJob) { this.cronJob.stop(); - console.log('[ScheduleExecutor] Cron job stopped'); } + + const activeRuns = [...this.activeRuns]; + if (activeRuns.length > 0) { + await Promise.allSettled(activeRuns); + } + + console.log('[ScheduleExecutor] Cron job stopped and active runs drained'); } /**