feat(NODE-6312): add error transformation for server timeouts (#4192)

src/cmap/connection.ts

@@ -21,6 +21,7 @@ import {
 } from '../constants';
 import {
   MongoCompatibilityError,
+  MONGODB_ERROR_CODES,
   MongoMissingDependencyError,
   MongoNetworkError,
   MongoNetworkTimeoutError,
@@ -545,6 +546,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)));
         }
 
@@ -618,6 +624,29 @@ 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)) {
+          // TODO(NODE-5684): test coverage to be added once cursors are enabling CSOT
+          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');

src/cmap/wire_protocol/responses.ts

@@ -10,7 +10,7 @@ import {
   pluckBSONSerializeOptions,
   type Timestamp
 } from '../../bson';
-import { MongoUnexpectedServerResponseError } from '../../error';
+import { MONGODB_ERROR_CODES, MongoUnexpectedServerResponseError } from '../../error';
 import { type ClusterTime } from '../../sdam/common';
 import { decorateDecryptionResult, ns } from '../../utils';
 import {
@@ -110,6 +110,40 @@ export class MongoDBResponse extends OnDemandDocument {
   // {ok:1}
   static empty = new MongoDBResponse(new Uint8Array([13, 0, 0, 0, 16, 111, 107, 0, 1, 0, 0, 0, 0]));
 
+  /**
+   * Returns true iff:
+   * - ok is 0 and the top-level code === 50
+   * - ok is 1 and the writeErrors array contains a code === 50
+   * - ok is 1 and the writeConcern object contains a code === 50
+   */
+  get isMaxTimeExpiredError() {
+    // {ok: 0, code: 50 ... }
+    const isTopLevel = this.ok === 0 && this.code === MONGODB_ERROR_CODES.MaxTimeMSExpired;
+    if (isTopLevel) return true;
+
+    if (this.ok === 0) return false;
+
+    // {ok: 1, writeConcernError: {code: 50 ... }}
+    const isWriteConcern =
+      this.get('writeConcernError', BSONType.object)?.getNumber('code') ===
+      MONGODB_ERROR_CODES.MaxTimeMSExpired;
+    if (isWriteConcern) return true;
+
+    const writeErrors = this.get('writeErrors', BSONType.array);
+    if (writeErrors?.size()) {
+      for (let i = 0; i < writeErrors.size(); i++) {
+        const isWriteError =
+          writeErrors.get(i, BSONType.object)?.getNumber('code') ===
+          MONGODB_ERROR_CODES.MaxTimeMSExpired;
+
+        // {ok: 1, writeErrors: [{code: 50 ... }]}
+        if (isWriteError) return true;
+      }
+    }
+
+    return false;
+  }
+
   /**
    * Drivers can safely assume that the `recoveryToken` field is always a BSON document but drivers MUST NOT modify the
    * contents of the document.

test/integration/client-side-operations-timeout/node_csot.test.ts

@@ -1,17 +1,23 @@
 /* Anything javascript specific relating to timeouts */
 import { expect } from 'chai';
+import * as semver from 'semver';
+import * as sinon from 'sinon';
 
 import {
+  BSON,
   type ClientSession,
   type Collection,
+  Connection,
   type Db,
   type FindCursor,
   LEGACY_HELLO_COMMAND,
   type MongoClient,
-  MongoOperationTimeoutError
+  MongoOperationTimeoutError,
+  MongoServerError
 } from '../../mongodb';
+import { type FailPoint } from '../../tools/utils';
 
-describe('CSOT driver tests', () => {
+describe('CSOT driver tests', { requires: { mongodb: '>=4.4' } }, () => {
   describe('timeoutMS inheritance', () => {
     let client: MongoClient;
     let db: Db;
@@ -161,4 +167,157 @@ describe('CSOT driver tests', () => {
       });
     });
   });
+
+  describe('server-side maxTimeMS errors are transformed', () => {
+    let client: MongoClient;
+    let commandsSucceeded;
+    let commandsFailed;
+
+    beforeEach(async function () {
+      client = this.configuration.newClient({ timeoutMS: 500_000, monitorCommands: true });
+      commandsSucceeded = [];
+      commandsFailed = [];
+      client.on('commandSucceeded', event => {
+        if (event.commandName === 'configureFailPoint') return;
+        commandsSucceeded.push(event);
+      });
+      client.on('commandFailed', event => commandsFailed.push(event));
+    });
+
+    afterEach(async function () {
+      await client
+        .db()
+        .collection('a')
+        .drop()
+        .catch(() => null);
+      await client.close();
+      commandsSucceeded = undefined;
+      commandsFailed = undefined;
+    });
+
+    describe('when a maxTimeExpired error is returned at the top-level', () => {
+      // {ok: 0, code: 50, codeName: "MaxTimeMSExpired", errmsg: "operation time limit exceeded"}
+      const failpoint: FailPoint = {
+        configureFailPoint: 'failCommand',
+        mode: { times: 1 },
+        data: {
+          failCommands: ['ping'],
+          errorCode: 50
+        }
+      };
+
+      beforeEach(async function () {
+        if (semver.satisfies(this.configuration.version, '>=4.4'))
+          await client.db('admin').command(failpoint);
+        else {
+          this.skipReason = 'Requires server version later than 4.4';
+          this.skip();
+        }
+      });
+
+      afterEach(async function () {
+        if (semver.satisfies(this.configuration.version, '>=4.4'))
+          await client.db('admin').command({ ...failpoint, mode: 'off' });
+      });
+
+      it('throws a MongoOperationTimeoutError error and emits command failed', async () => {
+        const error = await client
+          .db()
+          .command({ ping: 1 })
+          .catch(error => error);
+        expect(error).to.be.instanceOf(MongoOperationTimeoutError);
+        expect(error.cause).to.be.instanceOf(MongoServerError);
+        expect(error.cause).to.have.property('code', 50);
+
+        expect(commandsFailed).to.have.lengthOf(1);
+        expect(commandsFailed).to.have.nested.property('[0].failure.cause.code', 50);
+      });
+    });
+
+    describe('when a maxTimeExpired error is returned inside a writeErrors array', () => {
+      // The server should always return one maxTimeExpiredError at the front of the writeErrors array
+      // But for the sake of defensive programming we will find any maxTime error in the array.
+
+      beforeEach(async () => {
+        const writeErrorsReply = BSON.serialize({
+          ok: 1,
+          writeErrors: [
+            { code: 2, codeName: 'MaxTimeMSExpired', errmsg: 'operation time limit exceeded' },
+            { code: 3, codeName: 'MaxTimeMSExpired', errmsg: 'operation time limit exceeded' },
+            { code: 4, codeName: 'MaxTimeMSExpired', errmsg: 'operation time limit exceeded' },
+            { code: 50, codeName: 'MaxTimeMSExpired', errmsg: 'operation time limit exceeded' }
+          ]
+        });
+        const commandSpy = sinon.spy(Connection.prototype, 'command');
+        const readManyStub = sinon
+          // @ts-expect-error: readMany is private
+          .stub(Connection.prototype, 'readMany')
+          .callsFake(async function* (...args) {
+            const realIterator = readManyStub.wrappedMethod.call(this, ...args);
+            const cmd = commandSpy.lastCall.args.at(1);
+            if ('giveMeWriteErrors' in cmd) {
+              await realIterator.next().catch(() => null); // dismiss response
+              yield { parse: () => writeErrorsReply };
+            } else {
+              yield (await realIterator.next()).value;
+            }
+          });
+      });
+
+      afterEach(() => sinon.restore());
+
+      it('throws a MongoOperationTimeoutError error and emits command succeeded', async () => {
+        const error = await client
+          .db('admin')
+          .command({ giveMeWriteErrors: 1 })
+          .catch(error => error);
+        expect(error).to.be.instanceOf(MongoOperationTimeoutError);
+        expect(error.cause).to.be.instanceOf(MongoServerError);
+        expect(error.cause).to.have.nested.property('writeErrors[3].code', 50);
+
+        expect(commandsSucceeded).to.have.lengthOf(1);
+        expect(commandsSucceeded).to.have.nested.property('[0].reply.writeErrors[3].code', 50);
+      });
+    });
+
+    describe('when a maxTimeExpired error is returned inside a writeConcernError embedded document', () => {
+      // {ok: 1, writeConcernError: {code: 50, codeName: "MaxTimeMSExpired"}}
+      const failpoint: FailPoint = {
+        configureFailPoint: 'failCommand',
+        mode: { times: 1 },
+        data: {
+          failCommands: ['insert'],
+          writeConcernError: { code: 50, errmsg: 'times up buster', errorLabels: [] }
+        }
+      };
+
+      beforeEach(async function () {
+        if (semver.satisfies(this.configuration.version, '>=4.4'))
+          await client.db('admin').command(failpoint);
+        else {
+          this.skipReason = 'Requires server version later than 4.4';
+          this.skip();
+        }
+      });
+
+      afterEach(async function () {
+        if (semver.satisfies(this.configuration.version, '>=4.4'))
+          await client.db('admin').command({ ...failpoint, mode: 'off' });
+      });
+
+      it('throws a MongoOperationTimeoutError error and emits command succeeded', async () => {
+        const error = await client
+          .db()
+          .collection('a')
+          .insertOne({})
+          .catch(error => error);
+        expect(error).to.be.instanceOf(MongoOperationTimeoutError);
+        expect(error.cause).to.be.instanceOf(MongoServerError);
+        expect(error.cause).to.have.nested.property('writeConcernError.code', 50);
+
+        expect(commandsSucceeded).to.have.lengthOf(1);
+        expect(commandsSucceeded).to.have.nested.property('[0].reply.writeConcernError.code', 50);
+      });
+    });
+  });
 });