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/core/sdam/monitor.js

@@ -8,6 +8,7 @@ const Connection = require('../../cmap/connection').Connection;
 const common = require('./common');
 const makeStateMachine = require('../utils').makeStateMachine;
 const MongoError = require('../error').MongoError;
+const makeInterruptableAsyncInterval = require('../../utils').makeInterruptableAsyncInterval;
 
 const sdamEvents = require('./events');
 const ServerHeartbeatStartedEvent = sdamEvents.ServerHeartbeatStartedEvent;
@@ -18,7 +19,6 @@ const kServer = Symbol('server');
 const kMonitorId = Symbol('monitorId');
 const kConnection = Symbol('connection');
 const kCancellationToken = Symbol('cancellationToken');
-const kLastCheckTime = Symbol('lastCheckTime');
 
 const STATE_CLOSED = common.STATE_CLOSED;
 const STATE_CLOSING = common.STATE_CLOSING;
@@ -33,6 +33,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);
@@ -41,6 +45,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,39 +94,34 @@ 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]);
-      rescheduleMonitoring(this, minHeartbeatFrequencyMS);
-      return;
-    }
-
-    if (this[kMonitorId]) {
-      clearTimeout(this[kMonitorId]);
-    }
-
-    monitorServer(this);
+    this[kMonitorId].wake();
   }
 
   close() {
-    if (this.s.state === STATE_CLOSED || this.s.state === STATE_CLOSING) {
+    if (isInCloseState(this)) {
       return;
     }
 
     stateTransition(this, STATE_CLOSING);
     this[kCancellationToken].emit('cancel');
     if (this[kMonitorId]) {
-      clearTimeout(this[kMonitorId]);
+      this[kMonitorId].stop();
+      this[kMonitorId] = null;
     }
 
     if (this[kConnection]) {
@@ -186,7 +186,7 @@ function checkServer(monitor, callback) {
       return;
     }
 
-    if (monitor.s.state === STATE_CLOSING || monitor.s.state === STATE_CLOSED) {
+    if (isInCloseState(monitor)) {
       conn.destroy({ force: true });
       failureHandler(new MongoError('monitor was destroyed'));
       return;
@@ -198,52 +198,44 @@ 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]));
+  return callback => {
+    stateTransition(monitor, STATE_MONITORING);
+    function done() {
+      if (!isInCloseState(monitor)) {
+        stateTransition(monitor, STATE_IDLE);
+      }
 
-  checkServer(monitor, e0 => {
-    if (e0 == null) {
-      rescheduleMonitoring(monitor);
-      return;
+      callback();
     }
 
-    // otherwise an error occured on initial discovery, also bail
-    if (monitor[kServer].description.type === ServerType.Unknown) {
-      monitor.emit('resetServer', e0);
-      rescheduleMonitoring(monitor);
-      return;
-    }
+    // TODO: the next line is a legacy event, remove in v4
+    process.nextTick(() => monitor.emit('monitoring', monitor[kServer]));
 
-    // According to the SDAM specification's "Network error during server check" section, if
-    // an ismaster call fails we reset the server's pool. If a server was once connected,
-    // change its type to `Unknown` only after retrying once.
-    monitor.emit('resetConnectionPool');
-
-    checkServer(monitor, e1 => {
-      if (e1) {
-        monitor.emit('resetServer', e1);
+    checkServer(monitor, e0 => {
+      if (e0 == null) {
+        return done();
       }
 
-      rescheduleMonitoring(monitor);
-    });
-  });
-}
+      // otherwise an error occured on initial discovery, also bail
+      if (monitor[kServer].description.type === ServerType.Unknown) {
+        monitor.emit('resetServer', e0);
+        return done();
+      }
 
-function rescheduleMonitoring(monitor, ms) {
-  const heartbeatFrequencyMS = monitor.options.heartbeatFrequencyMS;
-  if (monitor.s.state === STATE_CLOSING || monitor.s.state === STATE_CLOSED) {
-    return;
-  }
+      // According to the SDAM specification's "Network error during server check" section, if
+      // an ismaster call fails we reset the server's pool. If a server was once connected,
+      // change its type to `Unknown` only after retrying once.
+      monitor.emit('resetConnectionPool');
 
-  stateTransition(monitor, STATE_IDLE);
+      checkServer(monitor, e1 => {
+        if (e1) {
+          monitor.emit('resetServer', e1);
+        }
 
-  monitor[kLastCheckTime] = process.hrtime();
-  monitor[kMonitorId] = setTimeout(() => {
-    monitor[kMonitorId] = undefined;
-    monitor.requestCheck();
-  }, ms || heartbeatFrequencyMS);
+        done();
+      });
+    });
+  };
 }
 
 module.exports = {

test/unit/sdam/server_selection/select_servers.test.js

@@ -85,7 +85,6 @@ describe('selectServer', function() {
       let completed = 0;
       function finish() {
         completed++;
-        console.log(completed);
         if (completed === toSelect) done();
       }
 

test/unit/utils.test.js

@@ -37,7 +37,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();
@@ -76,7 +76,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);
@@ -100,7 +100,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();