fix(NODE-6845): ensure internal rejections are handled

.gitignore

@@ -103,6 +103,7 @@ secrets-export.fish
 mo-expansion.sh
 mo-expansion.yml
 expansions.sh
+uri.txt
 
 .drivers-tools/
 

.mocharc.js

@@ -7,6 +7,7 @@ module.exports = {
  require: [
  'source-map-support/register',
  'ts-node/register',
+ 'test/tools/runner/throw_rejections.cjs',
  'test/tools/runner/chai_addons.ts',
  'test/tools/runner/ee_checker.ts'
  ],

src/cursor/abstract_cursor.ts

@@ -1146,50 +1146,59 @@ class ReadableCursorStream extends Readable {
  return;
  }
 
- this._cursor.next().then(
- result => {
- if (result == null) {
- this.push(null);
- } else if (this.destroyed) {
- this._cursor.close().then(undefined, squashError);
- } else {
- if (this.push(result)) {
- return this._readNext();
+ this._cursor
+ .next()
+ .then(
+ // result from next()
+ result => {
+ if (result == null) {
+ this.push(null);
+ } else if (this.destroyed) {
+ this._cursor.close().then(undefined, squashError);
+ } else {
+ if (this.push(result)) {
+ return this._readNext();
+ }
+
+ this._readInProgress = false;
+ }
+ },
+ // error from next()
+ err => {
+ // NOTE: This is questionable, but we have a test backing the behavior. It seems the
+ // desired behavior is that a stream ends cleanly when a user explicitly closes
+ // a client during iteration. Alternatively, we could do the "right" thing and
+ // propagate the error message by removing this special case.
+ if (err.message.match(/server is closed/)) {
+ this._cursor.close().then(undefined, squashError);
+ return this.push(null);
  }
 
- this._readInProgress = false;
- }
- },
- err => {
- // NOTE: This is questionable, but we have a test backing the behavior. It seems the
- // desired behavior is that a stream ends cleanly when a user explicitly closes
- // a client during iteration. Alternatively, we could do the "right" thing and
- // propagate the error message by removing this special case.
- if (err.message.match(/server is closed/)) {
- this._cursor.close().then(undefined, squashError);
- return this.push(null);
- }
+ // NOTE: This is also perhaps questionable. The rationale here is that these errors tend
+ // to be "operation was interrupted", where a cursor has been closed but there is an
+ // active getMore in-flight. This used to check if the cursor was killed but once
+ // that changed to happen in cleanup legitimate errors would not destroy the
+ // stream. There are change streams test specifically test these cases.
+ if (err.message.match(/operation was interrupted/)) {
+ return this.push(null);
+ }
 
- // NOTE: This is also perhaps questionable. The rationale here is that these errors tend
- // to be "operation was interrupted", where a cursor has been closed but there is an
- // active getMore in-flight. This used to check if the cursor was killed but once
- // that changed to happen in cleanup legitimate errors would not destroy the
- // stream. There are change streams test specifically test these cases.
- if (err.message.match(/operation was interrupted/)) {
- return this.push(null);
+ // NOTE: The two above checks on the message of the error will cause a null to be pushed
+ // to the stream, thus closing the stream before the destroy call happens. This means
+ // that either of those error messages on a change stream will not get a proper
+ // 'error' event to be emitted (the error passed to destroy). Change stream resumability
+ // relies on that error event to be emitted to create its new cursor and thus was not
+ // working on 4.4 servers because the error emitted on failover was "interrupted at
+ // shutdown" while on 5.0+ it is "The server is in quiesce mode and will shut down".
+ // See NODE-4475.
+ return this.destroy(err);
  }
-
- // NOTE: The two above checks on the message of the error will cause a null to be pushed
- // to the stream, thus closing the stream before the destroy call happens. This means
- // that either of those error messages on a change stream will not get a proper
- // 'error' event to be emitted (the error passed to destroy). Change stream resumability
- // relies on that error event to be emitted to create its new cursor and thus was not
- // working on 4.4 servers because the error emitted on failover was "interrupted at
- // shutdown" while on 5.0+ it is "The server is in quiesce mode and will shut down".
- // See NODE-4475.
- return this.destroy(err);
- }
- );
+ )
+ // if either of the above handlers throw
+ .catch(error => {
+ this._readInProgress = false;
+ this.destroy(error);
+ });
  }
 }
 

src/sdam/server.ts

@@ -48,6 +48,7 @@ import {
  maxWireVersion,
  type MongoDBNamespace,
  noop,
+ squashError,
  supportsRetryableWrites
 } from '../utils';
 import { throwIfWriteConcernError } from '../write_concern';
@@ -345,9 +346,10 @@ export class Server extends TypedEventEmitter<ServerEvents> {
  operationError instanceof MongoError &&
  operationError.code === MONGODB_ERROR_CODES.Reauthenticate
  ) {
- reauthPromise = this.pool.reauthenticate(conn).catch(error => {
+ reauthPromise = this.pool.reauthenticate(conn);
+ reauthPromise.then(undefined, error => {
  reauthPromise = null;
- throw error;
+ squashError(error);
  });
 
  await abortable(reauthPromise, options);
@@ -368,9 +370,10 @@ export class Server extends TypedEventEmitter<ServerEvents> {
  if (session?.pinnedConnection !== conn) {
  if (reauthPromise != null) {
  // The reauth promise only exists if it hasn't thrown.
- void reauthPromise.finally(() => {
+ const checkBackIn = () => {
  this.pool.checkIn(conn);
- });
+ };
+ void reauthPromise.then(checkBackIn, checkBackIn);
  } else {
  this.pool.checkIn(conn);
  }

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/change-streams/change_stream.test.ts

@@ -819,6 +819,7 @@ describe('Change Streams', function () {
  const write = lastWrite();
 
  const nextP = changeStream.next();
+ nextP.catch(() => null);
 
  await changeStream.close();
 

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

@@ -1023,7 +1023,8 @@ describe('CSOT driver tests', metadata, () => {
 
  beforeEach(async function () {
  cs = client.db('db').collection('coll').watch([], { timeoutMS: 120 });
- const _changePromise = once(cs, 'change');
+ cs.once('change', () => null);
+
  await once(cs.cursor, 'init');
 
  await internalClient.db().admin().command(failpoint);

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

@@ -753,7 +753,7 @@ describe('AbortSignal support', () => {
  if (args[1].find != null) {
  commandStub.restore();
  controller.abort();
- throw new ReAuthenticationError({});
+ throw new ReAuthenticationError({ message: 'This is a fake reauthentication error' });
  }
  return commandStub.wrappedMethod.apply(this, args);
  });
@@ -792,8 +792,9 @@ describe('AbortSignal support', () => {
  describe('if reauth throws', () => {
  beforeEach(() => {
  sinon.stub(ConnectionPool.prototype, 'reauthenticate').callsFake(async function () {
+ const error = new Error('Rejecting reauthenticate for testing');
  await sleep(1000);
- throw new Error();
+ throw error;
  });
  });
 

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

@@ -60,7 +60,9 @@ maybeDescribe('examples(change-stream):', function () {
  it('Open A Change Stream', {
  metadata: { requires: { topology: ['replicaset'], mongodb: '>=3.6.0' } },
  test: async function () {
- const looper = new Looper(() => db.collection('inventory').insertOne({ a: 1 }));
+ const looper = new Looper(async () => {
+ await db.collection('inventory').insertOne({ a: 1 });
+ });
  looper.run();
 
  // Start Changestream Example 1

test/integration/shared.js

@@ -90,32 +90,25 @@ function ignoreNsNotFound(err) {
  if (!err.message.match(/ns not found/)) throw err;
 }
 
-function setupDatabase(configuration, dbsToClean) {
+async function setupDatabase(configuration, dbsToClean) {
  dbsToClean = Array.isArray(dbsToClean) ? dbsToClean : [];
- var configDbName = configuration.db;
- var client = configuration.newClient(configuration.writeConcernMax(), {
- maxPoolSize: 1
- });
+ const configDbName = configuration.db;
 
  dbsToClean.push(configDbName);
 
- return client
- .connect()
- .then(() =>
- dbsToClean.reduce(
- (result, dbName) =>
- result
- .then(() =>
- client.db(dbName).command({ dropAllUsersFromDatabase: 1, writeConcern: { w: 1 } })
- )
- .then(() => client.db(dbName).dropDatabase({ writeConcern: { w: 1 } })),
- Promise.resolve()
- )
- )
- .then(
- () => client.close(),
- err => client.close(() => Promise.reject(err))
- );
+ const client = configuration.newClient();
+ try {
+ for (const dbName of dbsToClean) {
+ const db = await client.db(dbName);
+ for await (const { name } of db.listCollections({}, { nameOnly: true })) {
+ const collection = db.collection(name);
+ await collection.deleteMany({}).catch(() => null);
+ await collection.drop().catch(() => null);
+ }
+ }
+ } finally {
+ await client.close();
+ }
 }
 
 /**

test/manual/mocharc.js

@@ -4,7 +4,11 @@ const [major] = process.versions.node.split('.');
 
 /** @type {import("mocha").MochaOptions} */
 module.exports = {
- require: ['ts-node/register', 'test/tools/runner/chai_addons.ts'],
+ require: [
+ 'ts-node/register',
+ 'test/tools/runner/throw_rejections.cjs',
+ 'test/tools/runner/chai_addons.ts'
+ ],
  reporter: 'test/tools/reporter/mongodb_reporter.js',
  failZero: true,
  color: true,

test/mocha_lambda.js

@@ -4,7 +4,10 @@ const [major] = process.versions.node.split('.');
 
 /** @type {import("mocha").MochaOptions} */
 module.exports = {
- require: ['test/integration/node-specific/examples/setup.js'],
+ require: [
+ 'test/tools/runner/throw_rejections.cjs',
+ 'test/integration/node-specific/examples/setup.js'
+ ],
  extension: ['js'],
  ui: 'test/tools/runner/metadata_ui.js',
  recursive: true,

test/mocha_mongodb.js

@@ -7,6 +7,7 @@ module.exports = {
  require: [
  'source-map-support/register',
  'ts-node/register',
+ 'test/tools/runner/throw_rejections.cjs',
  'test/tools/runner/chai_addons.ts',
  'test/tools/runner/ee_checker.ts',
  'test/tools/runner/hooks/configuration.ts',

test/tools/runner/throw_rejections.cjs

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