feat(backend): graceful shutdown for HTTP server, DB pool, and Redis

backend/src/controllers/healthController.ts

@@ -13,6 +13,22 @@ export const redis: Redis | null = config.REDIS_URL
     })
   : null;
 
+export async function closeHealthDependencies(): Promise<void> {
+  const closers: Promise<unknown>[] = [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;

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<void> } | undefined;
+let integrityCheckJob: { stop(): Promise<void> } | undefined;
+let auditCacheCleanup: ReturnType<typeof setInterval> | undefined;
+let idempotencyCleanup: ReturnType<typeof setInterval> | undefined;
+const activeCleanupRuns = new Set<Promise<void>>();
+let isShuttingDown = false;
+
+function trackCleanupRun(run: Promise<void>): 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'));

backend/src/jobs/part49Jobs.ts

@@ -26,40 +26,80 @@ const ADVISORY_LOCK_INTEGRITY_CHECK = 84_901_002;
  */
 
 export interface ScheduledJob {
-  stop(): void;
+  stop(): Promise<void>;
+}
+
+function trackRun(activeRuns: Set<Promise<void>>, run: Promise<void>): void {
+  activeRuns.add(run);
+  void run.then(
+    () => {
+      activeRuns.delete(run);
+    },
+    () => {
+      activeRuns.delete(run);
+    }
+  );
+}
+
+async function stopAndDrain(
+  task: ScheduledTask,
+  activeRuns: Set<Promise<void>>,
+  label: string
+): Promise<void> {
+  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<Promise<void>>();
+
   // 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<Promise<void>>();
+  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');
     },
   };
 }