refactor(NODE-4632): async await in MongoClient, ClientSession, and AbstractCursor (#3428)

Author: nbbeeken

src/change_stream.ts

@@ -1,5 +1,4 @@
 import type { Readable } from 'stream';
-import { promisify } from 'util';
 
 import type { Binary, Document, Timestamp } from './bson';
 import { Collection } from './collection';
@@ -239,7 +238,7 @@ export interface ChangeStreamInsertDocument<TSchema extends Document = Document>
   operationType: 'insert';
   /** This key will contain the document being inserted */
   fullDocument: TSchema;
-  /** Namespace the insert event occured on */
+  /** Namespace the insert event occurred on */
   ns: ChangeStreamNameSpace;
 }
 
@@ -262,7 +261,7 @@ export interface ChangeStreamUpdateDocument<TSchema extends Document = Document>
   fullDocument?: TSchema;
   /** Contains a description of updated and removed fields in this operation */
   updateDescription: UpdateDescription<TSchema>;
-  /** Namespace the update event occured on */
+  /** Namespace the update event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -285,7 +284,7 @@ export interface ChangeStreamReplaceDocument<TSchema extends Document = Document
   operationType: 'replace';
   /** The fullDocument of a replace event represents the document after the insert of the replacement document */
   fullDocument: TSchema;
-  /** Namespace the replace event occured on */
+  /** Namespace the replace event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -307,7 +306,7 @@ export interface ChangeStreamDeleteDocument<TSchema extends Document = Document>
     ChangeStreamDocumentCollectionUUID {
   /** Describes the type of operation represented in this change notification */
   operationType: 'delete';
-  /** Namespace the delete event occured on */
+  /** Namespace the delete event occurred on */
   ns: ChangeStreamNameSpace;
   /**
    * Contains the pre-image of the modified or deleted document if the
@@ -328,7 +327,7 @@ export interface ChangeStreamDropDocument
     ChangeStreamDocumentCollectionUUID {
   /** Describes the type of operation represented in this change notification */
   operationType: 'drop';
-  /** Namespace the drop event occured on */
+  /** Namespace the drop event occurred on */
   ns: ChangeStreamNameSpace;
 }
 
@@ -343,7 +342,7 @@ export interface ChangeStreamRenameDocument
   operationType: 'rename';
   /** The new name for the `ns.coll` collection */
   to: { db: string; coll: string };
-  /** The "from" namespace that the rename occured on */
+  /** The "from" namespace that the rename occurred on */
   ns: ChangeStreamNameSpace;
 }
 
@@ -918,36 +917,31 @@ export class ChangeStream<
     }
   }
 
-  /**
-   * @internal
-   *
-   * TODO(NODE-4320): promisify selectServer and refactor this code to be async
-   *
-   * we promisify _processErrorIteratorModeCallback until we have a promisifed version of selectServer.
-   */
-  // eslint-disable-next-line @typescript-eslint/unbound-method
-  private _processErrorIteratorMode = promisify(this._processErrorIteratorModeCallback);
-
   /** @internal */
-  private _processErrorIteratorModeCallback(changeStreamError: AnyError, callback: Callback) {
+  private async _processErrorIteratorMode(changeStreamError: AnyError) {
     if (this[kClosed]) {
       // TODO(NODE-3485): Replace with MongoChangeStreamClosedError
-      return callback(new MongoAPIError(CHANGESTREAM_CLOSED_ERROR));
+      throw new MongoAPIError(CHANGESTREAM_CLOSED_ERROR);
     }
 
-    if (isResumableError(changeStreamError, this.cursor.maxWireVersion)) {
-      this.cursor.close().catch(() => null);
-
-      const topology = getTopology(this.parent);
-      topology.selectServer(this.cursor.readPreference, {}, serverSelectionError => {
-        // if the topology can't reconnect, close the stream
-        if (serverSelectionError) return this.close(() => callback(changeStreamError));
+    if (!isResumableError(changeStreamError, this.cursor.maxWireVersion)) {
+      try {
+        await this.close();
+      } catch {
+        // ignore errors from close
+      }
+      throw changeStreamError;
+    }
 
-        this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
-        callback();
-      });
-    } else {
-      this.close(() => callback(changeStreamError));
+    await this.cursor.close().catch(() => null);
+    const topology = getTopology(this.parent);
+    try {
+      await topology.selectServerAsync(this.cursor.readPreference, {});
+      this.cursor = this._createChangeStreamCursor(this.cursor.resumeOptions);
+    } catch {
+      // if the topology can't reconnect, close the stream
+      await this.close();
+      throw changeStreamError;
     }
   }
 }

src/collection.ts

@@ -1478,7 +1478,7 @@ export class Collection<TSchema extends Document = Document> {
    * Create a new Change Stream, watching for new changes (insertions, updates, replacements, deletions, and invalidations) in this collection.
    *
    * @remarks
-   * watch() accepts two generic arguments for distinct usecases:
+   * watch() accepts two generic arguments for distinct use cases:
    * - The first is to override the schema that may be defined for this specific collection
    * - The second is to override the shape of the change stream document entirely, if it is not provided the type will default to ChangeStreamDocument of the first argument
    * @example
@@ -1603,7 +1603,7 @@ export class Collection<TSchema extends Document = Document> {
    *
    * @throws MongoNotConnectedError
    * @remarks
-   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implemenation.
+   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implementation.
    * However, `collection.bulkWrite()` provides an equivalent API that does not require prior connecting.
    */
   initializeUnorderedBulkOp(options?: BulkWriteOptions): UnorderedBulkOperation {
@@ -1615,7 +1615,7 @@ export class Collection<TSchema extends Document = Document> {
    *
    * @throws MongoNotConnectedError
    * @remarks
-   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implemenation.
+   * **NOTE:** MongoClient must be connected prior to calling this method due to a known limitation in this legacy implementation.
    * However, `collection.bulkWrite()` provides an equivalent API that does not require prior connecting.
    */
   initializeOrderedBulkOp(options?: BulkWriteOptions): OrderedBulkOperation {

src/connection_string.ts

@@ -30,7 +30,6 @@ import { ReadPreference, ReadPreferenceMode } from './read_preference';
 import type { TagSet } from './sdam/server_description';
 import {
   AnyOptions,
-  Callback,
   DEFAULT_PK_FACTORY,
   emitWarning,
   emitWarningOnce,
@@ -70,97 +69,89 @@ function matchesParentDomain(srvAddress: string, parentDomain: string): boolean
  * @param uri - The connection string to parse
  * @param options - Optional user provided connection string options
  */
-export function resolveSRVRecord(options: MongoOptions, callback: Callback<HostAddress[]>): void {
+export async function resolveSRVRecord(options: MongoOptions): Promise<HostAddress[]> {
   if (typeof options.srvHost !== 'string') {
-    return callback(new MongoAPIError('Option "srvHost" must not be empty'));
+    throw new MongoAPIError('Option "srvHost" must not be empty');
   }
 
   if (options.srvHost.split('.').length < 3) {
     // TODO(NODE-3484): Replace with MongoConnectionStringError
-    return callback(new MongoAPIError('URI must include hostname, domain name, and tld'));
+    throw new MongoAPIError('URI must include hostname, domain name, and tld');
   }
 
   // Resolve the SRV record and use the result as the list of hosts to connect to.
   const lookupAddress = options.srvHost;
-  dns.resolveSrv(`_${options.srvServiceName}._tcp.${lookupAddress}`, (err, addresses) => {
-    if (err) return callback(err);
+  const addresses = await dns.promises.resolveSrv(
+    `_${options.srvServiceName}._tcp.${lookupAddress}`
+  );
 
-    if (addresses.length === 0) {
-      return callback(new MongoAPIError('No addresses found at host'));
-    }
+  if (addresses.length === 0) {
+    throw new MongoAPIError('No addresses found at host');
+  }
 
-    for (const { name } of addresses) {
-      if (!matchesParentDomain(name, lookupAddress)) {
-        return callback(new MongoAPIError('Server record does not share hostname with parent URI'));
-      }
+  for (const { name } of addresses) {
+    if (!matchesParentDomain(name, lookupAddress)) {
+      throw new MongoAPIError('Server record does not share hostname with parent URI');
     }
+  }
 
-    const hostAddresses = addresses.map(r =>
-      HostAddress.fromString(`${r.name}:${r.port ?? 27017}`)
-    );
+  const hostAddresses = addresses.map(r => HostAddress.fromString(`${r.name}:${r.port ?? 27017}`));
+
+  validateLoadBalancedOptions(hostAddresses, options, true);
 
-    const lbError = validateLoadBalancedOptions(hostAddresses, options, true);
-    if (lbError) {
-      return callback(lbError);
+  // Resolve TXT record and add options from there if they exist.
+  let record;
+  try {
+    record = await dns.promises.resolveTxt(lookupAddress);
+  } catch (error) {
+    if (error.code !== 'ENODATA' && error.code !== 'ENOTFOUND') {
+      throw error;
     }
+    return hostAddresses;
+  }
 
-    // Resolve TXT record and add options from there if they exist.
-    dns.resolveTxt(lookupAddress, (err, record) => {
-      if (err) {
-        if (err.code !== 'ENODATA' && err.code !== 'ENOTFOUND') {
-          return callback(err);
-        }
-      } else {
-        if (record.length > 1) {
-          return callback(new MongoParseError('Multiple text records not allowed'));
-        }
+  if (record.length > 1) {
+    throw new MongoParseError('Multiple text records not allowed');
+  }
 
-        const txtRecordOptions = new URLSearchParams(record[0].join(''));
-        const txtRecordOptionKeys = [...txtRecordOptions.keys()];
-        if (txtRecordOptionKeys.some(key => !VALID_TXT_RECORDS.includes(key))) {
-          return callback(
-            new MongoParseError(`Text record may only set any of: ${VALID_TXT_RECORDS.join(', ')}`)
-          );
-        }
+  const txtRecordOptions = new URLSearchParams(record[0].join(''));
+  const txtRecordOptionKeys = [...txtRecordOptions.keys()];
+  if (txtRecordOptionKeys.some(key => !VALID_TXT_RECORDS.includes(key))) {
+    throw new MongoParseError(`Text record may only set any of: ${VALID_TXT_RECORDS.join(', ')}`);
+  }
 
-        if (VALID_TXT_RECORDS.some(option => txtRecordOptions.get(option) === '')) {
-          return callback(new MongoParseError('Cannot have empty URI params in DNS TXT Record'));
-        }
+  if (VALID_TXT_RECORDS.some(option => txtRecordOptions.get(option) === '')) {
+    throw new MongoParseError('Cannot have empty URI params in DNS TXT Record');
+  }
 
-        const source = txtRecordOptions.get('authSource') ?? undefined;
-        const replicaSet = txtRecordOptions.get('replicaSet') ?? undefined;
-        const loadBalanced = txtRecordOptions.get('loadBalanced') ?? undefined;
-
-        if (
-          !options.userSpecifiedAuthSource &&
-          source &&
-          options.credentials &&
-          !AUTH_MECHS_AUTH_SRC_EXTERNAL.has(options.credentials.mechanism)
-        ) {
-          options.credentials = MongoCredentials.merge(options.credentials, { source });
-        }
+  const source = txtRecordOptions.get('authSource') ?? undefined;
+  const replicaSet = txtRecordOptions.get('replicaSet') ?? undefined;
+  const loadBalanced = txtRecordOptions.get('loadBalanced') ?? undefined;
 
-        if (!options.userSpecifiedReplicaSet && replicaSet) {
-          options.replicaSet = replicaSet;
-        }
+  if (
+    !options.userSpecifiedAuthSource &&
+    source &&
+    options.credentials &&
+    !AUTH_MECHS_AUTH_SRC_EXTERNAL.has(options.credentials.mechanism)
+  ) {
+    options.credentials = MongoCredentials.merge(options.credentials, { source });
+  }
 
-        if (loadBalanced === 'true') {
-          options.loadBalanced = true;
-        }
+  if (!options.userSpecifiedReplicaSet && replicaSet) {
+    options.replicaSet = replicaSet;
+  }
 
-        if (options.replicaSet && options.srvMaxHosts > 0) {
-          return callback(new MongoParseError('Cannot combine replicaSet option with srvMaxHosts'));
-        }
+  if (loadBalanced === 'true') {
+    options.loadBalanced = true;
+  }
 
-        const lbError = validateLoadBalancedOptions(hostAddresses, options, true);
-        if (lbError) {
-          return callback(lbError);
-        }
-      }
+  if (options.replicaSet && options.srvMaxHosts > 0) {
+    throw new MongoParseError('Cannot combine replicaSet option with srvMaxHosts');
+  }
 
-      callback(undefined, hostAddresses);
-    });
-  });
+  validateLoadBalancedOptions(hostAddresses, options, true);
+
+  return hostAddresses;
 }
 
 /**
@@ -442,10 +433,8 @@ export function parseOptions(
     PromiseProvider.set(options.promiseLibrary);
   }
 
-  const lbError = validateLoadBalancedOptions(hosts, mongoOptions, isSRV);
-  if (lbError) {
-    throw lbError;
-  }
+  validateLoadBalancedOptions(hosts, mongoOptions, isSRV);
+
   if (mongoClient && mongoOptions.autoEncryption) {
     Encrypter.checkForMongoCrypt();
     mongoOptions.encrypter = new Encrypter(mongoClient, uri, options);
@@ -522,24 +511,33 @@ export function parseOptions(
   return mongoOptions;
 }
 
+/**
+ * #### Throws if LB mode is true:
+ * - hosts contains more than one host
+ * - there is a replicaSet name set
+ * - directConnection is set
+ * - if srvMaxHosts is used when an srv connection string is passed in
+ *
+ * @throws MongoParseError
+ */
 function validateLoadBalancedOptions(
   hosts: HostAddress[] | string[],
   mongoOptions: MongoOptions,
   isSrv: boolean
-): MongoParseError | undefined {
+): void {
   if (mongoOptions.loadBalanced) {
     if (hosts.length > 1) {
-      return new MongoParseError(LB_SINGLE_HOST_ERROR);
+      throw new MongoParseError(LB_SINGLE_HOST_ERROR);
     }
     if (mongoOptions.replicaSet) {
-      return new MongoParseError(LB_REPLICA_SET_ERROR);
+      throw new MongoParseError(LB_REPLICA_SET_ERROR);
     }
     if (mongoOptions.directConnection) {
-      return new MongoParseError(LB_DIRECT_CONNECTION_ERROR);
+      throw new MongoParseError(LB_DIRECT_CONNECTION_ERROR);
     }
 
     if (isSrv && mongoOptions.srvMaxHosts > 0) {
-      return new MongoParseError('Cannot limit srv hosts with loadBalanced enabled');
+      throw new MongoParseError('Cannot limit srv hosts with loadBalanced enabled');
     }
   }
   return;