feat(NODE-7932): add serverSelector option to target one member

src/connection_string.ts

@@ -345,6 +345,12 @@ export function parseOptions(
     );
   }
 
+  if (urlOptions.has('serverSelector')) {
+    throw new MongoParseError(
+      'URI cannot contain `serverSelector`, it can only be passed to the client'
+    );
+  }
+
   if (urlOptions.has('srvHostValidator')) {
     throw new MongoParseError(
       'URI cannot contain `srvHostValidator`, it can only be passed to the client'
@@ -1187,6 +1193,14 @@ export const OPTIONS = {
     default: 30000,
     type: 'uint'
   },
+  serverSelector: {
+    transform({ values: [value] }) {
+      if (typeof value !== 'function') {
+        throw new MongoParseError('serverSelector must be a function');
+      }
+      return value;
+    }
+  },
   servername: {
     type: 'string'
   },

src/index.ts

@@ -593,7 +593,11 @@ export type {
   TagSet,
   TopologyVersion
 } from './sdam/server_description';
-export type { DeprioritizedServers, ServerSelector } from './sdam/server_selection';
+export type {
+  DeprioritizedServers,
+  ServerSelector,
+  ServerSelectorFunction
+} from './sdam/server_selection';
 export type { SrvPoller, SrvPollerEvents, SrvPollerOptions } from './sdam/srv_polling';
 export type {
   ConnectOptions,

src/mongo_client.ts

@@ -49,7 +49,11 @@ import { ReadPreference, type ReadPreferenceMode } from './read_preference';
 import { type Runtime, type RuntimeAdapters } from './runtime_adapters';
 import type { ServerMonitoringMode } from './sdam/monitor';
 import type { TagSet } from './sdam/server_description';
-import { DeprioritizedServers, readPreferenceServerSelector } from './sdam/server_selection';
+import {
+  DeprioritizedServers,
+  readPreferenceServerSelector,
+  type ServerSelectorFunction
+} from './sdam/server_selection';
 import type { SrvPoller } from './sdam/srv_polling';
 import { Topology, type TopologyEvents } from './sdam/topology';
 import { ClientSession, type ClientSessionOptions, ServerSessionPool } from './sessions';
@@ -259,6 +263,104 @@ export interface MongoClientOptions extends BSONSerializeOptions, SupportedNodeC
   localThresholdMS?: number;
   /** Specifies how long (in milliseconds) to block for server selection before throwing an exception.  */
   serverSelectionTimeoutMS?: number;
+  /**
+   * Further filters the servers that are already eligible for an operation.
+   *
+   * The driver first applies the operation's read preference, tag sets, max
+   * staleness, and deprioritized-server rules. It then calls this function
+   * with copies of the servers that remain, and only after the function
+   * returns does it apply `localThresholdMS`. Array order is not a preference.
+   * Inside the latency window the driver still chooses by current operation
+   * count. Return a one-element array to send the operation to that member.
+   * Returning one slower member keeps it: that member becomes the low end of
+   * the window. Returning it together with a faster member can drop it again.
+   *
+   * Return only servers from the array you were given. A server from an
+   * earlier call, or any other address, throws `MongoInvalidArgumentError`.
+   * An empty array means none of the eligible servers are acceptable.
+   * Selection then waits until `serverSelectionTimeoutMS` and throws
+   * `MongoServerSelectionError`. Prefer `return preferred ? [preferred] : servers`
+   * when the member is a preference. Use `[]` only when the operation must
+   * not run anywhere else.
+   *
+   * The descriptions are copies. Editing `roundTripTime`, `tags`, `hosts`,
+   * `electionId`, `topologyVersion`, `$clusterTime`, or `error` does not
+   * change the topology. The driver selects the original server with the same
+   * address, not the object you return. Do not keep the copies for a later
+   * call. The next selection builds new ones, and a saved description whose
+   * address is no longer eligible is rejected.
+   *
+   * What a call is allowed to see depends on the operation:
+   *
+   * - `insertOne`, `updateOne`, `deleteOne`, and other ordinary writes are
+   *   offered writable servers only. On a replica set that is the primary, so
+   *   this function cannot send those writes to a secondary.
+   * - An aggregate that writes (`$out` or `$merge`) can be offered secondaries
+   *   when its read preference allows that and the members are new enough.
+   *   Returning one of those secondaries does send the aggregate there.
+   * - A `find` or a read aggregate is offered the servers that match its read
+   *   preference. `primary` sees only the primary. `nearest` sees every
+   *   suitable member, including members outside `localThresholdMS`.
+   *
+   * The function is not called for a Single topology (`directConnection`), a
+   * LoadBalanced topology, or when no server is eligible. It is not called for
+   * an operation pinned to a server (`getMore`, `killCursors`). It is not
+   * called for the connect handshake or for `endSessions`, so a function that
+   * returns an empty array can still `connect()` and `close()` can still end
+   * pooled sessions.
+   *
+   * Transactions on a replica set use read preference primary, so the function
+   * only sees the primary. On a sharded cluster the function runs for the
+   * first command in the transaction. The driver then pins that mongos, and
+   * later commands in the transaction reuse it without calling the function.
+   *
+   * This is not a connection-string option. Pass it to the `MongoClient`
+   * constructor. It matches PyMongo's `server_selector` keyword.
+   *
+   * @example Prefer one member, and keep the rest of the eligible list when it is absent
+   * ```ts
+   * // nearest reads that can see db-2 use db-2. Writes still see only the
+   * // primary, so they return that primary instead of waiting out the
+   * // selection timeout.
+   * new MongoClient(uri, {
+   *   readPreference: 'nearest',
+   *   serverSelector(servers) {
+   *     const preferred = servers.find(server => server.address === 'db-2.example.com:27017');
+   *     return preferred ? [preferred] : servers;
+   *   }
+   * });
+   * ```
+   *
+   * @example Require one member for operations that are allowed to use it
+   * ```ts
+   * // [] waits until serverSelectionTimeoutMS. A primary write is only offered
+   * // the primary, so this times out those writes while db-2 is a secondary.
+   * // Use it when a read must not fall back to another member.
+   * new MongoClient(uri, {
+   *   readPreference: 'nearest',
+   *   serverSelector(servers) {
+   *     const member = servers.find(server => server.address === 'db-2.example.com:27017');
+   *     return member ? [member] : [];
+   *   }
+   * });
+   * ```
+   *
+   * @example Choose one mongos, or one secondary for a writing aggregate
+   * ```ts
+   * // Every eligible mongos is passed in. Returning one sends the command
+   * // there and leaves the others for the next call that does not match.
+   * // The same shape works for an aggregate with `$out` or `$merge`: the
+   * // eligible list may contain secondaries, and returning one of them runs
+   * // that aggregate on it.
+   * new MongoClient(uri, {
+   *   serverSelector(servers) {
+   *     const chosen = servers.find(server => server.address === 'mongos-a.example.com:27017');
+   *     return chosen ? [chosen] : servers;
+   *   }
+   * });
+   * ```
+   */
+  serverSelector?: ServerSelectorFunction;
   /** heartbeatFrequencyMS controls when the driver checks the state of the MongoDB deployment. Specify the interval (in milliseconds) between checks, counted from the end of the previous check until the beginning of the next one. */
   heartbeatFrequencyMS?: number;
   /** Sets the minimum heartbeat frequency. In the event that the driver has to frequently re-check a server's availability, it will wait at least this long since the previous check to avoid wasted effort. */
@@ -849,7 +951,8 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
       { description: topologyDescription }: Topology
     ) {
       // If we would attempt to select a server and get nothing back we short circuit
-      // to avoid the server selection timeout.
+      // to avoid the server selection timeout. This check ignores `serverSelector`:
+      // session cleanup must run even when the filter would select no server.
       const selector = readPreferenceServerSelector(ReadPreference.primaryPreferred);
       const serverDescriptions = Array.from(topologyDescription.servers.values());
       const servers = selector(topologyDescription, serverDescriptions, new DeprioritizedServers());
@@ -1096,6 +1199,8 @@ export interface MongoOptions
   proxyUsername?: string;
   proxyPassword?: string;
   serverMonitoringMode: ServerMonitoringMode;
+  /** @see MongoClientOptions.serverSelector */
+  serverSelector?: ServerSelectorFunction;
   /** @internal */
   connectionType?: typeof Connection;
   /** @internal */