feat(NODE-6258): add signal support to cursor APIs

Author: nbbeeken

src/client-side-encryption/auto_encrypter.ts

@@ -11,6 +11,7 @@ import { kDecorateResult } from '../constants';
 import { getMongoDBClientEncryption } from '../deps';
 import { MongoRuntimeError } from '../error';
 import { MongoClient, type MongoClientOptions } from '../mongo_client';
+import { type Abortable } from '../mongo_types';
 import { MongoDBCollectionNamespace } from '../utils';
 import { autoSelectSocketOptions } from './client_encryption';
 import * as cryptoCallbacks from './crypto_callbacks';
@@ -372,8 +373,10 @@ export class AutoEncrypter {
   async encrypt(
     ns: string,
     cmd: Document,
-    options: CommandOptions = {}
+    options: CommandOptions & Abortable = {}
   ): Promise<Document | Uint8Array> {
+    options.signal?.throwIfAborted();
+
     if (this._bypassEncryption) {
       // If `bypassAutoEncryption` has been specified, don't encrypt
       return cmd;
@@ -398,7 +401,7 @@ export class AutoEncrypter {
       socketOptions: autoSelectSocketOptions(this._client.s.options)
     });
 
-    return deserialize(await stateMachine.execute(this, context, options.timeoutContext), {
+    return deserialize(await stateMachine.execute(this, context, options), {
       promoteValues: false,
       promoteLongs: false
     });
@@ -407,7 +410,12 @@ export class AutoEncrypter {
   /**
    * Decrypt a command response
    */
-  async decrypt(response: Uint8Array, options: CommandOptions = {}): Promise<Uint8Array> {
+  async decrypt(
+    response: Uint8Array,
+    options: CommandOptions & Abortable = {}
+  ): Promise<Uint8Array> {
+    options.signal?.throwIfAborted();
+
     const context = this._mongocrypt.makeDecryptionContext(response);
 
     context.id = this._contextCounter++;
@@ -419,11 +427,7 @@ export class AutoEncrypter {
       socketOptions: autoSelectSocketOptions(this._client.s.options)
     });
 
-    return await stateMachine.execute(
-      this,
-      context,
-      options.timeoutContext?.csotEnabled() ? options.timeoutContext : undefined
-    );
+    return await stateMachine.execute(this, context, options);
   }
 
   /**

src/client-side-encryption/client_encryption.ts

@@ -225,7 +225,7 @@ export class ClientEncryption {
       TimeoutContext.create(resolveTimeoutOptions(this._client, { timeoutMS: this._timeoutMS }));
 
     const dataKey = deserialize(
-      await stateMachine.execute(this, context, timeoutContext)
+      await stateMachine.execute(this, context, { timeoutContext })
     ) as DataKey;
 
     const { db: dbName, collection: collectionName } = MongoDBCollectionNamespace.fromString(
@@ -293,7 +293,9 @@ export class ClientEncryption {
       resolveTimeoutOptions(this._client, { timeoutMS: this._timeoutMS })
     );
 
-    const { v: dataKeys } = deserialize(await stateMachine.execute(this, context, timeoutContext));
+    const { v: dataKeys } = deserialize(
+      await stateMachine.execute(this, context, { timeoutContext })
+    );
     if (dataKeys.length === 0) {
       return {};
     }
@@ -696,7 +698,7 @@ export class ClientEncryption {
         ? TimeoutContext.create(resolveTimeoutOptions(this._client, { timeoutMS: this._timeoutMS }))
         : undefined;
 
-    const { v } = deserialize(await stateMachine.execute(this, context, timeoutContext));
+    const { v } = deserialize(await stateMachine.execute(this, context, { timeoutContext }));
 
     return v;
   }
@@ -780,7 +782,7 @@ export class ClientEncryption {
       this._timeoutMS != null
         ? TimeoutContext.create(resolveTimeoutOptions(this._client, { timeoutMS: this._timeoutMS }))
         : undefined;
-    const { v } = deserialize(await stateMachine.execute(this, context, timeoutContext));
+    const { v } = deserialize(await stateMachine.execute(this, context, { timeoutContext }));
     return v;
   }
 }

src/client-side-encryption/state_machine.ts

@@ -15,8 +15,15 @@ import { CursorTimeoutContext } from '../cursor/abstract_cursor';
 import { getSocks, type SocksLib } from '../deps';
 import { MongoOperationTimeoutError } from '../error';
 import { type MongoClient, type MongoClientOptions } from '../mongo_client';
+import { type Abortable } from '../mongo_types';
 import { Timeout, type TimeoutContext, TimeoutError } from '../timeout';
-import { BufferPool, MongoDBCollectionNamespace, promiseWithResolvers } from '../utils';
+import {
+  addAbortListener,
+  BufferPool,
+  kDispose,
+  MongoDBCollectionNamespace,
+  promiseWithResolvers
+} from '../utils';
 import { autoSelectSocketOptions, type DataKey } from './client_encryption';
 import { MongoCryptError } from './errors';
 import { type MongocryptdManager } from './mongocryptd_manager';
@@ -189,7 +196,7 @@ export class StateMachine {
   async execute(
     executor: StateMachineExecutable,
     context: MongoCryptContext,
-    timeoutContext?: TimeoutContext
+    options: { timeoutContext?: TimeoutContext } & Abortable
   ): Promise<Uint8Array> {
     const keyVaultNamespace = executor._keyVaultNamespace;
     const keyVaultClient = executor._keyVaultClient;
@@ -199,6 +206,7 @@ export class StateMachine {
     let result: Uint8Array | null = null;
 
     while (context.state !== MONGOCRYPT_CTX_DONE && context.state !== MONGOCRYPT_CTX_ERROR) {
+      options.signal?.throwIfAborted();
       debug(`[context#${context.id}] ${stateToString.get(context.state) || context.state}`);
 
       switch (context.state) {
@@ -214,7 +222,7 @@ export class StateMachine {
             metaDataClient,
             context.ns,
             filter,
-            timeoutContext
+            options
           );
           if (collInfo) {
             context.addMongoOperationResponse(collInfo);
@@ -235,9 +243,9 @@ export class StateMachine {
           // When we are using the shared library, we don't have a mongocryptd manager.
           const markedCommand: Uint8Array = mongocryptdManager
             ? await mongocryptdManager.withRespawn(
-                this.markCommand.bind(this, mongocryptdClient, context.ns, command, timeoutContext)
+                this.markCommand.bind(this, mongocryptdClient, context.ns, command, options)
               )
-            : await this.markCommand(mongocryptdClient, context.ns, command, timeoutContext);
+            : await this.markCommand(mongocryptdClient, context.ns, command, options);
 
           context.addMongoOperationResponse(markedCommand);
           context.finishMongoOperation();
@@ -246,12 +254,7 @@ export class StateMachine {
 
         case MONGOCRYPT_CTX_NEED_MONGO_KEYS: {
           const filter = context.nextMongoOperation();
-          const keys = await this.fetchKeys(
-            keyVaultClient,
-            keyVaultNamespace,
-            filter,
-            timeoutContext
-          );
+          const keys = await this.fetchKeys(keyVaultClient, keyVaultNamespace, filter, options);
 
           if (keys.length === 0) {
             // See docs on EMPTY_V
@@ -273,7 +276,7 @@ export class StateMachine {
         }
 
         case MONGOCRYPT_CTX_NEED_KMS: {
-          await Promise.all(this.requests(context, timeoutContext));
+          await Promise.all(this.requests(context, options));
           context.finishKMSRequests();
           break;
         }
@@ -315,11 +318,13 @@ export class StateMachine {
    * @param kmsContext - A C++ KMS context returned from the bindings
    * @returns A promise that resolves when the KMS reply has be fully parsed
    */
-  async kmsRequest(request: MongoCryptKMSRequest, timeoutContext?: TimeoutContext): Promise<void> {
+  async kmsRequest(
+    request: MongoCryptKMSRequest,
+    options?: { timeoutContext?: TimeoutContext } & Abortable
+  ): Promise<void> {
     const parsedUrl = request.endpoint.split(':');
     const port = parsedUrl[1] != null ? Number.parseInt(parsedUrl[1], 10) : HTTPS_PORT;
-    const socketOptions = autoSelectSocketOptions(this.options.socketOptions || {});
-    const options: tls.ConnectionOptions & {
+    const socketOptions: tls.ConnectionOptions & {
       host: string;
       port: number;
       autoSelectFamily?: boolean;
@@ -328,7 +333,7 @@ export class StateMachine {
       host: parsedUrl[0],
       servername: parsedUrl[0],
       port,
-      ...socketOptions
+      ...autoSelectSocketOptions(this.options.socketOptions || {})
     };
     const message = request.message;
     const buffer = new BufferPool();
@@ -363,7 +368,7 @@ export class StateMachine {
           throw error;
         }
         try {
-          await this.setTlsOptions(providerTlsOptions, options);
+          await this.setTlsOptions(providerTlsOptions, socketOptions);
         } catch (err) {
           throw onerror(err);
         }
@@ -380,23 +385,25 @@ export class StateMachine {
       .once('close', () => rejectOnNetSocketError(onclose()))
       .once('connect', () => resolveOnNetSocketConnect());
 
+    let abortListener;
+
     try {
       if (this.options.proxyOptions && this.options.proxyOptions.proxyHost) {
         const netSocketOptions = {
+          ...socketOptions,
           host: this.options.proxyOptions.proxyHost,
-          port: this.options.proxyOptions.proxyPort || 1080,
-          ...socketOptions
+          port: this.options.proxyOptions.proxyPort || 1080
         };
         netSocket.connect(netSocketOptions);
         await willConnect;
 
         try {
           socks ??= loadSocks();
-          options.socket = (
+          socketOptions.socket = (
             await socks.SocksClient.createConnection({
               existing_socket: netSocket,
               command: 'connect',
-              destination: { host: options.host, port: options.port },
+              destination: { host: socketOptions.host, port: socketOptions.port },
               proxy: {
                 // host and port are ignored because we pass existing_socket
                 host: 'iLoveJavaScript',
@@ -412,7 +419,7 @@ export class StateMachine {
         }
       }
 
-      socket = tls.connect(options, () => {
+      socket = tls.connect(socketOptions, () => {
         socket.write(message);
       });
 
@@ -422,6 +429,11 @@ export class StateMachine {
         resolve
       } = promiseWithResolvers<void>();
 
+      abortListener = addAbortListener(options?.signal, function () {
+        destroySockets();
+        rejectOnTlsSocketError(this.reason);
+      });
+
       socket
         .once('error', err => rejectOnTlsSocketError(onerror(err)))
         .once('close', () => rejectOnTlsSocketError(onclose()))
@@ -436,8 +448,11 @@ export class StateMachine {
             resolve();
           }
         });
-      await (timeoutContext?.csotEnabled()
-        ? Promise.all([willResolveKmsRequest, Timeout.expires(timeoutContext?.remainingTimeMS)])
+      await (options?.timeoutContext?.csotEnabled()
+        ? Promise.all([
+            willResolveKmsRequest,
+            Timeout.expires(options.timeoutContext?.remainingTimeMS)
+          ])
         : willResolveKmsRequest);
     } catch (error) {
       if (error instanceof TimeoutError)
@@ -446,16 +461,17 @@ export class StateMachine {
     } finally {
       // There's no need for any more activity on this socket at this point.
       destroySockets();
+      abortListener?.[kDispose]();
     }
   }
 
-  *requests(context: MongoCryptContext, timeoutContext?: TimeoutContext) {
+  *requests(context: MongoCryptContext, options?: { timeoutContext?: TimeoutContext } & Abortable) {
     for (
       let request = context.nextKMSRequest();
       request != null;
       request = context.nextKMSRequest()
     ) {
-      yield this.kmsRequest(request, timeoutContext);
+      yield this.kmsRequest(request, options);
     }
   }
 
@@ -516,14 +532,16 @@ export class StateMachine {
     client: MongoClient,
     ns: string,
     filter: Document,
-    timeoutContext?: TimeoutContext
+    options?: { timeoutContext?: TimeoutContext } & Abortable
   ): Promise<Uint8Array | null> {
     const { db } = MongoDBCollectionNamespace.fromString(ns);
 
     const cursor = client.db(db).listCollections(filter, {
       promoteLongs: false,
       promoteValues: false,
-      timeoutContext: timeoutContext && new CursorTimeoutContext(timeoutContext, Symbol())
+      timeoutContext:
+        options?.timeoutContext && new CursorTimeoutContext(options?.timeoutContext, Symbol()),
+      signal: options?.signal
     });
 
     // There is always exactly zero or one matching documents, so this should always exhaust the cursor
@@ -547,17 +565,30 @@ export class StateMachine {
     client: MongoClient,
     ns: string,
     command: Uint8Array,
-    timeoutContext?: TimeoutContext
+    options?: { timeoutContext?: TimeoutContext } & Abortable
   ): Promise<Uint8Array> {
     const { db } = MongoDBCollectionNamespace.fromString(ns);
     const bsonOptions = { promoteLongs: false, promoteValues: false };
     const rawCommand = deserialize(command, bsonOptions);
 
+    const commandOptions: {
+      timeoutMS?: number;
+      signal?: AbortSignal;
+    } = {
+      timeoutMS: undefined,
+      signal: undefined
+    };
+
+    if (options?.timeoutContext?.csotEnabled()) {
+      commandOptions.timeoutMS = options.timeoutContext.remainingTimeMS;
+    }
+    if (options?.signal) {
+      commandOptions.signal = options.signal;
+    }
+
     const response = await client.db(db).command(rawCommand, {
       ...bsonOptions,
-      ...(timeoutContext?.csotEnabled()
-        ? { timeoutMS: timeoutContext?.remainingTimeMS }
-        : undefined)
+      ...commandOptions
     });
 
     return serialize(response, this.bsonOptions);
@@ -575,17 +606,30 @@ export class StateMachine {
     client: MongoClient,
     keyVaultNamespace: string,
     filter: Uint8Array,
-    timeoutContext?: TimeoutContext
+    options?: { timeoutContext?: TimeoutContext } & Abortable
   ): Promise<Array<DataKey>> {
     const { db: dbName, collection: collectionName } =
       MongoDBCollectionNamespace.fromString(keyVaultNamespace);
 
+    const commandOptions: {
+      timeoutContext?: CursorTimeoutContext;
+      signal?: AbortSignal;
+    } = {
+      timeoutContext: undefined,
+      signal: undefined
+    };
+
+    if (options?.timeoutContext != null) {
+      commandOptions.timeoutContext = new CursorTimeoutContext(options.timeoutContext, Symbol());
+    }
+    if (options?.signal != null) {
+      commandOptions.signal = options.signal;
+    }
+
     return client
       .db(dbName)
       .collection<DataKey>(collectionName, { readConcern: { level: 'majority' } })
-      .find(deserialize(filter), {
-        timeoutContext: timeoutContext && new CursorTimeoutContext(timeoutContext, Symbol())
-      })
+      .find(deserialize(filter), commandOptions)
       .toArray();
   }
 }

src/cmap/connection.ts

@@ -33,7 +33,7 @@ import {
 import type { ServerApi, SupportedNodeConnectionOptions } from '../mongo_client';
 import { type MongoClientAuthProviders } from '../mongo_client_auth_providers';
 import { MongoLoggableComponent, type MongoLogger, SeverityLevel } from '../mongo_logger';
-import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
+import { type Abortable, type CancellationToken, TypedEventEmitter } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import { ServerType } from '../sdam/common';
 import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
@@ -438,7 +438,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
   private async *sendWire(
     message: WriteProtocolMessageType,
-    options: CommandOptions,
+    options: CommandOptions & Abortable,
     responseType?: MongoDBResponseConstructor
   ): AsyncGenerator<MongoDBResponse> {
     this.throwIfAborted();
@@ -453,7 +453,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       await this.writeCommand(message, {
         agreedCompressor: this.description.compressor ?? 'none',
         zlibCompressionLevel: this.description.zlibCompressionLevel,
-        timeoutContext: options.timeoutContext
+        timeoutContext: options.timeoutContext,
+        signal: options.signal
       });
 
       if (options.noResponse || message.moreToCome) {
@@ -473,7 +474,10 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         );
       }
 
-      for await (const response of this.readMany({ timeoutContext: options.timeoutContext })) {
+      for await (const response of this.readMany({
+        timeoutContext: options.timeoutContext,
+        signal: options.signal
+      })) {
         this.socket.setTimeout(0);
         const bson = response.parse();
 
@@ -676,7 +680,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       agreedCompressor?: CompressorName;
       zlibCompressionLevel?: number;
       timeoutContext?: TimeoutContext;
-    }
+    } & Abortable
   ): Promise<void> {
     const finalCommand =
       options.agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)
@@ -701,7 +705,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
     if (this.socket.write(buffer)) return;
 
-    const drainEvent = once<void>(this.socket, 'drain');
+    const drainEvent = once<void>(this.socket, 'drain', { signal: options.signal });
     const timeout = options?.timeoutContext?.timeoutForSocketWrite;
     if (timeout) {
       try {
@@ -729,9 +733,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    *
    * Note that `for-await` loops call `return` automatically when the loop is exited.
    */
-  private async *readMany(options: {
-    timeoutContext?: TimeoutContext;
-  }): AsyncGenerator<OpMsgResponse | OpReply> {
+  private async *readMany(
+    options: {
+      timeoutContext?: TimeoutContext;
+    } & Abortable
+  ): AsyncGenerator<OpMsgResponse | OpReply> {
     try {
       this.dataEvents = onData(this.messageStream, options);
       this.messageStream.resume();

src/cmap/connection_pool.ts

@@ -25,10 +25,18 @@ import {
   MongoRuntimeError,
   MongoServerError
 } from '../error';
-import { CancellationToken, TypedEventEmitter } from '../mongo_types';
+import { type Abortable, CancellationToken, TypedEventEmitter } from '../mongo_types';
 import type { Server } from '../sdam/server';
 import { type TimeoutContext, TimeoutError } from '../timeout';
-import { type Callback, List, makeCounter, now, promiseWithResolvers } from '../utils';
+import {
+  addAbortListener,
+  type Callback,
+  kDispose,
+  List,
+  makeCounter,
+  now,
+  promiseWithResolvers
+} from '../utils';
 import { connect } from './connect';
 import { Connection, type ConnectionEvents, type ConnectionOptions } from './connection';
 import {
@@ -316,7 +324,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
    * will be held by the pool. This means that if a connection is checked out it MUST be checked back in or
    * explicitly destroyed by the new owner.
    */
-  async checkOut(options: { timeoutContext: TimeoutContext }): Promise<Connection> {
+  async checkOut(options: { timeoutContext: TimeoutContext } & Abortable): Promise<Connection> {
     const checkoutTime = now();
     this.emitAndLog(
       ConnectionPool.CONNECTION_CHECK_OUT_STARTED,
@@ -334,6 +342,11 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       checkoutTime
     };
 
+    const abortListener = addAbortListener(options.signal, function () {
+      waitQueueMember.cancelled = true;
+      reject(this.reason);
+    });
+
     this.waitQueue.push(waitQueueMember);
     process.nextTick(() => this.processWaitQueue());
 
@@ -364,6 +377,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       }
       throw error;
     } finally {
+      abortListener?.[kDispose]();
       timeout?.clear();
     }
   }

src/cmap/wire_protocol/on_data.ts

@@ -1,7 +1,8 @@
 import { type EventEmitter } from 'events';
 
+import { type Abortable } from '../../mongo_types';
 import { type TimeoutContext } from '../../timeout';
-import { List, promiseWithResolvers } from '../../utils';
+import { addAbortListener, kDispose, List, promiseWithResolvers } from '../../utils';
 
 /**
  * @internal
@@ -21,8 +22,10 @@ type PendingPromises = Omit<
  */
 export function onData(
   emitter: EventEmitter,
-  { timeoutContext }: { timeoutContext?: TimeoutContext }
+  { timeoutContext, signal }: { timeoutContext?: TimeoutContext } & Abortable
 ) {
+  signal?.throwIfAborted();
+
   // Setup pending events and pending promise lists
   /**
    * When the caller has not yet called .next(), we store the
@@ -90,6 +93,9 @@ export function onData(
   // Adding event handlers
   emitter.on('data', eventHandler);
   emitter.on('error', errorHandler);
+  const abortListener = addAbortListener(signal, function () {
+    errorHandler(this.reason);
+  });
 
   const timeoutForSocketRead = timeoutContext?.timeoutForSocketRead;
   timeoutForSocketRead?.throwIfExpired();
@@ -115,6 +121,7 @@ export function onData(
     // Adding event handlers
     emitter.off('data', eventHandler);
     emitter.off('error', errorHandler);
+    abortListener?.[kDispose]();
     finished = true;
     timeoutForSocketRead?.clear();
     const doneResult = { value: undefined, done: finished } as const;

src/collection.ts

@@ -14,6 +14,7 @@ import type { Db } from './db';
 import { MongoInvalidArgumentError, MongoOperationTimeoutError } from './error';
 import type { MongoClient, PkFactory } from './mongo_client';
 import type {
+  Abortable,
   Filter,
   Flatten,
   OptionalUnlessRequiredId,
@@ -505,20 +506,20 @@ export class Collection<TSchema extends Document = Document> {
   async findOne(filter: Filter<TSchema>): Promise<WithId<TSchema> | null>;
   async findOne(
     filter: Filter<TSchema>,
-    options: Omit<FindOptions, 'timeoutMode'>
+    options: Omit<FindOptions, 'timeoutMode'> & Abortable
   ): Promise<WithId<TSchema> | null>;
 
   // allow an override of the schema.
   async findOne<T = TSchema>(): Promise<T | null>;
   async findOne<T = TSchema>(filter: Filter<TSchema>): Promise<T | null>;
   async findOne<T = TSchema>(
     filter: Filter<TSchema>,
-    options?: Omit<FindOptions, 'timeoutMode'>
+    options?: Omit<FindOptions, 'timeoutMode'> & Abortable
   ): Promise<T | null>;
 
   async findOne(
     filter: Filter<TSchema> = {},
-    options: FindOptions = {}
+    options: FindOptions & Abortable = {}
   ): Promise<WithId<TSchema> | null> {
     const cursor = this.find(filter, options).limit(-1).batchSize(1);
     const res = await cursor.next();
@@ -532,9 +533,15 @@ export class Collection<TSchema extends Document = Document> {
    * @param filter - The filter predicate. If unspecified, then all documents in the collection will match the predicate
    */
   find(): FindCursor<WithId<TSchema>>;
-  find(filter: Filter<TSchema>, options?: FindOptions): FindCursor<WithId<TSchema>>;
-  find<T extends Document>(filter: Filter<TSchema>, options?: FindOptions): FindCursor<T>;
-  find(filter: Filter<TSchema> = {}, options: FindOptions = {}): FindCursor<WithId<TSchema>> {
+  find(filter: Filter<TSchema>, options?: FindOptions & Abortable): FindCursor<WithId<TSchema>>;
+  find<T extends Document>(
+    filter: Filter<TSchema>,
+    options?: FindOptions & Abortable
+  ): FindCursor<T>;
+  find(
+    filter: Filter<TSchema> = {},
+    options: FindOptions & Abortable = {}
+  ): FindCursor<WithId<TSchema>> {
     return new FindCursor<WithId<TSchema>>(
       this.client,
       this.s.namespace,
@@ -792,7 +799,7 @@ export class Collection<TSchema extends Document = Document> {
    */
   async countDocuments(
     filter: Filter<TSchema> = {},
-    options: CountDocumentsOptions = {}
+    options: CountDocumentsOptions & Abortable = {}
   ): Promise<number> {
     const pipeline = [];
     pipeline.push({ $match: filter });
@@ -1006,7 +1013,7 @@ export class Collection<TSchema extends Document = Document> {
    */
   aggregate<T extends Document = Document>(
     pipeline: Document[] = [],
-    options?: AggregateOptions
+    options?: AggregateOptions & Abortable
   ): AggregationCursor<T> {
     if (!Array.isArray(pipeline)) {
       throw new MongoInvalidArgumentError(

src/cursor/abstract_cursor.ts

@@ -12,7 +12,7 @@ import {
   MongoTailableCursorError
 } from '../error';
 import type { MongoClient } from '../mongo_client';
-import { TypedEventEmitter } from '../mongo_types';
+import { type Abortable, TypedEventEmitter } from '../mongo_types';
 import { executeOperation } from '../operations/execute_operation';
 import { GetMoreOperation } from '../operations/get_more';
 import { KillCursorsOperation } from '../operations/kill_cursors';
@@ -22,7 +22,13 @@ import { type AsyncDisposable, configureResourceManagement } from '../resource_m
 import type { Server } from '../sdam/server';
 import { ClientSession, maybeClearPinnedConnection } from '../sessions';
 import { type CSOTTimeoutContext, type Timeout, TimeoutContext } from '../timeout';
-import { type MongoDBNamespace, squashError } from '../utils';
+import {
+  addAbortListener,
+  type Disposable,
+  kDispose,
+  type MongoDBNamespace,
+  squashError
+} from '../utils';
 
 /**
  * @internal
@@ -247,12 +253,14 @@ export abstract class AbstractCursor<
 
   /** @internal */
   protected deserializationOptions: OnDemandDocumentDeserializeOptions;
+  protected signal: AbortSignal | undefined;
+  private abortListener: Disposable | undefined;
 
   /** @internal */
   protected constructor(
     client: MongoClient,
     namespace: MongoDBNamespace,
-    options: AbstractCursorOptions = {}
+    options: AbstractCursorOptions & Abortable = {}
   ) {
     super();
 
@@ -352,6 +360,9 @@ export abstract class AbstractCursor<
     };
 
     this.timeoutContext = options.timeoutContext;
+    this.signal = options.signal;
+    // eslint-disable-next-line @typescript-eslint/no-misused-promises
+    this.abortListener = addAbortListener(this.signal, this.close.bind(this, undefined));
   }
 
   /**
@@ -455,6 +466,8 @@ export abstract class AbstractCursor<
   }
 
   async *[Symbol.asyncIterator](): AsyncGenerator<TSchema, void, void> {
+    this.signal?.throwIfAborted();
+
     if (this.closed) {
       return;
     }
@@ -481,6 +494,8 @@ export abstract class AbstractCursor<
         }
 
         yield document;
+
+        this.signal?.throwIfAborted();
       }
     } finally {
       // Only close the cursor if it has not already been closed. This finally clause handles
@@ -496,9 +511,16 @@ export abstract class AbstractCursor<
   }
 
   stream(options?: CursorStreamOptions): Readable & AsyncIterable<TSchema> {
+    const readable = new ReadableCursorStream(this);
+    const abortListener = addAbortListener(this.signal, function () {
+      readable.destroy(this.reason);
+    });
+    readable.once('end', () => {
+      abortListener?.[kDispose]();
+    });
+
     if (options?.transform) {
       const transform = options.transform;
-      const readable = new ReadableCursorStream(this);
 
       const transformedStream = readable.pipe(
         new Transform({
@@ -522,10 +544,12 @@ export abstract class AbstractCursor<
       return transformedStream;
     }
 
-    return new ReadableCursorStream(this);
+    return readable;
   }
 
   async hasNext(): Promise<boolean> {
+    this.signal?.throwIfAborted();
+
     if (this.cursorId === Long.ZERO) {
       return false;
     }
@@ -551,6 +575,8 @@ export abstract class AbstractCursor<
 
   /** Get the next available document from the cursor, returns null if no more documents are available. */
   async next(): Promise<TSchema | null> {
+    this.signal?.throwIfAborted();
+
     if (this.cursorId === Long.ZERO) {
       throw new MongoCursorExhaustedError();
     }
@@ -581,6 +607,8 @@ export abstract class AbstractCursor<
    * Try to get the next available document from the cursor or `null` if an empty batch is returned
    */
   async tryNext(): Promise<TSchema | null> {
+    this.signal?.throwIfAborted();
+
     if (this.cursorId === Long.ZERO) {
       throw new MongoCursorExhaustedError();
     }
@@ -620,6 +648,8 @@ export abstract class AbstractCursor<
    * @deprecated - Will be removed in a future release. Use for await...of instead.
    */
   async forEach(iterator: (doc: TSchema) => boolean | void): Promise<void> {
+    this.signal?.throwIfAborted();
+
     if (typeof iterator !== 'function') {
       throw new MongoInvalidArgumentError('Argument "iterator" must be a function');
     }
@@ -645,6 +675,8 @@ export abstract class AbstractCursor<
    * cursor.rewind() can be used to reset the cursor.
    */
   async toArray(): Promise<TSchema[]> {
+    this.signal?.throwIfAborted();
+
     const array: TSchema[] = [];
     // at the end of the loop (since readBufferedDocuments is called) the buffer will be empty
     // then, the 'await of' syntax will run a getMore call
@@ -968,6 +1000,7 @@ export abstract class AbstractCursor<
 
   /** @internal */
   private async cleanup(timeoutMS?: number, error?: Error) {
+    this.abortListener?.[kDispose]();
     this.isClosed = true;
     const session = this.cursorSession;
     const timeoutContextForKillCursors = (): CursorTimeoutContext | undefined => {

src/cursor/aggregation_cursor.ts

@@ -8,6 +8,7 @@ import {
   validateExplainTimeoutOptions
 } from '../explain';
 import type { MongoClient } from '../mongo_client';
+import { type Abortable } from '../mongo_types';
 import { AggregateOperation, type AggregateOptions } from '../operations/aggregate';
 import { executeOperation } from '../operations/execute_operation';
 import type { ClientSession } from '../sessions';
@@ -39,7 +40,7 @@ export class AggregationCursor<TSchema = any> extends ExplainableCursor<TSchema>
     client: MongoClient,
     namespace: MongoDBNamespace,
     pipeline: Document[] = [],
-    options: AggregateOptions = {}
+    options: AggregateOptions & Abortable = {}
   ) {
     super(client, namespace, options);
 

src/cursor/find_cursor.ts

@@ -72,7 +72,8 @@ export class FindCursor<TSchema = any> extends ExplainableCursor<TSchema> {
     const options = {
       ...this.findOptions, // NOTE: order matters here, we may need to refine this
       ...this.cursorOptions,
-      session
+      session,
+      signal: this.signal
     };
 
     if (options.explain) {

src/db.ts

@@ -8,7 +8,7 @@ import { ListCollectionsCursor } from './cursor/list_collections_cursor';
 import { RunCommandCursor, type RunCursorCommandOptions } from './cursor/run_command_cursor';
 import { MongoInvalidArgumentError } from './error';
 import type { MongoClient, PkFactory } from './mongo_client';
-import type { TODO_NODE_3286 } from './mongo_types';
+import type { Abortable, TODO_NODE_3286 } from './mongo_types';
 import type { AggregateOptions } from './operations/aggregate';
 import { CollectionsOperation } from './operations/collections';
 import {
@@ -273,7 +273,7 @@ export class Db {
    * @param command - The command to run
    * @param options - Optional settings for the command
    */
-  async command(command: Document, options?: RunCommandOptions): Promise<Document> {
+  async command(command: Document, options?: RunCommandOptions & Abortable): Promise<Document> {
     // Intentionally, we do not inherit options from parent for this operation.
     return await executeOperation(
       this.client,
@@ -284,7 +284,8 @@ export class Db {
           ...resolveBSONOptions(options),
           timeoutMS: options?.timeoutMS ?? this.timeoutMS,
           session: options?.session,
-          readPreference: options?.readPreference
+          readPreference: options?.readPreference,
+          signal: options?.signal
         })
       )
     );
@@ -351,22 +352,25 @@ export class Db {
    */
   listCollections(
     filter: Document,
-    options: Exclude<ListCollectionsOptions, 'nameOnly'> & { nameOnly: true }
+    options: Exclude<ListCollectionsOptions, 'nameOnly'> & { nameOnly: true } & Abortable
   ): ListCollectionsCursor<Pick<CollectionInfo, 'name' | 'type'>>;
   listCollections(
     filter: Document,
-    options: Exclude<ListCollectionsOptions, 'nameOnly'> & { nameOnly: false }
+    options: Exclude<ListCollectionsOptions, 'nameOnly'> & { nameOnly: false } & Abortable
   ): ListCollectionsCursor<CollectionInfo>;
   listCollections<
     T extends Pick<CollectionInfo, 'name' | 'type'> | CollectionInfo =
       | Pick<CollectionInfo, 'name' | 'type'>
       | CollectionInfo
-  >(filter?: Document, options?: ListCollectionsOptions): ListCollectionsCursor<T>;
+  >(filter?: Document, options?: ListCollectionsOptions & Abortable): ListCollectionsCursor<T>;
   listCollections<
     T extends Pick<CollectionInfo, 'name' | 'type'> | CollectionInfo =
       | Pick<CollectionInfo, 'name' | 'type'>
       | CollectionInfo
-  >(filter: Document = {}, options: ListCollectionsOptions = {}): ListCollectionsCursor<T> {
+  >(
+    filter: Document = {},
+    options: ListCollectionsOptions & Abortable = {}
+  ): ListCollectionsCursor<T> {
     return new ListCollectionsCursor<T>(this, filter, resolveOptions(this, options));
   }
 

src/index.ts

@@ -430,6 +430,7 @@ export type {
   SeverityLevel
 } from './mongo_logger';
 export type {
+  Abortable,
   CommonEvents,
   EventsDescription,
   GenericListener,

src/mongo_types.ts

@@ -474,6 +474,38 @@ export class TypedEventEmitter<Events extends EventsDescription> extends EventEm
 /** @public */
 export class CancellationToken extends TypedEventEmitter<{ cancel(): void }> {}
 
+/** @public */
+export type Abortable = {
+  /**
+   * When provided the corresponding `AbortController` can be used to cancel an asynchronous action.
+   *
+   * The driver will convert the abort event into a promise rejection with an error that has the name `'AbortError'`.
+   *
+   * The cause of the error will be set to `signal.reason`
+   *
+   * @example
+   * ```js
+   * const controller = new AbortController();
+   * const { signal } = controller;
+   * req,on('close', () => controller.abort(new Error('Request aborted by user')));
+   *
+   * try {
+   *   const res = await fetch('...', { signal });
+   *   await collection.insertOne(await res.json(), { signal });
+   * catch (error) {
+   *   if (error.name === 'AbortError') {
+   *     // error is MongoAbortError or DOMException,
+   *     // both represent the signal being aborted
+   *     error.cause === signal.reason; // true
+   *   }
+   * }
+   * ```
+   *
+   * @see MongoAbortError
+   */
+  signal?: AbortSignal | undefined;
+};
+
 /**
  * Helper types for dot-notation filter attributes
  */

src/operations/execute_operation.ts

@@ -64,7 +64,10 @@ export async function executeOperation<
     throw new MongoRuntimeError('This method requires a valid operation instance');
   }
 
+  // Like CSOT, an operation signal interruption does not relate to auto-connect
+  operation.options.signal?.throwIfAborted();
   const topology = await autoConnect(client);
+  operation.options.signal?.throwIfAborted();
 
   // The driver sessions spec mandates that we implicitly create sessions for operations
   // that are not explicitly provided with a session.
@@ -198,7 +201,8 @@ async function tryOperation<
   let server = await topology.selectServer(selector, {
     session,
     operationName: operation.commandName,
-    timeoutContext
+    timeoutContext,
+    signal: operation.options.signal
   });
 
   const hasReadAspect = operation.hasAspect(Aspect.READ_OPERATION);
@@ -260,7 +264,8 @@ async function tryOperation<
       server = await topology.selectServer(selector, {
         session,
         operationName: operation.commandName,
-        previousServer
+        previousServer,
+        signal: operation.options.signal
       });
 
       if (hasWriteAspect && !supportsRetryableWrites(server)) {

src/operations/list_collections.ts

@@ -2,6 +2,7 @@ import type { Binary, Document } from '../bson';
 import { CursorResponse } from '../cmap/wire_protocol/responses';
 import { type CursorTimeoutContext, type CursorTimeoutMode } from '../cursor/abstract_cursor';
 import type { Db } from '../db';
+import { type Abortable } from '../mongo_types';
 import type { Server } from '../sdam/server';
 import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
@@ -10,7 +11,9 @@ import { CommandOperation, type CommandOperationOptions } from './command';
 import { Aspect, defineAspects } from './operation';
 
 /** @public */
-export interface ListCollectionsOptions extends Omit<CommandOperationOptions, 'writeConcern'> {
+export interface ListCollectionsOptions
+  extends Omit<CommandOperationOptions, 'writeConcern'>,
+    Abortable {
   /** Since 4.0: If true, will only return the collection name in the response, and will omit additional info */
   nameOnly?: boolean;
   /** Since 4.0: If true and nameOnly is true, allows a user without the required privilege (i.e. listCollections action on the database) to run the command when access control is enforced. */

src/operations/operation.ts

@@ -1,4 +1,5 @@
 import { type BSONSerializeOptions, type Document, resolveBSONOptions } from '../bson';
+import { type Abortable } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import type { Server } from '../sdam/server';
 import type { ClientSession } from '../sessions';
@@ -59,7 +60,7 @@ export abstract class AbstractOperation<TResult = any> {
   // BSON serialization options
   bsonOptions?: BSONSerializeOptions;
 
-  options: OperationOptions;
+  options: OperationOptions & Abortable;
 
   /** Specifies the time an operation will run until it throws a timeout error. */
   timeoutMS?: number;
@@ -68,7 +69,7 @@ export abstract class AbstractOperation<TResult = any> {
 
   static aspects?: Set<symbol>;
 
-  constructor(options: OperationOptions = {}) {
+  constructor(options: OperationOptions & Abortable = {}) {
     this.readPreference = this.hasAspect(Aspect.WRITE_OPERATION)
       ? ReadPreference.primary
       : (ReadPreference.fromOptions(options) ?? ReadPreference.primary);

src/sdam/server.ts

@@ -36,7 +36,7 @@ import {
   needsRetryableWriteLabel
 } from '../error';
 import type { ServerApi } from '../mongo_client';
-import { TypedEventEmitter } from '../mongo_types';
+import { type Abortable, TypedEventEmitter } from '../mongo_types';
 import type { GetMoreOptions } from '../operations/get_more';
 import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
@@ -107,7 +107,7 @@ export type ServerEvents = {
 /** @internal */
 export type ServerCommandOptions = Omit<CommandOptions, 'timeoutContext' | 'socketTimeoutMS'> & {
   timeoutContext: TimeoutContext;
-};
+} & Abortable;
 
 /** @internal */
 export class Server extends TypedEventEmitter<ServerEvents> {
@@ -285,7 +285,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
   public async command(
     ns: MongoDBNamespace,
     cmd: Document,
-    options: ServerCommandOptions,
+    paramOpts: ServerCommandOptions,
     responseType?: MongoDBResponseConstructor
   ): Promise<Document> {
     if (ns.db == null || typeof ns === 'string') {
@@ -297,24 +297,25 @@ export class Server extends TypedEventEmitter<ServerEvents> {
     }
 
     // Clone the options
-    const finalOptions = Object.assign({}, options, {
+    const options = {
+      ...paramOpts,
       wireProtocolCommand: false,
       directConnection: this.topology.s.options.directConnection
-    });
+    };
 
     // There are cases where we need to flag the read preference not to get sent in
     // the command, such as pre-5.0 servers attempting to perform an aggregate write
     // with a non-primary read preference. In this case the effective read preference
     // (primary) is not the same as the provided and must be removed completely.
-    if (finalOptions.omitReadPreference) {
-      delete finalOptions.readPreference;
+    if (options.omitReadPreference) {
+      delete options.readPreference;
     }
 
     if (this.description.iscryptd) {
-      finalOptions.omitMaxTimeMS = true;
+      options.omitMaxTimeMS = true;
     }
 
-    const session = finalOptions.session;
+    const session = options.session;
     let conn = session?.pinnedConnection;
 
     this.incrementOperationCount();
@@ -333,11 +334,11 @@ export class Server extends TypedEventEmitter<ServerEvents> {
 
     try {
       try {
-        const res = await conn.command(ns, cmd, finalOptions, responseType);
+        const res = await conn.command(ns, cmd, options, responseType);
         throwIfWriteConcernError(res);
         return res;
       } catch (commandError) {
-        throw this.decorateCommandError(conn, cmd, finalOptions, commandError);
+        throw this.decorateCommandError(conn, cmd, options, commandError);
       }
     } catch (operationError) {
       if (
@@ -346,11 +347,11 @@ export class Server extends TypedEventEmitter<ServerEvents> {
       ) {
         await this.pool.reauthenticate(conn);
         try {
-          const res = await conn.command(ns, cmd, finalOptions, responseType);
+          const res = await conn.command(ns, cmd, options, responseType);
           throwIfWriteConcernError(res);
           return res;
         } catch (commandError) {
-          throw this.decorateCommandError(conn, cmd, finalOptions, commandError);
+          throw this.decorateCommandError(conn, cmd, options, commandError);
         }
       } else {
         throw operationError;

src/sdam/topology.ts

@@ -31,15 +31,17 @@ import {
 } from '../error';
 import type { MongoClient, ServerApi } from '../mongo_client';
 import { MongoLoggableComponent, type MongoLogger, SeverityLevel } from '../mongo_logger';
-import { TypedEventEmitter } from '../mongo_types';
+import { type Abortable, TypedEventEmitter } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import type { ClientSession } from '../sessions';
 import { Timeout, TimeoutContext, TimeoutError } from '../timeout';
 import type { Transaction } from '../transactions';
 import {
+  addAbortListener,
   type Callback,
   type EventEmitterWithState,
   HostAddress,
+  kDispose,
   List,
   makeStateMachine,
   now,
@@ -525,7 +527,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
    */
   async selectServer(
     selector: string | ReadPreference | ServerSelector,
-    options: SelectServerOptions
+    options: SelectServerOptions & Abortable
   ): Promise<Server> {
     let serverSelector;
     if (typeof selector !== 'function') {
@@ -602,6 +604,11 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       previousServer: options.previousServer
     };
 
+    const abortListener = addAbortListener(options.signal, function () {
+      waitQueueMember.cancelled = true;
+      reject(this.reason);
+    });
+
     this.waitQueue.push(waitQueueMember);
     processWaitQueue(this);
 
@@ -647,6 +654,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       // Other server selection error
       throw error;
     } finally {
+      abortListener?.[kDispose]();
       if (options.timeoutContext?.clearServerSelectionTimeout) timeout?.clear();
     }
   }

src/utils.ts

@@ -27,6 +27,7 @@ import {
   MongoRuntimeError
 } from './error';
 import type { MongoClient } from './mongo_client';
+import { type Abortable } from './mongo_types';
 import type { CommandOperationOptions, OperationParent } from './operations/command';
 import type { Hint, OperationOptions } from './operations/operation';
 import { ReadConcern } from './read_concern';
@@ -1349,19 +1350,24 @@ export const randomBytes = promisify(crypto.randomBytes);
  * @param ee - An event emitter that may emit `ev`
  * @param name - An event name to wait for
  */
-export async function once<T>(ee: EventEmitter, name: string): Promise<T> {
+export async function once<T>(ee: EventEmitter, name: string, options?: Abortable): Promise<T> {
+  options?.signal?.throwIfAborted();
+
   const { promise, resolve, reject } = promiseWithResolvers<T>();
   const onEvent = (data: T) => resolve(data);
   const onError = (error: Error) => reject(error);
+  const abortListener = addAbortListener(options?.signal, function () {
+    reject(this.reason);
+  });
 
   ee.once(name, onEvent).once('error', onError);
+
   try {
-    const res = await promise;
-    ee.off('error', onError);
-    return res;
-  } catch (error) {
+    return await promise;
+  } finally {
     ee.off(name, onEvent);
-    throw error;
+    ee.off('error', onError);
+    abortListener?.[kDispose]();
   }
 }
 
@@ -1468,3 +1474,17 @@ export function decorateDecryptionResult(
     decorateDecryptionResult(decrypted[k], originalValue, false);
   }
 }
+
+export const kDispose: unique symbol = (Symbol.dispose as any) ?? Symbol('dispose');
+export interface Disposable {
+  [kDispose](): void;
+}
+
+export function addAbortListener(
+  signal: AbortSignal | undefined | null,
+  listener: (this: AbortSignal, event: Event) => void
+): Disposable | undefined {
+  if (signal == null) return;
+  signal.addEventListener('abort', listener);
+  return { [kDispose]: () => signal.removeEventListener('abort', listener) };
+}

test/integration/client-side-encryption/driver.test.ts

@@ -831,12 +831,12 @@ describe('CSOT', function () {
   describe('State machine', function () {
     const stateMachine = new StateMachine({} as any);
 
-    const timeoutContext = () => {
-      return new CSOTTimeoutContext({
+    const timeoutContext = () => ({
+      timeoutContext: new CSOTTimeoutContext({
         timeoutMS: 1000,
         serverSelectionTimeoutMS: 30000
-      });
-    };
+      })
+    });
 
     const timeoutMS = 1000;
 
@@ -1001,7 +1001,7 @@ describe('CSOT', function () {
 
             const { result: error } = await measureDuration(() =>
               stateMachine
-                .fetchKeys(client, 'test.test', BSON.serialize({}), timeoutContext)
+                .fetchKeys(client, 'test.test', BSON.serialize({}), { timeoutContext })
                 .catch(e => e)
             );
             expect(error).to.be.instanceOf(MongoOperationTimeoutError);

test/integration/client-side-operations-timeout/client_side_operations_timeout.unit.test.ts

@@ -138,7 +138,7 @@ describe('CSOT spec unit tests', function () {
             timeoutMS: 500,
             serverSelectionTimeoutMS: 30000
           });
-          const err = await stateMachine.kmsRequest(request, timeoutContext).catch(e => e);
+          const err = await stateMachine.kmsRequest(request, { timeoutContext }).catch(e => e);
           expect(err).to.be.instanceOf(MongoOperationTimeoutError);
           expect(err.errmsg).to.equal('KMS request timed out');
         });

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

@@ -622,7 +622,7 @@ export async function clearFailPoint(configuration: TestConfiguration, url = con
 
 export async function makeMultiBatchWrite(
   configuration: TestConfiguration
-): Promise<AnyClientBulkWriteModel[]> {
+): Promise<AnyClientBulkWriteModel<any>[]> {
   const { maxBsonObjectSize, maxMessageSizeBytes } = await configuration.hello();
 
   const length = maxMessageSizeBytes / maxBsonObjectSize + 1;
@@ -637,10 +637,10 @@ export async function makeMultiBatchWrite(
 
 export async function makeMultiResponseBatchModelArray(
   configuration: TestConfiguration
-): Promise<AnyClientBulkWriteModel[]> {
+): Promise<AnyClientBulkWriteModel<any>[]> {
   const { maxBsonObjectSize } = await configuration.hello();
   const namespace = `foo.${new BSON.ObjectId().toHexString()}`;
-  const models: AnyClientBulkWriteModel[] = [
+  const models: AnyClientBulkWriteModel<any>[] = [
     {
       name: 'updateOne',
       namespace,
@@ -693,3 +693,42 @@ export function mergeTestMetadata(
     }
   };
 }
+
+export function findLast<T, S extends T>(
+  array: T[],
+  predicate: (value: T, index: number, array: T[]) => value is S,
+  thisArg?: any
+): S | undefined;
+export function findLast<T>(
+  array: T[],
+  predicate: (value: T, index: number, array: T[]) => boolean,
+  thisArg?: any
+): T | undefined;
+export function findLast(
+  array: unknown[],
+  predicate: (value: unknown, index: number, array: unknown[]) => boolean,
+  thisArg?: any
+): unknown | undefined {
+  if (typeof array.findLast === 'function') return array.findLast(predicate, thisArg);
+
+  for (let i = array.length - 1; i >= 0; i--) {
+    if (predicate.call(thisArg, array[i], i, array)) {
+      return array[i];
+    }
+  }
+
+  return undefined;
+}
+
+// Node.js 16 doesn't make this global, but it can still be obtained.
+export const DOMException: {
+  new: (
+    message?: string,
+    nameOrOptions?: string | { name?: string; cause?: unknown }
+  ) => DOMException;
+} = (() => {
+  if (globalThis.DOMException != null) return globalThis.DOMException;
+  const ac = new AbortController();
+  ac.abort();
+  return ac.signal.reason.constructor;
+})();

test/tools/utils.ts

@@ -81,7 +81,12 @@ describe('StateMachine', function () {
       a: new Long('0'),
       b: new Int32(0)
     };
-    const options = { promoteLongs: false, promoteValues: false };
+    const options = {
+      promoteLongs: false,
+      promoteValues: false,
+      signal: undefined,
+      timeoutMS: undefined
+    };
     const serializedCommand = serialize(command);
     const stateMachine = new StateMachine({} as any);
 
@@ -493,7 +498,7 @@ describe('StateMachine', function () {
           });
 
           await stateMachine
-            .fetchKeys(client, 'keyVault', BSON.serialize({ a: 1 }), context)
+            .fetchKeys(client, 'keyVault', BSON.serialize({ a: 1 }), { timeoutContext: context })
             .catch(e => squashError(e));
 
           const { timeoutContext } = findSpy.getCalls()[0].args[1] as FindOptions;
@@ -535,7 +540,7 @@ describe('StateMachine', function () {
           });
           await sleep(300);
           await stateMachine
-            .markCommand(client, 'keyVault', BSON.serialize({ a: 1 }), timeoutContext)
+            .markCommand(client, 'keyVault', BSON.serialize({ a: 1 }), { timeoutContext })
             .catch(e => squashError(e));
           expect(dbCommandSpy.getCalls()[0].args[1].timeoutMS).to.not.be.undefined;
           expect(dbCommandSpy.getCalls()[0].args[1].timeoutMS).to.be.lessThanOrEqual(205);
@@ -576,7 +581,9 @@ describe('StateMachine', function () {
             });
             await sleep(300);
             await stateMachine
-              .fetchCollectionInfo(client, 'keyVault', BSON.serialize({ a: 1 }), context)
+              .fetchCollectionInfo(client, 'keyVault', BSON.serialize({ a: 1 }), {
+                timeoutContext: context
+              })
               .catch(e => squashError(e));
             const [_filter, { timeoutContext }] = listCollectionsSpy.getCalls()[0].args;
             expect(timeoutContext).to.exist;