wip

Author: nbbeeken

.eslintrc.json

@@ -127,6 +127,10 @@
       {
         "selector": "CallExpression[callee.name='clearTimeout']",
         "message": "clearTimeout must remove abort listener"
+      },
+      {
+        "selector": "CallExpression[callee.property.name='removeAllListeners'][arguments.length=0]",
+        "message": "removeAllListeners can remove error listeners leading to uncaught errors"
       }
     ],
     "@typescript-eslint/no-unused-vars": "error",

src/change_stream.ts

@@ -956,7 +956,9 @@ export class ChangeStream<
   private _endStream(): void {
     const cursorStream = this.cursorStream;
     if (cursorStream) {
-      ['data', 'close', 'end', 'error'].forEach(event => cursorStream.removeAllListeners(event));
+      cursorStream.removeAllListeners('data');
+      cursorStream.removeAllListeners('close');
+      cursorStream.removeAllListeners('end');
       cursorStream.destroy();
     }
 

src/cmap/connect.ts

@@ -441,7 +441,6 @@ export async function makeSocket(
     throw error;
   } finally {
     socket.setTimeout(0);
-    socket.removeAllListeners();
     if (cancellationHandler != null) {
       options.cancellationToken?.removeListener('cancel', cancellationHandler);
     }

src/cmap/connection_pool.ts

@@ -374,6 +374,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
 
     try {
       timeout?.throwIfExpired();
+      timeout?.ref();
       return await (timeout ? Promise.race([promise, timeout]) : promise);
     } catch (error) {
       if (TimeoutError.is(error)) {
@@ -399,6 +400,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
       }
       throw error;
     } finally {
+      timeout?.unref();
       abortListener?.[kDispose]();
       timeout?.clear();
     }

src/index.ts

@@ -53,6 +53,7 @@ export {
   MongoClientBulkWriteCursorError,
   MongoClientBulkWriteError,
   MongoClientBulkWriteExecutionError,
+  MongoClientClosedError,
   MongoCompatibilityError,
   MongoCursorExhaustedError,
   MongoCursorInUseError,

src/mongo_client.ts

@@ -22,7 +22,7 @@ import { MONGO_CLIENT_EVENTS } from './constants';
 import { type AbstractCursor } from './cursor/abstract_cursor';
 import { Db, type DbOptions } from './db';
 import type { Encrypter } from './encrypter';
-import { MongoInvalidArgumentError } from './error';
+import { MongoClientClosedError, MongoInvalidArgumentError } from './error';
 import { MongoClientAuthProviders } from './mongo_client_auth_providers';
 import {
   type LogComponentSeveritiesClientOptions,
@@ -692,7 +692,6 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
   /* @internal */
   private async _close(force = false): Promise<void> {
     try {
-      this.closeController.abort();
       // There's no way to set hasBeenClosed back to false
       Object.defineProperty(this.s, 'hasBeenClosed', {
         value: true,
@@ -701,6 +700,12 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
         writable: false
       });
 
+      if (this.options.maxPoolSize === 1) {
+        // If maxPoolSize is 1 we won't be able to run anything
+        // unless we interrupt whatever is using the one connection.
+        this.closeController.abort(new MongoClientClosedError());
+      }
+
       const activeCursorCloses = Array.from(this.s.activeCursors, cursor => cursor.close());
       this.s.activeCursors.clear();
 
@@ -749,7 +754,9 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
         await encrypter.close(this, force);
       }
     } finally {
-      // ignore
+      if (!this.closeController.signal.aborted) {
+        this.closeController.abort(new MongoClientClosedError());
+      }
     }
   }
 

src/sdam/monitor.ts

@@ -387,6 +387,7 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
   const makeMonitoringConnection = async () => {
     const socket = await makeSocket(monitor.connectOptions, monitor.closeSignal);
     const connection = makeConnection(monitor.connectOptions, socket);
+    connection.unref();
     // The start time is after socket creation but before the handshake
     start = now();
     try {
@@ -447,15 +448,11 @@ function monitorServer(monitor: Monitor) {
 
       // if the check indicates streaming is supported, immediately reschedule monitoring
       if (useStreamingProtocol(monitor, hello?.topologyVersion)) {
-        clearOnAbortTimeout(
-          () => {
-            if (!isInCloseState(monitor)) {
-              monitor.monitorId?.wake();
-            }
-          },
-          0,
-          monitor.closeSignal
-        );
+        queueMicrotask(() => {
+          if (!isInCloseState(monitor)) {
+            monitor.monitorId?.wake();
+          }
+        });
       }
 
       done();
@@ -554,6 +551,7 @@ export class RTTPinger {
     if (connection == null) {
       connect(this.monitor.connectOptions, this.closeSignal).then(
         connection => {
+          connection.unref();
           this.measureAndReschedule(start, connection);
         },
         () => {

src/sdam/topology.ts

@@ -621,6 +621,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
 
     try {
       timeout?.throwIfExpired();
+      timeout?.ref();
       const server = await (timeout ? Promise.race([serverPromise, timeout]) : serverPromise);
       if (options.timeoutContext?.csotEnabled() && server.description.minRoundTripTime !== 0) {
         options.timeoutContext.minRoundTripTime = server.description.minRoundTripTime;
@@ -661,6 +662,7 @@ export class Topology extends TypedEventEmitter<TopologyEvents> {
       // Other server selection error
       throw error;
     } finally {
+      timeout?.unref();
       abortListener?.[kDispose]();
       if (options.timeoutContext?.clearServerSelectionTimeout) timeout?.clear();
     }

src/timeout.ts

@@ -39,7 +39,7 @@ export function clearOnAbortTimeout(
   }, ms);
 
   if ('unref' in id && typeof id.unref === 'function') {
-    id.unref();
+    // id.unref();
   }
 
   const abortListener = addAbortListener(closeSignal, function clearId() {
@@ -116,7 +116,9 @@ export class Timeout extends Promise<never> {
     this.start = Math.trunc(performance.now());
 
     if (rejection == null && this.duration > 0) {
-      if (options.closeSignal == null) throw new Error('incorrect timer use detected!');
+      if (options.closeSignal == null) {
+        throw new Error('You must provide a close signal to timeoutContext');
+      }
 
       this.id = clearOnAbortTimeout(
         () => {

src/utils.ts

@@ -1571,7 +1571,7 @@ export function addAbortSignalToStream(
 
   const abortListener = addAbortListener(signal, function () {
     stream.off('close', abortListener[kDispose]).off('error', abortListener[kDispose]);
-    stream.destroy(this.reason);
+    if (!stream.destroyed) stream.destroy(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]);