Fix heartbeat for process restarts

src/heartbeat.js

@@ -24,13 +24,14 @@ class HeartbeatMonitor extends EventEmitter {
     this.timestamps = {};
 
     this.start = this.start.bind(this);
+    this.clearHeartbeatOnWorkerDown = this.clearHeartbeatOnWorkerDown.bind(this);
     this.monitorHeartbeat = this.monitorHeartbeat.bind(this);
   }
 
   recordHeartbeat(workerId) {
-    const past = this.timestamps[workerId] || this.heartbeatStartedAt;
+    const past = this.timestamps[workerId];
 
-    if (past !== null) {
+    if (past != null) {
       this.emit(
         'info:workerHeartbeatDelta',
         workerId,
@@ -40,16 +41,24 @@ class HeartbeatMonitor extends EventEmitter {
     this.timestamps[workerId] = process.hrtime();
   }
 
+  clearHeartbeatOnWorkerDown(workerId) {
+    this.timestamps[workerId] = null;
+  }
+
   start() {
     this.heartbeatStartedAt = process.hrtime();
 
+    this.supervisor.on('workerDown', this.clearHeartbeatOnWorkerDown);
+
     this.monitorHeartbeat();
   }
 
   stop() {
     if (this.monitorHeartbeatTimeout) {
       clearTimeout(this.monitorHeartbeatTimeout);
     }
+
+    this.supervisor.removeListener('workerDown', this.clearHeartbeatOnWorkerDown);
   }
 
   monitorHeartbeat() {
@@ -58,13 +67,17 @@ class HeartbeatMonitor extends EventEmitter {
     }
 
     Object.entries(this.supervisor.workers)
-      .filter(([, worker]) => !worker.isDead()).forEach(([workerId]) => {
-        const ts = this.timestamps[workerId] || this.heartbeatStartAt;
-        const delta = hrtimeToMilliseconds(process.hrtime(ts));
-
-        if (delta > this.stallTolerance) {
-          this.emit('workerStall', workerId);
-          this.emit('info:workerHeartbeatDelta', workerId, delta);
+      .filter(([, worker]) => !worker.isDead())
+      .filter(([, worker]) => worker.state === 'up')
+      .forEach(([workerId]) => {
+        const ts = this.timestamps[workerId];
+        if (ts != null) {
+          const delta = hrtimeToMilliseconds(process.hrtime(ts));
+
+          if (delta > this.stallTolerance) {
+            this.emit('workerStall', workerId);
+            this.emit('info:workerHeartbeatDelta', workerId, delta);
+          }
         }
       });
 
@@ -94,7 +107,7 @@ export function workerHeartbeat(worker, { workerInterval }) {
 
   function heartbeatFn() {
     if (obj.heartbeatTimeout) {
-      clearTimeout(this.heartbeatTimeout);
+      clearTimeout(obj.heartbeatTimeout);
     }
     worker.send({ act: 'poolHallHeartbeat' });
     obj.heartbeatTimeout = setTimeout(heartbeatFn, interval);

tests/heartbeat.test.js

@@ -83,12 +83,15 @@ describe('heartbeat', () => {
         });
         Object.values(supervisor.workers).forEach((worker) => {
           worker.emit('message', { act: 'poolHallHeartbeat' });
-          worker.emit('up');
+          worker.process.emit('message', { poolHallInternal: true, act: 'ready' });
         });
       });
 
       it('reports stalled workers', (done) => {
         supervisor.start();
+        Object.values(supervisor.workers).forEach((worker) => {
+          worker.state = 'up'; // eslint-disable-line no-param-reassign
+        });
         const strId = Object.keys(supervisor.workers)[0];
         monitor.heartbeatStartedAt = now; // timer won't be installed immediately
         monitor.timestamps[strId] = [now[0] - 10, now[1]];
@@ -99,9 +102,26 @@ describe('heartbeat', () => {
         monitor.monitorHeartbeat();
       });
 
+      it('does not report down workers as stalled', () => {
+        supervisor.start();
+        Object.values(supervisor.workers).forEach((worker) => {
+          worker.state = 'down'; // eslint-disable-line no-param-reassign
+        });
+        const strId = Object.keys(supervisor.workers)[0];
+        monitor.heartbeatStartedAt = now; // timer won't be installed immediately
+        monitor.timestamps[strId] = [now[0] - 10, now[1]];
+        monitor.once('workerStall', () => {
+          assert.fail();
+        });
+        monitor.monitorHeartbeat();
+      });
+
       it('does not report workers not stalled', () => {
         supervisor.start();
         monitor.heartbeatStartedAt = now;
+        Object.values(supervisor.workers).forEach((worker) => {
+          worker.state = 'up'; // eslint-disable-line no-param-reassign
+        });
         Object.keys(supervisor.workers)
           .forEach((strId) => {
             monitor.timestamps[strId] = now;
@@ -128,7 +148,7 @@ describe('heartbeat', () => {
 
         Object.values(supervisor.workers).forEach((worker) => {
           worker.emit('message', { act: 'poolHallHeartbeat' });
-          worker.emit('up');
+          worker.process.emit('message', { poolHallInternal: true, act: 'ready' });
         });
         setImmediate(jest.runOnlyPendingTimers);
       });
@@ -139,12 +159,34 @@ describe('heartbeat', () => {
         monitor.once('info:workerHeartbeatDelta', () => done());
         Object.values(supervisor.workers).forEach((worker) => {
           worker.emit('message', { act: 'poolHallHeartbeat' });
-          worker.emit('up');
+          worker.process.emit('message', { poolHallInternal: true, act: 'ready' });
         });
         Object.values(supervisor.workers).forEach((worker) => {
           worker.emit('message', { act: 'poolHallHeartbeat' });
         });
       });
+
+      it('clears heartbeat timestamp on workerDown', () => {
+        supervisor.start();
+        monitor.start();
+
+        Object.values(supervisor.workers).forEach((worker) => {
+          worker.emit('message', { act: 'poolHallHeartbeat' });
+          worker.process.emit('message', { poolHallInternal: true, act: 'ready' });
+        });
+
+        Object.keys(supervisor.workers).forEach((workerId) => {
+          assert(monitor.timestamps[workerId] != null);
+        });
+
+        Object.values(supervisor.workers).forEach((worker) => {
+          worker.emit('down');
+        });
+
+        Object.keys(supervisor.workers).forEach((workerId) => {
+          assert.equal(monitor.timestamps[workerId], null);
+        });
+      });
     });
   });
 });