Merge branch 'main' into NODE-7298

Author: tadjik1

drivers-evergreen-tools

@@ -18,6 +18,7 @@ import { MongoClient } from './mongo_client';
 import { type InferIdType, TypedEventEmitter } from './mongo_types';
 import type { AggregateOptions } from './operations/aggregate';
 import type { OperationParent } from './operations/command';
+import { DeprioritizedServers } from './sdam/server_selection';
 import type { ServerSessionId } from './sessions';
 import { CSOTTimeoutContext, type TimeoutContext } from './timeout';
 import { type AnyOptions, getTopology, type MongoDBNamespace, squashError } from './utils';
@@ -1073,7 +1074,8 @@ export class ChangeStream<
     try {
       await topology.selectServer(this.cursor.readPreference, {
         operationName: 'reconnect topology in change stream',
-        timeoutContext: this.timeoutContext
+        timeoutContext: this.timeoutContext,
+        deprioritizedServers: new DeprioritizedServers()
       });
       this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
     } catch {

src/change_stream.ts

@@ -1,4 +1,5 @@
 import * as dns from 'dns';
+import * as os from 'os';
 
 import { getKerberos, type Kerberos, type KerberosClient } from '../../deps';
 import { MongoInvalidArgumentError, MongoMissingCredentialsError } from '../../error';
@@ -97,7 +98,7 @@ async function makeKerberosClient(authContext: AuthContext): Promise<KerberosCli
   }
 
   const spnHost = mechanismProperties.SERVICE_HOST ?? host;
-  let spn = `${serviceName}${process.platform === 'win32' ? '/' : '@'}${spnHost}`;
+  let spn = `${serviceName}${os.platform() === 'win32' ? '/' : '@'}${spnHost}`;
   if ('SERVICE_REALM' in mechanismProperties) {
     spn = `${spn}@${mechanismProperties.SERVICE_REALM}`;
   }

src/cmap/auth/gssapi.ts

@@ -35,6 +35,11 @@ import {
 /** @public */
 export type Stream = Socket | TLSSocket;
 
+function applyBackpressureLabels(error: MongoError) {
+  error.addErrorLabel(MongoErrorLabel.SystemOverloadedError);
+  error.addErrorLabel(MongoErrorLabel.RetryableError);
+}
+
 export async function connect(options: ConnectionOptions): Promise<Connection> {
   let connection: Connection | null = null;
   try {
@@ -103,6 +108,8 @@ export async function performInitialHandshake(
   const authContext = new AuthContext(conn, credentials, options);
   conn.authContext = authContext;
 
+  // If we encounter an error preparing the handshake document, do NOT apply backpressure labels.  Errors
+  // encountered building the handshake document are all client-side, and do not indicate an overloaded server.
   const handshakeDoc = await prepareHandshakeDocument(authContext);
 
   // @ts-expect-error: TODO(NODE-5141): The options need to be filtered properly, Connection options differ from Command options
@@ -163,12 +170,15 @@ export async function performInitialHandshake(
     try {
       await provider.auth(authContext);
     } catch (error) {
+      // NOTE: If we encounter an error authenticating a connection, do NOT apply backpressure labels.
+
       if (error instanceof MongoError) {
         error.addErrorLabel(MongoErrorLabel.HandshakeError);
         if (needsRetryableWriteLabel(error, response.maxWireVersion, conn.description.type)) {
           error.addErrorLabel(MongoErrorLabel.RetryableWriteError);
         }
       }
+
       throw error;
     }
   }
@@ -189,6 +199,11 @@ export async function performInitialHandshake(
       if (error instanceof MongoError) {
         error.addErrorLabel(MongoErrorLabel.HandshakeError);
       }
+      // If we encounter a network error executing the initial handshake, apply backpressure labels.
+      if (error instanceof MongoNetworkError) {
+        applyBackpressureLabels(error);
+      }
+
       throw error;
     }
   }
@@ -424,6 +439,10 @@ export async function makeSocket(options: MakeConnectionOptions): Promise<Stream
     socket = await connectedSocket;
     return socket;
   } catch (error) {
+    // If we encounter an error while establishing a socket, apply the backpressure labels to it.  We cannot
+    // differentiate between DNS, TLS errors and network errors without refactoring our connection establishment to
+    // handle all three steps separately.
+    applyBackpressureLabels(error);
     socket.destroy();
     throw error;
   } finally {

src/cmap/connect.ts

@@ -231,7 +231,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this.mongoLogger = this.server.topology.client?.mongoLogger;
     this.component = 'connection';
 
-    process.nextTick(() => {
+    queueMicrotask(() => {
       this.emitAndLog(ConnectionPool.CONNECTION_POOL_CREATED, new ConnectionPoolCreatedEvent(this));
     });
   }
@@ -342,7 +342,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     });
 
     this.waitQueue.push(waitQueueMember);
-    process.nextTick(() => this.processWaitQueue());
+    queueMicrotask(() => this.processWaitQueue());
 
     try {
       timeout?.throwIfExpired();
@@ -405,7 +405,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       this.destroyConnection(connection, reason);
     }
 
-    process.nextTick(() => this.processWaitQueue());
+    queueMicrotask(() => this.processWaitQueue());
   }
 
   /**
@@ -461,7 +461,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     }
 
     if (interruptInUseConnections) {
-      process.nextTick(() => this.interruptInUseConnections(oldGeneration));
+      queueMicrotask(() => this.interruptInUseConnections(oldGeneration));
     }
 
     this.processWaitQueue();
@@ -702,7 +702,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       this.createConnection((err, connection) => {
         if (!err && connection) {
           this.connections.push(connection);
-          process.nextTick(() => this.processWaitQueue());
+          queueMicrotask(() => this.processWaitQueue());
         }
         if (this.poolState === PoolState.ready) {
           clearTimeout(this.minPoolSizeTimer);
@@ -809,7 +809,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
             waitQueueMember.resolve(connection);
           }
         }
-        process.nextTick(() => this.processWaitQueue());
+        queueMicrotask(() => this.processWaitQueue());
       });
     }
     this.processingWaitQueue = false;

src/cmap/connection_pool.ts

@@ -157,8 +157,8 @@ export async function makeClientMetadata(
 
   // Note: order matters, os.type is last so it will be removed last if we're at maxSize
   const osInfo = new Map()
-    .set('name', process.platform)
-    .set('architecture', process.arch)
+    .set('name', os.platform())
+    .set('architecture', os.arch())
     .set('version', os.release())
     .set('type', os.type());
 

src/cmap/handshake/client_metadata.ts

@@ -99,7 +99,9 @@ export const MongoErrorLabel = Object.freeze({
   ResetPool: 'ResetPool',
   PoolRequestedRetry: 'PoolRequestedRetry',
   InterruptInUseConnections: 'InterruptInUseConnections',
-  NoWritesPerformed: 'NoWritesPerformed'
+  NoWritesPerformed: 'NoWritesPerformed',
+  RetryableError: 'RetryableError',
+  SystemOverloadedError: 'SystemOverloadedError'
 } as const);
 
 /** @public */

src/error.ts

@@ -165,7 +165,7 @@ export class GridFSBucketWriteStream extends Writable {
         }
       );
     } else {
-      return process.nextTick(callback);
+      return queueMicrotask(callback);
     }
   }
 
@@ -188,7 +188,7 @@ export class GridFSBucketWriteStream extends Writable {
   /** @internal */
   override _final(callback: (error?: Error | null) => void): void {
     if (this.state.streamEnd) {
-      return process.nextTick(callback);
+      return queueMicrotask(callback);
     }
     this.state.streamEnd = true;
     writeRemnant(this, callback);
@@ -220,11 +220,11 @@ export class GridFSBucketWriteStream extends Writable {
 
 function handleError(stream: GridFSBucketWriteStream, error: Error, callback: Callback): void {
   if (stream.state.errored) {
-    process.nextTick(callback);
+    queueMicrotask(callback);
     return;
   }
   stream.state.errored = true;
-  process.nextTick(callback, error);
+  queueMicrotask(() => callback(error));
 }
 
 function createChunkDoc(filesId: ObjectId, n: number, data: Buffer): GridFSChunk {
@@ -283,7 +283,7 @@ async function checkChunksIndex(stream: GridFSBucketWriteStream): Promise<void>
 
 function checkDone(stream: GridFSBucketWriteStream, callback: Callback): void {
   if (stream.done) {
-    return process.nextTick(callback);
+    return queueMicrotask(callback);
   }
 
   if (stream.state.streamEnd && stream.state.outstandingRequests === 0 && !stream.state.errored) {
@@ -327,7 +327,7 @@ function checkDone(stream: GridFSBucketWriteStream, callback: Callback): void {
     return;
   }
 
-  process.nextTick(callback);
+  queueMicrotask(callback);
 }
 
 async function checkIndexes(stream: GridFSBucketWriteStream): Promise<void> {
@@ -425,7 +425,7 @@ function doWrite(
   if (stream.pos + inputBuf.length < stream.chunkSizeBytes) {
     inputBuf.copy(stream.bufToStore, stream.pos);
     stream.pos += inputBuf.length;
-    process.nextTick(callback);
+    queueMicrotask(callback);
     return;
   }
 
@@ -530,7 +530,7 @@ function writeRemnant(stream: GridFSBucketWriteStream, callback: Callback): void
 
 function isAborted(stream: GridFSBucketWriteStream, callback: Callback<void>): boolean {
   if (stream.state.aborted) {
-    process.nextTick(callback, new MongoAPIError('Stream has been aborted'));
+    queueMicrotask(() => callback(new MongoAPIError('Stream has been aborted')));
     return true;
   }
   return false;

src/gridfs/upload.ts

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

src/index.ts

@@ -48,7 +48,7 @@ import type { ReadConcern, ReadConcernLevel, ReadConcernLike } from './read_conc
 import { ReadPreference, type ReadPreferenceMode } from './read_preference';
 import type { ServerMonitoringMode } from './sdam/monitor';
 import type { TagSet } from './sdam/server_description';
-import { readPreferenceServerSelector } from './sdam/server_selection';
+import { DeprioritizedServers, readPreferenceServerSelector } 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';
@@ -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, new DeprioritizedServers());
       if (servers.length !== 0) {
         const endSessions = Array.from(client.s.sessionPool.sessions, ({ id }) => id);
         if (endSessions.length !== 0) {