fix: run indexer reliably under pm2

indexer/nest-cli.json

@@ -8,7 +8,7 @@
     "assets": [
       {
         "include": "_static/**/*",
-        "outDir": "dist",
+        "outDir": "dist/src",
         "watchAssets": true
       }
     ]
@@ -21,7 +21,7 @@
         "assets": [
           {
             "include": "_static/**/*",
-            "outDir": "dist/sepolia",
+            "outDir": "dist/sepolia/src",
             "watchAssets": true
           }
         ]
@@ -34,7 +34,7 @@
         "assets": [
           {
             "include": "_static/**/*",
-            "outDir": "dist/mainnet",
+            "outDir": "dist/mainnet/src",
             "watchAssets": true
           }
         ]

indexer/package.json

@@ -7,9 +7,9 @@
     "build:sepolia": "NETWORK=sepolia nest build sepolia",
     "start:sepolia": "NETWORK=sepolia nest start --watch",
     "start:mainnet": "NETWORK=mainnet nest start --watch",
-    "engage": "yarn build && pm2 start pm2.config.js && pm2 log",
-    "engage:mainnet": "NETWORK=mainnet yarn build:mainnet && pm2 start pm2.config.js --only mainnet && pm2 log",
-    "engage:sepolia": "NETWORK=sepolia yarn build:sepolia && pm2 start pm2.config.js --only sepolia && pm2 log",
+    "engage": "yarn build && ./node_modules/.bin/pm2 startOrRestart pm2.config.js && ./node_modules/.bin/pm2 log",
+    "engage:mainnet": "NETWORK=mainnet yarn build:mainnet && ./node_modules/.bin/pm2 startOrRestart pm2.config.js --only mainnet && ./node_modules/.bin/pm2 log",
+    "engage:sepolia": "NETWORK=sepolia yarn build:sepolia && ./node_modules/.bin/pm2 startOrRestart pm2.config.js --only sepolia && ./node_modules/.bin/pm2 log",
     "lint": "eslint \"{src,apps,libs,test}/**/*.ts\" --fix",
     "test": "jest"
   },
@@ -72,6 +72,7 @@
     "eslint-config-prettier": "^8.3.0",
     "eslint-plugin-prettier": "^3.4.0",
     "jest": "^29.7.0",
+    "pm2": "^6.0.14",
     "prettier": "^2.3.2",
     "source-map-support": "^0.5.20",
     "supertest": "^6.1.3",

indexer/pm2.config.js

@@ -20,15 +20,15 @@ module.exports = {
   apps: [
     {
       name: "mainnet",
-      script: "dist/mainnet/main.js",
+      script: "dist/mainnet/src/main.js",
       env: {
         ...supabaseConfig,
         ...loadEnvFile('.env.mainnet'),
       },
     },
     {
       name: "sepolia",
-      script: "dist/sepolia/main.js",
+      script: "dist/sepolia/src/main.js",
       env: {
         ...supabaseConfig,
         ...loadEnvFile('.env.sepolia'),

indexer/src/main.ts

@@ -6,6 +6,25 @@ import { AppModule } from '@/app.module';
 import { AppConfigService } from '@/config/config.service';
 import { CustomLogger } from '@/modules/shared/services/logger.service';
 
+function formatUnhandledReason(reason: unknown): string {
+  if (reason instanceof Error) {
+    return reason.stack ?? `${reason.name}: ${reason.message}`;
+  }
+
+  try {
+    return JSON.stringify(reason, null, 2);
+  } catch {
+    return String(reason);
+  }
+}
+
+function registerProcessErrorHandlers() {
+  process.on('unhandledRejection', (reason) => {
+    Logger.error(formatUnhandledReason(reason), '', 'UnhandledRejection');
+    process.exit(1);
+  });
+}
+
 async function listenWithRetries(app, startPort: number, maxRetries = 10): Promise<number> {
   let port = startPort;
   for (let attempt = 0; attempt < maxRetries; attempt++) {
@@ -25,6 +44,8 @@ async function listenWithRetries(app, startPort: number, maxRetries = 10): Promi
 }
 
 async function bootstrap() {
+  registerProcessErrorHandlers();
+
   const app = await NestFactory.create(AppModule);
   const configSvc = app.get(AppConfigService);
 

indexer/src/modules/queue/queues/block-processing.queue.ts

@@ -25,15 +25,35 @@ export class BlockProcessingQueue {
 
     const existingJob = await this.queue.getJob(jobId);
     if (existingJob) {
-      await existingJob.remove();
-      Logger.warn('⚠️', `Updated existing job [${jobId}]`);
+      const state = await existingJob.getState();
+
+      // Active/waiting jobs cannot be removed safely and should be reused.
+      if (['active', 'waiting', 'delayed', 'paused'].includes(state)) {
+        return;
+      }
+
+      try {
+        await existingJob.remove();
+        Logger.warn('⚠️', `Updated existing job [${jobId}]`);
+      } catch (error) {
+        Logger.warn(`Skipped removing existing job [${jobId}]: ${error?.message || error}`);
+        return;
+      }
     }
 
-    await this.queue.add(
-      'BlockNumQueue',
-      { blockNum, chain: this.configSvc.chain.chainIdL1, timestamp, retryCount: 0, maxRetries },
-      { jobId, removeOnComplete: true, removeOnFail: true }
-    );
+    try {
+      await this.queue.add(
+        'BlockNumQueue',
+        { blockNum, chain: this.configSvc.chain.chainIdL1, timestamp, retryCount: 0, maxRetries },
+        { jobId, removeOnComplete: true, removeOnFail: true }
+      );
+    } catch (error) {
+      const message = String(error?.message || error);
+      if (message.includes('jobId') || message.includes('JobId') || message.includes('already exists')) {
+        return;
+      }
+      throw error;
+    }
 
     if (blockNum % 1000 === 0) Logger.debug(`Added block ${blockNum} to queue`);
   }