feat(NODE-6258): add signal support to find and aggregate

src/operations/execute_operation.ts

@@ -25,7 +25,7 @@ import {
 import type { Topology } from '../sdam/topology';
 import type { ClientSession } from '../sessions';
 import { TimeoutContext } from '../timeout';
-import { supportsRetryableWrites } from '../utils';
+import { abortable, supportsRetryableWrites } from '../utils';
 import { AbstractOperation, Aspect } from './operation';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
@@ -65,9 +65,7 @@ export async function executeOperation<
   }
 
   // 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();
+  const topology = await abortable(autoConnect(client), operation.options);
 
   // The driver sessions spec mandates that we implicitly create sessions for operations
   // that are not explicitly provided with a session.

src/sdam/server.ts

@@ -42,6 +42,7 @@ import type { ClientSession } from '../sessions';
 import { type TimeoutContext } from '../timeout';
 import { isTransactionCommand } from '../transactions';
 import {
+  abortable,
   type EventEmitterWithState,
   makeStateMachine,
   maxWireVersion,
@@ -345,7 +346,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
         operationError instanceof MongoError &&
         operationError.code === MONGODB_ERROR_CODES.Reauthenticate
       ) {
-        await this.pool.reauthenticate(conn);
+        await abortable(this.pool.reauthenticate(conn), options);
         try {
           const res = await conn.command(ns, cmd, options, responseType);
           throwIfWriteConcernError(res);

src/utils.ts

@@ -1475,11 +1475,28 @@ export function decorateDecryptionResult(
   }
 }
 
+/** @internal */
 export const kDispose: unique symbol = (Symbol.dispose as any) ?? Symbol('dispose');
+
+/** @internal */
 export interface Disposable {
   [kDispose](): void;
 }
 
+/**
+ * A utility that helps with writing listener code idiomatically
+ *
+ * @example
+ * ```js
+ * using listener = addAbortListener(signal, function () {
+ *   console.log('aborted', this.reason);
+ * });
+ * ```
+ *
+ * @param signal - if exists adds an abort listener
+ * @param listener - the listener to be added to signal
+ * @returns A disposable that will remove the abort listener
+ */
 export function addAbortListener(
   signal: AbortSignal | undefined | null,
   listener: (this: AbortSignal, event: Event) => void
@@ -1488,3 +1505,39 @@ export function addAbortListener(
   signal.addEventListener('abort', listener, { once: true });
   return { [kDispose]: () => signal.removeEventListener('abort', listener) };
 }
+
+/**
+ * Takes a promise and races it with a promise wrapping the abort event of the optionally provided signal.
+ * The given promise is _always_ ordered before the signal's abort promise.
+ * When given an already rejected promise and an already aborted signal, the promise's rejection takes precedence.
+ *
+ * This is useful **ONLY** when you do not care about cancelling the operation represented by `promise`.
+ * Proper cancellation must be implemented by the promise returning API.
+ *
+ * @see https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/race
+ *
+ * @param promise - A promise to discard if the signal aborts
+ * @param options - An options object carrying an optional signal
+ */
+export async function abortable<T>(
+  promise: Promise<T>,
+  { signal }: { signal?: AbortSignal }
+): Promise<T> {
+  if (signal == null) {
+    return await promise;
+  }
+
+  const { promise: aborted, reject } = promiseWithResolvers<never>();
+
+  const abortListener = signal.aborted
+    ? reject(signal.reason)
+    : addAbortListener(signal, function () {
+        reject(this.reason);
+      });
+
+  try {
+    return await Promise.race([promise, aborted]);
+  } finally {
+    abortListener?.[kDispose]();
+  }
+}

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

@@ -9,11 +9,13 @@ import {
   type AbstractCursor,
   AggregationCursor,
   type Collection,
+  Connection,
   type Db,
   FindCursor,
   ListCollectionsCursor,
   type Log,
   type MongoClient,
+  MongoServerError,
   promiseWithResolvers,
   ReadPreference,
   setDifference,
@@ -545,6 +547,160 @@ describe('AbortSignal support', () => {
     });
   });
 
+  describe('when auto connecting and the signal aborts', () => {
+    let client: MongoClient;
+    let db: Db;
+    let collection: Collection<{ a: number; ssn: string }>;
+    const logs: Log[] = [];
+    let connectStarted;
+    let controller: AbortController;
+    let signal: AbortSignal;
+    let cursor: AbstractCursor<{ a: number }>;
+
+    describe('when connect succeeds', () => {
+      beforeEach(async function () {
+        logs.length = 0;
+
+        const promise = promiseWithResolvers<void>();
+        connectStarted = promise.promise;
+
+        client = this.configuration.newClient(
+          {},
+          {
+            __enableMongoLogger: true,
+            __internalLoggerConfig: { MONGODB_LOG_SERVER_SELECTION: 'debug' },
+            mongodbLogPath: {
+              write: log => {
+                if (log.c === 'serverSelection' && log.operation === 'ping') {
+                  controller.abort();
+                  promise.resolve();
+                }
+                logs.push(log);
+              }
+            },
+            serverSelectionTimeoutMS: 1000
+          }
+        );
+        db = client.db('abortSignal');
+        collection = db.collection('support');
+
+        controller = new AbortController();
+        signal = controller.signal;
+
+        cursor = collection.find({}, { signal });
+      });
+
+      afterEach(async function () {
+        logs.length = 0;
+        await client?.close();
+      });
+
+      it('escapes auto connect without interrupting it', async () => {
+        const toArray = cursor.toArray().catch(error => error);
+        await connectStarted;
+        expect(await toArray).to.be.instanceOf(DOMException);
+        await sleep(1100);
+        expect(client.topology).to.exist;
+        expect(client.topology.description).to.have.property('type').not.equal('Unknown');
+      });
+    });
+
+    describe('when connect fails', () => {
+      beforeEach(async function () {
+        logs.length = 0;
+
+        const promise = promiseWithResolvers<void>();
+        connectStarted = promise.promise;
+
+        client = this.configuration.newClient('mongodb://iLoveJavaScript', {
+          __enableMongoLogger: true,
+          __internalLoggerConfig: { MONGODB_LOG_SERVER_SELECTION: 'debug' },
+          mongodbLogPath: {
+            write: log => {
+              if (log.c === 'serverSelection' && log.operation === 'ping') {
+                controller.abort();
+                promise.resolve();
+              }
+              logs.push(log);
+            }
+          },
+          serverSelectionTimeoutMS: 200,
+          maxPoolSize: 1
+        });
+        db = client.db('abortSignal');
+        collection = db.collection('support');
+
+        controller = new AbortController();
+        signal = controller.signal;
+
+        cursor = collection.find({}, { signal });
+      });
+
+      afterEach(async function () {
+        logs.length = 0;
+        await client?.close();
+      });
+
+      it('escapes auto connect without interrupting it', async () => {
+        const toArray = cursor.toArray().catch(error => error);
+        await connectStarted;
+        expect(await toArray).to.be.instanceOf(DOMException);
+        await sleep(500);
+        expect(client.topology).to.exist;
+        expect(client.topology.description).to.have.property('type', 'Unknown');
+        expect(findLast(logs, l => l.failure.includes('ENOTFOUND'))).to.exist;
+      });
+    });
+  });
+
+  describe('when reauthenticating and the signal aborts', () => {
+    let client: MongoClient;
+    let collection: Collection;
+    let cursor;
+    let controller: AbortController;
+    let signal: AbortSignal;
+
+    class ReAuthenticationError extends MongoServerError {
+      override code = 391; // reauth code.
+    }
+
+    beforeEach(async function () {
+      client = this.configuration.newClient();
+      await client.connect();
+
+      const db = client.db('abortSignal');
+      collection = db.collection('support');
+
+      controller = new AbortController();
+      signal = controller.signal;
+
+      cursor = collection.find({}, { signal });
+
+      const commandStub = sinon.stub(Connection.prototype, 'command').callsFake(async function (
+        ...args
+      ) {
+        if (args[1].find != null) {
+          sinon.restore();
+          controller.abort();
+          throw new ReAuthenticationError({});
+        }
+        return commandStub.wrappedMethod.apply(this, args);
+      });
+    });
+
+    afterEach(async function () {
+      logs.length = 0;
+      await client?.close();
+    });
+
+    it('escapes reauth without interrupting it', async () => {
+      const checkIn = events.once(client, 'connectionCheckedIn');
+      const toArray = cursor.toArray().catch(error => error);
+      expect(await toArray).to.be.instanceOf(DOMException);
+      await checkIn; // checks back in despite the abort
+    });
+  });
+
   describe('KMS requests', function () {
     const stateMachine = new StateMachine({} as any);
     const request = {