feat(NODE-6313): add CSOT support to sessions and transactions (#4199)

Author: nbbeeken

package-lock.json

@@ -97,7 +97,7 @@
     "mocha": "^10.4.0",
     "mocha-sinon": "^2.1.2",
     "mongodb-client-encryption": "^6.1.0",
-    "mongodb-legacy": "^6.1.0",
+    "mongodb-legacy": "^6.1.1",
     "nyc": "^15.1.0",
     "prettier": "^3.3.3",
     "semver": "^7.6.3",

package.json

@@ -750,6 +750,13 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
           return;
         }
       }
+    } catch (readError) {
+      if (TimeoutError.is(readError)) {
+        throw new MongoOperationTimeoutError(
+          `Timed out during socket read (${readError.duration}ms)`
+        );
+      }
+      throw readError;
     } finally {
       this.dataEvents = null;
       this.messageStream.pause();

src/cmap/connection.ts

@@ -1,7 +1,6 @@
 import { type EventEmitter } from 'events';
 
-import { MongoOperationTimeoutError } from '../../error';
-import { type TimeoutContext, TimeoutError } from '../../timeout';
+import { type TimeoutContext } from '../../timeout';
 import { List, promiseWithResolvers } from '../../utils';
 
 /**
@@ -91,8 +90,11 @@ export function onData(
   // Adding event handlers
   emitter.on('data', eventHandler);
   emitter.on('error', errorHandler);
+
+  const timeoutForSocketRead = timeoutContext?.timeoutForSocketRead;
+  timeoutForSocketRead?.throwIfExpired();
   // eslint-disable-next-line github/no-then
-  timeoutContext?.timeoutForSocketRead?.then(undefined, errorHandler);
+  timeoutForSocketRead?.then(undefined, errorHandler);
 
   return iterator;
 
@@ -104,12 +106,9 @@ export function onData(
 
   function errorHandler(err: Error) {
     const promise = unconsumedPromises.shift();
-    const timeoutError = TimeoutError.is(err)
-      ? new MongoOperationTimeoutError('Timed out during socket read')
-      : undefined;
 
-    if (promise != null) promise.reject(timeoutError ?? err);
-    else error = timeoutError ?? err;
+    if (promise != null) promise.reject(err);
+    else error = err;
     void closeHandler();
   }
 

src/cmap/wire_protocol/on_data.ts

@@ -470,10 +470,14 @@ export class Collection<TSchema extends Document = Document> {
     // Intentionally, we do not inherit options from parent for this operation.
     return await executeOperation(
       this.client,
-      new RenameOperation(this as TODO_NODE_3286, newName, {
-        ...options,
-        readPreference: ReadPreference.PRIMARY
-      }) as TODO_NODE_3286
+      new RenameOperation(
+        this as TODO_NODE_3286,
+        newName,
+        resolveOptions(undefined, {
+          ...options,
+          readPreference: ReadPreference.PRIMARY
+        })
+      ) as TODO_NODE_3286
     );
   }
 

src/collection.ts

@@ -275,12 +275,16 @@ export class Db {
     // Intentionally, we do not inherit options from parent for this operation.
     return await executeOperation(
       this.client,
-      new RunCommandOperation(this, command, {
-        ...resolveBSONOptions(options),
-        timeoutMS: options?.timeoutMS ?? this.timeoutMS,
-        session: options?.session,
-        readPreference: options?.readPreference
-      })
+      new RunCommandOperation(
+        this,
+        command,
+        resolveOptions(undefined, {
+          ...resolveBSONOptions(options),
+          timeoutMS: options?.timeoutMS ?? this.timeoutMS,
+          session: options?.session,
+          readPreference: options?.readPreference
+        })
+      )
     );
   }
 
@@ -385,7 +389,11 @@ export class Db {
       new RenameOperation(
         this.collection<TSchema>(fromCollection) as TODO_NODE_3286,
         toCollection,
-        { ...options, new_collection: true, readPreference: ReadPreference.primary }
+        resolveOptions(undefined, {
+          ...options,
+          new_collection: true,
+          readPreference: ReadPreference.primary
+        })
       ) as TODO_NODE_3286
     );
   }

src/db.ts

@@ -128,6 +128,9 @@ function isAggregateError(e: unknown): e is Error & { errors: Error[] } {
  * mongodb-client-encryption has a dependency on this error, it uses the constructor with a string argument
  */
 export class MongoError extends Error {
+  get [Symbol.toStringTag]() {
+    return this.name;
+  }
   /** @internal */
   [kErrorLabels]: Set<string>;
   /**

src/error.ts

@@ -58,7 +58,7 @@ type ResultTypeFromOperation<TOperation> =
 export async function executeOperation<
   T extends AbstractOperation<TResult>,
   TResult = ResultTypeFromOperation<T>
->(client: MongoClient, operation: T, timeoutContext?: TimeoutContext): Promise<TResult> {
+>(client: MongoClient, operation: T, timeoutContext?: TimeoutContext | null): Promise<TResult> {
   if (!(operation instanceof AbstractOperation)) {
     // TODO(NODE-3483): Extend MongoRuntimeError
     throw new MongoRuntimeError('This method requires a valid operation instance');
@@ -81,11 +81,6 @@ export async function executeOperation<
   } else if (session.client !== client) {
     throw new MongoInvalidArgumentError('ClientSession must be from the same MongoClient');
   }
-  if (session.explicit && session?.timeoutMS != null && operation.options.timeoutMS != null) {
-    throw new MongoInvalidArgumentError(
-      'Do not specify timeoutMS on operation if already specified on an explicit session'
-    );
-  }
 
   const readPreference = operation.readPreference ?? ReadPreference.primary;
   const inTransaction = !!session?.inTransaction();
@@ -107,6 +102,7 @@ export async function executeOperation<
   }
 
   timeoutContext ??= TimeoutContext.create({
+    session,
     serverSelectionTimeoutMS: client.s.options.serverSelectionTimeoutMS,
     waitQueueTimeoutMS: client.s.options.waitQueueTimeoutMS,
     timeoutMS: operation.options.timeoutMS