WIP

Author: baileympearson

src/cmap/connection.ts

@@ -582,6 +582,9 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         this.throwIfAborted();
       }
     } catch (error) {
+      if (options.session != null) {
+        updateSessionFromResponse(options.session, MongoDBResponse.empty);
+      }
       if (this.shouldEmitAndLogCommand) {
         this.emitAndLogCommand(
           this.monitorCommands,

src/operations/execute_operation.ts

@@ -37,7 +37,7 @@ import {
   supportsRetryableWrites
 } from '../utils';
 import { AggregateOperation } from './aggregate';
-import { AbstractOperation, Aspect } from './operation';
+import { AbstractOperation, Aspect, RetryContext } from './operation';
 
 const MMAPv1_RETRY_WRITES_ERROR_CODE = MONGODB_ERROR_CODES.IllegalOperation;
 const MMAPv1_RETRY_WRITES_ERROR_MESSAGE =
@@ -254,13 +254,18 @@ async function executeOperationWithRetries<
     2 // backoff rate
   );
 
+  const retryContext =
+    operation.retryContext ??
+    new RetryContext(willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1);
   for (
-    let attempt = 0, maxAttempts = willRetry ? (timeoutContext.csotEnabled() ? Infinity : 2) : 1;
-    attempt <= maxAttempts;
+    let attempt = 0;
+    attempt <= retryContext.maxAttempts;
     attempt++,
-      maxAttempts = previousOperationError?.hasErrorLabel(MongoErrorLabel.SystemOverloadedError)
+      retryContext.maxAttempts = previousOperationError?.hasErrorLabel(
+        MongoErrorLabel.SystemOverloadedError
+      )
         ? 5
-        : maxAttempts
+        : retryContext.maxAttempts
   ) {
     if (previousOperationError) {
       if (hasWriteAspect && previousOperationError.code === MMAPv1_RETRY_WRITES_ERROR_CODE) {
@@ -275,8 +280,8 @@ async function executeOperationWithRetries<
         // bulk write commands are retryable if all operations in the batch are retryable
         (operation.hasAspect(Aspect.COMMAND_BATCHING) && operation.canRetryWrite) ||
         // if we have a retryable read or write operation, we can retry
-        (hasWriteAspect && isRetryableWriteError(previousOperationError)) ||
-        (hasReadAspect && isRetryableReadError(previousOperationError)) ||
+        (hasWriteAspect && willRetryWrite && isRetryableWriteError(previousOperationError)) ||
+        (hasReadAspect && willRetryRead && isRetryableReadError(previousOperationError)) ||
         // if we have a retryable, system overloaded error, we can retry
         (previousOperationError.hasErrorLabel(MongoErrorLabel.SystemOverloadedError) &&
           previousOperationError.hasErrorLabel(MongoErrorLabel.RetryableError));

src/operations/operation.ts

@@ -45,6 +45,10 @@ export interface OperationOptions extends BSONSerializeOptions {
   timeoutMS?: number;
 }
 
+export class RetryContext {
+  constructor(public maxAttempts: number) {}
+}
+
 /**
  * This class acts as a parent class for any operation and is responsible for setting this.options,
  * as well as setting and getting a session.
@@ -66,6 +70,8 @@ export abstract class AbstractOperation<TResult = any> {
   /** Specifies the time an operation will run until it throws a timeout error. */
   timeoutMS?: number;
 
+  retryContext?: RetryContext;
+
   private _session: ClientSession | undefined;
 
   static aspects?: Set<symbol>;

src/sessions.ts

@@ -23,6 +23,7 @@ import {
 import type { MongoClient, MongoOptions } from './mongo_client';
 import { TypedEventEmitter } from './mongo_types';
 import { executeOperation } from './operations/execute_operation';
+import { RetryContext } from './operations/operation';
 import { RunCommandOperation } from './operations/run_command';
 import { ReadConcernLevel } from './read_concern';
 import { ReadPreference } from './read_preference';
@@ -466,7 +467,11 @@ export class ClientSession
       } else {
         const wcKeys = Object.keys(wc);
         if (wcKeys.length > 2 || (!wcKeys.includes('wtimeoutMS') && !wcKeys.includes('wTimeoutMS')))
-          // if the write concern was specified with wTimeoutMS, then we set both wtimeoutMS and wTimeoutMS, guaranteeing at least two keys, so if we have more than two keys, then we can automatically assume that we should add the write concern to the command. If it has 2 or fewer keys, we need to check that those keys aren't the wtimeoutMS or wTimeoutMS options before we add the write concern to the command
+          // if the write concern was specified with wTimeoutMS, then we set both wtimeoutMS
+          // and wTimeoutMS, guaranteeing at least two keys, so if we have more than two keys,
+          // then we can automatically assume that we should add the write concern to the command.
+          // If it has 2 or fewer keys, we need to check that those keys aren't the wtimeoutMS
+          // or wTimeoutMS options before we add the write concern to the command
           WriteConcern.apply(command, { ...wc, wtimeoutMS: undefined });
       }
     }
@@ -487,11 +492,14 @@ export class ClientSession
       command.recoveryToken = this.transaction.recoveryToken;
     }
 
+    const retryContext = new RetryContext(5);
+
     const operation = new RunCommandOperation(new MongoDBNamespace('admin'), command, {
       session: this,
       readPreference: ReadPreference.primary,
       bypassPinningCheck: true
     });
+    operation.retryContext = retryContext;
 
     const timeoutContext =
       this.timeoutContext ??
@@ -516,15 +524,13 @@ export class ClientSession
         this.unpin({ force: true });
 
         try {
-          await executeOperation(
-            this.client,
-            new RunCommandOperation(new MongoDBNamespace('admin'), command, {
-              session: this,
-              readPreference: ReadPreference.primary,
-              bypassPinningCheck: true
-            }),
-            timeoutContext
-          );
+          const op = new RunCommandOperation(new MongoDBNamespace('admin'), command, {
+            session: this,
+            readPreference: ReadPreference.primary,
+            bypassPinningCheck: true
+          });
+          op.retryContext = retryContext;
+          await executeOperation(this.client, op, timeoutContext);
           return;
         } catch (retryCommitError) {
           // If the retry failed, we process that error instead of the original
@@ -957,6 +963,11 @@ export class ServerSession {
   id: ServerSessionId;
   lastUse: number;
   txnNumber: number;
+
+  /*
+   * Indicates that a network error has been encountered while using this session.
+   * Once a session is marked as dirty, it is always dirty.
+   */
   isDirty: boolean;
 
   /** @internal */
@@ -1050,16 +1061,15 @@ export class ServerSessionPool {
    * @param session - The session to release to the pool
    */
   release(session: ServerSession): void {
-    const sessionTimeoutMinutes = this.client.topology?.logicalSessionTimeoutMinutes ?? 10;
+    if (this.client.topology?.loadBalanced) {
+      if (session.isDirty) return;
 
-    if (this.client.topology?.loadBalanced && !sessionTimeoutMinutes) {
       this.sessions.unshift(session);
-    }
-
-    if (!sessionTimeoutMinutes) {
       return;
     }
 
+    const sessionTimeoutMinutes = this.client.topology?.logicalSessionTimeoutMinutes ?? 10;
+
     this.sessions.prune(session => session.hasTimedOut(sessionTimeoutMinutes));
 
     if (!session.hasTimedOut(sessionTimeoutMinutes)) {
@@ -1147,9 +1157,9 @@ export function applySession(
   command.autocommit = false;
 
   if (session.transaction.state === TxnState.STARTING_TRANSACTION) {
-    session.transaction.transition(TxnState.TRANSACTION_IN_PROGRESS);
     command.startTransaction = true;
 
+    // TODO: read concern only applied if it is not the same as the server's default
     const readConcern =
       session.transaction.options.readConcern || session?.clientOptions?.readConcern;
     if (readConcern) {
@@ -1185,4 +1195,17 @@ export function updateSessionFromResponse(session: ClientSession, document: Mong
       session.snapshotTime = atClusterTime;
     }
   }
+
+  if (session.transaction.state === TxnState.STARTING_TRANSACTION) {
+    if (document.ok === 1) {
+      session.transaction.transition(TxnState.TRANSACTION_IN_PROGRESS);
+    } else {
+      const error = new MongoServerError(document.toObject());
+      const isBackpressureError = error.hasErrorLabel(MongoErrorLabel.RetryableError);
+
+      if (!isBackpressureError) {
+        session.transaction.transition(TxnState.TRANSACTION_IN_PROGRESS);
+      }
+    }
+  }
 }

sync.sh

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

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

@@ -11,6 +11,6 @@ function shouldSkip({ description }: Test) {
   return skippedTests[description] ?? false;
 }
 
-describe('Client Backpressure (spec)', function () {
+describe.only('Client Backpressure (spec)', function () {
   runUnifiedSuite(loadSpecTests('client-backpressure'), shouldSkip);
 });

test/integration/transactions/transactions.spec.test.ts

@@ -13,7 +13,7 @@ const SKIPPED_TESTS = [
   'client bulkWrite with writeConcern in a transaction causes a transaction error'
 ];
 
-describe('Transactions Spec Unified Tests', function () {
+describe.only('Transactions Spec Unified Tests', function () {
   runUnifiedSuite(loadSpecTests(path.join('transactions', 'unified')), test => {
     return SKIPPED_TESTS.includes(test.description)
       ? 'TODO(NODE-5924): Skipping failing transaction tests'

test/spec/client-backpressure/backpressure-retry-loop.json

@@ -31,15 +31,15 @@
     },
     {
       "database": {
-        "id": "database",
+        "id": "internal_db",
         "client": "internal_client",
         "databaseName": "retryable-writes-tests"
       }
     },
     {
       "collection": {
         "id": "retryable-writes-tests",
-        "database": "database",
+        "database": "internal_db",
         "collectionName": "coll"
       }
     },

test/spec/client-backpressure/backpressure-retry-loop.yml

@@ -23,7 +23,7 @@ createEntities:
 
   -
     database:
-      id: &internal_db database
+      id: &internal_db internal_db
       client: *internal_client
       databaseName: &database_name retryable-writes-tests
 

test/spec/client-backpressure/backpressure-retry-loop.yml.template

@@ -23,7 +23,7 @@ createEntities:
 
   -
     database:
-      id: &internal_db database
+      id: &internal_db internal_db
       client: *internal_client
       databaseName: &database_name retryable-writes-tests