feat(NODE-6305): Add CSOT support to tailable cursors

etc/notes/errors.md

@@ -67,7 +67,7 @@ Children of `MongoError` include:
 ### `MongoDriverError`
 
 This class represents errors which originate in the driver itself or when the user incorrectly uses the driver. This class should **never** be directly instantiated.
-Its children are the main classes of errors that most users will interact with: [**`MongoAPIError`**](#MongoAPIError) and [**`MongoRuntimeError`**](#MongoRuntimeError).
+Its children are the main classes of errors that most users will interact with: [**`MongoAPIError`**](#MongoAPIError), [**`MongoRuntimeError`**](#MongoRuntimeError) and [**`MongoOperationTimeoutError`**](#MongoOperationTimeoutError).
 
 ### `MongoAPIError`
 
@@ -109,6 +109,10 @@ This class should **never** be directly instantiated.
 | **MongoGridFSChunkError**   | Thrown when a malformed or invalid chunk is encountered when reading from a GridFS Stream. |
 | **MongoUnexpectedServerResponseError**   | Thrown when the driver receives a **parsable** response it did not expect from the server. |
 
+### `MongoOperationTimeoutError`
+
+- TODO(NODE-5688): Add MongoOperationTimeoutError documentation
+
 ### MongoUnexpectedServerResponseError
 
 Intended for the scenario where the MongoDB returns an unexpected response in relation to some state the driver is in.

package.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",

src/admin.ts

@@ -78,7 +78,8 @@ export class Admin {
       new RunAdminCommandOperation(command, {
         ...resolveBSONOptions(options),
         session: options?.session,
-        readPreference: options?.readPreference
+        readPreference: options?.readPreference,
+        timeoutMS: options?.timeoutMS ?? this.s.db.timeoutMS
       })
     );
   }
@@ -154,7 +155,10 @@ export class Admin {
    * @param options - Optional settings for the command
    */
   async listDatabases(options?: ListDatabasesOptions): Promise<ListDatabasesResult> {
-    return await executeOperation(this.s.db.client, new ListDatabasesOperation(this.s.db, options));
+    return await executeOperation(
+      this.s.db.client,
+      new ListDatabasesOperation(this.s.db, { timeoutMS: this.s.db.timeoutMS, ...options })
+    );
   }
 
   /**

src/bulk/common.ts

@@ -20,6 +20,7 @@ import { makeUpdateStatement, UpdateOperation, type UpdateStatement } from '../o
 import type { Server } from '../sdam/server';
 import type { Topology } from '../sdam/topology';
 import type { ClientSession } from '../sessions';
+import { type TimeoutContext } from '../timeout';
 import {
   applyRetryableWrites,
   type Callback,
@@ -873,6 +874,9 @@ export interface BulkWriteOptions extends CommandOperationOptions {
   forceServerObjectId?: boolean;
   /** Map of parameter names and values that can be accessed using $$var (requires MongoDB 5.0). */
   let?: Document;
+
+  /** @internal */
+  timeoutContext?: TimeoutContext;
 }
 
 const executeCommandsAsync = promisify(executeCommands);

src/cmap/connection.ts

@@ -16,9 +16,11 @@ import {
 } from '../constants';
 import {
   MongoCompatibilityError,
+  MONGODB_ERROR_CODES,
   MongoMissingDependencyError,
   MongoNetworkError,
   MongoNetworkTimeoutError,
+  MongoOperationTimeoutError,
   MongoParseError,
   MongoServerError,
   MongoUnexpectedServerResponseError
@@ -30,6 +32,7 @@ import { type CancellationToken, TypedEventEmitter } from '../mongo_types';
 import { ReadPreference, type ReadPreferenceLike } from '../read_preference';
 import { ServerType } from '../sdam/common';
 import { applySession, type ClientSession, updateSessionFromResponse } from '../sessions';
+import { type TimeoutContext, TimeoutError } from '../timeout';
 import {
   BufferPool,
   calculateDurationInMs,
@@ -83,6 +86,7 @@ export interface CommandOptions extends BSONSerializeOptions {
   documentsReturnedIn?: string;
   noResponse?: boolean;
   omitReadPreference?: boolean;
+  omitMaxTimeMS?: boolean;
 
   // TODO(NODE-2802): Currently the CommandOptions take a property willRetryWrite which is a hint
   // from executeOperation that the txnNum should be applied to this command.
@@ -94,6 +98,9 @@ export interface CommandOptions extends BSONSerializeOptions {
   writeConcern?: WriteConcern;
 
   directConnection?: boolean;
+
+  /** @internal */
+  timeoutContext?: TimeoutContext;
 }
 
 /** @public */
@@ -413,6 +420,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
       ...options
     };
 
+    if (!options.omitMaxTimeMS && options.timeoutContext?.csotEnabled()) {
+      const { maxTimeMS } = options.timeoutContext;
+      if (maxTimeMS > 0 && Number.isFinite(maxTimeMS)) cmd.maxTimeMS = maxTimeMS;
+    }
+
     const message = this.supportsOpMsg
       ? new OpMsgRequest(db, cmd, commandOptions)
       : new OpQueryRequest(db, cmd, commandOptions);
@@ -427,7 +439,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
   ): AsyncGenerator<MongoDBResponse> {
     this.throwIfAborted();
 
-    if (typeof options.socketTimeoutMS === 'number') {
+    if (options.timeoutContext?.csotEnabled()) {
+      this.socket.setTimeout(0);
+    } else if (typeof options.socketTimeoutMS === 'number') {
       this.socket.setTimeout(options.socketTimeoutMS);
     } else if (this.socketTimeoutMS !== 0) {
       this.socket.setTimeout(this.socketTimeoutMS);
@@ -436,7 +450,8 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     try {
       await this.writeCommand(message, {
         agreedCompressor: this.description.compressor ?? 'none',
-        zlibCompressionLevel: this.description.zlibCompressionLevel
+        zlibCompressionLevel: this.description.zlibCompressionLevel,
+        timeoutContext: options.timeoutContext
       });
 
       if (options.noResponse || message.moreToCome) {
@@ -446,7 +461,17 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
       this.throwIfAborted();
 
-      for await (const response of this.readMany()) {
+      if (
+        options.timeoutContext?.csotEnabled() &&
+        options.timeoutContext.minRoundTripTime != null &&
+        options.timeoutContext.remainingTimeMS < options.timeoutContext.minRoundTripTime
+      ) {
+        throw new MongoOperationTimeoutError(
+          'Server roundtrip time is greater than the time remaining'
+        );
+      }
+
+      for await (const response of this.readMany({ timeoutContext: options.timeoutContext })) {
         this.socket.setTimeout(0);
         const bson = response.parse();
 
@@ -515,6 +540,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         }
 
         if (document.ok === 0) {
+          if (options.timeoutContext?.csotEnabled() && document.isMaxTimeExpiredError) {
+            throw new MongoOperationTimeoutError('Server reported a timeout error', {
+              cause: new MongoServerError((object ??= document.toObject(bsonOptions)))
+            });
+          }
           throw new MongoServerError((object ??= document.toObject(bsonOptions)));
         }
 
@@ -588,6 +618,28 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
   ): Promise<Document> {
     this.throwIfAborted();
     for await (const document of this.sendCommand(ns, command, options, responseType)) {
+      if (options.timeoutContext?.csotEnabled()) {
+        if (MongoDBResponse.is(document)) {
+          if (document.isMaxTimeExpiredError) {
+            throw new MongoOperationTimeoutError('Server reported a timeout error', {
+              cause: new MongoServerError(document.toObject())
+            });
+          }
+        } else {
+          if (
+            (Array.isArray(document?.writeErrors) &&
+              document.writeErrors.some(
+                error => error?.code === MONGODB_ERROR_CODES.MaxTimeMSExpired
+              )) ||
+            document?.writeConcernError?.code === MONGODB_ERROR_CODES.MaxTimeMSExpired
+          ) {
+            throw new MongoOperationTimeoutError('Server reported a timeout error', {
+              cause: new MongoServerError(document)
+            });
+          }
+        }
+      }
+
       return document;
     }
     throw new MongoUnexpectedServerResponseError('Unable to get response from server');
@@ -623,7 +675,11 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    */
   private async writeCommand(
     command: WriteProtocolMessageType,
-    options: { agreedCompressor?: CompressorName; zlibCompressionLevel?: number }
+    options: {
+      agreedCompressor?: CompressorName;
+      zlibCompressionLevel?: number;
+      timeoutContext?: TimeoutContext;
+    }
   ): Promise<void> {
     const finalCommand =
       options.agreedCompressor === 'none' || !OpCompressedRequest.canCompress(command)
@@ -635,8 +691,32 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
 
     const buffer = Buffer.concat(await finalCommand.toBin());
 
+    if (options.timeoutContext?.csotEnabled()) {
+      if (
+        options.timeoutContext.minRoundTripTime != null &&
+        options.timeoutContext.remainingTimeMS < options.timeoutContext.minRoundTripTime
+      ) {
+        throw new MongoOperationTimeoutError(
+          'Server roundtrip time is greater than the time remaining'
+        );
+      }
+    }
+
     if (this.socket.write(buffer)) return;
-    return await once(this.socket, 'drain');
+
+    const drainEvent = once<void>(this.socket, 'drain');
+    const timeout = options?.timeoutContext?.timeoutForSocketWrite;
+    if (timeout) {
+      try {
+        return await Promise.race([drainEvent, timeout]);
+      } catch (error) {
+        if (TimeoutError.is(error)) {
+          throw new MongoOperationTimeoutError('Timed out at socket write');
+        }
+        throw error;
+      }
+    }
+    return await drainEvent;
   }
 
   /**
@@ -648,9 +728,12 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
    *
    * Note that `for-await` loops call `return` automatically when the loop is exited.
    */
-  private async *readMany(): AsyncGenerator<OpMsgResponse | OpReply> {
+  private async *readMany(options: {
+    timeoutContext?: TimeoutContext;
+  }): AsyncGenerator<OpMsgResponse | OpReply> {
     try {
-      this.dataEvents = onData(this.messageStream);
+      this.dataEvents = onData(this.messageStream, options);
+
       for await (const message of this.dataEvents) {
         const response = await decompressResponse(message);
         yield response;
@@ -659,6 +742,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();