refactor(NODE-5915): topology close logic to be synchronous (#4021)

Author: W-A-James

src/cmap/connect.ts

@@ -43,7 +43,7 @@ export async function connect(options: ConnectionOptions): Promise<Connection> {
     await performInitialHandshake(connection, options);
     return connection;
   } catch (error) {
-    connection?.destroy({ force: false });
+    connection?.destroy();
     throw error;
   }
 }

src/cmap/connection.ts

@@ -126,12 +126,6 @@ export interface ConnectionOptions
   mongoLogger?: MongoLogger | undefined;
 }
 
-/** @internal */
-export interface DestroyOptions {
-  /** Force the destruction. */
-  force: boolean;
-}
-
 /** @public */
 export type ConnectionEvents = {
   commandStarted(event: CommandStartedEvent): void;
@@ -301,14 +295,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     }, 1).unref(); // No need for this timer to hold the event loop open
   }
 
-  public destroy(options: DestroyOptions, callback?: Callback): void {
+  public destroy(): void {
     if (this.closed) {
-      if (typeof callback === 'function') process.nextTick(callback);
       return;
     }
-    if (typeof callback === 'function') {
-      this.once('close', () => process.nextTick(() => callback()));
-    }
 
     // load balanced mode requires that these listeners remain on the connection
     // after cleanup on timeouts, errors or close so we remove them before calling

src/cmap/connection_pool.ts

@@ -28,7 +28,6 @@ import { CancellationToken, TypedEventEmitter } from '../mongo_types';
 import type { Server } from '../sdam/server';
 import {
   type Callback,
-  eachAsync,
   List,
   makeCounter,
   promiseWithResolvers,
@@ -493,25 +492,16 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
   private interruptInUseConnections(minGeneration: number) {
     for (const connection of this[kCheckedOut]) {
       if (connection.generation <= minGeneration) {
-        this.checkIn(connection);
         connection.onError(new PoolClearedOnNetworkError(this));
+        this.checkIn(connection);
       }
     }
   }
 
   /** Close the pool */
-  close(callback: Callback<void>): void;
-  close(options: CloseOptions, callback: Callback<void>): void;
-  close(_options?: CloseOptions | Callback<void>, _cb?: Callback<void>): void {
-    let options = _options as CloseOptions;
-    const callback = (_cb ?? _options) as Callback<void>;
-    if (typeof options === 'function') {
-      options = {};
-    }
-
-    options = Object.assign({ force: false }, options);
+  close(): void {
     if (this.closed) {
-      return callback();
+      return;
     }
 
     // immediately cancel any in-flight connections
@@ -526,21 +516,15 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     this.clearMinPoolSizeTimer();
     this.processWaitQueue();
 
-    eachAsync<Connection>(
-      this[kConnections].toArray(),
-      (conn, cb) => {
-        this.emitAndLog(
-          ConnectionPool.CONNECTION_CLOSED,
-          new ConnectionClosedEvent(this, conn, 'poolClosed')
-        );
-        conn.destroy({ force: !!options.force }, cb);
-      },
-      err => {
-        this[kConnections].clear();
-        this.emitAndLog(ConnectionPool.CONNECTION_POOL_CLOSED, new ConnectionPoolClosedEvent(this));
-        callback(err);
-      }
-    );
+    for (const conn of this[kConnections]) {
+      this.emitAndLog(
+        ConnectionPool.CONNECTION_CLOSED,
+        new ConnectionClosedEvent(this, conn, 'poolClosed')
+      );
+      conn.destroy();
+    }
+    this[kConnections].clear();
+    this.emitAndLog(ConnectionPool.CONNECTION_POOL_CLOSED, new ConnectionPoolClosedEvent(this));
   }
 
   /**
@@ -592,7 +576,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       new ConnectionClosedEvent(this, connection, reason)
     );
     // destroy the connection
-    process.nextTick(() => connection.destroy({ force: false }));
+    connection.destroy();
   }
 
   private connectionIsStale(connection: Connection) {
@@ -648,7 +632,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
         // The pool might have closed since we started trying to create a connection
         if (this[kPoolState] !== PoolState.ready) {
           this[kPending]--;
-          connection.destroy({ force: true });
+          connection.destroy();
           callback(this.closed ? new PoolClosedError(this) : new PoolClearedError(this));
           return;
         }

src/index.ts

@@ -275,7 +275,6 @@ export type {
   Connection,
   ConnectionEvents,
   ConnectionOptions,
-  DestroyOptions,
   ProxyOptions
 } from './cmap/connection';
 export type {

src/mongo_client.ts

@@ -552,7 +552,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
       try {
         await promisify(callback => this.topology?.connect(options, callback))();
       } catch (error) {
-        this.topology?.close({ force: true });
+        this.topology?.close();
         throw error;
       }
     };
@@ -614,19 +614,12 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> {
     const topology = this.topology;
     this.topology = undefined;
 
-    await new Promise<void>((resolve, reject) => {
-      topology.close({ force }, error => {
-        if (error) return reject(error);
-        const { encrypter } = this[kOptions];
-        if (encrypter) {
-          return encrypter.closeCallback(this, force, error => {
-            if (error) return reject(error);
-            resolve();
-          });
-        }
-        resolve();
-      });
-    });
+    topology.close();
+
+    const { encrypter } = this[kOptions];
+    if (encrypter) {
+      await encrypter.close(this, force);
+    }
   }
 
   /**

src/sdam/monitor.ts

@@ -214,7 +214,7 @@ function resetMonitorState(monitor: Monitor) {
 
   monitor[kCancellationToken].emit('cancel');
 
-  monitor.connection?.destroy({ force: true });
+  monitor.connection?.destroy();
   monitor.connection = null;
 }
 
@@ -247,7 +247,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
   );
 
   function onHeartbeatFailed(err: Error) {
-    monitor.connection?.destroy({ force: true });
+    monitor.connection?.destroy();
     monitor.connection = null;
 
     monitor.emitAndLogHeartbeat(
@@ -366,13 +366,13 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
       await performInitialHandshake(connection, monitor.connectOptions);
       return connection;
     } catch (error) {
-      connection.destroy({ force: false });
+      connection.destroy();
       throw error;
     }
   })().then(
     connection => {
       if (isInCloseState(monitor)) {
-        connection.destroy({ force: true });
+        connection.destroy();
         return;
       }
 
@@ -479,7 +479,7 @@ export class RTTPinger {
     this.closed = true;
     clearTimeout(this[kMonitorId]);
 
-    this.connection?.destroy({ force: true });
+    this.connection?.destroy();
     this.connection = undefined;
   }
 }
@@ -495,7 +495,7 @@ function measureRoundTripTime(rttPinger: RTTPinger, options: RTTPingerOptions) {
 
   function measureAndReschedule(conn?: Connection) {
     if (rttPinger.closed) {
-      conn?.destroy({ force: true });
+      conn?.destroy();
       return;
     }
 
@@ -529,7 +529,7 @@ function measureRoundTripTime(rttPinger: RTTPinger, options: RTTPingerOptions) {
   connection.command(ns('admin.$cmd'), { [commandName]: 1 }, undefined).then(
     () => measureAndReschedule(),
     () => {
-      rttPinger.connection?.destroy({ force: true });
+      rttPinger.connection?.destroy();
       rttPinger.connection = undefined;
       rttPinger[kRoundTripTime] = 0;
       return;

src/sdam/server.ts

@@ -1,6 +1,6 @@
 import type { Document } from '../bson';
 import { type AutoEncrypter } from '../client-side-encryption/auto_encrypter';
-import { type CommandOptions, Connection, type DestroyOptions } from '../cmap/connection';
+import { type CommandOptions, Connection } from '../cmap/connection';
 import {
   ConnectionPool,
   type ConnectionPoolEvents,
@@ -41,7 +41,6 @@ import type { GetMoreOptions } from '../operations/get_more';
 import type { ClientSession } from '../sessions';
 import { isTransactionCommand } from '../transactions';
 import {
-  type Callback,
   type EventEmitterWithState,
   makeStateMachine,
   maxWireVersion,
@@ -236,18 +235,8 @@ export class Server extends TypedEventEmitter<ServerEvents> {
   }
 
   /** Destroy the server connection */
-  destroy(options?: DestroyOptions, callback?: Callback): void {
-    if (typeof options === 'function') {
-      callback = options;
-      options = { force: false };
-    }
-    options = Object.assign({}, { force: false }, options);
-
+  destroy(): void {
     if (this.s.state === STATE_CLOSED) {
-      if (typeof callback === 'function') {
-        callback();
-      }
-
       return;
     }
 
@@ -257,13 +246,9 @@ export class Server extends TypedEventEmitter<ServerEvents> {
       this.monitor?.close();
     }
 
-    this.pool.close(options, err => {
-      stateTransition(this, STATE_CLOSED);
-      this.emit('closed');
-      if (typeof callback === 'function') {
-        callback(err);
-      }
-    });
+    this.pool.close();
+    stateTransition(this, STATE_CLOSED);
+    this.emit('closed');
   }
 
   /**