fix(sessions): transfer control of session management to session pool

Part of HELP-5834

.eslintrc

@@ -7,7 +7,8 @@
     "mocha": true
   },
   "globals": {
-    "Promise": true
+    "Promise": true,
+    "Set": true
   },
   "parserOptions": {
     "ecmaVersion": 2017

lib/mongo_client.js

@@ -327,14 +327,6 @@ MongoClient.prototype.close = function(force, callback) {
   // Remove listeners after emit
   self.removeAllListeners('close');
 
-  // If we have sessions, we want to send a single `endSessions` command for them,
-  // and then individually clean them up.  They will be removed from the internal state
-  // when they emit their `ended` events.
-  if (this.s.sessions.length) {
-    this.topology.endSessions(this.s.sessions);
-    this.s.sessions.forEach(session => session.endSession({ skipCommand: true }));
-  }
-
   // Callback after next event loop tick
   if (typeof callback === 'function')
     return process.nextTick(function() {
@@ -508,13 +500,7 @@ MongoClient.prototype.startSession = function(options) {
     throw new MongoError('Current topology does not support sessions');
   }
 
-  const session = this.topology.startSession(options);
-  session.once('ended', () => {
-    this.s.sessions = this.s.sessions.filter(s => s.equals(session));
-  });
-
-  this.s.sessions.push(session);
-  return session;
+  return this.topology.startSession(options);
 };
 
 var mergeOptions = function(target, source, flatten) {

lib/topologies/replset.js

@@ -371,22 +371,9 @@ class ReplSet extends TopologyBase {
   }
 
   close(forceClosed) {
-    var self = this;
-    // Call destroy on the topology
-    this.s.coreTopology.destroy({
-      force: typeof forceClosed === 'boolean' ? forceClosed : false
-    });
-
-    // We need to wash out all stored processes
-    if (forceClosed === true) {
-      this.s.storeOptions.force = forceClosed;
-      this.s.store.flush();
-    }
+    super.close(forceClosed);
 
-    var events = ['timeout', 'error', 'close', 'joined', 'left'];
-    events.forEach(function(e) {
-      self.removeAllListeners(e);
-    });
+    ['timeout', 'error', 'close', 'joined', 'left'].forEach(e => this.removeAllListeners(e));
   }
 }
 

lib/topologies/topology_base.js

@@ -4,9 +4,7 @@ const EventEmitter = require('events'),
   MongoError = require('mongodb-core').MongoError,
   f = require('util').format,
   os = require('os'),
-  translateReadPreference = require('../utils').translateReadPreference,
-  ClientSession = require('mongodb-core').Sessions.ClientSession;
-
+  translateReadPreference = require('../utils').translateReadPreference;
 // The store of ops
 var Store = function(topology, storeOptions) {
   var self = this;
@@ -290,7 +288,12 @@ class TopologyBase extends EventEmitter {
   }
 
   startSession(options) {
-    return new ClientSession(this, this.s.sessionPool, options);
+    if (this.s.sessionPool) {
+      return this.s.sessionPool.startSession(options);
+    }
+
+    // TODO: Should we have a better error message here?
+    throw new Error('Sessions are not supported');
   }
 
   endSessions(sessions, callback) {
@@ -388,6 +391,10 @@ class TopologyBase extends EventEmitter {
   }
 
   close(forceClosed) {
+    if (this.s.sessionPool) {
+      this.s.sessionPool.endAllSessions();
+    }
+
     this.s.coreTopology.destroy({
       force: typeof forceClosed === 'boolean' ? forceClosed : false
     });

package.json

@@ -32,6 +32,7 @@
     "mongodb-test-runner": "^1.1.18",
     "prettier": "^1.5.3",
     "semver": "5.4.1",
+    "sinon": "^4.3.0",
     "worker-farm": "^1.5.0"
   },
   "author": "Christian Kvalheim",

test/functional/apm_tests.js

@@ -1036,6 +1036,10 @@ describe('APM', function() {
         // Get the result
         result = results.successes.shift();
 
+        if (result.commandName === 'endSessions') {
+          result = results.successes.shift();
+        }
+
         // Validate the test
         expect(commandName).to.equal(result.commandName);
         // Do we have a getMore command
@@ -1054,6 +1058,10 @@ describe('APM', function() {
         results.failures = filterSessionsCommands(results.failures);
         result = results.failures.shift();
 
+        if (result.commandName === 'endSessions') {
+          result = results.failures.shift();
+        }
+
         // Validate the test
         expect(commandName).to.equal(result.commandName);
       }

test/functional/crud_api_tests.js

@@ -830,7 +830,7 @@ describe('CRUD API', function() {
         test.equal(null, err);
 
         // Delete all items with no selector
-        db.collection('t6_1').deleteMany(function(err) {
+        db.collection('t6_1').deleteMany({}, function(err) {
           test.equal(null, err);
 
           client.close();

test/functional/index_tests.js

@@ -271,6 +271,8 @@ describe('Indexes', function() {
   });
 
   /**
+   * TODO: This test is non-deterministic, and I don't think it is doing
+   * what it thinks it is doing. The try-catch statement does nothing.
    * @ignore
    */
   it('shouldThrowErrorOnAttemptingSafeCreateIndexWithNoCallback', {
@@ -302,6 +304,8 @@ describe('Indexes', function() {
   });
 
   /**
+   * TODO: This test is non-deterministic, and I don't think it is doing
+   * what it thinks it is doing. The try-catch statement does nothing.
    * @ignore
    */
   it('shouldThrowErrorOnAttemptingSafeEnsureIndexWithNoCallback', {

test/functional/session_leak_test.js

@@ -0,0 +1,51 @@
+'use strict';
+
+const expect = require('chai').expect;
+const sinon = require('sinon');
+const core = require('mongodb-core');
+const Server = core.Server;
+const ReplSet = core.ReplSet;
+const Mongos = core.Mongos;
+const ServerSessionPool = core.Sessions.ServerSessionPool;
+
+(() => {
+  const sandbox = sinon.createSandbox();
+
+  beforeEach('Session Leak Before Each - setup session tracking', function() {
+    sandbox.spy(Server.prototype, 'endSessions');
+    sandbox.spy(ReplSet.prototype, 'endSessions');
+    sandbox.spy(Mongos.prototype, 'endSessions');
+    sandbox.spy(ServerSessionPool.prototype, 'acquire');
+  });
+
+  afterEach('Session Leak After Each - ensure no leaks', function() {
+    const poolCalls = ServerSessionPool.prototype.acquire.getCalls();
+    const endCalls = Server.prototype.endSessions
+      .getCalls()
+      .concat(ReplSet.prototype.endSessions.getCalls())
+      .concat(Mongos.prototype.endSessions.getCalls());
+
+    const sessions = new Set();
+    poolCalls.forEach(call => sessions.add(call.returnValue.id));
+    // const totalSessionCount = set.size;
+
+    endCalls.forEach(call => {
+      const arg = call.args[0];
+      const ids = Array.isArray(arg) ? arg : [arg];
+
+      ids.forEach(id => sessions.delete(id));
+    });
+
+    const leakedSessionCount = sessions.size;
+    try {
+      expect(
+        leakedSessionCount,
+        `test is leaking ${leakedSessionCount} sessions, when it should be leaking 0`
+      ).to.equal(0);
+    } catch (e) {
+      this.test.error(e);
+    }
+  });
+
+  afterEach('Session Leak After Each - restore sandbox', () => sandbox.restore());
+})();