fix(mongos): fix connection leak when mongos reconnects

lib/topologies/mongos.js

@@ -568,13 +568,13 @@ function reconnectProxies(self, proxies, callback) {
           _self.on('parseError', handleEvent(self, 'parseError'));
 
           // Move to the connected servers
-          moveServerFrom(self.disconnectedProxies, self.connectedProxies, _self);
+          moveServerFrom(self.connectingProxies, self.connectedProxies, _self);
           // Emit topology Change
           emitTopologyDescriptionChanged(self);
           // Emit joined event
           self.emit('joined', 'mongos', _self);
         });
-      } else if (event === 'connect' && self.authenticating) {
+      } else {
         // Move from connectingProxies
         moveServerFrom(self.connectingProxies, self.disconnectedProxies, _self);
         this.destroy();
@@ -613,6 +613,9 @@ function reconnectProxies(self, proxies, callback) {
         })
       );
 
+      _server.destroy();
+      removeProxyFrom(self.disconnectedProxies, _server);
+
       // Relay the server description change
       server.on('serverDescriptionChanged', function(event) {
         self.emit('serverDescriptionChanged', event);
@@ -635,6 +638,7 @@ function reconnectProxies(self, proxies, callback) {
       relayEvents(server, self, ['commandStarted', 'commandSucceeded', 'commandFailed']);
 
       // Connect to proxy
+      self.connectingProxies.push(server);
       server.connect(self.s.connectOptions);
     }, i);
   }

lib/topologies/server.js

@@ -1004,6 +1004,8 @@ var listeners = ['close', 'error', 'timeout', 'parseError', 'connect'];
  * @param {boolean} [options.force=false] Force destroy the pool
  */
 Server.prototype.destroy = function(options) {
+  if (this._destroyed) return;
+
   options = options || {};
   var self = this;
 
@@ -1016,7 +1018,10 @@ Server.prototype.destroy = function(options) {
   }
 
   // No pool, return
-  if (!self.s.pool) return;
+  if (!self.s.pool) {
+    this._destroyed = true;
+    return;
+  }
 
   // Emit close event
   if (options.emitClose) {
@@ -1051,6 +1056,7 @@ Server.prototype.destroy = function(options) {
 
   // Destroy the pool
   this.s.pool.destroy(options.force);
+  this._destroyed = true;
 };
 
 /**

test/tests/unit/mongos/reconnect_tests.js

@@ -0,0 +1,106 @@
+'use strict';
+
+const Mongos = require('../../../../lib/topologies/mongos');
+const expect = require('chai').expect;
+const mock = require('mongodb-mock-server');
+const genClusterTime = require('../common').genClusterTime;
+
+const Connection = require('../../../../lib/connection/connection');
+
+describe('Reconnect (Mongos)', function() {
+  const fixture = {};
+
+  function startServer() {
+    return mock.createServer(fixture.port).then(mockServer => {
+      mockServer.setMessageHandler(request => {
+        request.reply(
+          Object.assign({}, mock.DEFAULT_ISMASTER, {
+            $clusterTime: genClusterTime(Date.now()),
+            msg: 'isdbgrid'
+          })
+        );
+      });
+      fixture.server = mockServer;
+      fixture.port = mockServer.port;
+    });
+  }
+
+  function stopServer() {
+    return mock.cleanup();
+  }
+
+  beforeEach(() => startServer());
+  beforeEach(() => Connection.enableConnectionAccounting());
+  afterEach(() => Connection.disableConnectionAccounting());
+  afterEach(() => stopServer());
+
+  it('should not connection swarm when reconnecting', function(done) {
+    const reconnectInterval = 500;
+    const socketTimeout = reconnectInterval * 5;
+    const haInterval = reconnectInterval * 10;
+    const reconnectTries = Number.MAX_VALUE;
+
+    const connectOptions = {
+      haInterval,
+      reconnectInterval,
+      socketTimeout,
+      reconnectTries,
+      reconnect: true,
+      poolSize: 500
+    };
+
+    const mongos = new Mongos([fixture.server.address()], connectOptions);
+
+    function runIsMaster(assertion) {
+      return new Promise((resolve, reject) => {
+        mongos.command('admin.$cmd', { ismaster: 1 }, {}, (err, response) => {
+          try {
+            assertion(err, response);
+            resolve();
+          } catch (e) {
+            reject(e);
+          }
+        });
+      });
+    }
+
+    function connectMongos() {
+      return new Promise((resolve, reject) => {
+        mongos.once('error', reject);
+        mongos.once('connect', resolve);
+        mongos.connect(connectOptions);
+      });
+    }
+
+    function assertSuccess(err, response) {
+      expect(err).to.not.exist;
+      expect(response).to.exist;
+    }
+
+    function assertError(err, response) {
+      expect(err).to.exist;
+      expect(response).to.not.exist;
+    }
+
+    function delay(ms) {
+      return new Promise(resolve => setTimeout(resolve, ms));
+    }
+
+    function cleanup(err) {
+      mongos.destroy();
+      return err;
+    }
+
+    Promise.resolve()
+      .then(() => connectMongos())
+      .then(() => runIsMaster(assertSuccess))
+      .then(() => stopServer())
+      .then(() => runIsMaster(assertError))
+      .then(() => delay(haInterval * 2))
+      .then(() => startServer())
+      .then(() => delay(haInterval * 2))
+      .then(() => expect(Object.keys(Connection.connections())).to.have.a.lengthOf(1))
+      .then(() => cleanup(), cleanup)
+      .then(done);
+  });
+});