test(NODE-3188): refactor transaction spec tests: (#2831)

Author: durran

src/cmap/connection.ts

@@ -368,6 +368,18 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
         clusterTime = session.clusterTime;
       }
 
+      // We need to unpin any read or write commands that happen outside of a pinned
+      // transaction, so we check if we have a pinned transaction that is no longer
+      // active, and unpin for all except start or commit.
+      if (
+        !session.transaction.isActive &&
+        session.transaction.isPinned &&
+        !finalCmd.startTransaction &&
+        !finalCmd.commitTransaction
+      ) {
+        session.transaction.unpinServer();
+      }
+
       const err = applySession(session, finalCmd, options as CommandOptions);
       if (err) {
         return callback(err);

src/cursor/abstract_cursor.ts

@@ -189,10 +189,15 @@ export abstract class AbstractCursor<
     return this[kOptions].readConcern;
   }
 
+  /** @internal */
   get session(): ClientSession | undefined {
     return this[kSession];
   }
 
+  set session(clientSession: ClientSession | undefined) {
+    this[kSession] = clientSession;
+  }
+
   /** @internal */
   get cursorOptions(): InternalAbstractCursorOptions {
     return this[kOptions];

src/sessions.ts

@@ -549,30 +549,30 @@ function endTransaction(session: ClientSession, commandName: string, callback: C
   }
 
   function commandHandler(e?: MongoError, r?: Document) {
-    if (commandName === 'commitTransaction') {
-      session.transaction.transition(TxnState.TRANSACTION_COMMITTED);
+    if (commandName !== 'commitTransaction') {
+      session.transaction.transition(TxnState.TRANSACTION_ABORTED);
+      // The spec indicates that we should ignore all errors on `abortTransaction`
+      return callback();
+    }
 
+    session.transaction.transition(TxnState.TRANSACTION_COMMITTED);
+    if (e) {
       if (
-        e &&
-        (e instanceof MongoNetworkError ||
-          e instanceof MongoWriteConcernError ||
-          isRetryableError(e) ||
-          isMaxTimeMSExpiredError(e))
+        e instanceof MongoNetworkError ||
+        e instanceof MongoWriteConcernError ||
+        isRetryableError(e) ||
+        isMaxTimeMSExpiredError(e)
       ) {
         if (isUnknownTransactionCommitResult(e)) {
           e.addErrorLabel('UnknownTransactionCommitResult');
 
           // per txns spec, must unpin session in this case
           session.transaction.unpinServer();
         }
+      } else if (e.hasErrorLabel('TransientTransactionError')) {
+        session.transaction.unpinServer();
       }
-    } else {
-      session.transaction.transition(TxnState.TRANSACTION_ABORTED);
-
-      // The spec indicates that we should ignore all errors on `abortTransaction`
-      return callback();
     }
-
     callback(e, r);
   }
 

src/transactions.ts

@@ -145,7 +145,11 @@ export class Transaction {
     const nextStates = stateMachine[this.state];
     if (nextStates && nextStates.includes(nextState)) {
       this.state = nextState;
-      if (this.state === TxnState.NO_TRANSACTION || this.state === TxnState.STARTING_TRANSACTION) {
+      if (
+        this.state === TxnState.NO_TRANSACTION ||
+        this.state === TxnState.STARTING_TRANSACTION ||
+        this.state === TxnState.TRANSACTION_ABORTED
+      ) {
         this.unpinServer();
       }
       return;

test/functional/transactions.test.js

@@ -1,10 +1,12 @@
 'use strict';
 
+const path = require('path');
 const { expect } = require('chai');
 const { Topology } = require('../../src/sdam/topology');
 const { ClientSession } = require('../../src/sessions');
 const { TestRunnerContext, generateTopologyTests } = require('./spec-runner');
 const { loadSpecTests } = require('../spec');
+const { runUnifiedTest } = require('./unified-spec-runner/runner');
 const { MongoNetworkError } = require('../../src/error');
 
 function ignoreNsNotFoundForListIndexes(err) {
@@ -79,14 +81,30 @@ class TransactionsRunnerContext extends TestRunnerContext {
   }
 }
 
+describe('Transactions Spec Unified Tests', function () {
+  for (const transactionTest of loadSpecTests(path.join('transactions', 'unified'))) {
+    expect(transactionTest).to.exist;
+    context(String(transactionTest.description), function () {
+      for (const test of transactionTest.tests) {
+        it(String(test.description), {
+          metadata: { sessions: { skipLeakTests: true } },
+          test: async function () {
+            await runUnifiedTest(this, transactionTest, test);
+          }
+        });
+      }
+    });
+  }
+});
+
 describe('Transactions', function () {
   const testContext = new TransactionsRunnerContext();
 
   [
-    { name: 'spec tests', specPath: 'transactions' },
+    { name: 'spec tests', specPath: path.join('transactions', 'legacy') },
     {
       name: 'withTransaction spec tests',
-      specPath: 'transactions/convenient-api'
+      specPath: path.join('transactions', 'convenient-api')
     }
   ].forEach(suiteSpec => {
     describe(suiteSpec.name, function () {

test/functional/unified-spec-runner/entities.ts

@@ -272,7 +272,6 @@ export class EntitiesMap<E = Entity> extends Map<string, E> {
             options.defaultTransactionOptions.maxCommitTimeMS = defaultOptions.maxCommitTimeMS;
           }
         }
-
         const session = client.startSession(options);
         map.set(entity.session.id, session);
       } else if ('bucket' in entity) {

test/functional/unified-spec-runner/operations.ts

@@ -1,6 +1,6 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 import { expect } from 'chai';
-import { Collection, Db, GridFSFile, MongoClient, ObjectId } from '../../../src';
+import { Collection, Db, GridFSFile, MongoClient, ObjectId, AbstractCursor } from '../../../src';
 import { ReadConcern } from '../../../src/read_concern';
 import { ReadPreference } from '../../../src/read_preference';
 import { WriteConcern } from '../../../src/write_concern';
@@ -11,6 +11,7 @@ import { expectErrorCheck, resultCheck } from './match';
 import type { OperationDescription } from './schema';
 import { CommandStartedEvent } from '../../../src/cmap/command_monitoring_events';
 import { translateOptions } from './unified-utils';
+import { getSymbolFrom } from '../../tools/utils';
 
 interface OperationFunctionParams {
   client: MongoClient;
@@ -21,6 +22,18 @@ interface OperationFunctionParams {
 type RunOperationFn = (p: OperationFunctionParams) => Promise<Document | boolean | number | void>;
 export const operations = new Map<string, RunOperationFn>();
 
+function executeWithPotentialSession(
+  entities: EntitiesMap,
+  operation: OperationDescription,
+  cursor: AbstractCursor
+) {
+  const session = entities.getEntity('session', operation.arguments.session, false);
+  if (session) {
+    cursor.session = session;
+  }
+  return cursor.toArray();
+}
+
 operations.set('abortTransaction', async ({ entities, operation }) => {
   const session = entities.getEntity('session', operation.object);
   return session.abortTransaction();
@@ -31,19 +44,18 @@ operations.set('aggregate', async ({ entities, operation }) => {
   if (!(dbOrCollection instanceof Db || dbOrCollection instanceof Collection)) {
     throw new Error(`Operation object '${operation.object}' must be a db or collection`);
   }
-  return dbOrCollection
-    .aggregate(operation.arguments.pipeline, {
-      allowDiskUse: operation.arguments.allowDiskUse,
-      batchSize: operation.arguments.batchSize,
-      bypassDocumentValidation: operation.arguments.bypassDocumentValidation,
-      maxTimeMS: operation.arguments.maxTimeMS,
-      maxAwaitTimeMS: operation.arguments.maxAwaitTimeMS,
-      collation: operation.arguments.collation,
-      hint: operation.arguments.hint,
-      let: operation.arguments.let,
-      out: operation.arguments.out
-    })
-    .toArray();
+  const cursor = dbOrCollection.aggregate(operation.arguments.pipeline, {
+    allowDiskUse: operation.arguments.allowDiskUse,
+    batchSize: operation.arguments.batchSize,
+    bypassDocumentValidation: operation.arguments.bypassDocumentValidation,
+    maxTimeMS: operation.arguments.maxTimeMS,
+    maxAwaitTimeMS: operation.arguments.maxAwaitTimeMS,
+    collation: operation.arguments.collation,
+    hint: operation.arguments.hint,
+    let: operation.arguments.let,
+    out: operation.arguments.out
+  });
+  return executeWithPotentialSession(entities, operation, cursor);
 });
 
 operations.set('assertCollectionExists', async ({ operation, client }) => {
@@ -232,7 +244,8 @@ operations.set('endSession', async ({ entities, operation }) => {
 operations.set('find', async ({ entities, operation }) => {
   const collection = entities.getEntity('collection', operation.object);
   const { filter, sort, batchSize, limit, let: vars } = operation.arguments;
-  return collection.find(filter, { sort, batchSize, limit, let: vars }).toArray();
+  const cursor = collection.find(filter, { sort, batchSize, limit, let: vars });
+  return executeWithPotentialSession(entities, operation, cursor);
 });
 
 operations.set('findOneAndReplace', async ({ entities, operation }) => {