refactor(topology): use a wait queue for server selection

The server selection loop is better modeled as a queue of requests
rather than manually rescheduling timeouts. This makes the loop
more readable and eliminates dependencies on SDAM events.

NODE-2398

Author: mbroadst

lib/core/sdam/server_selection.js

@@ -3,12 +3,6 @@ const ServerType = require('./common').ServerType;
 const TopologyType = require('./common').TopologyType;
 const ReadPreference = require('../topologies/read_preference');
 const MongoError = require('../error').MongoError;
-const calculateDurationInMs = require('../utils').calculateDurationInMs;
-const MongoServerSelectionError = require('../error').MongoServerSelectionError;
-
-const common = require('./common');
-const STATE_CLOSED = common.STATE_CLOSED;
-const clearAndRemoveTimerFrom = common.clearAndRemoveTimerFrom;
 
 // max staleness constants
 const IDLE_WRITE_PERIOD = 10000;
@@ -246,90 +240,7 @@ function readPreferenceServerSelector(readPreference) {
   };
 }
 
-/**
- * Selects servers using the provided selector
- *
- * @private
- * @param {Topology} topology The topology to select servers from
- * @param {function} selector The predicate used for selecting servers
- * @param {Number} timeout The max time we are willing wait for selection
- * @param {Number} start A high precision timestamp for the start of the selection process
- * @param {function} callback The callback used to convey errors or the resultant servers
- */
-function selectServers(topology, selector, timeout, start, callback) {
-  const duration = calculateDurationInMs(start);
-  if (duration >= timeout) {
-    return callback(
-      new MongoServerSelectionError(`Server selection timed out after ${timeout} ms`),
-      topology.description
-    );
-  }
-
-  // explicitly disallow selection if client is closed
-  if (topology.s.state === STATE_CLOSED) {
-    callback(new MongoError('Topology is closed, please connect'));
-    return;
-  }
-
-  // otherwise, attempt server selection
-  const serverDescriptions = Array.from(topology.description.servers.values());
-  let descriptions;
-
-  // support server selection by options with readPreference
-  if (typeof selector === 'object') {
-    const readPreference = selector.readPreference
-      ? selector.readPreference
-      : ReadPreference.primary;
-
-    selector = readPreferenceServerSelector(readPreference);
-  }
-
-  try {
-    descriptions = selector
-      ? selector(topology.description, serverDescriptions)
-      : serverDescriptions;
-  } catch (e) {
-    return callback(e, null);
-  }
-
-  if (descriptions.length) {
-    const servers = descriptions.map(description => topology.s.servers.get(description.address));
-    return callback(null, servers);
-  }
-
-  const retrySelection = () => {
-    // ensure all server monitors attempt monitoring soon
-    topology.s.servers.forEach(server => process.nextTick(() => server.requestCheck()));
-
-    const iterationTimer = setTimeout(() => {
-      topology.removeListener('topologyDescriptionChanged', descriptionChangedHandler);
-      callback(
-        new MongoServerSelectionError(
-          `Server selection timed out after ${timeout} ms`,
-          topology.description
-        )
-      );
-    }, timeout - duration);
-
-    const descriptionChangedHandler = () => {
-      // successful iteration, clear the check timer
-      clearAndRemoveTimerFrom(iterationTimer, topology.s.iterationTimers);
-
-      // topology description has changed due to monitoring, reattempt server selection
-      selectServers(topology, selector, timeout, start, callback);
-    };
-
-    // track this timer in case we need to clean it up outside this loop
-    topology.s.iterationTimers.add(iterationTimer);
-
-    topology.once('topologyDescriptionChanged', descriptionChangedHandler);
-  };
-
-  retrySelection();
-}
-
 module.exports = {
-  selectServers,
   writableServerSelector,
   readPreferenceServerSelector
 };

lib/core/sdam/topology.js

@@ -1,4 +1,5 @@
 'use strict';
+const Denque = require('denque');
 const EventEmitter = require('events');
 const ServerDescription = require('./server_description').ServerDescription;
 const ServerType = require('./common').ServerType;
@@ -18,6 +19,7 @@ const isNodeShuttingDownError = require('../error').isNodeShuttingDownError;
 const maxWireVersion = require('../utils').maxWireVersion;
 const ClientSession = require('../sessions').ClientSession;
 const MongoError = require('../error').MongoError;
+const MongoServerSelectionError = require('../error').MongoServerSelectionError;
 const resolveClusterTime = require('../topologies/shared').resolveClusterTime;
 const SrvPoller = require('./srv_polling').SrvPoller;
 const getMMAPError = require('../topologies/shared').getMMAPError;
@@ -35,7 +37,7 @@ const clearAndRemoveTimerFrom = common.clearAndRemoveTimerFrom;
 const serverSelection = require('./server_selection');
 const readPreferenceServerSelector = serverSelection.readPreferenceServerSelector;
 const writableServerSelector = serverSelection.writableServerSelector;
-const selectServers = serverSelection.selectServers;
+// const selectServers = serverSelection.selectServers;
 
 // Global state
 let globalTopologyCounter = 0;
@@ -74,6 +76,9 @@ const DEPRECATED_OPTIONS = new Set([
   'bufferMaxEntries'
 ]);
 
+const kCancelled = Symbol('cancelled');
+const kWaitQueue = Symbol('waitQueue');
+
 /**
  * A container of server instances representing a connection to a MongoDB topology.
  *
@@ -140,6 +145,7 @@ class Topology extends EventEmitter {
       return result;
     }, new Map());
 
+    this[kWaitQueue] = new Denque();
     this.s = {
       // the id of this topology
       id: topologyId,
@@ -195,7 +201,6 @@ class Topology extends EventEmitter {
       clusterTime: null,
 
       // timer management
-      iterationTimers: new Set(),
       connectionTimers: new Set()
     };
 
@@ -335,8 +340,7 @@ class Topology extends EventEmitter {
       return;
     }
 
-    // clear all existing monitor timers
-    drainTimerQueue(this.s.iterationTimers);
+    drainWaitQueue(this[kWaitQueue], new MongoError('Topology closed'));
     drainTimerQueue(this.s.connectionTimers);
 
     if (this.s.srvPoller) {
@@ -416,26 +420,43 @@ class Topology extends EventEmitter {
     const transaction = session && session.transaction;
 
     if (isSharded && transaction && transaction.server) {
-      callback(null, transaction.server);
+      callback(undefined, transaction.server);
       return;
     }
 
-    selectServers(
-      this,
-      selector,
-      options.serverSelectionTimeoutMS,
-      process.hrtime(),
-      (err, servers) => {
-        if (err) return callback(err);
-
-        const selectedServer = randomSelection(servers);
-        if (isSharded && transaction && transaction.isActive) {
-          transaction.pinServer(selectedServer);
-        }
+    // support server selection by options with readPreference
+    let serverSelector = selector;
+    if (typeof selector === 'object') {
+      const readPreference = selector.readPreference
+        ? selector.readPreference
+        : ReadPreference.primary;
 
-        callback(null, selectedServer);
-      }
-    );
+      serverSelector = readPreferenceServerSelector(readPreference);
+    }
+
+    const waitQueueMember = {
+      serverSelector,
+      transaction,
+      callback
+    };
+
+    const serverSelectionTimeoutMS = options.serverSelectionTimeoutMS;
+    if (serverSelectionTimeoutMS) {
+      waitQueueMember.timer = setTimeout(() => {
+        waitQueueMember[kCancelled] = true;
+        waitQueueMember.timer = undefined;
+        const timeoutError = new MongoServerSelectionError(
+          `Server selection timed out after ${serverSelectionTimeoutMS} ms`,
+          this.description
+        );
+
+        waitQueueMember.callback(timeoutError);
+      }, serverSelectionTimeoutMS);
+    }
+
+    // place the member at the front of the wait queue
+    this[kWaitQueue].unshift(waitQueueMember);
+    processWaitQueue(this);
   }
 
   // Sessions related methods
@@ -545,6 +566,11 @@ class Topology extends EventEmitter {
     // update server list from updated descriptions
     updateServers(this, serverDescription);
 
+    // attempt to resolve any outstanding server selection attempts
+    if (this[kWaitQueue].length > 0) {
+      processWaitQueue(this);
+    }
+
     this.emit(
       'topologyDescriptionChanged',
       new events.TopologyDescriptionChangedEvent(
@@ -1012,6 +1038,64 @@ function srvPollingHandler(topology) {
   };
 }
 
+function drainWaitQueue(queue, err) {
+  while (queue.length) {
+    const waitQueueMember = queue.pop();
+    clearTimeout(waitQueueMember.timer);
+    if (!waitQueueMember[kCancelled]) {
+      waitQueueMember.callback(err);
+    }
+  }
+}
+
+function processWaitQueue(topology) {
+  if (topology.s.state === STATE_CLOSED) {
+    drainWaitQueue(topology[kWaitQueue], new MongoError('Topology is closed, please connect'));
+    return;
+  }
+
+  const isSharded = topology.description.type === TopologyType.Sharded;
+  const serverDescriptions = Array.from(topology.description.servers.values());
+  for (let i = 0; i < topology[kWaitQueue].length; ++i) {
+    const waitQueueMember = topology[kWaitQueue].shift();
+    if (waitQueueMember[kCancelled]) {
+      continue;
+    }
+
+    let selectedDescriptions;
+    try {
+      const serverSelector = waitQueueMember.serverSelector;
+      selectedDescriptions = serverSelector
+        ? serverSelector(topology.description, serverDescriptions)
+        : serverDescriptions;
+    } catch (e) {
+      clearTimeout(waitQueueMember.timer);
+      waitQueueMember.callback(e);
+      break;
+    }
+
+    if (selectedDescriptions.length === 0) {
+      topology[kWaitQueue].push(waitQueueMember);
+      break;
+    }
+
+    const selectedServerDescription = randomSelection(selectedDescriptions);
+    const selectedServer = topology.s.servers.get(selectedServerDescription.address);
+    const transaction = waitQueueMember.transaction;
+    if (isSharded && transaction && transaction.isActive) {
+      transaction.pinServer(selectedServer);
+    }
+
+    clearTimeout(waitQueueMember.timer);
+    waitQueueMember.callback(undefined, selectedServer);
+  }
+
+  if (topology[kWaitQueue].length > 0) {
+    // ensure all server monitors attempt monitoring soon
+    topology.s.servers.forEach(server => process.nextTick(() => server.requestCheck()));
+  }
+}
+
 /**
  * A server opening SDAM monitoring event
  *

test/functional/scram_sha_256.test.js

@@ -180,7 +180,7 @@ describe('SCRAM-SHA-256 auth', function() {
         },
         authSource: this.configuration.db,
         authMechanism: 'SCRAM-SHA-1',
-        serverSelectionTimeoutMS: 2000
+        serverSelectionTimeoutMS: 100
       };
 
       return withClient(
@@ -204,7 +204,7 @@ describe('SCRAM-SHA-256 auth', function() {
           password: 'pencil'
         },
         authSource: 'admin',
-        serverSelectionTimeoutMS: 2000
+        serverSelectionTimeoutMS: 100
       };
 
       const badPasswordOptions = {
@@ -213,7 +213,7 @@ describe('SCRAM-SHA-256 auth', function() {
           password: 'pencil'
         },
         authSource: 'admin',
-        serverSelectionTimeoutMS: 2000
+        serverSelectionTimeoutMS: 100
       };
 
       const getErrorMsg = options =>