NODE-2274: unified failover server loss

lib/core/connection/pool.js

@@ -25,12 +25,14 @@ const makeStateMachine = require('../utils').makeStateMachine;
 const DISCONNECTED = 'disconnected';
 const CONNECTING = 'connecting';
 const CONNECTED = 'connected';
+const DRAINING = 'draining';
 const DESTROYING = 'destroying';
 const DESTROYED = 'destroyed';
 const stateTransition = makeStateMachine({
-  [DISCONNECTED]: [CONNECTING, DESTROYING, DISCONNECTED],
-  [CONNECTING]: [CONNECTING, DESTROYING, CONNECTED, DISCONNECTED],
-  [CONNECTED]: [CONNECTED, DISCONNECTED, DESTROYING],
+  [DISCONNECTED]: [CONNECTING, DRAINING, DISCONNECTED],
+  [CONNECTING]: [CONNECTING, CONNECTED, DRAINING, DISCONNECTED],
+  [CONNECTED]: [CONNECTED, DISCONNECTED, DRAINING],
+  [DRAINING]: [DRAINING, DESTROYING, DESTROYED],
   [DESTROYING]: [DESTROYING, DESTROYED],
   [DESTROYED]: [DESTROYED]
 });
@@ -239,7 +241,10 @@ function resetPoolState(pool) {
 
 function connectionFailureHandler(pool, event, err, conn) {
   if (conn) {
-    if (conn._connectionFailHandled) return;
+    if (conn._connectionFailHandled) {
+      return;
+    }
+
     conn._connectionFailHandled = true;
     conn.destroy();
 
@@ -270,8 +275,10 @@ function connectionFailureHandler(pool, event, err, conn) {
 
   // No more socket available propegate the event
   if (pool.socketCount() === 0) {
-    if (pool.state !== DESTROYED && pool.state !== DESTROYING) {
-      stateTransition(pool, DISCONNECTED);
+    if (pool.state !== DESTROYED && pool.state !== DESTROYING && pool.state !== DRAINING) {
+      if (pool.options.reconnect) {
+        stateTransition(pool, DISCONNECTED);
+      }
     }
 
     // Do not emit error events, they are always close events
@@ -426,7 +433,7 @@ function messageHandler(self) {
           updateSessionFromResponse(session, document);
         }
 
-        if (document.$clusterTime) {
+        if (self.topology && document.$clusterTime) {
           self.topology.clusterTime = document.$clusterTime;
         }
       }
@@ -537,14 +544,20 @@ Pool.prototype.isDisconnected = function() {
 /**
  * Connect pool
  */
-Pool.prototype.connect = function() {
+Pool.prototype.connect = function(callback) {
   if (this.state !== DISCONNECTED) {
     throw new MongoError('connection in unlawful state ' + this.state);
   }
 
   stateTransition(this, CONNECTING);
   createConnection(this, (err, conn) => {
     if (err) {
+      if (typeof callback === 'function') {
+        this.destroy();
+        callback(err);
+        return;
+      }
+
       if (this.state === CONNECTING) {
         this.emit('error', err);
       }
@@ -554,14 +567,19 @@ Pool.prototype.connect = function() {
     }
 
     stateTransition(this, CONNECTED);
-    this.emit('connect', this, conn);
 
     // create min connections
     if (this.minSize) {
       for (let i = 0; i < this.minSize; i++) {
         createConnection(this);
       }
     }
+
+    if (typeof callback === 'function') {
+      callback(null, conn);
+    } else {
+      this.emit('connect', this, conn);
+    }
   });
 };
 
@@ -596,6 +614,8 @@ Pool.prototype.unref = function() {
 
 // Destroy the connections
 function destroy(self, connections, options, callback) {
+  stateTransition(self, DESTROYING);
+
   eachAsync(
     connections,
     (conn, cb) => {
@@ -626,14 +646,19 @@ function destroy(self, connections, options, callback) {
  */
 Pool.prototype.destroy = function(force, callback) {
   var self = this;
+  if (typeof force === 'function') {
+    callback = force;
+    force = false;
+  }
+
   // Do not try again if the pool is already dead
   if (this.state === DESTROYED || self.state === DESTROYING) {
     if (typeof callback === 'function') callback(null, null);
     return;
   }
 
-  // Set state to destroyed
-  stateTransition(this, DESTROYING);
+  // Set state to draining
+  stateTransition(this, DRAINING);
 
   // Are we force closing
   if (force) {
@@ -660,6 +685,14 @@ Pool.prototype.destroy = function(force, callback) {
 
   // Wait for the operations to drain before we close the pool
   function checkStatus() {
+    if (self.state === DESTROYED || self.state === DESTROYING) {
+      if (typeof callback === 'function') {
+        callback();
+      }
+
+      return;
+    }
+
     flushMonitoringOperations(self.queue);
 
     if (self.queue.length === 0) {
@@ -676,7 +709,6 @@ Pool.prototype.destroy = function(force, callback) {
       }
 
       destroy(self, connections, { force: false }, callback);
-      // } else if (self.queue.length > 0 && !this.reconnectId) {
     } else {
       // Ensure we empty the queue
       _execute(self)();
@@ -695,6 +727,14 @@ Pool.prototype.destroy = function(force, callback) {
  * @param {function} [callback]
  */
 Pool.prototype.reset = function(callback) {
+  if (this.s.state !== CONNECTED) {
+    if (typeof callback === 'function') {
+      callback(new MongoError('pool is not connected, reset aborted'));
+    }
+
+    return;
+  }
+
   const connections = this.availableConnections.concat(this.inUseConnections);
   eachAsync(
     connections,
@@ -715,12 +755,12 @@ Pool.prototype.reset = function(callback) {
 
       resetPoolState(this);
 
-      // create an initial connection, and kick off execution again
-      createConnection(this);
-
-      if (typeof callback === 'function') {
-        callback(null, null);
-      }
+      // create a new connection, this will ultimately trigger execution
+      createConnection(this, () => {
+        if (typeof callback === 'function') {
+          callback(null, null);
+        }
+      });
     }
   );
 };
@@ -788,17 +828,12 @@ Pool.prototype.write = function(command, options, cb) {
 
   // Pool was destroyed error out
   if (this.state === DESTROYED || this.state === DESTROYING) {
-    // Callback with an error
-    if (cb) {
-      try {
-        cb(new MongoError('pool destroyed'));
-      } catch (err) {
-        process.nextTick(function() {
-          throw err;
-        });
-      }
-    }
+    cb(new MongoError('pool destroyed'));
+    return;
+  }
 
+  if (this.state === DRAINING) {
+    cb(new MongoError('pool is draining, new operations prohibited'));
     return;
   }
 
@@ -846,10 +881,6 @@ Pool.prototype.write = function(command, options, cb) {
   // Optional per operation socketTimeout
   operation.socketTimeout = options.socketTimeout;
   operation.monitoring = options.monitoring;
-  // Custom socket Timeout
-  if (options.socketTimeout) {
-    operation.socketTimeout = options.socketTimeout;
-  }
 
   // Get the requestId
   operation.requestId = command.requestId;
@@ -948,15 +979,6 @@ function createConnection(pool, callback) {
         pool.logger.debug(`connection attempt failed with error [${JSON.stringify(err)}]`);
       }
 
-      if (pool.options.legacyCompatMode === false) {
-        // The unified topology uses the reported `error` from a pool to track what error
-        // reason is returned to the user during selection timeout. We only want to emit
-        // this if the pool is active because the listeners are removed on destruction.
-        if (pool.state !== DESTROYED && pool.state !== DESTROYING) {
-          pool.emit('error', err);
-        }
-      }
-
       // check if reconnect is enabled, and attempt retry if so
       if (!pool.reconnectId && pool.options.reconnect) {
         if (pool.state === CONNECTING && pool.options.legacyCompatMode) {
@@ -1047,6 +1069,12 @@ function _execute(self) {
       if (self.availableConnections.length === 0) {
         // Flush any monitoring operations
         flushMonitoringOperations(self.queue);
+
+        // Try to create a new connection to execute stuck operation
+        if (totalConnections < self.options.size && self.queue.length > 0) {
+          createConnection(self);
+        }
+
         break;
       }
 
@@ -1111,10 +1139,7 @@ function _execute(self) {
           }
 
           // Re-execute the operation
-          setTimeout(function() {
-            _execute(self)();
-          }, 10);
-
+          setTimeout(() => _execute(self)(), 10);
           break;
         }
       }

lib/core/sdam/monitoring.js

@@ -4,8 +4,8 @@ const ServerDescription = require('./server_description').ServerDescription;
 const calculateDurationInMs = require('../utils').calculateDurationInMs;
 
 // pulled from `Server` implementation
-const STATE_DISCONNECTED = 'disconnected';
-const STATE_DISCONNECTING = 'disconnecting';
+const STATE_CLOSED = 'closed';
+const STATE_CLOSING = 'closing';
 
 /**
  * Published when server description changes, but does NOT include changes to the RTT.
@@ -180,7 +180,7 @@ function monitorServer(server, options) {
 
     // emit an event indicating that our description has changed
     server.emit('descriptionReceived', new ServerDescription(server.description.address, isMaster));
-    if (server.s.state === STATE_DISCONNECTED || server.s.state === STATE_DISCONNECTING) {
+    if (server.s.state === STATE_CLOSED || server.s.state === STATE_CLOSING) {
       return;
     }