correct timing issues around transactions

`endSession` must wait for all the machinery behind the scenes to
check out a connection and write a message before considering its
job finished.

Author: mbroadst

lib/core/sessions.js

@@ -13,6 +13,7 @@ const Transaction = require('./transactions').Transaction;
 const TxnState = require('./transactions').TxnState;
 const isPromiseLike = require('./utils').isPromiseLike;
 const ReadPreference = require('./topologies/read_preference');
+const maybePromise = require('../utils').maybePromise;
 const isTransactionCommand = require('./transactions').isTransactionCommand;
 const resolveClusterTime = require('./topologies/shared').resolveClusterTime;
 const isSharded = require('./wireprotocol/shared').isSharded;
@@ -125,25 +126,35 @@ class ClientSession extends EventEmitter {
     if (typeof options === 'function') (callback = options), (options = {});
     options = options || {};
 
-    if (this.hasEnded) {
-      if (typeof callback === 'function') callback(null, null);
-      return;
-    }
+    return maybePromise(this, callback, done => {
+      if (this.hasEnded) {
+        return done();
+      }
 
-    if (this.serverSession && this.inTransaction()) {
-      this.abortTransaction(); // pass in callback?
-    }
+      function completeEndSession() {
+        // release the server session back to the pool
+        this.sessionPool.release(this.serverSession);
+        this[kServerSession] = undefined;
 
-    // release the server session back to the pool
-    this.sessionPool.release(this.serverSession);
-    this[kServerSession] = undefined;
+        // mark the session as ended, and emit a signal
+        this.hasEnded = true;
+        this.emit('ended', this);
+
+        // spec indicates that we should ignore all errors for `endSessions`
+        done();
+      }
+
+      if (this.serverSession && this.inTransaction()) {
+        this.abortTransaction(err => {
+          if (err) return done(err);
+          completeEndSession();
+        });
 
-    // mark the session as ended, and emit a signal
-    this.hasEnded = true;
-    this.emit('ended', this);
+        return;
+      }
 
-    // spec indicates that we should ignore all errors for `endSessions`
-    if (typeof callback === 'function') callback(null, null);
+      completeEndSession();
+    });
   }
 
   /**
@@ -227,16 +238,7 @@ class ClientSession extends EventEmitter {
    * @return {Promise} A promise is returned if no callback is provided
    */
   commitTransaction(callback) {
-    if (typeof callback === 'function') {
-      endTransaction(this, 'commitTransaction', callback);
-      return;
-    }
-
-    return new Promise((resolve, reject) => {
-      endTransaction(this, 'commitTransaction', (err, reply) =>
-        err ? reject(err) : resolve(reply)
-      );
-    });
+    return maybePromise(this, callback, done => endTransaction(this, 'commitTransaction', done));
   }
 
   /**
@@ -246,16 +248,7 @@ class ClientSession extends EventEmitter {
    * @return {Promise} A promise is returned if no callback is provided
    */
   abortTransaction(callback) {
-    if (typeof callback === 'function') {
-      endTransaction(this, 'abortTransaction', callback);
-      return;
-    }
-
-    return new Promise((resolve, reject) => {
-      endTransaction(this, 'abortTransaction', (err, reply) =>
-        err ? reject(err) : resolve(reply)
-      );
-    });
+    return maybePromise(this, callback, done => endTransaction(this, 'abortTransaction', done));
   }
 
   /**

test/functional/spec-runner/index.js

@@ -7,6 +7,7 @@ const expect = chai.expect;
 const EJSON = require('mongodb-extjson');
 const TestRunnerContext = require('./context').TestRunnerContext;
 const resolveConnectionString = require('./utils').resolveConnectionString;
+const delay = require('../shared').delay;
 
 chai.use(require('chai-subset'));
 chai.use(require('./matcher').default);
@@ -347,11 +348,12 @@ function runTestSuiteTest(configuration, spec, context) {
         throw err;
       })
       .then(() => {
-        if (session0) session0.endSession();
-        if (session1) session1.endSession();
-
-        return validateExpectations(context.commandEvents, spec, savedSessionData);
-      });
+        const promises = [];
+        if (session0) promises.push(session0.endSession());
+        if (session1) promises.push(session1.endSession());
+        return Promise.all(promises);
+      })
+      .then(() => validateExpectations(context.commandEvents, spec, savedSessionData));
   });
 }