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

package.json

@@ -99,7 +99,7 @@
     "mocha": "^10.4.0",
     "mocha-sinon": "^2.1.2",
     "mongodb-client-encryption": "^6.1.0",
-    "mongodb-legacy": "^6.0.1",
+    "mongodb-legacy": "^6.1.1",
     "nyc": "^15.1.0",
     "prettier": "^2.8.8",
     "semver": "^7.6.0",

src/cmap/connection.ts

@@ -737,6 +737,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.throwIfAborted();

src/cmap/wire_protocol/on_data.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/error.ts

@@ -765,9 +765,22 @@ export class MongoUnexpectedServerResponseError extends MongoRuntimeError {
  * @internal
  */
 export class MongoOperationTimeoutError extends MongoRuntimeError {
+  get [Symbol.toStringTag]() {
+    return 'MongoOperationTimeoutError';
+  }
+
   override get name(): string {
     return 'MongoOperationTimeoutError';
   }
+
+  static is(error: unknown): error is MongoOperationTimeoutError {
+    return (
+      error != null &&
+      typeof error === 'object' &&
+      Symbol.toStringTag in error &&
+      error[Symbol.toStringTag] === 'MongoOperationTimeoutError'
+    );
+  }
 }
 
 /**

src/operations/execute_operation.ts

@@ -82,11 +82,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,11 +102,13 @@ export async function executeOperation<
     session.unpin();
   }
 
-  timeoutContext ??= TimeoutContext.create({
-    serverSelectionTimeoutMS: client.s.options.serverSelectionTimeoutMS,
-    waitQueueTimeoutMS: client.s.options.waitQueueTimeoutMS,
-    timeoutMS: operation.options.timeoutMS
-  });
+  timeoutContext ??=
+    session.timeoutContext ??
+    TimeoutContext.create({
+      serverSelectionTimeoutMS: client.s.options.serverSelectionTimeoutMS,
+      waitQueueTimeoutMS: client.s.options.waitQueueTimeoutMS,
+      timeoutMS: operation.options.timeoutMS ?? session.timeoutMS
+    });
 
   try {
     return await tryOperation(operation, {