fix(pool): support a drain event for use with unified topology

Until we integrate the new connection pool conforming to the CMAP
specification, any type of error/close/timeout coming from a pool
signals pool invalidation. If the pool is configured for unified
then it should invalidate the server by emitting a `drain` event
which is translated to an `error in the Server instance, causing
the topology to cycle the server.

NODE-2332

Author: mbroadst

lib/core/connection/pool.js

@@ -256,6 +256,15 @@ function connectionFailureHandler(pool, event, err, conn) {
       const workItem = conn.workItems.shift();
       if (workItem.cb) workItem.cb(err);
     }
+
+    if (pool.state !== DRAINING && pool.options.legacyCompatMode === false) {
+      // since an error/close/timeout means pool invalidation in a
+      // pre-CMAP world, we will issue a custom `drain` event here to
+      // signal that the server should be recycled
+      stateTransition(pool, DRAINING);
+      pool.emit('drain', err);
+      return;
+    }
   }
 
   // Did we catch a timeout, increment the numberOfConsecutiveTimeouts

lib/core/error.js

@@ -207,7 +207,9 @@ function isNodeShuttingDownError(err) {
  * @param {Server} server
  */
 function isSDAMUnrecoverableError(error, server) {
-  if (error instanceof MongoParseError) {
+  // NOTE: null check is here for a strictly pre-CMAP world, a timeout or
+  //       close event are considered unrecoverable
+  if (error instanceof MongoParseError || error == null) {
     return true;
   }
 

lib/core/sdam/server.js

@@ -158,6 +158,10 @@ class Server extends EventEmitter {
     // setup listeners
     this.s.pool.on('parseError', parseErrorEventHandler(this));
 
+    this.s.pool.on('drain', err => {
+      this.emit('error', err);
+    });
+
     // it is unclear whether consumers should even know about these events
     // this.s.pool.on('timeout', timeoutEventHandler(this));
     // this.s.pool.on('reconnect', reconnectEventHandler(this));

test/functional/spec-runner/index.js

@@ -221,7 +221,7 @@ function parseSessionOptions(options) {
   return result;
 }
 
-const IGNORED_COMMANDS = new Set(['ismaster', 'configureFailPoint']);
+const IGNORED_COMMANDS = new Set(['ismaster', 'configureFailPoint', 'endSessions']);
 
 let displayCommands = false;
 function runTestSuiteTest(configuration, spec, context) {