wip

Author: nbbeeken

.evergreen/run-tests.sh

@@ -65,4 +65,5 @@ export MONGODB_URI=${MONGODB_URI}
 export LOAD_BALANCER=${LOAD_BALANCER}
 export TEST_CSFLE=${TEST_CSFLE}
 export COMPRESSOR=${COMPRESSOR}
+export NODE_OPTIONS="${NODE_OPTIONS} --trace-uncaught"
 npm run "${TEST_NPM_SCRIPT}"

src/change_stream.ts

@@ -664,6 +664,8 @@ export class ChangeStream<
     this.isClosed = false;
     this.mode = false;
 
+    this.on('error', () => null);
+
     // Listen for any `change` listeners being added to ChangeStream
     this.on('newListener', eventName => {
       if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {

src/client-side-encryption/state_machine.ts

@@ -351,12 +351,8 @@ export class StateMachine {
     let socket: tls.TLSSocket;
 
     function destroySockets() {
-      for (const sock of [socket, netSocket]) {
-        if (sock) {
-          sock.removeAllListeners();
-          sock.destroy();
-        }
-      }
+      socket?.destroy();
+      netSocket?.destroy();
     }
 
     function onerror(cause: Error) {

src/cmap/connect.ts

@@ -389,6 +389,7 @@ export async function makeSocket(
 
   addAbortSignalToStream(closeSignal, socket);
 
+  socket.unref();
   socket.setKeepAlive(true, 300000);
   socket.setTimeout(connectTimeoutMS);
   socket.setNoDelay(noDelay);

src/cmap/connection.ts

@@ -298,6 +298,14 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     );
   }
 
+  unref() {
+    this.socket.unref();
+  }
+
+  ref() {
+    this.socket.ref();
+  }
+
   public markAvailable(): void {
     this.lastUseTime = now();
   }
@@ -353,7 +361,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       return;
     }
 
-    this.socket.destroy();
+    if (!this.socket.destroyed) this.socket.destroy();
     this.error = error;
 
     this.dataEvents?.throw(error).then(undefined, squashError);

src/cmap/connection_pool.ts

@@ -413,6 +413,8 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
     if (!this.checkedOut.has(connection)) {
       return;
     }
+
+    connection.unref();
     const poolClosed = this.closed;
     const stale = this.connectionIsStale(connection);
     const willDestroy = !!(poolClosed || stale || connection.closed);
@@ -788,6 +790,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
         );
 
         this.waitQueue.shift();
+        connection.ref();
         waitQueueMember.resolve(connection);
       }
     }

src/mongo_client.ts

@@ -691,14 +691,15 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
 
   /* @internal */
   private async _close(force = false): Promise<void> {
-    this.closeController.abort();
-    // There's no way to set hasBeenClosed back to false
-    Object.defineProperty(this.s, 'hasBeenClosed', {
-      value: true,
-      enumerable: true,
-      configurable: false,
-      writable: false
-    });
+    try {
+      this.closeController.abort();
+      // There's no way to set hasBeenClosed back to false
+      Object.defineProperty(this.s, 'hasBeenClosed', {
+        value: true,
+        enumerable: true,
+        configurable: false,
+        writable: false
+      });
 
       const activeCursorCloses = Array.from(this.s.activeCursors, cursor => cursor.close());
       this.s.activeCursors.clear();

src/sdam/monitor.ts

@@ -384,20 +384,13 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
   }
 
   // connecting does an implicit `hello`
-  (async () => {
+  const makeMonitoringConnection = async () => {
     const socket = await makeSocket(monitor.connectOptions, monitor.closeSignal);
     const connection = makeConnection(monitor.connectOptions, socket);
     // The start time is after socket creation but before the handshake
     start = now();
     try {
       await performInitialHandshake(connection, monitor.connectOptions, monitor.closeSignal);
-      return connection;
-    } catch (error) {
-      connection.destroy();
-      throw error;
-    }
-  })().then(
-    connection => {
       if (isInCloseState(monitor)) {
         connection.destroy();
         return;
@@ -417,15 +410,16 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
           useStreamingProtocol(monitor, connection.hello?.topologyVersion)
         )
       );
-
-      callback(undefined, connection.hello);
-    },
-    error => {
+      return connection.hello;
+    } catch (error) {
+      connection.destroy();
       monitor.connection = null;
       awaited = false;
-      onHeartbeatFailed(error);
+      throw error;
     }
-  );
+  };
+
+  makeMonitoringConnection().then(callback.bind(undefined, undefined), onHeartbeatFailed);
 }
 
 function monitorServer(monitor: Monitor) {

src/utils.ts

@@ -1,3 +1,4 @@
+/* eslint-disable no-console */
 import * as crypto from 'crypto';
 import type { SrvRecord } from 'dns';
 import { type EventEmitter } from 'events';
@@ -1571,7 +1572,37 @@ export function addAbortSignalToStream(
 
   const abortListener = addAbortListener(signal, function () {
     stream.off('close', abortListener[kDispose]).off('error', abortListener[kDispose]);
-    stream.destroy(this.reason);
+    const removeAll = stream.removeAllListeners.bind(stream);
+
+    //@ts-expect-error: overwrite
+    stream.removeAllListeners = function (...args) {
+      const removingError = args.length === 0 || args.includes('error');
+      if (removingError) {
+        console.log('removing all', args, new Error('why are u removing my error listener'));
+      }
+      return removeAll(...args);
+    };
+
+    stream.on('removeListener', (name, listener) => {
+      if (name === 'error') {
+        console.log(
+          'who doth remove my error listener',
+          new Error('error listener gone missing!!'),
+          listener
+        );
+      }
+    });
+
+    const error = new Error(
+      //@ts-expect-error: we added these
+      `sad: ${stream.___socketId}: error listeners: ${stream.listenerCount('error')} + ${stream.___stack}`,
+      { cause: this.reason }
+    );
+
+    //@ts-expect-error: adding this for debug
+    error.stream = stream;
+
+    stream.destroy(error);
   });
   // not nearly as complex as node's eos() but... do we need all that?? sobbing emoji.
   stream.once('close', abortListener[kDispose]).once('error', abortListener[kDispose]);

test/integration/change-streams/change_stream.test.ts

@@ -18,6 +18,7 @@ import {
   MongoChangeStreamError,
   type MongoClient,
   MongoServerError,
+  promiseWithResolvers,
   ReadPreference,
   type ResumeToken
 } from '../../mongodb';
@@ -62,6 +63,7 @@ describe('Change Streams', function () {
     await csDb.createCollection('test').catch(() => null);
     collection = csDb.collection('test');
     changeStream = collection.watch();
+    changeStream.once('error', error => this.error(error));
   });
 
   afterEach(async () => {
@@ -695,10 +697,18 @@ describe('Change Streams', function () {
     async test() {
       await initIteratorMode(changeStream);
 
+      const { promise, resolve, reject } = promiseWithResolvers<void>();
+
       const outStream = new PassThrough({ objectMode: true });
 
-      // @ts-expect-error: transform requires a Document return type
-      changeStream.stream({ transform: JSON.stringify }).pipe(outStream);
+      const csStream = changeStream
+        // @ts-expect-error: transform requires a Document return type
+        .stream({ transform: JSON.stringify });
+
+      csStream.once('error', reject).pipe(outStream).once('error', reject);
+
+      outStream.on('close', resolve);
+      csStream.on('close', resolve);
 
       const willBeData = once(outStream, 'data');
 
@@ -709,6 +719,8 @@ describe('Change Streams', function () {
       expect(parsedEvent).to.have.nested.property('fullDocument.a', 1);
 
       outStream.destroy();
+      csStream.destroy();
+      await promise;
     }
   });