chore: fix more unhandling

Author: nbbeeken

src/sdam/server.ts

@@ -346,8 +346,11 @@ export class Server extends TypedEventEmitter<ServerEvents> {
         operationError instanceof MongoError &&
         operationError.code === MONGODB_ERROR_CODES.Reauthenticate
       ) {
-        // TODO: what do we do with this error? the signal has aborted so where should it be delivered?
-        reauthPromise = this.pool.reauthenticate(conn).then(undefined, squashError);
+        reauthPromise = this.pool.reauthenticate(conn);
+        reauthPromise.then(undefined, error => {
+          reauthPromise = null;
+          squashError(error);
+        });
 
         await abortable(reauthPromise, options);
         reauthPromise = null; // only reachable if reauth succeeds

src/sdam/topology.ts

@@ -559,11 +559,6 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
         new ServerSelectionStartedEvent(selector, this.description, options.operationName)
       );
     }
-    let timeout;
-    if (options.timeoutContext) timeout = options.timeoutContext.serverSelectionTimeout;
-    else {
-      timeout = Timeout.expires(options.serverSelectionTimeoutMS ?? 0);
-    }
 
     const isSharded = this.description.type === TopologyType.Sharded;
     const session = options.session;
@@ -586,10 +581,16 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
           )
         );
       }
-      if (options.timeoutContext?.clearServerSelectionTimeout) timeout?.clear();
+
       return transaction.server;
     }
 
+    let timeout;
+    if (options.timeoutContext) timeout = options.timeoutContext.serverSelectionTimeout;
+    else {
+      timeout = Timeout.expires(options.serverSelectionTimeoutMS ?? 0);
+    }
+
     const { promise: serverPromise, resolve, reject } = promiseWithResolvers<Server>();
 
     const waitQueueMember: ServerSelectionRequest = {

src/timeout.ts

@@ -3,7 +3,7 @@ import { clearTimeout, setTimeout } from 'timers';
 import { type Document } from './bson';
 import { MongoInvalidArgumentError, MongoOperationTimeoutError, MongoRuntimeError } from './error';
 import { type ClientSession } from './sessions';
-import { csotMin, noop } from './utils';
+import { csotMin, noop, squashError } from './utils';
 
 /** @internal */
 export class TimeoutError extends Error {
@@ -102,7 +102,13 @@ export class Timeout extends Promise<never> {
   }
 
   throwIfExpired(): void {
-    if (this.timedOut) throw new TimeoutError('Timed out', { duration: this.duration });
+    if (this.timedOut) {
+      // This method is invoked when someone wants to throw immediately instead of await the result of this promise
+      // Since they won't be handling the rejection from the promise (because we're about to throw here)
+      // attach handling to prevent this from bubbling up to Node.js
+      this.then(undefined, squashError);
+      throw new TimeoutError('Timed out', { duration: this.duration });
+    }
   }
 
   public static expires(duration: number, unref?: true): Timeout {

test/integration/client-side-operations-timeout/node_csot.test.ts

@@ -1023,7 +1023,9 @@ describe('CSOT driver tests', metadata, () => {
 
         beforeEach(async function () {
           cs = client.db('db').collection('coll').watch([], { timeoutMS: 120 });
-          const _changePromise = once(cs, 'change');
+          const changePromise = once(cs, 'change');
+          void changePromise.then(undefined, () => null); // need to handled when this rejects with the timeout error
+
           await once(cs.cursor, 'init');
 
           await internalClient.db().admin().command(failpoint);

test/integration/node-specific/examples/change_streams.test.js

@@ -2,7 +2,6 @@
 'use strict';
 
 const { setTimeout } = require('timers');
-const setupDatabase = require('../../shared').setupDatabase;
 const expect = require('chai').expect;
 
 // TODO: NODE-3819: Unskip flaky MacOS/Windows tests.
@@ -11,13 +10,9 @@ maybeDescribe('examples(change-stream):', function () {
   let client;
   let db;
 
-  before(async function () {
-    await setupDatabase(this.configuration);
-  });
-
   beforeEach(async function () {
     client = await this.configuration.newClient().connect();
-    db = client.db(this.configuration.db);
+    db = client.db('change_stream_examples');
 
     // ensure database exists, we need this for 3.6
     await db.collection('inventory').insertOne({});

test/tools/runner/throw_rejections.cjs

@@ -1,6 +1,8 @@
+/* eslint-disable no-console */
 // eslint-disable-next-line @typescript-eslint/no-require-imports
 const process = require('process');
 
-process.on('unhandledRejection', error => {
+process.on('unhandledRejection', (error, promise) => {
+  console.log('promise:', promise, 'unhandledRejection:', error);
   throw error;
 });

test/unit/cmap/auth/auth_provider.test.ts

@@ -8,12 +8,14 @@ describe('AuthProvider', function () {
       const provider = new AuthProvider();
       const context = { reauthenticating: true };
 
-      it('returns an error', function () {
-        provider.reauth(context, error => {
-          expect(error).to.exist;
-          expect(error).to.be.instanceOf(MongoRuntimeError);
-          expect(error?.message).to.equal('Reauthentication already in progress.');
-        });
+      it('returns an error', async function () {
+        const error = await provider.reauth(context).then(
+          () => null,
+          error => error
+        );
+        expect(error).to.exist;
+        expect(error).to.be.instanceOf(MongoRuntimeError);
+        expect(error?.message).to.equal('Reauthentication already in progress.');
       });
     });
   });