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

@@ -636,7 +636,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()}.
@@ -672,6 +674,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,7 @@ 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
+ expect(error).to.be.instanceOf(ConnectionPoolClosedError);
  });
 
  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();