Merge branch 'main' into lint-related-deps

Author: durran

CONTRIBUTORS.md

@@ -52,10 +52,10 @@
 - heroic
 - gitfy
 - Eric Adum
+- Neal Beeken - https://github.com/nbbeeken
 
 ## Maintainers
 
-- Neal Beeken <<[email protected]>>
 - Durran Jordan <<[email protected]>>
 - Daria Pardue <<[email protected]>>
 - Bailey Pearson <<[email protected]>>

src/client-side-encryption/auto_encrypter.ts

@@ -375,8 +375,8 @@ export class AutoEncrypter {
   /**
    * Cleans up the `_mongocryptdClient`, if present.
    */
-  async teardown(force: boolean): Promise<void> {
-    await this._mongocryptdClient?.close(force);
+  async close(): Promise<void> {
+    await this._mongocryptdClient?.close();
   }
 
   /**

src/cmap/connection_pool.ts

@@ -17,6 +17,7 @@ import {
 } from '../constants';
 import {
   type AnyError,
+  MongoClientClosedError,
   type MongoError,
   MongoInvalidArgumentError,
   MongoMissingCredentialsError,
@@ -484,11 +485,17 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     for (const connection of this.checkedOut) {
       if (connection.generation <= minGeneration) {
         connection.onError(new PoolClearedOnNetworkError(this));
-        this.checkIn(connection);
       }
     }
   }
 
+  /** For MongoClient.close() procedures */
+  public closeCheckedOutConnections() {
+    for (const conn of this.checkedOut) {
+      conn.onError(new MongoClientClosedError());
+    }
+  }
+
   /** Close the pool */
   close(): void {
     if (this.closed) {

src/encrypter.ts

@@ -1,11 +1,8 @@
-import { callbackify } from 'util';
-
 import { AutoEncrypter, type AutoEncryptionOptions } from './client-side-encryption/auto_encrypter';
 import { MONGO_CLIENT_EVENTS } from './constants';
 import { getMongoDBClientEncryption } from './deps';
 import { MongoInvalidArgumentError, MongoMissingDependencyError } from './error';
 import { MongoClient, type MongoClientOptions } from './mongo_client';
-import { type Callback } from './utils';
 
 /** @internal */
 export interface EncrypterOptions {
@@ -98,20 +95,16 @@ export class Encrypter {
     }
   }
 
-  closeCallback(client: MongoClient, force: boolean, callback: Callback<void>) {
-    callbackify(this.close.bind(this))(client, force, callback);
-  }
-
-  async close(client: MongoClient, force: boolean): Promise<void> {
+  async close(client: MongoClient): Promise<void> {
     let error;
     try {
-      await this.autoEncrypter.teardown(force);
+      await this.autoEncrypter.close();
     } catch (autoEncrypterError) {
       error = autoEncrypterError;
     }
     const internalClient = this.internalClient;
     if (internalClient != null && client !== internalClient) {
-      return await internalClient.close(force);
+      return await internalClient.close();
     }
     if (error != null) {
       throw error;

src/error.ts

@@ -1018,6 +1018,34 @@ export class MongoTopologyClosedError extends MongoAPIError {
   }
 }
 
+/**
+ * An error generated when the MongoClient is closed and async
+ * operations are interrupted.
+ *
+ * @public
+ * @category Error
+ */
+export class MongoClientClosedError extends MongoAPIError {
+  /**
+   * **Do not use this constructor!**
+   *
+   * Meant for internal use only.
+   *
+   * @remarks
+   * This class is only meant to be constructed within the driver. This constructor is
+   * not subject to semantic versioning compatibility guarantees and may change at any time.
+   *
+   * @public
+   **/
+  constructor() {
+    super('Operation interrupted because client was closed');
+  }
+
+  override get name(): string {
+    return 'MongoClientClosedError';
+  }
+}
+
 /** @public */
 export interface MongoNetworkErrorOptions {
   /** Indicates the timeout happened before a connection handshake completed */

src/index.ts

@@ -53,6 +53,7 @@ export {
   MongoClientBulkWriteCursorError,
   MongoClientBulkWriteError,
   MongoClientBulkWriteExecutionError,
+  MongoClientClosedError,
   MongoCompatibilityError,
   MongoCursorExhaustedError,
   MongoCursorInUseError,

src/mongo_client.ts

@@ -641,25 +641,27 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
   }
 
   /**
-   * Cleans up client-side resources used by the MongoCLient and .  This includes:
+   * Cleans up client-side resources used by the MongoClient.
    *
-   * - Closes all open, unused connections (see note).
+   * This includes:
+   *
+   * - Closes in-use connections.
+   * - Closes all active cursors.
    * - Ends all in-use sessions with {@link ClientSession#endSession|ClientSession.endSession()}.
+   *   - aborts in progress transactions if is one related to the session.
    * - Ends all unused sessions server-side.
+   * - Closes all remaining idle connections.
    * - Cleans up any resources being used for auto encryption if auto encryption is enabled.
    *
-   * @remarks Any in-progress operations are not killed and any connections used by in progress operations
-   * will be cleaned up lazily as operations finish.
-   *
-   * @param force - Force close, emitting no events
+   * @param _force - currently an unused flag that has no effect. Defaults to `false`.
    */
-  async close(force = false): Promise<void> {
+  async close(_force = false): Promise<void> {
     if (this.closeLock) {
       return await this.closeLock;
     }
 
     try {
-      this.closeLock = this._close(force);
+      this.closeLock = this._close();
       await this.closeLock;
     } finally {
       // release
@@ -668,7 +670,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
   }
 
   /* @internal */
-  private async _close(force = false): Promise<void> {
+  private async _close(): Promise<void> {
     // There's no way to set hasBeenClosed back to false
     Object.defineProperty(this.s, 'hasBeenClosed', {
       value: true,
@@ -677,6 +679,8 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
       writable: false
     });
 
+    this.topology?.closeCheckedOutConnections();
+
     const activeCursorCloses = Array.from(this.s.activeCursors, cursor => cursor.close());
     this.s.activeCursors.clear();
 
@@ -722,7 +726,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
 
     const { encrypter } = this.options;
     if (encrypter) {
-      await encrypter.close(this, force);
+      await encrypter.close(this);
     }
   }
 

src/sdam/server.ts

@@ -246,8 +246,12 @@ export class Server extends TypedEventEmitter<ServerEvents> {
     }
   }
 
+  closeCheckedOutConnections() {
+    return this.pool.closeCheckedOutConnections();
+  }
+
   /** Destroy the server connection */
-  destroy(): void {
+  close(): void {
     if (this.s.state === STATE_CLOSED) {
       return;
     }

src/sdam/topology.ts

@@ -490,14 +490,20 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
     }
   }
 
+  closeCheckedOutConnections() {
+    for (const server of this.s.servers.values()) {
+      return server.closeCheckedOutConnections();
+    }
+  }
+
   /** Close this topology */
   close(): void {
     if (this.s.state === STATE_CLOSED || this.s.state === STATE_CLOSING) {
       return;
     }
 
     for (const server of this.s.servers.values()) {
-      destroyServer(server, this);
+      closeServer(server, this);
     }
 
     this.s.servers.clear();
@@ -791,12 +797,12 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
 }
 
 /** Destroys a server, and removes all event listeners from the instance */
-function destroyServer(server: Server, topology: Topology) {
+function closeServer(server: Server, topology: Topology) {
   for (const event of LOCAL_SERVER_EVENTS) {
     server.removeAllListeners(event);
   }
 
-  server.destroy();
+  server.close();
   topology.emitAndLog(
     Topology.SERVER_CLOSED,
     new ServerClosedEvent(topology.s.id, server.description.address)
@@ -903,7 +909,7 @@ function updateServers(topology: Topology, incomingServerDescription?: ServerDes
 
     // prepare server for garbage collection
     if (server) {
-      destroyServer(server, topology);
+      closeServer(server, topology);
     }
   }
 }

test/integration/change-streams/change_stream.test.ts

@@ -63,6 +63,7 @@ describe('Change Streams', function () {
     await csDb.createCollection('test').catch(() => null);
     collection = csDb.collection('test');
     changeStream = collection.watch();
+    changeStream.on('error', () => null);
   });
 
   afterEach(async () => {
@@ -702,15 +703,19 @@ describe('Change Streams', function () {
 
       const outStream = new PassThrough({ objectMode: true });
 
-      // @ts-expect-error: transform requires a Document return type
-      changeStream.stream({ transform: JSON.stringify }).pipe(outStream);
+      const transform = doc => ({ doc: JSON.stringify(doc) });
+      changeStream
+        .stream({ transform })
+        .on('error', () => null)
+        .pipe(outStream)
+        .on('error', () => null);
 
       const willBeData = once(outStream, 'data');
 
       await collection.insertMany([{ a: 1 }]);
 
       const [data] = await willBeData;
-      const parsedEvent = JSON.parse(data);
+      const parsedEvent = JSON.parse(data.doc);
       expect(parsedEvent).to.have.nested.property('fullDocument.a', 1);
 
       outStream.destroy();