test(NODE-3188): backport transaction pinning tests (#2839)

Author: durran

lib/core/sessions.js

@@ -472,17 +472,20 @@ function endTransaction(session, commandName, callback) {
     if (commandName === 'commitTransaction') {
       session.transaction.transition(TxnState.TRANSACTION_COMMITTED);
 
-      if (
-        e &&
-        (e instanceof MongoNetworkError ||
+      if (e) {
+        if (
+          e instanceof MongoNetworkError ||
           e instanceof MongoWriteConcernError ||
           isRetryableError(e) ||
-          isMaxTimeMSExpiredError(e))
-      ) {
-        if (isUnknownTransactionCommitResult(e)) {
-          e.addErrorLabel('UnknownTransactionCommitResult');
+          isMaxTimeMSExpiredError(e)
+        ) {
+          if (isUnknownTransactionCommitResult(e)) {
+            e.addErrorLabel('UnknownTransactionCommitResult');
 
-          // per txns spec, must unpin session in this case
+            // per txns spec, must unpin session in this case
+            session.transaction.unpinServer();
+          }
+        } else if (e.hasErrorLabel('TransientTransactionError')) {
           session.transaction.unpinServer();
         }
       }

lib/core/transactions.js

@@ -150,7 +150,11 @@ class Transaction {
     const nextStates = stateMachine[this.state];
     if (nextStates && nextStates.indexOf(nextState) !== -1) {
       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;

lib/core/wireprotocol/command.js

@@ -60,6 +60,18 @@ function _command(server, ns, cmd, options, callback) {
       clusterTime = sessionClusterTime;
     }
 
+    // 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);
     if (err) {
       return callback(err);

lib/cursor.js

@@ -165,6 +165,10 @@ class Cursor extends CoreCursor {
     return this.cmd.sort;
   }
 
+  set session(clientSession) {
+    this.cursorState.session = clientSession;
+  }
+
   _initializeCursor(callback) {
     if (this.operation && this.operation.session != null) {
       this.cursorState.session = this.operation.session;

test/functional/transactions.test.js

@@ -1,11 +1,13 @@
 'use strict';
 
+const path = require('path');
 const chai = require('chai');
 const expect = chai.expect;
 const core = require('../../lib/core');
 const sessions = core.Sessions;
 const TestRunnerContext = require('./spec-runner').TestRunnerContext;
 const loadSpecTests = require('../spec').loadSpecTests;
+const runUnifiedTest = require('./unified-spec-runner/runner').runUnifiedTest;
 const generateTopologyTests = require('./spec-runner').generateTopologyTests;
 const MongoNetworkError = require('../../lib/core').MongoNetworkError;
 const semver = require('semver');
@@ -82,14 +84,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() {
+            return 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/operations.ts

@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/no-unused-vars */
 import { expect } from 'chai';
 import { Collection, Db, GridFSFile, MongoClient, ObjectId } from '../../../index';
+import { CoreCursor } from '../../../lib/core/cursor';
 import ReadConcern from '../../../lib/read_concern';
 import ReadPreference from '../../../lib/core/topologies/read_preference';
 import WriteConcern from '../../../lib/write_concern';
@@ -79,6 +80,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: CoreCursor
+) {
+  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();
@@ -89,18 +102,17 @@ 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,
-      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,
+    out: operation.arguments.out
+  });
+  return executeWithPotentialSession(entities, operation, cursor);
 });
 
 operations.set('assertCollectionExists', async ({ operation, client }) => {
@@ -288,7 +300,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 } = operation.arguments;
-  return collection.find(filter, { sort, batchSize, limit }).toArray();
+  const cursor = collection.find(filter, { sort, batchSize, limit });
+  return executeWithPotentialSession(entities, operation, cursor);
 });
 
 operations.set('findOneAndReplace', async ({ entities, operation }) => {

test/manual/atlas_connectivity.test.js

@@ -29,8 +29,15 @@ describe('Atlas Connectivity', function() {
     context(configName, function() {
       CONFIGS[configName].forEach(connectionString => {
         const name = connectionString.indexOf('mongodb+srv') >= 0 ? 'mongodb+srv' : 'normal';
-        it(`${name} (unified)`, makeConnectionTest(connectionString, { useUnifiedTopology: true }));
-        it(`${name} (legacy)`, makeConnectionTest(connectionString, { useUnifiedTopology: false }));
+        // TODO: Skipping until NODE-3308 is merged.
+        it.skip(
+          `${name} (unified)`,
+          makeConnectionTest(connectionString, { useUnifiedTopology: true })
+        );
+        it.skip(
+          `${name} (legacy)`,
+          makeConnectionTest(connectionString, { useUnifiedTopology: false })
+        );
       });
     });
   });