refactor: always emit success/failure events after started

Previously we were emitting a success or failure event for
heartbeats after the entire check, which would hide an extra
attempt from the user. Now we emit started => success/failure
events for every individual server check.

Author: mbroadst

lib/core/sdam/monitor.js

@@ -136,8 +136,27 @@ function checkServer(monitor, callback) {
     monitor[kConnection] = undefined;
   }
 
+  const start = process.hrtime();
   monitor.emit('serverHeartbeatStarted', new ServerHeartbeatStartedEvent(monitor.address));
 
+  function failureHandler(err) {
+    monitor.emit(
+      'serverHeartbeatFailed',
+      new ServerHeartbeatFailedEvent(calculateDurationInMs(start), err, monitor.address)
+    );
+
+    callback(err);
+  }
+
+  function successHandler(isMaster) {
+    monitor.emit(
+      'serverHeartbeatSucceeded',
+      new ServerHeartbeatSucceededEvent(calculateDurationInMs(start), isMaster, monitor.address)
+    );
+
+    return callback(undefined, isMaster);
+  }
+
   if (monitor[kConnection] != null) {
     const connectTimeoutMS = monitor.options.connectTimeoutMS;
     monitor[kConnection].command(
@@ -146,10 +165,11 @@ function checkServer(monitor, callback) {
       { socketTimeout: connectTimeoutMS },
       (err, result) => {
         if (err) {
-          return callback(err);
+          failureHandler(err);
+          return;
         }
 
-        return callback(undefined, result.result);
+        successHandler(result.result);
       }
     );
 
@@ -160,37 +180,37 @@ function checkServer(monitor, callback) {
   connect(monitor.connectOptions, monitor[kCancellationToken], (err, conn) => {
     if (err) {
       monitor[kConnection] = undefined;
-      callback(err);
+      failureHandler(err);
       return;
     }
 
     if (monitor.s.state === STATE_CLOSING || monitor.s.state === STATE_CLOSED) {
       conn.destroy({ force: true });
-      callback(new MongoError('monitor was destroyed'));
+      failureHandler(new MongoError('monitor was destroyed'));
       return;
     }
 
     monitor[kConnection] = conn;
-    callback(undefined, conn.ismaster);
+    successHandler(conn.ismaster);
   });
 }
 
 function monitorServer(monitor) {
-  const start = process.hrtime();
   stateTransition(monitor, STATE_MONITORING);
 
   // TODO: the next line is a legacy event, remove in v4
   process.nextTick(() => monitor.emit('monitoring', monitor[kServer]));
 
-  checkServer(monitor, (e0, isMaster) => {
-    if (isMaster) {
-      successHandler(monitor, start, isMaster);
+  checkServer(monitor, e0 => {
+    if (e0 == null) {
+      rescheduleMonitoring(monitor);
       return;
     }
 
     // otherwise an error occured on initial discovery, also bail
     if (monitor[kServer].description.type === ServerType.Unknown) {
-      failureHandler(monitor, start, e0);
+      monitor.emit('resetServer', e0);
+      rescheduleMonitoring(monitor);
       return;
     }
 
@@ -199,13 +219,12 @@ function monitorServer(monitor) {
     // change its type to `Unknown` only after retrying once.
     monitor.emit('resetConnectionPool');
 
-    checkServer(monitor, (e1, isMaster) => {
+    checkServer(monitor, e1 => {
       if (e1) {
-        failureHandler(monitor, start, e1);
-        return;
+        monitor.emit('resetServer', e1);
       }
 
-      successHandler(monitor, start, isMaster);
+      rescheduleMonitoring(monitor);
     });
   });
 }
@@ -225,28 +244,6 @@ function rescheduleMonitoring(monitor, ms) {
   }, ms || heartbeatFrequencyMS);
 }
 
-function successHandler(monitor, start, isMaster) {
-  rescheduleMonitoring(monitor);
-
-  process.nextTick(() =>
-    monitor.emit(
-      'serverHeartbeatSucceeded',
-      new ServerHeartbeatSucceededEvent(calculateDurationInMs(start), isMaster, monitor.address)
-    )
-  );
-}
-
-function failureHandler(monitor, start, err) {
-  rescheduleMonitoring(monitor);
-
-  process.nextTick(() =>
-    monitor.emit(
-      'serverHeartbeatFailed',
-      new ServerHeartbeatFailedEvent(calculateDurationInMs(start), err, monitor.address)
-    )
-  );
-}
-
 module.exports = {
   Monitor
 };

lib/core/sdam/server.js

@@ -136,15 +136,12 @@ class Server extends EventEmitter {
       this.s.pool.clear();
     });
 
-    this[kMonitor].on('serverHeartbeatFailed', event => {
+    this[kMonitor].on('resetServer', error => {
       // Revert to an `Unknown` state by emitting a default description with no isMaster, and the
       // error from the heartbeat attempt
       this.emit(
         'descriptionReceived',
-        new ServerDescription(this.description.address, null, {
-          roundTripTime: calculateRoundTripTime(this.description.roundTripTime, event.duration),
-          error: event.failure
-        })
+        new ServerDescription(this.description.address, null, { error })
       );
     });
 

lib/core/sdam/server_selection.js

@@ -299,7 +299,7 @@ function selectServers(topology, selector, timeout, start, callback) {
 
   const retrySelection = () => {
     // ensure all server monitors attempt monitoring soon
-    topology.s.servers.forEach(server => server.requestCheck());
+    topology.s.servers.forEach(server => process.nextTick(() => server.requestCheck()));
 
     const iterationTimer = setTimeout(() => {
       topology.removeListener('topologyDescriptionChanged', descriptionChangedHandler);

test/unit/sdam/monitoring.test.js

@@ -237,9 +237,13 @@ describe('monitoring', function() {
       });
       this.defer(() => monitor.close());
 
-      monitor.on('serverHeartbeatFailed', () => done(new Error('unexpected heartbeat failure')));
-
       let resetRequested = false;
+      monitor.on('serverHeartbeatFailed', () => {
+        if (resetRequested) {
+          done(new Error('unexpected heartbeat failure'));
+        }
+      });
+
       monitor.on('resetConnectionPool', () => (resetRequested = true));
       monitor.on('serverHeartbeatSucceeded', () => {
         if (server.description.type === ServerType.Unknown) {
@@ -272,9 +276,11 @@ describe('monitoring', function() {
           if (failedCount === 0) {
             failedCount++;
             request.reply({ ok: 0, errmsg: 'first error message' });
-          } else {
+          } else if (failedCount === 1) {
             failedCount++;
             request.reply({ ok: 0, errmsg: 'second error message' });
+          } else {
+            request.reply(mock.DEFAULT_ISMASTER_36);
           }
         }
       });
@@ -288,22 +294,89 @@ describe('monitoring', function() {
 
       let resetRequested = false;
       monitor.on('resetConnectionPool', () => (resetRequested = true));
-      monitor.on('serverHeartbeatSucceeded', () => {
-        if (server.description.type === ServerType.Unknown) {
-          // this is the first successful heartbeat, set the server type
-          server.description.type = ServerType.Standalone;
-          return;
+      monitor.once('serverHeartbeatSucceeded', () => {
+        // this is the first successful heartbeat, set the server type
+        server.description.type = ServerType.Standalone;
+
+        let failureCount = 0;
+        monitor.on('serverHeartbeatFailed', event => {
+          failureCount++;
+          if (failureCount === 2) {
+            expect(resetRequested).to.be.true;
+            expect(event)
+              .property('failure')
+              .to.match(/second error message/);
+            done();
+          }
+        });
+      });
+
+      monitor.connect();
+    });
+
+    it('should report events in the correct order during monitoring failure', function(done) {
+      let failedCount = 0;
+      let initialConnectCompleted = false;
+      mockServer.setMessageHandler(request => {
+        const doc = request.document;
+        if (doc.ismaster) {
+          if (!initialConnectCompleted) {
+            request.reply(mock.DEFAULT_ISMASTER_36);
+            initialConnectCompleted = true;
+            return;
+          }
+
+          if (failedCount === 0) {
+            failedCount++;
+            request.reply({ ok: 0, errmsg: 'first error message' });
+          } else {
+            failedCount++;
+            request.reply({ ok: 0, errmsg: 'second error message' });
+          }
         }
+      });
 
-        done(new Error('unexpected heartbeat success'));
+      const server = new MockServer(mockServer.address());
+      const monitor = new Monitor(server, {
+        heartbeatFrequencyMS: 250,
+        minHeartbeatFrequencyMS: 50
       });
+      this.defer(() => monitor.close());
 
-      monitor.on('serverHeartbeatFailed', event => {
-        expect(resetRequested).to.be.true;
-        expect(event)
-          .property('failure')
-          .to.match(/second error message/);
-        done();
+      let poolResetRequested = false;
+      let serverResetRequested = false;
+      monitor.on('resetConnectionPool', () => (poolResetRequested = true));
+      monitor.on('resetServer', () => (serverResetRequested = true));
+
+      const events = [];
+      monitor.once('serverHeartbeatSucceeded', () => {
+        // this is the first successful heartbeat, set the server type
+        server.description.type = ServerType.Standalone;
+
+        monitor.on('serverHeartbeatStarted', event => events.push(event));
+        monitor.on('serverHeartbeatFailed', event => events.push(event));
+        monitor.once('resetServer', err => {
+          expect(poolResetRequested).to.be.true;
+          expect(serverResetRequested).to.be.true;
+          expect(events.map(e => e.constructor.name)).to.eql([
+            'ServerHeartbeatStartedEvent',
+            'ServerHeartbeatFailedEvent',
+            'ServerHeartbeatStartedEvent',
+            'ServerHeartbeatFailedEvent'
+          ]);
+
+          expect(events[1])
+            .property('failure')
+            .to.match(/first error message/);
+          expect(events[3])
+            .property('failure')
+            .to.match(/second error message/);
+          expect(events[3])
+            .property('failure')
+            .to.eql(err);
+
+          done();
+        });
       });
 
       monitor.connect();