watcher: add supervisor to monitor watchers

Author: panoel

watcher/src/consts.ts

@@ -1,8 +1,13 @@
-import { ChainName, CONTRACTS, Network } from '@certusone/wormhole-sdk/lib/cjs/utils/consts';
+import { ChainName, CONTRACTS } from '@certusone/wormhole-sdk/lib/cjs/utils/consts';
 import { Environment } from '@wormhole-foundation/wormhole-monitor-common';
 import { AxiosRequestConfig } from 'axios';
 
 export const TIMEOUT = 0.5 * 1000;
+export const HB_INTERVAL = 5 * 60 * 1000; // 5 Minutes
+export type WorkerData = {
+  network: Environment;
+  chain: ChainName;
+};
 
 // Notes about RPCs
 // Ethereum

watcher/src/index.ts

@@ -10,6 +10,7 @@ import {
   getEnvironment,
   getMode,
 } from '@wormhole-foundation/wormhole-monitor-common';
+import { startSupervisor } from './workers/supervisor';
 
 initDb();
 
@@ -100,3 +101,4 @@ if (mode === 'vaa') {
 } else {
   throw new Error(`Unknown mode: ${mode}`);
 }
+startSupervisor(supportedChains);

watcher/src/watchers/Watcher.ts

@@ -9,6 +9,7 @@ import { TIMEOUT } from '../consts';
 import { VaasByBlock } from '../databases/types';
 import { getResumeBlockByChain, storeLatestBlock, storeVaasByBlock } from '../databases/utils';
 import { getLogger, WormholeLogger } from '../utils/logger';
+import { parentPort } from 'worker_threads';
 
 export class Watcher {
   chain: ChainName;
@@ -105,6 +106,9 @@ export class Watcher {
         this.logger.warn(`backing off for ${expoBacko}ms`);
         await sleep(expoBacko);
       }
+      if (parentPort) {
+        parentPort.postMessage('heartbeat');
+      }
     }
   }
 }

watcher/src/workers/supervisor.ts

@@ -0,0 +1,63 @@
+import { ChainName } from '@certusone/wormhole-sdk';
+import { Worker } from 'worker_threads';
+import { HB_INTERVAL, WorkerData } from '../consts';
+import { getLogger } from '../utils/logger';
+import { Environment, getEnvironment } from '@wormhole-foundation/wormhole-monitor-common';
+
+interface WorkerInfo {
+  worker: Worker;
+  data: WorkerData;
+  lastHB: number;
+}
+
+const workers: { [key: string]: WorkerInfo } = {};
+const logger = getLogger('supervisor');
+const network: Environment = getEnvironment();
+
+function spawnWorker(data: WorkerData) {
+  const workerName = `${data.chain}Worker`;
+  logger.info(`Spawning worker ${workerName} on network ${network}...`);
+  const worker = new Worker('./dist/src/workers/worker.js', { workerData: data });
+
+  worker.on('message', (message) => {
+    if (message === 'heartbeat') {
+      logger.debug(`Worker ${workerName} sent HB`);
+      workers[workerName].lastHB = Date.now();
+    }
+  });
+
+  worker.on('exit', (code) => {
+    logger.warn(`Worker ${workerName} exited with code ${code}`);
+    if (code !== 0) {
+      logger.error(`Restarting worker ${workerName}...`);
+      spawnWorker(data);
+    }
+  });
+
+  workers[workerName] = { worker, data, lastHB: Date.now() };
+  logger.debug('Finished spawning worker:', workerName);
+}
+
+function monitorWorkers() {
+  setInterval(() => {
+    for (const [workerName, workerInfo] of Object.entries(workers)) {
+      logger.debug(
+        `Checking worker ${workerName} with lastHB of ${new Date(workerInfo.lastHB)}...`
+      );
+      if (Date.now() - workerInfo.lastHB > HB_INTERVAL) {
+        logger.error(`Worker ${workerName} missed HB, restarting...`);
+        workerInfo.worker.terminate();
+        spawnWorker(workerInfo.data);
+      }
+    }
+  }, HB_INTERVAL);
+}
+
+export function startSupervisor(supportedChains: ChainName[]) {
+  supportedChains.forEach((chain) => {
+    const workerData: WorkerData = { network, chain };
+    spawnWorker(workerData);
+  });
+
+  monitorWorkers();
+}

watcher/src/workers/worker.ts

@@ -0,0 +1,10 @@
+import { initDb } from '../databases/utils';
+import { makeFinalizedWatcher } from '../watchers/utils';
+import { workerData } from 'worker_threads';
+
+initDb(false);
+const network = workerData.network;
+const chain = workerData.chain;
+console.log(`Making watcher for ${network} ${chain}...`);
+makeFinalizedWatcher(network, chain).watch();
+console.log(`Watcher for ${network} ${chain} started.`);