fix: use async interruptable interval for server monitoring

The existing implementation of monitoring check scheduling is prone
to failure during high load due to an error recording the last
check time. Refactoring to use the newly introduced async
interruptable interval timer eliminates this bug, as well as
optimizes in cases where multiple simultaneous requests were not
debounced.

NODE-2643

Author: mbroadst

lib/sdam/monitor.js

@@ -1,7 +1,7 @@
 'use strict';
 
 const { ServerType, STATE_CLOSED, STATE_CLOSING } = require('./common');
-const { makeStateMachine, calculateDurationInMs } = require('../utils');
+const { makeStateMachine, calculateDurationInMs, makeInterruptableAsyncInterval } = require('../utils');
 const EventEmitter = require('events');
 const connect = require('../cmap/connect');
 const { Connection } = require('../cmap/connection');
@@ -17,7 +17,6 @@ const kServer = Symbol('server');
 const kMonitorId = Symbol('monitorId');
 const kConnection = Symbol('connection');
 const kCancellationToken = Symbol('cancellationToken');
-const kLastCheckTime = Symbol('lastCheckTime');
 const kRTTPinger = Symbol('rttPinger');
 const kRoundTripTime = Symbol('roundTripTime');
 
@@ -32,6 +31,10 @@ const stateTransition = makeStateMachine({
 
 const INVALID_REQUEST_CHECK_STATES = new Set([STATE_CLOSING, STATE_CLOSED, STATE_MONITORING]);
 
+function isInCloseState(monitor) {
+  return monitor.s.state === STATE_CLOSED || monitor.s.state === STATE_CLOSING;
+}
+
 class Monitor extends EventEmitter {
   constructor(server, options) {
     super(options);
@@ -40,6 +43,7 @@ class Monitor extends EventEmitter {
     this[kConnection] = undefined;
     this[kCancellationToken] = new EventEmitter();
     this[kCancellationToken].setMaxListeners(Infinity);
+    this[kMonitorId] = null;
     this.s = {
       state: STATE_CLOSED
     };
@@ -89,33 +93,26 @@ class Monitor extends EventEmitter {
       return;
     }
 
-    monitorServer(this);
+    // start
+    const heartbeatFrequencyMS = this.options.heartbeatFrequencyMS;
+    const minHeartbeatFrequencyMS = this.options.minHeartbeatFrequencyMS;
+    this[kMonitorId] = makeInterruptableAsyncInterval(monitorServer(this), {
+      interval: heartbeatFrequencyMS,
+      minInterval: minHeartbeatFrequencyMS,
+      immediate: true
+    });
   }
 
   requestCheck() {
     if (INVALID_REQUEST_CHECK_STATES.has(this.s.state)) {
       return;
     }
 
-    const heartbeatFrequencyMS = this.options.heartbeatFrequencyMS;
-    const minHeartbeatFrequencyMS = this.options.minHeartbeatFrequencyMS;
-    const remainingTime = heartbeatFrequencyMS - calculateDurationInMs(this[kLastCheckTime]);
-    if (remainingTime > minHeartbeatFrequencyMS && this[kMonitorId]) {
-      clearTimeout(this[kMonitorId]);
-      this[kMonitorId] = undefined;
-
-      rescheduleMonitoring(this, minHeartbeatFrequencyMS);
-      return;
-    }
-
-    clearTimeout(this[kMonitorId]);
-    this[kMonitorId] = undefined;
-
-    monitorServer(this);
+    this[kMonitorId].wake();
   }
 
   reset() {
-    if (this.s.state === STATE_CLOSED || this.s.state === STATE_CLOSING) {
+    if (isInCloseState(this)) {
       return;
     }
 
@@ -124,12 +121,19 @@ class Monitor extends EventEmitter {
 
     // restart monitor
     stateTransition(this, STATE_IDLE);
+
+    // restart monitoring
     const heartbeatFrequencyMS = this.options.heartbeatFrequencyMS;
-    this[kMonitorId] = setTimeout(() => this.requestCheck(), heartbeatFrequencyMS);
+    const minHeartbeatFrequencyMS = this.options.minHeartbeatFrequencyMS;
+    this[kMonitorId] = makeInterruptableAsyncInterval(monitorServer(this), {
+      interval: heartbeatFrequencyMS,
+      minInterval: minHeartbeatFrequencyMS,
+      immediate: true
+    });
   }
 
   close() {
-    if (this.s.state === STATE_CLOSED || this.s.state === STATE_CLOSING) {
+    if (isInCloseState(this)) {
       return;
     }
 
@@ -144,6 +148,11 @@ class Monitor extends EventEmitter {
 
 function resetMonitorState(monitor) {
   stateTransition(monitor, STATE_CLOSING);
+  if (monitor[kMonitorId]) {
+    monitor[kMonitorId].stop();
+    monitor[kMonitorId] = null;
+  }
+
   if (monitor[kRTTPinger]) {
     monitor[kRTTPinger].close();
     monitor[kRTTPinger] = undefined;
@@ -214,7 +223,7 @@ function checkServer(monitor, callback) {
         new ServerHeartbeatSucceededEvent(duration, isMaster, monitor.address)
       );
 
-      // if we are streaming ismaster responses then we immediately issue another started
+      // if we are using the streaming protocol then we immediately issue another `started`
       // event, otherwise the "check" is complete and return to the main monitor loop
       if (isAwaitable && isMaster.topologyVersion) {
         monitor.emit('serverHeartbeatStarted', new ServerHeartbeatStartedEvent(monitor.address));
@@ -234,7 +243,7 @@ function checkServer(monitor, callback) {
 
   // connecting does an implicit `ismaster`
   connect(monitor.connectOptions, monitor[kCancellationToken], (err, conn) => {
-    if (conn && (monitor.s.state === STATE_CLOSED || monitor.s.state === STATE_CLOSING)) {
+    if (conn && isInCloseState(monitor)) {
       conn.destroy({ force: true });
       return;
     }
@@ -266,46 +275,40 @@ function checkServer(monitor, callback) {
 }
 
 function monitorServer(monitor) {
-  stateTransition(monitor, STATE_MONITORING);
-
-  // TODO: the next line is a legacy event, remove in v4
-  process.nextTick(() => monitor.emit('monitoring', monitor[kServer]));
-
-  const heartbeatFrequencyMS = monitor.options.heartbeatFrequencyMS;
-  checkServer(monitor, (err, isMaster) => {
-    if (err) {
-      if (monitor[kServer].description.type !== ServerType.Unknown) {
-        rescheduleMonitoring(monitor);
-        return;
+  return callback => {
+    stateTransition(monitor, STATE_MONITORING);
+    function done() {
+      if (!isInCloseState(monitor)) {
+        stateTransition(monitor, STATE_IDLE);
       }
-    }
 
-    // if the check indicates streaming is supported, immediately reschedule monitoring
-    if (isMaster && isMaster.topologyVersion) {
-      rescheduleMonitoring(monitor);
-      return;
+      callback();
     }
 
-    rescheduleMonitoring(monitor, heartbeatFrequencyMS);
-  });
-}
+    // TODO: the next line is a legacy event, remove in v4
+    process.nextTick(() => monitor.emit('monitoring', monitor[kServer]));
 
-function rescheduleMonitoring(monitor, ms) {
-  if (monitor.s.state === STATE_CLOSING || monitor.s.state === STATE_CLOSED) {
-    return;
-  }
+    checkServer(monitor, (err, isMaster) => {
+      if (err) {
+        // otherwise an error occured on initial discovery, also bail
+        if (monitor[kServer].description.type === ServerType.Unknown) {
+          monitor.emit('resetServer', err);
+          return done();
+        }
+      }
 
-  stateTransition(monitor, STATE_IDLE);
-  if (monitor[kMonitorId]) {
-    clearTimeout(monitor[kMonitorId]);
-    monitor[kMonitorId] = undefined;
-  }
+      // if the check indicates streaming is supported, immediately reschedule monitoring
+      if (isMaster && isMaster.topologyVersion) {
+        setTimeout(() => {
+          if (!isInCloseState(monitor)) {
+            monitor[kMonitorId].wake();
+          }
+        });
+      }
 
-  monitor[kLastCheckTime] = process.hrtime();
-  monitor[kMonitorId] = setTimeout(() => {
-    monitor[kMonitorId] = undefined;
-    monitor.requestCheck();
-  }, ms);
+      done();
+    });
+  };
 }
 
 function makeTopologyVersion(tv) {

lib/utils.js

@@ -951,6 +951,12 @@ function makeInterruptableAsyncInterval(fn, options) {
     const timeUntilNextCall = Math.max(interval - timeSinceLastCall, 0);
     lastWakeTime = currentTime;
 
+    // For the streaming protocol: there is nothing obviously stopping this
+    // interval from being woken up again while we are waiting "infinitely"
+    // for `fn` to be called again`. Since the function effectively
+    // never completes, the `timeUntilNextCall` will continue to grow
+    // negatively unbounded, so it will never trigger a reschedule here.
+
     // debounce multiple calls to wake within the `minInterval`
     if (timeSinceLastWake < minInterval) {
       return;
@@ -983,6 +989,7 @@ function makeInterruptableAsyncInterval(fn, options) {
   function executeAndReschedule() {
     lastWakeTime = 0;
     lastCallTime = now();
+
     fn(err => {
       if (err) throw err;
       reschedule(interval);

test/functional/sdam.test.js

@@ -21,11 +21,13 @@ class SDAMRunnerContext extends TestRunnerContext {
   }
 
   waitForPrimaryChange(client) {
+    const currentPrimary = this.currentPrimary;
+
     return new Promise(resolve => {
       function eventHandler(event) {
         if (
           event.newDescription.type === 'RSPrimary' &&
-          event.newDescription.address !== this.currentPrimary
+          event.newDescription.address !== currentPrimary
         ) {
           resolve();
           client.removeListener('serverDescriptionChanged', eventHandler);

test/functional/spec-runner/index.js

@@ -540,7 +540,7 @@ const kOperations = new Map([
   [
     'waitForPrimaryChange',
     (operation, testRunner, context /*, options */) => {
-      testRunner.waitForPrimaryChange(context.client);
+      return testRunner.waitForPrimaryChange(context.client);
     }
   ],
   [

test/unit/utils.test.js

@@ -35,7 +35,7 @@ describe('utils', function() {
   });
 
   context('makeInterruptableAsyncInterval', function() {
-    const roundToNearestMultipleOfTen = x => Math.floor(x / 10) * 10;
+    const roundToNearestMultipleOfTen = x => Math.round(x / 10) * 10;
 
     it('should execute a method in an repeating interval', function(done) {
       let lastTime = now();
@@ -74,7 +74,7 @@ describe('utils', function() {
 
       setTimeout(() => {
         const roundedMarks = marks.map(roundToNearestMultipleOfTen);
-        expect(roundedMarks[0]).to.equal(10);
+        expect(roundedMarks[0]).to.be.lessThan(50);
         executor.stop();
         done();
       }, 50);
@@ -98,7 +98,7 @@ describe('utils', function() {
 
       setTimeout(() => {
         const roundedMarks = marks.map(roundToNearestMultipleOfTen);
-        expect(roundedMarks[0]).to.equal(10);
+        expect(roundedMarks[0]).to.be.lessThan(50);
         expect(roundedMarks.slice(1).every(mark => mark === 50)).to.be.true;
         executor.stop();
         done();