wip

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/cmap/auth/mongodb_oidc.ts

@@ -100,7 +100,8 @@ export interface Workflow {
   execute(
     connection: Connection,
     credentials: MongoCredentials,
-    response?: Document
+    response?: Document,
+    closeSignal: AbortSignal
   ): Promise<void>;
 
   /**

src/cmap/connect.ts

@@ -16,7 +16,7 @@ import {
   MongoRuntimeError,
   needsRetryableWriteLabel
 } from '../error';
-import { addAbortSignalToStream, HostAddress, ns, promiseWithResolvers } from '../utils';
+import { addAbortSignalToStream, HostAddress, noop, ns, promiseWithResolvers } from '../utils';
 import { AuthContext } from './auth/auth_provider';
 import { AuthMechanism } from './auth/providers';
 import {
@@ -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();
   }

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/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

@@ -1571,7 +1571,16 @@ export function addAbortSignalToStream(
 
   const abortListener = addAbortListener(signal, function () {
     stream.off('close', abortListener[kDispose]).off('error', abortListener[kDispose]);
-    stream.destroy(this.reason);
+    stream.destroy(
+      new (class extends Error {
+        s = stream;
+      })(
+        `sad: ${stream.___socketId}: error listeners: ${stream.listenerCount('error')} + ${stream.___stack}`,
+        {
+          cause: this.reason
+        }
+      )
+    );
   });
   // 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;
     }
   });
 

test/integration/client-side-encryption/client_side_encryption.prose.18.azure_kms_mock_server.test.ts

@@ -30,6 +30,8 @@ const metadata: MongoDBMetadataUI = {
   }
 };
 
+const closeSignal = new AbortController().signal;
+
 context('Azure KMS Mock Server Tests', function () {
   context('Case 1: Success', metadata, function () {
     // 	Do not set an ``X-MongoDB-HTTP-TestParams`` header.
@@ -44,7 +46,7 @@ context('Azure KMS Mock Server Tests', function () {
     // 5. The token will have a resource of ``"https://vault.azure.net"``
 
     it('returns a properly formatted access token', async () => {
-      const credentials = await fetchAzureKMSToken(new KMSRequestOptions());
+      const credentials = await fetchAzureKMSToken(new KMSRequestOptions(), closeSignal);
       expect(credentials).to.have.property('accessToken', 'magic-cookie');
     });
   });
@@ -59,7 +61,10 @@ context('Azure KMS Mock Server Tests', function () {
     // The test case should ensure that this error condition is handled gracefully.
 
     it('returns an error', async () => {
-      const error = await fetchAzureKMSToken(new KMSRequestOptions('empty-json')).catch(e => e);
+      const error = await fetchAzureKMSToken(
+        new KMSRequestOptions('empty-json'),
+        closeSignal
+      ).catch(e => e);
 
       expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
     });
@@ -74,7 +79,9 @@ context('Azure KMS Mock Server Tests', function () {
     // The test case should ensure that this error condition is handled gracefully.
 
     it('returns an error', async () => {
-      const error = await fetchAzureKMSToken(new KMSRequestOptions('bad-json')).catch(e => e);
+      const error = await fetchAzureKMSToken(new KMSRequestOptions('bad-json'), closeSignal).catch(
+        e => e
+      );
 
       expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
     });
@@ -89,7 +96,9 @@ context('Azure KMS Mock Server Tests', function () {
     // 2. The response body is unspecified.
     // The test case should ensure that this error condition is handled gracefully.
     it('returns an error', async () => {
-      const error = await fetchAzureKMSToken(new KMSRequestOptions('404')).catch(e => e);
+      const error = await fetchAzureKMSToken(new KMSRequestOptions('404'), closeSignal).catch(
+        e => e
+      );
 
       expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
     });
@@ -104,7 +113,9 @@ context('Azure KMS Mock Server Tests', function () {
     // 2. The response body is unspecified.
     // The test case should ensure that this error condition is handled gracefully.
     it('returns an error', async () => {
-      const error = await fetchAzureKMSToken(new KMSRequestOptions('500')).catch(e => e);
+      const error = await fetchAzureKMSToken(new KMSRequestOptions('500'), closeSignal).catch(
+        e => e
+      );
 
       expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
     });
@@ -117,7 +128,9 @@ context('Azure KMS Mock Server Tests', function () {
     // The HTTP response from the ``fake_azure`` server will take at least 1000 seconds
     // to complete. The request should fail with a timeout.
     it('returns an error after the request times out', async () => {
-      const error = await fetchAzureKMSToken(new KMSRequestOptions('slow')).catch(e => e);
+      const error = await fetchAzureKMSToken(new KMSRequestOptions('slow'), closeSignal).catch(
+        e => e
+      );
 
       expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
     });

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

@@ -829,12 +829,14 @@ describe('CSOT', function () {
   });
 
   describe('State machine', function () {
-    const stateMachine = new StateMachine({} as any);
+    const signal = new AbortController().signal;
+    const stateMachine = new StateMachine({} as any, undefined, signal);
 
     const timeoutContext = () => ({
       timeoutContext: new CSOTTimeoutContext({
         timeoutMS: 1000,
-        serverSelectionTimeoutMS: 30000
+        serverSelectionTimeoutMS: 30000,
+        closeSignal: signal
       })
     });
 

test/mocha_mongodb.json

@@ -17,7 +17,6 @@
   "recursive": true,
   "timeout": 60000,
   "failZero": true,
-  "reporter": "test/tools/reporter/mongodb_reporter.js",
   "sort": true,
   "color": true,
   "ignore": [

test/tools/cluster_setup.sh

@@ -13,8 +13,8 @@ SHARDED_DIR=${SHARDED_DIR:-$DATA_DIR/sharded_cluster}
 
 if [[ $1 == "replica_set" ]]; then
     mkdir -p $REPLICASET_DIR # user / password
-    mlaunch init --dir $REPLICASET_DIR --ipv6 --auth --username "bob" --password "pwd123" --replicaset --nodes 3 --arbiter --name rs --port 31000 --enableMajorityReadConcern --setParameter enableTestCommands=1
-    echo "mongodb://bob:pwd123@localhost:31000,localhost:31001,localhost:31002/?replicaSet=rs"
+    mlaunch init --dir $REPLICASET_DIR --ipv6 --auth --username "bob" --password "pwd123" --replicaset --nodes 3 --arbiter --name "repl0" --port 27017 --enableMajorityReadConcern --setParameter enableTestCommands=1
+    echo "mongodb://bob:pwd123@localhost:27017,localhost:27018,localhost:27019/?replicaSet=repl0"
 elif [[ $1 == "sharded_cluster" ]]; then
     mkdir -p $SHARDED_DIR
     mlaunch init --dir $SHARDED_DIR --ipv6 --auth --username "bob" --password "pwd123" --replicaset --nodes 3 --name rs --port 51000 --enableMajorityReadConcern --setParameter enableTestCommands=1 --sharded 1 --mongos 2

test/tools/cmap_spec_runner.ts

@@ -191,11 +191,14 @@ const compareInputToSpec = (input, expected, message) => {
   expect(input, message).to.equal(expected);
 };
 
+const closeSignal = new AbortController().signal;
+
 const getTestOpDefinitions = (threadContext: ThreadContext) => ({
   checkOut: async function (op) {
     const timeoutContext = TimeoutContext.create({
       serverSelectionTimeoutMS: 0,
-      waitQueueTimeoutMS: threadContext.pool.options.waitQueueTimeoutMS
+      waitQueueTimeoutMS: threadContext.pool.options.waitQueueTimeoutMS,
+      closeSignal
     });
     const connection: Connection = await ConnectionPool.prototype.checkOut.call(
       threadContext.pool,
@@ -470,8 +473,6 @@ export function runCmapTestSuite(
         client: MongoClient;
 
       beforeEach(async function () {
-        let utilClient: MongoClient;
-
         const skipDescription = options?.testsToSkip?.find(
           ({ description }) => description === test.description
         );
@@ -486,20 +487,17 @@ export function runCmapTestSuite(
           }
         }
 
-        if (this.configuration.isLoadBalanced) {
-          // The util client can always point at the single mongos LB frontend.
-          utilClient = this.configuration.newClient(this.configuration.singleMongosLoadBalancerUri);
-        } else {
-          utilClient = this.configuration.newClient();
-        }
+        const utilClient = this.configuration.isLoadBalanced
+          ? this.configuration.newClient(this.configuration.singleMongosLoadBalancerUri)
+          : this.configuration.newClient();
 
         await utilClient.connect();
 
         const allRequirements = test.runOn || [];
 
         const someRequirementMet =
           !allRequirements.length ||
-          (await isAnyRequirementSatisfied(this.currentTest.ctx, allRequirements, utilClient));
+          (await isAnyRequirementSatisfied(this.currentTest.ctx, allRequirements));
 
         if (!someRequirementMet) {
           await utilClient.close();

test/tools/runner/config.ts

@@ -86,6 +86,7 @@ export class TestConfiguration {
   serverApi?: ServerApi;
   activeResources: number;
   isSrv: boolean;
+  shards: { host: string }[];
 
   constructor(
     private uri: string,
@@ -103,6 +104,7 @@ export class TestConfiguration {
     this.topologyType = this.isLoadBalanced ? TopologyType.LoadBalanced : context.topologyType;
     this.buildInfo = context.buildInfo;
     this.serverApi = context.serverApi;
+    this.shards = context.shards;
     this.isSrv = uri.indexOf('mongodb+srv') > -1;
     this.options = {
       hosts,

test/tools/runner/ee_checker.ts

@@ -24,7 +24,7 @@ events.EventEmitter = class RequireErrorListenerEventEmitter extends EventEmitte
     process.nextTick(() => {
       const isCS = this.constructor.name.toLowerCase().includes('ChangeStream'.toLowerCase());
       if (isCS) {
-        // consider adding a warning.
+        // consider adding a warning. something related to mode === 'iterator' should skip this.
         return;
       }
       if (this.listenerCount('error') === 0) {

test/tools/runner/hooks/configuration.ts

@@ -153,6 +153,11 @@ const testConfigBeforeHook = async function () {
     .command({ getParameter: '*' })
     .catch(error => ({ noReply: error }));
 
+  context.shards =
+    context.topologyType === 'sharded'
+      ? await client.db('config').collection('shards').find({}).toArray()
+      : [];
+
   this.configuration = new TestConfiguration(
     loadBalanced ? SINGLE_MONGOS_LB_URI : MONGODB_URI,
     context

test/tools/runner/hooks/leak_checker.ts

@@ -140,8 +140,9 @@ const leakCheckerAfterEach = async function () {
   }
 };
 
-const TRACE_SOCKETS = process.env.TRACE_SOCKETS === 'true' ? true : false;
-const kSocketId = Symbol('socketId');
+const TRACE_SOCKETS = true; // process.env.TRACE_SOCKETS === 'true' ? true : false;
+const kSocketId = '___socketId';
+const kStack = '___stack';
 const originalCreateConnection = net.createConnection;
 let socketCounter = 0n;
 
@@ -150,6 +151,7 @@ const socketLeakCheckBeforeAll = function socketLeakCheckBeforeAll() {
   net.createConnection = options => {
     const socket = originalCreateConnection(options);
     socket[kSocketId] = socketCounter.toString().padStart(5, '0');
+    socket[kStack] = new Error('').stack;
     socketCounter++;
     return socket;
   };