worker restarts

packages/bitcore-node/src/services/worker.ts

@@ -14,27 +14,18 @@ export class WorkerService extends EventEmitter {
     worker: ClusterWorker;
     active: boolean;
     started: Promise<any>;
+    workerId: number;
+    restartCount: number;
   }>();
 
+  private shuttingDown = false;
+
   async start() {
     if (cluster.isPrimary) {
       logger.verbose(`Master ${process.pid} is running`);
       if (!args.DEBUG) {
         for (let worker = 0; worker < config.numWorkers; worker++) {
-          const newWorker = cluster.fork();
-          logger.verbose(`Starting worker number ${worker}`);
-          newWorker.on('message', (msg: any) => {
-            this.emit(msg.id, msg);
-          });
-          newWorker.on('exit', (code, _signal) => {
-            logger[code == 0 ? 'info' : 'warn'](`Worker ${newWorker.process.pid} stopped with code ${code}`);
-          });
-          const started = new Promise<void>(resolve => {
-            newWorker.on('listening', () => {
-              resolve();
-            });
-          });
-          this.workers.push({ worker: newWorker, active: false, started });
+          this.startWorker(worker, 0);
         }
       }
       const startedPromises = this.workers.map(worker => worker.started);
@@ -45,7 +36,73 @@ export class WorkerService extends EventEmitter {
     }
   }
 
-  async stop() {}
+  async stop() {
+    this.shuttingDown = true;
+
+    // Disconnect all workers gracefully
+    for (const workerData of this.workers) {
+      if (workerData.worker.isConnected()) {
+        workerData.worker.disconnect();
+      }
+    }
+  }
+
+  private startWorker(workerId: number, restartCount: number = 0) {
+    const isRestart = restartCount > 0;
+    if (this.shuttingDown) {
+      if (isRestart) {
+        logger.info(`Not restarting worker ${workerId} - service is shutting down`);
+      }
+      return;
+    }
+
+    const newWorker = cluster.fork();
+
+    if (isRestart) {
+      logger.warn(`Restarting worker ${workerId} (restart #${restartCount})`);
+    } else {
+      logger.verbose(`Starting worker ${workerId}`);
+    }
+
+    newWorker.on('message', (msg: any) => {
+      this.emit(msg.id, msg);
+    });
+
+    newWorker.on('exit', (code, signal) => {
+      const exitReason = code !== 0 || signal ? 'crashed' : 'stopped gracefully';
+      logger[code == 0 ? 'info' : 'error'](
+        `Worker ${newWorker.process.pid} ${exitReason} (code: ${code}, signal: ${signal})`
+      );
+
+      const workerIndex = this.workers.findIndex(w => w.worker === newWorker);
+      if (workerIndex > -1) {
+        const workerData = this.workers[workerIndex];
+        this.workers.splice(workerIndex, 1);
+
+        if ((code !== 0 || signal) && !this.shuttingDown) {
+          logger.info(`Scheduling worker ${workerData.workerId} restart in 5 seconds...`);
+          setTimeout(() => {
+            this.startWorker(workerData.workerId, workerData.restartCount + 1);
+          }, 5 * 1000);
+        }
+      }
+    });
+
+    const started = new Promise<void>(resolve => {
+      newWorker.on('listening', () => {
+        logger.info(`Worker ${workerId} successfully ${isRestart ? 're' : ''}started (pid: ${newWorker.process.pid})`);
+        resolve();
+      });
+    });
+
+    this.workers.push({
+      worker: newWorker,
+      active: false,
+      started,
+      workerId,
+      restartCount
+    });
+  }
 
   sendTask(task: any, argument: any, done: CallbackType) {
     const worker = this.workers.shift();