Merge branch 'main' into DRIVERS-3344

Author: baileympearson

drivers-evergreen-tools

@@ -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

@@ -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

@@ -426,7 +426,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
 function monitorServer(monitor: Monitor) {
   return (callback: Callback) => {
     if (monitor.s.state === STATE_MONITORING) {
-      process.nextTick(callback);
+      queueMicrotask(callback);
       return;
     }
     stateTransition(monitor, STATE_MONITORING);

src/sdam/monitor.ts

@@ -400,7 +400,8 @@ export class Server extends TypedEventEmitter<ServerEvents> {
       error instanceof MongoNetworkError && !(error instanceof MongoNetworkTimeoutError);
     const isNetworkTimeoutBeforeHandshakeError =
       error instanceof MongoNetworkError && error.beforeHandshake;
-    const isAuthHandshakeError = error.hasErrorLabel(MongoErrorLabel.HandshakeError);
+    const isAuthOrEstablishmentHandshakeError = error.hasErrorLabel(MongoErrorLabel.HandshakeError);
+    const isSystemOverloadError = error.hasErrorLabel(MongoErrorLabel.SystemOverloadedError);
 
     // Perhaps questionable and divergent from the spec, but considering MongoParseErrors like state change errors was legacy behavior.
     if (isStateChangeError(error) || error instanceof MongoParseError) {
@@ -414,7 +415,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
           error.addErrorLabel(MongoErrorLabel.ResetPool);
         }
         markServerUnknown(this, error);
-        process.nextTick(() => this.requestCheck());
+        queueMicrotask(() => this.requestCheck());
         return;
       }
 
@@ -424,8 +425,12 @@ export class Server extends TypedEventEmitter<ServerEvents> {
     } else if (
       isNetworkNonTimeoutError ||
       isNetworkTimeoutBeforeHandshakeError ||
-      isAuthHandshakeError
+      isAuthOrEstablishmentHandshakeError
     ) {
+      // Do NOT clear the pool if we encounter a system overloaded error.
+      if (isSystemOverloadError) {
+        return;
+      }
       // from the SDAM spec: The driver MUST synchronize clearing the pool with updating the topology.
       // In load balanced mode: there is no monitoring, so there is no topology to update.  We simply clear the pool.
       // For other topologies: the `ResetPool` label instructs the topology to clear the server's pool in `updateServer()`.

src/sdam/server.ts

@@ -1079,7 +1079,7 @@ function processWaitQueue(topology: Topology) {
   if (topology.waitQueue.length > 0) {
     // ensure all server monitors attempt monitoring soon
     for (const [, server] of topology.s.servers) {
-      process.nextTick(function scheduleServerCheck() {
+      queueMicrotask(function scheduleServerCheck() {
         return server.requestCheck();
       });
     }

src/sdam/topology.ts

@@ -1,13 +1,18 @@
 import { expect } from 'chai';
 import { once } from 'events';
 
-import { type MongoClient } from '../../../src';
+import {
+  type ConnectionCheckOutFailedEvent,
+  type ConnectionPoolClearedEvent,
+  type MongoClient
+} from '../../../src';
 import {
   CONNECTION_POOL_CLEARED,
   CONNECTION_POOL_READY,
   SERVER_HEARTBEAT_FAILED,
   SERVER_HEARTBEAT_SUCCEEDED
 } from '../../../src/constants';
+import { sleep } from '../../tools/utils';
 
 describe('Server Discovery and Monitoring Prose Tests', function () {
   context('Monitors sleep at least minHeartbeatFrequencyMS between checks', function () {
@@ -187,4 +192,93 @@ describe('Server Discovery and Monitoring Prose Tests', function () {
       }
     });
   });
+
+  context('Connection Pool Backpressure', function () {
+    let client: MongoClient;
+    const checkoutFailedEvents: Array<ConnectionCheckOutFailedEvent> = [];
+    const poolClearedEvents: Array<ConnectionPoolClearedEvent> = [];
+
+    beforeEach(async function () {
+      // 1. Create a test client that listens to CMAP events, with maxConnecting=100.
+      client = this.configuration.newClient({}, { maxConnecting: 100 });
+
+      client.on('connectionCheckOutFailed', e => checkoutFailedEvents.push(e));
+      client.on('connectionPoolCleared', e => poolClearedEvents.push(e));
+
+      await client.connect();
+
+      // 2. Run the following commands to set up the rate limiter.
+      // ```python
+      // client.admin.command("setParameter", 1, ingressConnectionEstablishmentRateLimiterEnabled=True)
+      // client.admin.command("setParameter", 1, ingressConnectionEstablishmentRatePerSec=20)
+      // client.admin.command("setParameter", 1, ingressConnectionEstablishmentBurstCapacitySecs=1)
+      // client.admin.command("setParameter", 1, ingressConnectionEstablishmentMaxQueueDepth=1)
+      // ```
+      const admin = client.db('admin').admin();
+      await admin.command({
+        setParameter: 1,
+        ingressConnectionEstablishmentRateLimiterEnabled: true
+      });
+      await admin.command({
+        setParameter: 1,
+        ingressConnectionEstablishmentRatePerSec: 20
+      });
+      await admin.command({
+        setParameter: 1,
+        ingressConnectionEstablishmentBurstCapacitySecs: 1
+      });
+      await admin.command({
+        setParameter: 1,
+        ingressConnectionEstablishmentMaxQueueDepth: 1
+      });
+
+      // 3. Add a document to the test collection so that the sleep operations will actually block:
+      // `client.test.test.insert_one({})`.
+      await client.db('test').collection('test').insertOne({});
+    });
+
+    afterEach(async function () {
+      // 7. Sleep for 1 second to clear the rate limiter.
+      await sleep(1000);
+
+      // 8. Ensure that the following command runs at test teardown even if the test fails.
+      // `client.admin("setParameter", 1, ingressConnectionEstablishmentRateLimiterEnabled=False)`.
+      const admin = client.db('admin').admin();
+      await admin.command({
+        setParameter: 1,
+        ingressConnectionEstablishmentRateLimiterEnabled: false
+      });
+
+      await client.close();
+    });
+
+    it(
+      'does not clear the pool when connections are closed due to connection storms',
+      {
+        requires: {
+          // This test requires MongoDB 7.0+.
+          mongodb: '>=7.0' // rate limiting added in 7.0
+        }
+      },
+      async function () {
+        // 4. Run the following find command on the collection in 100 parallel threads/coroutines. Run these commands concurrently
+        // but block on their completion, and ignore errors raised by the command.
+        // `client.test.test.find_one({"$where": "function() { sleep(2000); return true; }})`
+        await Promise.allSettled(
+          Array.from({ length: 100 }).map(() =>
+            client
+              .db('test')
+              .collection('test')
+              .findOne({ $where: 'function() { sleep(2000); return true; }' })
+          )
+        );
+
+        // 5. Assert that at least 10 `ConnectionCheckOutFailedEvent` occurred.
+        expect(checkoutFailedEvents.length).to.be.at.least(10);
+
+        // 6. Assert that 0 `PoolClearedEvent` occurred.
+        expect(poolClearedEvents).to.be.empty;
+      }
+    );
+  });
 });

test/integration/server-discovery-and-monitoring/server_discovery_and_monitoring.prose.test.ts

@@ -107,8 +107,18 @@ describe('Sessions Prose Tests', () => {
       expect(allResults).to.have.lengthOf(operations.length);
       expect(events).to.have.lengthOf(operations.length);
 
-      // This is a guarantee in node, unless you are performing a transaction (which is not being done in this test)
-      expect(new Set(events.map(ev => ev.command.lsid.id.toString('hex')))).to.have.lengthOf(1);
+      // Previous version of this test was too strict: we were expecting that only one session be used for this scenario.
+      // That was possible at the time because the operations were simple enough and the server fast enough that the operations would complete serially.
+      //
+      // However, with a more complex operation bulkWrite (like `Array.from({ length: 100_000 }).map(() => ({ insertOne: { document: { a: 1 } } })),`),
+      // it's entirely possible and expected that bulkWrite would introduce a second session due to the time it takes to process all the inserts.
+      //
+      // The important bit of the test is that the number of sessions is less than the number of concurrent operations, so now instead of expecting exactly 1 session,
+      // we just expect less than operations.length - 1 sessions.
+      //
+      const uniqueSessionIds = new Set(events.map(ev => ev.command.lsid.id.toString('hex')));
+      const expectedMaxSessions = operations.length - 1;
+      expect(uniqueSessionIds).to.have.length.lessThan(expectedMaxSessions);
     });
   });