Add restoreNodeTimeout option to PoolCluster

closes #880
closes #906

Author: dougwilson

lib/PoolCluster.js

@@ -17,6 +17,7 @@ function PoolCluster(config) {
   this._canRetry = typeof config.canRetry === 'undefined' ? true : config.canRetry;
   this._defaultSelector = config.defaultSelector || 'RR';
   this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
+  this._restoreNodeTimeout = config.restoreNodeTimeout || 0;
 
   this._closed = false;
   this._findCaches = Object.create(null);
@@ -45,9 +46,10 @@ PoolCluster.prototype.add = function add(id, config) {
     : new PoolConfig(id);
 
   this._nodes[nodeId] = {
-    id: nodeId,
-    errorCount: 0,
-    pool: new Pool({config: poolConfig})
+    id            : nodeId,
+    errorCount    : 0,
+    pool          : new Pool({config: poolConfig}),
+    _offlineUntil : 0
   };
 
   this._clearFindCaches();
@@ -114,17 +116,13 @@ PoolCluster.prototype.of = function(pattern, selector) {
 };
 
 PoolCluster.prototype.remove = function remove(pattern) {
-  var foundNodeIds = this._findNodeIds(pattern);
+  var foundNodeIds = this._findNodeIds(pattern, true);
 
   for (var i = 0; i < foundNodeIds.length; i++) {
     var node = this._getNode(foundNodeIds[i]);
 
     if (node) {
-      delete this._nodes[node.id];
-
-      this._clearFindCaches();
-
-      node.pool.end(_noop);
+      this._removeNode(node);
     }
   }
 };
@@ -150,56 +148,86 @@ PoolCluster.prototype._clearFindCaches = function _clearFindCaches() {
   this._findCaches = Object.create(null);
 };
 
-PoolCluster.prototype._findNodeIds = function(pattern) {
-  if (this._findCaches[pattern] !== undefined) {
-    return this._findCaches[pattern];
+PoolCluster.prototype._decreaseErrorCount = function _decreaseErrorCount(node) {
+  var errorCount = node.errorCount;
+
+  if (errorCount > this._removeNodeErrorCount) {
+    errorCount = this._removeNodeErrorCount;
+  }
+
+  if (errorCount < 1) {
+    errorCount = 1;
   }
 
-  var foundNodeIds;
-  var nodeIds = Object.keys(this._nodes);
+  node.errorCount = errorCount - 1;
 
-  if (pattern === '*') {
-    // all
-    foundNodeIds = nodeIds;
-  } else if (nodeIds.indexOf(pattern) != -1) {
-    // one
-    foundNodeIds = [pattern];
-  } else if (pattern[pattern.length - 1] === '*') {
-    // wild-card matching
-    var keyword = pattern.substring(pattern.length - 1, 0);
+  if (node._offlineUntil) {
+    node._offlineUntil = 0;
+    this.emit('online', node.id);
+  }
+};
 
-    foundNodeIds = nodeIds.filter(function (id) {
-      return id.indexOf(keyword) === 0;
-    });
-  } else {
-    foundNodeIds = [];
+PoolCluster.prototype._findNodeIds = function _findNodeIds(pattern, includeOffline) {
+  var currentTime  = 0;
+  var foundNodeIds = this._findCaches[pattern];
+
+  if (foundNodeIds === undefined) {
+    var nodeIds  = Object.keys(this._nodes);
+    var wildcard = pattern.substr(-1) === '*';
+    var keyword  = wildcard
+      ? pattern.substr(0, pattern.length - 1)
+      : pattern;
+
+    if (wildcard) {
+      foundNodeIds = keyword.length !== 0
+        ? nodeIds.filter(function (id) { return id.substr(0, keyword.length) === keyword; })
+        : nodeIds;
+    } else {
+      var index = nodeIds.indexOf(keyword);
+      foundNodeIds = nodeIds.slice(index, index + 1);
+    }
+
+    this._findCaches[pattern] = foundNodeIds;
+  }
+
+  if (includeOffline) {
+    return foundNodeIds;
   }
 
-  this._findCaches[pattern] = foundNodeIds;
+  return foundNodeIds.filter(function (nodeId) {
+    var node = this._getNode(nodeId);
+
+    if (!node._offlineUntil) {
+      return true;
+    }
+
+    if (!currentTime) {
+      currentTime = getMonotonicMilliseconds();
+    }
 
-  return foundNodeIds;
+    return node._offlineUntil <= currentTime;
+  }, this);
 };
 
-PoolCluster.prototype._getNode = function(id) {
+PoolCluster.prototype._getNode = function _getNode(id) {
   return this._nodes[id] || null;
 };
 
 PoolCluster.prototype._increaseErrorCount = function _increaseErrorCount(node) {
-  if (++node.errorCount >= this._removeNodeErrorCount) {
-    delete this._nodes[node.id];
-
-    this._clearFindCaches();
+  var errorCount = ++node.errorCount;
 
-    node.pool.end(_noop);
-
-    this.emit('remove', node.id);
+  if (this._removeNodeErrorCount > errorCount) {
+    return;
   }
-};
 
-PoolCluster.prototype._decreaseErrorCount = function(node) {
-  if (node.errorCount > 0) {
-    --node.errorCount;
+  if (this._restoreNodeTimeout > 0) {
+    node._offlineUntil = getMonotonicMilliseconds() + this._restoreNodeTimeout;
+    this.emit('offline', node.id);
+    return;
   }
+
+  this._removeNode(node);
+  this.emit('remove', node.id);
 };
 
 PoolCluster.prototype._getConnection = function(node, cb) {
@@ -220,6 +248,27 @@ PoolCluster.prototype._getConnection = function(node, cb) {
   });
 };
 
+PoolCluster.prototype._removeNode = function _removeNode(node) {
+  delete this._nodes[node.id];
+
+  this._clearFindCaches();
+
+  node.pool.end(_noop);
+};
+
+function getMonotonicMilliseconds() {
+  var ms;
+
+  if (typeof process.hrtime === 'function') {
+    ms = process.hrtime();
+    ms = ms[0] * 1e3 + ms[1] * 1e-6;
+  } else {
+    ms = process.uptime() * 1000;
+  }
+
+  return Math.floor(ms);
+}
+
 function _cb(err) {
   if (err) {
     throw err;

lib/PoolNamespace.js

@@ -17,8 +17,16 @@ PoolNamespace.prototype.getConnection = function(cb) {
   var namespace   = this;
 
   if (clusterNode === null) {
-    var err = new Error('Pool does not exist.')
-    err.code = 'POOL_NOEXIST';
+    var err = null;
+
+    if (this._cluster._findNodeIds(this._pattern, true).length !== 0) {
+      err = new Error('Pool does not have online node.');
+      err.code = 'POOL_NONEONLINE';
+    } else {
+      err = new Error('Pool does not exist.');
+      err.code = 'POOL_NOEXIST';
+    }
+
     return cb(err);
   }
 

test/unit/pool-cluster/test-restore-events.js

@@ -0,0 +1,67 @@
+var assert  = require('assert');
+var common  = require('../../common');
+var cluster = common.createPoolCluster({
+  canRetry             : true,
+  removeNodeErrorCount : 2,
+  restoreNodeTimeout   : 100
+});
+var server  = common.createFakeServer();
+
+var connCount     = 0;
+var offline       = true;
+var offlineEvents = 0;
+var onlineEvents  = 0;
+var poolConfig    = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('MASTER', poolConfig);
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  cluster.on('offline', function (id) {
+    assert.equal(++offlineEvents, 1);
+    assert.equal(id, 'MASTER');
+    assert.equal(connCount, 2);
+
+    cluster.getConnection('MASTER', function (err) {
+      assert.ok(err);
+      assert.equal(err.code, 'POOL_NONEONLINE');
+
+      offline = false;
+    });
+
+    setTimeout(function () {
+      cluster.getConnection('MASTER', function (err, conn) {
+        assert.ifError(err);
+        conn.release();
+      });
+    }, 200);
+  });
+
+  cluster.on('online', function (id) {
+    assert.equal(++onlineEvents, 1);
+    assert.equal(id, 'MASTER');
+    assert.equal(connCount, 3);
+
+    cluster.end(function (err) {
+      assert.ifError(err);
+      server.destroy();
+    });
+  });
+
+  cluster.getConnection('MASTER', function (err) {
+    assert.ok(err);
+    assert.equal(err.code, 'PROTOCOL_CONNECTION_LOST');
+    assert.equal(err.fatal, true);
+    assert.equal(connCount, 2);
+  });
+});
+
+server.on('connection', function (conn) {
+  connCount += 1;
+
+  if (offline) {
+    conn.destroy();
+  } else {
+    conn.handshake();
+  }
+});

test/unit/pool-cluster/test-restore.js

@@ -0,0 +1,54 @@
+var assert  = require('assert');
+var common  = require('../../common');
+var cluster = common.createPoolCluster({
+  canRetry             : true,
+  removeNodeErrorCount : 2,
+  restoreNodeTimeout   : 100
+});
+var server  = common.createFakeServer();
+
+var connCount  = 0;
+var offline    = true;
+var poolConfig = common.getTestConfig({port: common.fakeServerPort});
+cluster.add('MASTER', poolConfig);
+
+server.listen(common.fakeServerPort, function (err) {
+  assert.ifError(err);
+
+  cluster.getConnection('MASTER', function (err) {
+    assert.ok(err);
+    assert.equal(err.code, 'PROTOCOL_CONNECTION_LOST');
+    assert.equal(err.fatal, true);
+    assert.equal(connCount, 2);
+
+    cluster.getConnection('MASTER', function (err) {
+      assert.ok(err);
+      assert.equal(err.code, 'POOL_NONEONLINE');
+
+      offline = false;
+    });
+
+    setTimeout(function () {
+      cluster.getConnection('MASTER', function (err, conn) {
+        assert.ifError(err);
+
+        conn.release();
+
+        cluster.end(function (err) {
+          assert.ifError(err);
+          server.destroy();
+        });
+      });
+    }, 200);
+  });
+});
+
+server.on('connection', function (conn) {
+  connCount += 1;
+
+  if (offline) {
+    conn.destroy();
+  } else {
+    conn.handshake();
+  }
+});