Initial POC - all tests passing but using full server descriptions for change tracking.

Author: baileympearson

src/change_stream.ts

@@ -1073,7 +1073,8 @@ export class ChangeStream<
     try {
       await topology.selectServer(this.cursor.readPreference, {
         operationName: 'reconnect topology in change stream',
-        timeoutContext: this.timeoutContext
+        timeoutContext: this.timeoutContext,
+        deprioritizedServers: []
       });
       this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
     } catch {

src/mongo_client.ts

@@ -789,7 +789,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
       // to avoid the server selection timeout.
       const selector = readPreferenceServerSelector(ReadPreference.primaryPreferred);
       const serverDescriptions = Array.from(topologyDescription.servers.values());
-      const servers = selector(topologyDescription, serverDescriptions);
+      const servers = selector(topologyDescription, serverDescriptions, []);
       if (servers.length !== 0) {
         const endSessions = Array.from(client.s.sessionPool.sessions, ({ id }) => id);
         if (endSessions.length !== 0) {

src/operations/execute_operation.ts

@@ -207,7 +207,8 @@ async function tryOperation<T extends AbstractOperation, TResult = ResultTypeFro
     session,
     operationName: operation.commandName,
     timeoutContext,
-    signal: operation.options.signal
+    signal: operation.options.signal,
+    deprioritizedServers: []
   });
 
   const hasReadAspect = operation.hasAspect(Aspect.READ_OPERATION);
@@ -234,7 +235,7 @@ async function tryOperation<T extends AbstractOperation, TResult = ResultTypeFro
 
   const maxTries = willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1;
   let previousOperationError: MongoError | undefined;
-  let previousServer: ServerDescription | undefined;
+  const deprioritizedServers: ServerDescription[] = [];
 
   for (let tries = 0; tries < maxTries; tries++) {
     if (previousOperationError) {
@@ -270,7 +271,7 @@ async function tryOperation<T extends AbstractOperation, TResult = ResultTypeFro
       server = await topology.selectServer(selector, {
         session,
         operationName: operation.commandName,
-        previousServer,
+        deprioritizedServers,
         signal: operation.options.signal
       });
 
@@ -303,7 +304,7 @@ async function tryOperation<T extends AbstractOperation, TResult = ResultTypeFro
       ) {
         throw previousOperationError;
       }
-      previousServer = server.description;
+      deprioritizedServers.push(server.description);
       previousOperationError = operationError;
 
       // Reset timeouts

src/sdam/server_selection.ts

@@ -15,21 +15,38 @@ export const MIN_SECONDARY_WRITE_WIRE_VERSION = 13;
 export type ServerSelector = (
   topologyDescription: TopologyDescription,
   servers: ServerDescription[],
-  deprioritized?: ServerDescription[]
+  deprioritized: ServerDescription[]
 ) => ServerDescription[];
 
+function sdEquals(a: ServerDescription, b: ServerDescription) {
+  return a.address === b.address && a.equals(b);
+}
+function filterDeprioritized(
+  candidates: ServerDescription[],
+  deprioritized: ServerDescription[]
+): ServerDescription[] {
+  const filtered = candidates.filter(
+    candidate => !deprioritized.some(serverDescription => sdEquals(serverDescription, candidate))
+  );
+
+  return filtered.length ? filtered : candidates;
+}
+
 /**
  * Returns a server selector that selects for writable servers
  */
 export function writableServerSelector(): ServerSelector {
   return function writableServer(
     topologyDescription: TopologyDescription,
-    servers: ServerDescription[]
+    servers: ServerDescription[],
+    deprioritized: ServerDescription[]
   ): ServerDescription[] {
-    return latencyWindowReducer(
-      topologyDescription,
-      servers.filter((s: ServerDescription) => s.isWritable)
+    const eligibleServers = filterDeprioritized(
+      servers.filter(({ isWritable }) => isWritable),
+      deprioritized
     );
+
+    return latencyWindowReducer(topologyDescription, eligibleServers);
   };
 }
 
@@ -39,8 +56,9 @@ export function writableServerSelector(): ServerSelector {
  */
 export function sameServerSelector(description?: ServerDescription): ServerSelector {
   return function sameServerSelector(
-    topologyDescription: TopologyDescription,
-    servers: ServerDescription[]
+    _topologyDescription: TopologyDescription,
+    servers: ServerDescription[],
+    _deprioritized: ServerDescription[]
   ): ServerDescription[] {
     if (!description) return [];
     // Filter the servers to match the provided description only if
@@ -218,10 +236,7 @@ function latencyWindowReducer(
   );
 
   const high = low + topologyDescription.localThresholdMS;
-  return servers.reduce((result: ServerDescription[], server: ServerDescription) => {
-    if (server.roundTripTime <= high && server.roundTripTime >= low) result.push(server);
-    return result;
-  }, []);
+  return servers.filter(server => server.roundTripTime <= high && server.roundTripTime >= low);
 }
 
 // filters
@@ -245,6 +260,18 @@ function loadBalancerFilter(server: ServerDescription): boolean {
   return server.type === ServerType.LoadBalancer;
 }
 
+function isDeprioritizedFactory(
+  deprioritized: ServerDescription[]
+): (server: ServerDescription) => boolean {
+  return server =>
+    // if any deprioritized servers equal the server, here we are.
+    !deprioritized.some(deprioritizedServer => {
+      const result = sdEquals(deprioritizedServer, server);
+      // console.error(result);
+      return result;
+    });
+}
+
 /**
  * Returns a function which selects servers based on a provided read preference
  *
@@ -258,7 +285,7 @@ export function readPreferenceServerSelector(readPreference: ReadPreference): Se
   return function readPreferenceServers(
     topologyDescription: TopologyDescription,
     servers: ServerDescription[],
-    deprioritized: ServerDescription[] = []
+    deprioritized: ServerDescription[]
   ): ServerDescription[] {
     if (topologyDescription.type === TopologyType.LoadBalanced) {
       return servers.filter(loadBalancerFilter);
@@ -273,38 +300,79 @@ export function readPreferenceServerSelector(readPreference: ReadPreference): Se
     }
 
     if (topologyDescription.type === TopologyType.Sharded) {
-      const filtered = servers.filter(server => {
-        return !deprioritized.includes(server);
-      });
-      const selectable = filtered.length > 0 ? filtered : deprioritized;
+      const selectable = filterDeprioritized(servers, deprioritized);
       return latencyWindowReducer(topologyDescription, selectable.filter(knownFilter));
     }
 
     const mode = readPreference.mode;
     if (mode === ReadPreference.PRIMARY) {
-      return servers.filter(primaryFilter);
+      return filterDeprioritized(servers.filter(primaryFilter), deprioritized);
     }
 
     if (mode === ReadPreference.PRIMARY_PREFERRED) {
-      const result = servers.filter(primaryFilter);
-      if (result.length) {
-        return result;
+      const primary = servers.filter(primaryFilter);
+
+      // If there is a primary and it is not deprioritized, use the primary.  Otherwise,
+      // check for secondaries.
+      const eligiblePrimary = primary.filter(isDeprioritizedFactory(deprioritized));
+      if (eligiblePrimary.length) {
+        return eligiblePrimary;
       }
-    }
 
-    const filter = mode === ReadPreference.NEAREST ? nearestFilter : secondaryFilter;
-    const selectedServers = latencyWindowReducer(
-      topologyDescription,
-      tagSetReducer(
+      const secondaries = tagSetReducer(
         readPreference,
-        maxStalenessReducer(readPreference, topologyDescription, servers.filter(filter))
-      )
+        maxStalenessReducer(readPreference, topologyDescription, servers.filter(secondaryFilter))
+      );
+      const deprioritizedSecondaries = secondaries.filter(isDeprioritizedFactory(deprioritized));
+
+      // console.error({ deprioritizedSecondaries, secondaries, deprioritized });
+      if (deprioritizedSecondaries.length)
+        return latencyWindowReducer(topologyDescription, deprioritizedSecondaries);
+
+      // if we make it here, we have no primaries or secondaries that not deprioritized.
+      // prefer the primary (which may not exist, if the topology has no primary).
+      // otherwise, return the secondaries (which also may not exist, but there is nothing else to check here).
+      return primary.length ? primary : latencyWindowReducer(topologyDescription, secondaries);
+    }
+
+    // TODO: should we be applying the latency window to nearest servers?
+    if (mode === 'nearest') {
+      // if read preference is nearest
+      return latencyWindowReducer(
+        topologyDescription,
+        filterDeprioritized(
+          tagSetReducer(
+            readPreference,
+            maxStalenessReducer(readPreference, topologyDescription, servers.filter(nearestFilter))
+          ),
+          deprioritized
+        )
+      );
+    }
+
+    const filter = secondaryFilter;
+
+    const secondaries = tagSetReducer(
+      readPreference,
+      maxStalenessReducer(readPreference, topologyDescription, servers.filter(filter))
     );
+    const eligibleSecondaries = secondaries.filter(isDeprioritizedFactory(deprioritized));
+
+    if (eligibleSecondaries.length)
+      return latencyWindowReducer(topologyDescription, eligibleSecondaries);
+
+    // we have no eligible secondaries, try for a primary.
+    if (mode === ReadPreference.SECONDARY_PREFERRED) {
+      const primary = servers.filter(primaryFilter);
+      const eligiblePrimary = primary.filter(isDeprioritizedFactory(deprioritized));
+
+      if (eligiblePrimary.length) return eligiblePrimary;
 
-    if (mode === ReadPreference.SECONDARY_PREFERRED && selectedServers.length === 0) {
-      return servers.filter(primaryFilter);
+      // we have no eligible primary nor secondaries that have not been deprioritized
+      return secondaries.length ? latencyWindowReducer(topologyDescription, secondaries) : primary;
     }
 
-    return selectedServers;
+    // return all secondaries in the latency window.
+    return latencyWindowReducer(topologyDescription, secondaries);
   };
 }

src/sdam/topology.ts

@@ -105,7 +105,7 @@ export interface ServerSelectionRequest {
   cancelled: boolean;
   operationName: string;
   waitingLogged: boolean;
-  previousServer?: ServerDescription;
+  deprioritizedServers: ServerDescription[];
 }
 
 /** @internal */
@@ -169,7 +169,7 @@ export interface SelectServerOptions {
   serverSelectionTimeoutMS?: number;
   session?: ClientSession;
   operationName: string;
-  previousServer?: ServerDescription;
+  deprioritizedServers: ServerDescription[];
   /**
    * @internal
    * TODO(NODE-6496): Make this required by making ChangeStream use LegacyTimeoutContext
@@ -455,7 +455,8 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     const selectServerOptions = {
       operationName: 'handshake',
       ...options,
-      timeoutContext
+      timeoutContext,
+      deprioritizedServers: []
     };
 
     try {
@@ -605,7 +606,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       startTime: processTimeMS(),
       operationName: options.operationName,
       waitingLogged: false,
-      previousServer: options.previousServer
+      deprioritizedServers: options.deprioritizedServers
     };
 
     const abortListener = addAbortListener(options.signal, function () {
@@ -957,13 +958,9 @@ function processWaitQueue(topology: Topology) {
     let selectedDescriptions;
     try {
       const serverSelector = waitQueueMember.serverSelector;
-      const previousServer = waitQueueMember.previousServer;
+      const deprioritizedServers = waitQueueMember.deprioritizedServers;
       selectedDescriptions = serverSelector
-        ? serverSelector(
-            topology.description,
-            serverDescriptions,
-            previousServer ? [previousServer] : []
-          )
+        ? serverSelector(topology.description, serverDescriptions, deprioritizedServers)
         : serverDescriptions;
     } catch (selectorError) {
       if (

test/spec/server-selection/server_selection/ReplicaSetNoPrimary/read/DeprioritizedNearest.json

@@ -0,0 +1,63 @@
+{
+  "topology_description": {
+    "type": "ReplicaSetNoPrimary",
+    "servers": [
+      {
+        "address": "b:27017",
+        "avg_rtt_ms": 5,
+        "type": "RSSecondary",
+        "tags": {
+          "data_center": "nyc"
+        }
+      },
+      {
+        "address": "c:27017",
+        "avg_rtt_ms": 100,
+        "type": "RSSecondary",
+        "tags": {
+          "data_center": "nyc"
+        }
+      }
+    ]
+  },
+  "operation": "read",
+  "read_preference": {
+    "mode": "Nearest",
+    "tag_sets": [
+      {
+        "data_center": "nyc"
+      }
+    ]
+  },
+  "deprioritized_servers": [
+    {
+      "address": "b:27017",
+      "avg_rtt_ms": 5,
+      "type": "RSSecondary",
+      "tags": {
+        "data_center": "nyc"
+      }
+    }
+  ],
+  "suitable_servers": [
+    {
+      "address": "c:27017",
+      "avg_rtt_ms": 100,
+      "type": "RSSecondary",
+      "tags": {
+        "data_center": "nyc"
+      }
+    }
+  ],
+  "in_latency_window": [
+    {
+      "address": "c:27017",
+      "avg_rtt_ms": 100,
+      "type": "RSSecondary",
+      "tags": {
+        "data_center": "nyc"
+      }
+    }
+  ]
+}
+

test/spec/server-selection/server_selection/ReplicaSetNoPrimary/read/DeprioritizedNearest.yml

@@ -0,0 +1,26 @@
+topology_description:
+  type: ReplicaSetNoPrimary
+  servers:
+  - &1
+    address: b:27017
+    avg_rtt_ms: 5
+    type: RSSecondary
+    tags:
+      data_center: nyc
+  - &2
+    address: c:27017
+    avg_rtt_ms: 100
+    type: RSSecondary
+    tags:
+      data_center: nyc
+operation: read
+read_preference:
+  mode: Nearest
+  tag_sets:
+  - data_center: nyc
+deprioritized_servers:
+- *1
+suitable_servers:
+- *2
+in_latency_window:
+- *2