fix: data fetcher add graceful shutdown timeout

packages/data-fetcher/.env.example

@@ -1,6 +1,8 @@
 LOG_LEVEL=debug
 PORT=3040
 
+GRACEFUL_SHUTDOWN_TIMEOUT_MS=0
+
 BLOCKCHAIN_RPC_URL=http://localhost:3050
 
 RPC_CALLS_DEFAULT_RETRY_TIMEOUT=30000

packages/data-fetcher/src/config.ts

@@ -13,6 +13,7 @@ export default () => {
     WS_MAX_CONNECTIONS,
     USE_WEBSOCKETS_FOR_TRANSACTIONS,
     MAX_BLOCKS_BATCH_SIZE,
+    GRACEFUL_SHUTDOWN_TIMEOUT_MS,
   } = process.env;
 
   return {
@@ -31,5 +32,6 @@ export default () => {
       useWebSocketsForTransactions: USE_WEBSOCKETS_FOR_TRANSACTIONS === "true",
     },
     maxBlocksBatchSize: parseInt(MAX_BLOCKS_BATCH_SIZE, 10) || 20,
+    gracefulShutdownTimeoutMs: parseInt(GRACEFUL_SHUTDOWN_TIMEOUT_MS, 10) || 0,
   };
 };

packages/data-fetcher/src/health/health.controller.ts

@@ -1,18 +1,23 @@
 import { Logger, Controller, Get, OnApplicationShutdown, BeforeApplicationShutdown } from "@nestjs/common";
 import { HealthCheckService, HealthCheck, HealthCheckResult } from "@nestjs/terminus";
 import { HttpAdapterHost } from "@nestjs/core";
+import { ConfigService } from "@nestjs/config";
 import { JsonRpcHealthIndicator } from "./jsonRpcProvider.health";
+import { setTimeout } from "node:timers/promises";
 
 @Controller(["health", "ready"])
 export class HealthController implements OnApplicationShutdown, BeforeApplicationShutdown {
   private readonly logger: Logger;
+  private readonly gracefulShutdownTimeoutMs: number;
 
   constructor(
     private readonly healthCheckService: HealthCheckService,
     private readonly jsonRpcHealthIndicator: JsonRpcHealthIndicator,
-    private readonly httpAdapter: HttpAdapterHost
+    private readonly httpAdapter: HttpAdapterHost,
+    configService: ConfigService
   ) {
     this.logger = new Logger(HealthController.name);
+    this.gracefulShutdownTimeoutMs = configService.get<number>("gracefulShutdownTimeoutMs");
   }
 
   @Get()
@@ -27,24 +32,30 @@ export class HealthController implements OnApplicationShutdown, BeforeApplicatio
   }
 
   async beforeApplicationShutdown(signal?: string): Promise<void> {
+    this.logger.debug(`Received termination signal ${signal || ""}`);
     const httpServer = this.httpAdapter.httpAdapter.getHttpServer();
     await new Promise((resolve) => {
       httpServer.getConnections((_, count) => {
         this.logger.debug({ message: `Number of active connections: ${count}` });
         resolve(true);
       });
     });
-    this.logger.debug({ message: "Before application shutdown called", signal });
+
+    if (signal === "SIGTERM") {
+      this.logger.debug(`Awaiting ${this.gracefulShutdownTimeoutMs}ms before shutdown`);
+      await setTimeout(this.gracefulShutdownTimeoutMs);
+      this.logger.debug(`Timeout reached, shutting down now`);
+    }
   }
 
   async onApplicationShutdown(signal?: string): Promise<void> {
+    this.logger.debug({ message: "On application shutdown called", signal });
     const httpServer = this.httpAdapter.httpAdapter.getHttpServer();
     await new Promise((resolve) => {
       httpServer.getConnections((_, count) => {
         this.logger.debug({ message: `Number of active connections: ${count}` });
         resolve(true);
       });
     });
-    this.logger.debug({ message: "On application shutdown called", signal });
   }
 }

packages/data-fetcher/src/main.ts

@@ -31,10 +31,6 @@ async function bootstrap() {
   app.enableShutdownHooks();
   app.useGlobalInterceptors(new ResponseTransformInterceptor());
   await app.listen(configService.get("port"));
-
-  setInterval(() => {
-    logger.debug({ message: "I'm alive", context: "Main" });
-  }, 1000);
 }
 
 bootstrap();

packages/worker/src/block/block.watcher.ts

@@ -90,18 +90,11 @@ export class BlockWatcher implements OnModuleInit, OnModuleDestroy {
   private async getBlockInfoFromBlockChain(blockNumber: number): Promise<BlockData> {
     this.logger.debug({ message: "Getting block from the blockchain", blockNumber });
 
-    //const stopGetBlockInfoDurationMetric = this.getBlockInfoDurationMetric.startTimer();
-    this.dataFetchService
-      .getBlockData(blockNumber)
-      .then((_) => {
-        this.logger.debug({ message: "Received block from the blockchain", blockNumber });
-      })
-      .catch((e) => {
-        this.logger.error(e);
-      });
-    //stopGetBlockInfoDurationMetric();
-
-    return null;
+    const stopGetBlockInfoDurationMetric = this.getBlockInfoDurationMetric.startTimer();
+    const blockData = await this.dataFetchService.getBlockData(blockNumber);
+    stopGetBlockInfoDurationMetric();
+
+    return blockData;
   }
 
   public async onModuleInit(): Promise<void> {