fix(indexer): allow clean exit by stopping worker (#90)

Signed-off-by: david <[email protected]>
across-protocol · Oct 30, 2024 · 50e239d · 50e239d
1 parent 62e02d1
commit 50e239d
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 5 deletions.
diff --git a/packages/indexer/src/data-indexing/service/Indexer.ts b/packages/indexer/src/data-indexing/service/Indexer.ts
@@ -74,7 +74,7 @@ export class Indexer {
   public stopGracefully() {
     this.logger.info({
       at: "Indexer::stopGracefully",
-      message: `Requesting indexer ${this.dataHandler.getDataIdentifier()} to be stoped`,
+      message: `Requesting indexer ${this.dataHandler.getDataIdentifier()} to be stopped`,
     });
     this.stopRequested = true;
   }

diff --git a/packages/indexer/src/generics/BaseIndexer.ts b/packages/indexer/src/generics/BaseIndexer.ts
@@ -61,7 +61,7 @@ export abstract class BaseIndexer {
   public stop(): void {
     this.logger.info({
       at: "BaseIndexer#stop",
-      message: `Requesting indexer ${this.name} to be stoped`,
+      message: `Requesting indexer ${this.name} to be stopped`,
     });
     this.stopRequested = true;
   }

diff --git a/packages/indexer/src/main.ts b/packages/indexer/src/main.ts
@@ -91,7 +91,12 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
 
   const indexerQueuesService = new IndexerQueuesService(redis);
   // Set up message workers
-  new IntegratorIdWorker(redis, postgres, logger, retryProvidersFactory);
+  const integratorIdWorker = new IntegratorIdWorker(
+    redis,
+    postgres,
+    logger,
+    retryProvidersFactory,
+  );
 
   const spokePoolIndexers = spokePoolChainsEnabled.map((chainId) => {
     const spokePoolIndexerDataHandler = new SpokePoolIndexerDataHandler(
@@ -156,11 +161,13 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
         at: "Indexer#Main",
         message: "Wait for shutdown, or press Ctrl+C again to forcefully exit.",
       });
+      integratorIdWorker.close();
       spokePoolIndexers.map((s) => s.stopGracefully());
       hubPoolIndexer.stopGracefully();
       bundleProcessor.stop();
       bundleBuilderProcessor.stop();
     } else {
+      integratorIdWorker.close();
       logger.info({ at: "Indexer#Main", message: "Forcing exit..." });
       redis?.quit();
       postgres?.destroy();
@@ -195,7 +202,7 @@ export async function Main(config: parseEnv.Config, logger: winston.Logger) {
         bundleBuilderResult.status === "fulfilled",
     },
   });
-
+  await integratorIdWorker.close();
   redis?.quit();
   postgres?.destroy();
   logger.info({ at: "Indexer#Main", message: "Exiting indexer" });

diff --git a/packages/indexer/src/messaging/IntegratorIdWorker.ts b/packages/indexer/src/messaging/IntegratorIdWorker.ts
@@ -47,7 +47,6 @@ export class IntegratorIdWorker {
       { connection: this.redis, concurrency: 10 },
     );
   }
-
   private async run(relayHash: string) {
     const repository = this.postgres.getRepository(entities.V3FundsDeposited);
     const deposit = await repository.findOne({
@@ -80,4 +79,7 @@ export class IntegratorIdWorker {
     }
     return;
   }
+  public async close() {
+    return this.worker.close();
+  }
 }