Merge remote-tracking branch 'upstream/master' into hotfix/fix_3091

# Conflicts:
#	test/esm/integration/pool-cluster/test-promise-wrapper.test.mjs

Author: jcmartineztiempo

lib/pool_cluster.js

@@ -13,17 +13,30 @@ const EventEmitter = require('events').EventEmitter;
 const makeSelector = {
   RR() {
     let index = 0;
-    return clusterIds => clusterIds[index++ % clusterIds.length];
+    return (clusterIds) => clusterIds[index++ % clusterIds.length];
   },
   RANDOM() {
-    return clusterIds =>
+    return (clusterIds) =>
       clusterIds[Math.floor(Math.random() * clusterIds.length)];
   },
   ORDER() {
-    return clusterIds => clusterIds[0];
+    return (clusterIds) => clusterIds[0];
   }
 };
 
+const getMonotonicMilliseconds = function () {
+  let 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);
+};
+
 class PoolNamespace {
   constructor(cluster, pattern, selector) {
     this._cluster = cluster;
@@ -34,15 +47,28 @@ class PoolNamespace {
   getConnection(cb) {
     const clusterNode = this._getClusterNode();
     if (clusterNode === null) {
-      return cb(new Error('Pool does Not exists.'));
+      let err = new Error('Pool does Not exist.');
+      err.code = 'POOL_NOEXIST';
+
+      if (this._cluster._findNodeIds(this._pattern, true).length !== 0) {
+        err = new Error('Pool does Not have online node.');
+        err.code = 'POOL_NONEONLINE';
+      }
+
+      return cb(err);
     }
     return this._cluster._getConnection(clusterNode, (err, connection) => {
       if (err) {
+        if (
+          this._cluster._canRetry &&
+          this._cluster._findNodeIds(this._pattern).length !== 0
+        ) {
+          this._cluster.emit('warn', err);
+          return this.getConnection(cb);
+        }
+
         return cb(err);
       }
-      if (connection === 'retry') {
-        return this.getConnection(cb);
-      }
       return cb(null, connection);
     });
   }
@@ -123,6 +149,7 @@ class PoolCluster extends EventEmitter {
     this._canRetry =
       typeof config.canRetry === 'undefined' ? true : config.canRetry;
     this._removeNodeErrorCount = config.removeNodeErrorCount || 5;
+    this._restoreNodeTimeout = config.restoreNodeTimeout || 0;
     this._defaultSelector = config.defaultSelector || 'RR';
     this._closed = false;
     this._lastId = 0;
@@ -155,13 +182,26 @@ class PoolCluster extends EventEmitter {
       this._nodes[id] = {
         id: id,
         errorCount: 0,
-        pool: new Pool({ config: new PoolConfig(config) })
+        pool: new Pool({ config: new PoolConfig(config) }),
+        _offlineUntil: 0
       };
       this._serviceableNodeIds.push(id);
       this._clearFindCaches();
     }
   }
 
+  remove(pattern) {
+    const foundNodeIds = this._findNodeIds(pattern, true);
+
+    for (let i = 0; i < foundNodeIds.length; i++) {
+      const node = this._getNode(foundNodeIds[i]);
+
+      if (node) {
+        this._removeNode(node);
+      }
+    }
+  }
+
   getConnection(pattern, selector, cb) {
     let namespace;
     if (typeof pattern === 'function') {
@@ -181,7 +221,7 @@ class PoolCluster extends EventEmitter {
     const cb =
       callback !== undefined
         ? callback
-        : err => {
+        : (err) => {
           if (err) {
             throw err;
           }
@@ -190,11 +230,12 @@ class PoolCluster extends EventEmitter {
       process.nextTick(cb);
       return;
     }
+
     this._closed = true;
 
     let calledBack = false;
     let waitingClose = 0;
-    const onEnd = err => {
+    const onEnd = (err) => {
       if (!calledBack && (err || --waitingClose <= 0)) {
         calledBack = true;
         return cb(err);
@@ -205,67 +246,98 @@ class PoolCluster extends EventEmitter {
       waitingClose++;
       this._nodes[id].pool.end(onEnd);
     }
+
     if (waitingClose === 0) {
       process.nextTick(onEnd);
     }
   }
 
-  _findNodeIds(pattern) {
-    if (typeof this._findCaches[pattern] !== 'undefined') {
-      return this._findCaches[pattern];
-    }
-    let foundNodeIds;
-    if (pattern === '*') {
-      // all
-      foundNodeIds = this._serviceableNodeIds;
-    } else if (this._serviceableNodeIds.indexOf(pattern) !== -1) {
-      // one
-      foundNodeIds = [pattern];
-    } else {
-      // wild matching
-      const keyword = pattern.substring(pattern.length - 1, 0);
-      foundNodeIds = this._serviceableNodeIds.filter(id =>
-        id.startsWith(keyword)
-      );
+  _findNodeIds(pattern, includeOffline) {
+    let currentTime = 0;
+    let foundNodeIds = this._findCaches[pattern];
+
+    if (typeof this._findCaches[pattern] === 'undefined') {
+      if (pattern === '*') {
+        // all
+        foundNodeIds = this._serviceableNodeIds;
+      } else if (this._serviceableNodeIds.indexOf(pattern) !== -1) {
+        // one
+        foundNodeIds = [pattern];
+      } else {
+        // wild matching
+        const keyword = pattern.substring(pattern.length - 1, 0);
+        foundNodeIds = this._serviceableNodeIds.filter((id) =>
+          id.startsWith(keyword)
+        );
+      }
     }
+
     this._findCaches[pattern] = foundNodeIds;
-    return foundNodeIds;
+
+    if (includeOffline) {
+      return foundNodeIds;
+    }
+
+    return foundNodeIds.filter((nodeId) => {
+      const node = this._getNode(nodeId);
+
+      if (!node._offlineUntil) {
+        return true;
+      }
+
+      if (!currentTime) {
+        currentTime = getMonotonicMilliseconds();
+      }
+
+      return node._offlineUntil <= currentTime;
+    });
   }
 
   _getNode(id) {
     return this._nodes[id] || null;
   }
 
   _increaseErrorCount(node) {
-    if (++node.errorCount >= this._removeNodeErrorCount) {
-      const index = this._serviceableNodeIds.indexOf(node.id);
-      if (index !== -1) {
-        this._serviceableNodeIds.splice(index, 1);
-        delete this._nodes[node.id];
-        this._clearFindCaches();
-        node.pool.end();
-        this.emit('remove', node.id);
-      }
+    const errorCount = ++node.errorCount;
+
+    if (this._removeNodeErrorCount > errorCount) {
+      return;
+    }
+
+    if (this._restoreNodeTimeout > 0) {
+      node._offlineUntil =
+        getMonotonicMilliseconds() + this._restoreNodeTimeout;
+      this.emit('offline', node.id);
+      return;
     }
+
+    this._removeNode(node);
+    this.emit('remove', node.id);
   }
 
   _decreaseErrorCount(node) {
-    if (node.errorCount > 0) {
-      --node.errorCount;
+    let errorCount = node.errorCount;
+
+    if (errorCount > this._removeNodeErrorCount) {
+      errorCount = this._removeNodeErrorCount;
+    }
+
+    if (errorCount < 1) {
+      errorCount = 1;
+    }
+
+    node.errorCount = errorCount - 1;
+
+    if (node._offlineUntil) {
+      node._offlineUntil = 0;
+      this.emit('online', node.id);
     }
   }
 
   _getConnection(node, cb) {
     node.pool.getConnection((err, connection) => {
       if (err) {
         this._increaseErrorCount(node);
-        if (this._canRetry) {
-          // REVIEW: this seems wrong?
-          this.emit('warn', err);
-          // eslint-disable-next-line no-console
-          console.warn(`[Error] PoolCluster : ${err}`);
-          return cb(null, 'retry');
-        }
         return cb(err);
       }
       this._decreaseErrorCount(node);
@@ -275,6 +347,16 @@ class PoolCluster extends EventEmitter {
     });
   }
 
+  _removeNode(node) {
+    const index = this._serviceableNodeIds.indexOf(node.id);
+    if (index !== -1) {
+      this._serviceableNodeIds.splice(index, 1);
+      delete this._nodes[node.id];
+      this._clearFindCaches();
+      node.pool.end();
+    }
+  }
+
   _clearFindCaches() {
     this._findCaches = {};
   }

package-lock.json

@@ -109,7 +109,7 @@ class PromisePoolCluster extends EventEmitter {
     super();
     this.poolCluster = poolCluster;
     this.Promise = thePromise || Promise;
-    inheritEvents(poolCluster, this, ['warn', 'remove']);
+    inheritEvents(poolCluster, this, ['warn', 'remove' , 'online', 'offline']);
   }
 
   getConnection(pattern, selector) {
@@ -205,7 +205,7 @@ class PromisePoolCluster extends EventEmitter {
       })(func);
     }
   }
-})(['add']);
+})(['add', 'remove']);
 
 function createPromisePoolCluster(opts) {
   const corePoolCluster = createPoolCluster(opts);