Add tests for retry loop and backoff.

Author: baileympearson

src/cmap/connect.ts

@@ -209,6 +209,7 @@ export interface HandshakeDocument extends Document {
   compression: string[];
   saslSupportedMechs?: string;
   loadBalanced?: boolean;
+  backpressure: true;
 }
 
 /**
@@ -226,6 +227,7 @@ export async function prepareHandshakeDocument(
 
   const handshakeDoc: HandshakeDocument = {
     [serverApi?.version || options.loadBalanced === true ? 'hello' : LEGACY_HELLO_COMMAND]: 1,
+    backpressure: true,
     helloOk: true,
     client: clientMetadata,
     compression: compressors

src/index.ts

@@ -87,6 +87,7 @@ export {
   MongoWriteConcernError,
   WriteConcernErrorResult
 } from './error';
+export { TokenBucket } from './token_bucket';
 export {
   AbstractCursor,
   // Actual driver classes exported

src/operations/execute_operation.ts

@@ -38,6 +38,7 @@ import {
 } from '../utils';
 import { AggregateOperation } from './aggregate';
 import { AbstractOperation, Aspect } from './operation';
+import { RunCommandOperation } from './run_command';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
 const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
@@ -265,22 +266,22 @@ async function executeOperationWithRetries<
       if (previousOperationError.hasErrorLabel(MongoErrorLabel.SystemOverloadedError)) {
         systemOverloadRetryAttempt += 1;
 
+        // if retryable writes or reads are not configured, throw.
+        const isOperationConfiguredForRetry =
+          (hasReadAspect && topology.s.options.retryReads) ||
+          (hasWriteAspect && topology.s.options.retryWrites);
+        const isRunCommand = operation instanceof RunCommandOperation;
+
         if (
           // if the SystemOverloadError is not retryable, throw.
           !previousOperationError.hasErrorLabel(MongoErrorLabel.RetryableError) ||
-          !(
-            // if retryable writes or reads are not configured, throw.
-            (
-              (hasReadAspect && topology.s.options.retryReads) ||
-              (hasWriteAspect && topology.s.options.retryWrites)
-            )
-          )
+          !(isOperationConfiguredForRetry || isRunCommand)
         ) {
           throw previousOperationError;
         }
 
         // if we have exhausted overload retry attempts, throw.
-        if (systemOverloadRetryAttempt >= maxSystemOverloadRetryAttempts) {
+        if (systemOverloadRetryAttempt > maxSystemOverloadRetryAttempts) {
           throw previousOperationError;
         }
 
@@ -361,7 +362,10 @@ async function executeOperationWithRetries<
 
     try {
       // If attempt > 0 and we are command batching we need to reset the batch.
-      if (nonOverloadRetryAttempt > 0 && operation.hasAspect(Aspect.COMMAND_BATCHING)) {
+      if (
+        (nonOverloadRetryAttempt > 0 || systemOverloadRetryAttempt > 0) &&
+        operation.hasAspect(Aspect.COMMAND_BATCHING)
+      ) {
         operation.resetBatch();
       }
 

src/token_bucket.ts

@@ -1,3 +1,6 @@
+/**
+ * @internal
+ */
 export class TokenBucket {
   private budget: number;
   constructor(allowance: number) {

sync.sh

@@ -0,0 +1,3 @@
+
+
+cp ~/dev/specifications/source/client-backpressure/tests/* ~/dev/node-mongodb-native/test/spec/client-backpressure

test/integration/client-backpressure/client-backpressure.prose.test.ts

@@ -0,0 +1,60 @@
+import { expect } from 'chai';
+import * as sinon from 'sinon';
+
+import { type Collection, type MongoClient, MongoServerError } from '../../../src';
+import { clearFailPoint, configureFailPoint, measureDuration } from '../../tools/utils';
+
+describe('Client Backpressure (Prose)', function () {
+  let client: MongoClient;
+  let collection: Collection;
+
+  beforeEach(async function () {
+    client = this.configuration.newClient();
+    await client.connect();
+
+    collection = client.db('foo').collection('bar');
+  });
+
+  afterEach(async function () {
+    await client.close();
+    await clearFailPoint(this.configuration);
+  });
+
+  it(
+    'Test 1: Operation Retry Uses Exponential Backoff',
+    {
+      requires: {
+        mongodb: '4.4'
+      }
+    },
+    async function () {
+      await configureFailPoint(this.configuration, {
+        configureFailPoint: 'failCommand',
+        mode: 'alwaysOn',
+        data: {
+          failCommands: ['insert'],
+          errorCode: 2,
+          errorLabels: ['SystemOverloadedError', 'RetryableError']
+        }
+      });
+
+      const stub = sinon.stub(Math, 'random');
+
+      stub.returns(0);
+
+      const { duration: durationNoBackoff } = await measureDuration(async () => {
+        const error = await collection.insertOne({ a: 1 }).catch(e => e);
+        expect(error).to.be.instanceof(MongoServerError);
+      });
+
+      stub.returns(1);
+
+      const { duration: durationBackoff } = await measureDuration(async () => {
+        const error = await collection.insertOne({ a: 1 }).catch(e => e);
+        expect(error).to.be.instanceof(MongoServerError);
+      });
+
+      expect(durationBackoff - durationNoBackoff).to.be.within(3100 - 1000, 3100 + 1000);
+    }
+  );
+});

test/integration/client-backpressure/client-backpressure.spec.test.ts

@@ -0,0 +1,16 @@
+import { loadSpecTests } from '../../spec';
+import { runUnifiedSuite } from '../../tools/unified-spec-runner/runner';
+import { type Test } from '../../tools/unified-spec-runner/schema';
+
+const skippedTests = {
+  'collection.dropIndexes retries at most maxAttempts=5 times':
+    'TODO(NODE-6517): dropIndexes squashes all errors other than ns not found'
+};
+
+function shouldSkip({ description }: Test) {
+  return skippedTests[description] ?? false;
+}
+
+describe('Client Backpressure (spec)', function () {
+  runUnifiedSuite(loadSpecTests('client-backpressure'), shouldSkip);
+});

test/integration/mongodb-handshake/mongodb-handshake.prose.test.ts

@@ -1,7 +1,14 @@
 import { expect } from 'chai';
 import * as sinon from 'sinon';
 
-import { type ClientMetadata, type DriverInfo, Int32, MongoClient } from '../../../src';
+import {
+  type ClientMetadata,
+  type Document,
+  type DriverInfo,
+  type HandshakeDocument,
+  Int32,
+  MongoClient
+} from '../../../src';
 import { Connection } from '../../../src/cmap/connection';
 import { getFAASEnv, isDriverInfoEqual } from '../../../src/cmap/handshake/client_metadata';
 import { LEGACY_HELLO_COMMAND } from '../../../src/constants';
@@ -939,3 +946,38 @@ describe('Client Metadata Update Prose Tests', function () {
     }
   });
 });
+
+// TODO: add prose test descriptions here to align the test with the spec.
+describe('Backpressure Metadata', function () {
+  let client: MongoClient;
+  let spy: sinon.SinonSpy<Parameters<typeof Connection.prototype.command>>;
+
+  beforeEach(async function () {
+    client = this.configuration.newClient();
+    spy = sinon.spy(Connection.prototype, 'command');
+    await client.connect();
+
+    // run an operation to force a connection establishment,
+    // if we're testing noauth load balanced mode.
+    await client.db('foo').collection('bar').insertOne({ name: 'bumpy' });
+  });
+
+  afterEach(async function () {
+    sinon.restore();
+    await client?.close();
+  });
+
+  it('includes backpressure in the handshake document', function () {
+    const isHello = (cmd: Document): cmd is HandshakeDocument =>
+      `hello` in cmd || LEGACY_HELLO_COMMAND in cmd;
+
+    const hellos = spy.args.map(([_ns, command, _options]) => command).filter(isHello);
+
+    expect(hellos.length).to.be.greaterThan(0);
+
+    expect(
+      hellos.every(hello => hello.backpressure === true),
+      `some handshake documents did not specify backpressure: true`
+    );
+  });
+});

test/spec/client-backpressure/README.md

@@ -0,0 +1,61 @@
+# Client Backpressure Tests
+
+______________________________________________________________________
+
+## Introduction
+
+The YAML and JSON files in this directory are platform-independent tests meant to exercise a driver's implementation of
+retryable reads. These tests utilize the [Unified Test Format](../../unified-test-format/unified-test-format.md).
+
+Several prose tests, which are not easily expressed in YAML, are also presented in this file. Those tests will need to
+be manually implemented by each driver.
+
+### Prose Tests
+
+#### Test 1: Operation Retry Uses Exponential Backoff
+
+Drivers should test that retries do not occur immediately when a SystemOverloadedError is encountered.
+
+1. let `client` be a `MongoClient`
+2. let `collection` be a collection
+3. Now, run transactions without backoff:
+    1. Configure the random number generator used for jitter to always return `0` -- this effectively disables backoff.
+
+    2. Configure the following failPoint:
+
+        ```javascript
+            {
+                configureFailPoint: 'failCommand',
+                mode: 'alwaysOn',
+                data: {
+                failCommands: ['insert'],
+                errorCode: 2,
+                errorLabels: ['SystemOverloadedError', 'RetryableError']
+                }
+            }
+        ```
+
+    3. Execute the following command. Expect that the command errors. Measure the duration of the command execution.
+
+        ```javascript
+           const start = performance.now();
+           expect(
+            await coll.insertOne({ a: 1 }).catch(e => e)
+           ).to.be.an.instanceof(MongoServerError);
+           const end = performance.now();
+        ```
+
+    4. Configure the random number generator used for jitter to always return `1`.
+
+    5. Execute step 3 again.
+
+    6. Compare the two time between the two runs.
+    ```python
+    assertTrue(with_backoff_time - no_backoff_time >= 2.1)
+    ```
+    The sum of 5 backoffs is 3.1 seconds. There is a 1-second window to account for potential variance between the two
+    runs.
+
+## Changelog
+
+- 2025-XX-XX: Initial version.