done

Author: baileympearson

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
@@ -231,10 +249,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
@@ -258,6 +273,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
  *
@@ -271,7 +298,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);
@@ -286,38 +313,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/unit/assorted/server_selection.spec.test.ts

@@ -11,7 +11,7 @@ import {
   runServerSelectionLogicTest
 } from './server_selection_logic_spec_utils';
 
-describe('Server Selection Logic (spec)', function () {
+describe.only('Server Selection Logic (spec)', function () {
   beforeEach(function () {
     if (this.currentTest.title.match(/Possible/)) {
       this.currentTest.skipReason = 'Nodejs driver does not support PossiblePrimary';

test/unit/assorted/server_selection_logic_spec_utils.ts

@@ -12,6 +12,7 @@ import { type ServerType, type TopologyType } from '../../../src/sdam/common';
 import { type ServerDescription, type TagSet } from '../../../src/sdam/server_description';
 import {
   readPreferenceServerSelector,
+  type ServerSelector,
   writableServerSelector
 } from '../../../src/sdam/server_selection';
 import { TopologyDescription } from '../../../src/sdam/topology_description';
@@ -41,6 +42,7 @@ interface ServerSelectionLogicTest {
    */
   suitable_servers: never;
   in_latency_window: ServerSelectionLogicTestServer[];
+  deprioritized_servers?: ServerSelectionLogicTestServer[];
 }
 
 function readPreferenceFromDefinition(definition) {
@@ -96,8 +98,11 @@ export function runServerSelectionLogicTest(testDefinition: ServerSelectionLogic
   const expectedServers = serverDescriptionsToMap(
     testDefinition.in_latency_window.map(s => serverDescriptionFromDefinition(s))
   );
+  const deprioritized =
+    testDefinition.deprioritized_servers?.map(s => serverDescriptionFromDefinition(s, allHosts)) ??
+    [];
 
-  let selector;
+  let selector: ServerSelector;
   if (testDefinition.operation === 'write') {
     selector = writableServerSelector();
   } else if (testDefinition.operation === 'read' || testDefinition.read_preference) {
@@ -107,10 +112,11 @@ export function runServerSelectionLogicTest(testDefinition: ServerSelectionLogic
     expect.fail('test operation was neither read nor write, and no read preference was provided.');
   }
 
-  const result = selector(topologyDescription, serversInTopology);
+  const result = selector(topologyDescription, serversInTopology, deprioritized);
 
   expect(result.length).to.equal(expectedServers.size);
 
+  // console.error({ result, expectedServers });
   for (const server of result) {
     const expectedServer = expectedServers.get(server.address);
     expect(expectedServer).to.exist;