feat(NODE-6882): close checked out connection in MongoClient.close()

Author: nbbeeken

src/cmap/connection_pool.ts

@@ -17,6 +17,7 @@ import {
 } from '../constants';
 import {
   type AnyError,
+  ConnectionPoolClosedError,
   type MongoError,
   MongoInvalidArgumentError,
   MongoMissingCredentialsError,
@@ -489,6 +490,16 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     }
   }
 
+  closeCheckedOutConnections() {
+    for (const conn of this.checkedOut) {
+      this.emitAndLog(
+        ConnectionPool.CONNECTION_CLOSED,
+        new ConnectionClosedEvent(this, conn, 'poolClosed')
+      );
+      conn.onError(new ConnectionPoolClosedError());
+    }
+  }
+
   /** Close the pool */
   close(): void {
     if (this.closed) {

src/error.ts

@@ -1018,6 +1018,62 @@ 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(message = 'MongoClient is closed') {
+    super(message);
+  }
+
+  override get name(): string {
+    return 'MongoClientClosedError';
+  }
+}
+
+/**
+ * An error generated when a ConnectionPool is closed and async
+ * operations are interrupted.
+ *
+ * @public
+ * @category Error
+ */
+export class ConnectionPoolClosedError 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(message = 'ConnectionPool is closed') {
+    super(message);
+  }
+
+  override get name(): string {
+    return 'ConnectionPoolClosedError';
+  }
+}
+
 /** @public */
 export interface MongoNetworkErrorOptions {
   /** Indicates the timeout happened before a connection handshake completed */

src/index.ts

@@ -45,6 +45,7 @@ export {
 export { ClientEncryption } from './client-side-encryption/client_encryption';
 export { ChangeStreamCursor } from './cursor/change_stream_cursor';
 export {
+  ConnectionPoolClosedError,
   MongoAPIError,
   MongoAWSError,
   MongoAzureError,
@@ -53,6 +54,7 @@ export {
   MongoClientBulkWriteCursorError,
   MongoClientBulkWriteError,
   MongoClientBulkWriteExecutionError,
+  MongoClientClosedError,
   MongoCompatibilityError,
   MongoCursorExhaustedError,
   MongoCursorInUseError,

src/mongo_client.ts

@@ -641,7 +641,9 @@ 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.
+   *
+   * This includes:
    *
    * - Closes all open, unused connections (see note).
    * - Ends all in-use sessions with {@link ClientSession#endSession|ClientSession.endSession()}.
@@ -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();
 

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();

test/integration/crud/misc_cursors.test.js

@@ -11,7 +11,7 @@ const { Writable } = require('stream');
 const { once, on } = require('events');
 const { setTimeout } = require('timers');
 const { ReadPreference } = require('../../mongodb');
-const { ServerType } = require('../../mongodb');
+const { ServerType, ConnectionPoolClosedError } = require('../../mongodb');
 const { formatSort } = require('../../mongodb');
 
 describe('Cursor', function () {
@@ -1872,7 +1872,11 @@ describe('Cursor', function () {
     expect(cursor).to.have.property('closed', true);
 
     const error = await rejectedEarlyBecauseClientClosed;
-    expect(error).to.be.null; // TODO(NODE-6632): This should throw again after the client signal aborts the in-progress next call
+    if (this.configuration.topologyType === 'LoadBalanced') {
+      expect(error).to.be.instanceOf(ConnectionPoolClosedError);
+    } else {
+      expect(error).to.be.null;
+    }
   });
 
   it('shouldAwaitData', {

test/integration/index_management.test.ts

@@ -770,20 +770,19 @@ describe('Indexes', function () {
 
           expect(events).to.be.an('array').with.lengthOf(1);
           expect(events[0]).nested.property('command.commitQuorum').to.equal(0);
-          await collection.drop(err => {
-            expect(err).to.not.exist;
-          });
+          await collection.drop();
         }
       };
     }
     it(
       'should run command with commitQuorum if specified on db.createIndex',
-      commitQuorumTest((db, collection) =>
-        db.createIndex(collection.collectionName, 'a', {
-          // @ts-expect-error revaluate this?
-          writeConcern: { w: 'majority' },
-          commitQuorum: 0
-        })
+      commitQuorumTest(
+        async (db, collection) =>
+          await db.createIndex(collection.collectionName, 'a', {
+            // @ts-expect-error revaluate this?
+            writeConcern: { w: 'majority' },
+            commitQuorum: 0
+          })
       )
     );
     it(

test/integration/node-specific/examples/causal_consistency.test.js

@@ -31,8 +31,8 @@ describe('examples(causal-consistency):', function () {
   it('supports causal consistency', async function () {
     const session = client.startSession({ causalConsistency: true });
 
-    collection.insertOne({ darmok: 'jalad' }, { session });
-    collection.updateOne({ darmok: 'jalad' }, { $set: { darmok: 'tanagra' } }, { session });
+    await collection.insertOne({ darmok: 'jalad' }, { session });
+    await collection.updateOne({ darmok: 'jalad' }, { $set: { darmok: 'tanagra' } }, { session });
 
     const results = await collection.find({}, { session }).toArray();
 

test/integration/node-specific/mongo_client.test.ts

@@ -1035,15 +1035,15 @@ describe('class MongoClient', function () {
         client.on('commandStarted', ev => ev.commandName === 'killCursors' && kills.push(ev));
       });
 
-      it('are all closed', async () => {
+      it('are all closed', async function () {
         const cursors = Array.from({ length: 30 }, (_, skip) =>
           collection.find({}, { skip, batchSize: 1 })
         );
         await Promise.all(cursors.map(c => c.next()));
         expect(client.s.activeCursors).to.have.lengthOf(30);
         await client.close();
         expect(client.s.activeCursors).to.have.lengthOf(0);
-        expect(kills).to.have.lengthOf(30);
+        expect(kills).to.have.lengthOf(this.configuration.topologyType === 'LoadBalanced' ? 0 : 30);
       });
 
       it('creating cursors after close adds to activeCursors', async () => {

test/unit/index.test.ts

@@ -36,6 +36,7 @@ const EXPECTED_EXPORTS = [
   'ConnectionClosedEvent',
   'ConnectionCreatedEvent',
   'ConnectionPoolClearedEvent',
+  'ConnectionPoolClosedError',
   'ConnectionPoolClosedEvent',
   'ConnectionPoolCreatedEvent',
   'ConnectionPoolMonitoringEvent',
@@ -71,6 +72,7 @@ const EXPECTED_EXPORTS = [
   'MongoClientBulkWriteCursorError',
   'MongoClientBulkWriteError',
   'MongoClientBulkWriteExecutionError',
+  'MongoClientClosedError',
   'MongoCompatibilityError',
   'MongoCryptAzureKMSRequestError',
   'MongoCryptCreateDataKeyError',