Merge pull request #264 from luin/try-again

feat(cluster): redirect on TRYAGAIN error
redis · Mar 13, 2016 · 4a119bf · 4a119bf
2 parents 1e6dd2e + 9b9267f
commit 4a119bf
Show file tree

Hide file tree

Showing 8 changed files with 138 additions and 100 deletions.
diff --git a/API.md b/API.md
@@ -58,8 +58,8 @@ Creates a Redis instance
 | [options.connectTimeout] | <code>number</code> | <code>10000</code> | The milliseconds before a timeout occurs during the initial connection to the Redis server. |
 | [options.autoResubscribe] | <code>boolean</code> | <code>true</code> | After reconnected, if the previous connection was in the subscriber mode, client will auto re-subscribe these channels. |
 | [options.autoResendUnfulfilledCommands] | <code>boolean</code> | <code>true</code> | If true, client will resend unfulfilled commands(e.g. block commands) in the previous connection when reconnected. |
-| [options.lazyConnect] | <code>boolean</code> | <code>false</code> | By default, When a new `Redis` instance is created, it will connect to Redis server automatically. If you want to keep disconnected util a command is called, you can pass the `lazyConnect` option to the constructor: |
-| [options.keyPrefix] | <code>string</code> | <code>&quot;&#x27;&#x27;&quot;</code> | The prefix to prepend to all keys in a command. ```javascript var redis = new Redis({ lazyConnect: true }); // No attempting to connect to the Redis server here. // Now let's connect to the Redis server redis.get('foo', function () { }); ``` |
+| [options.lazyConnect] | <code>boolean</code> | <code>false</code> | By default, When a new `Redis` instance is created, it will connect to Redis server automatically. If you want to keep disconnected util a command is called, you can pass the `lazyConnect` option to the constructor: ```javascript var redis = new Redis({ lazyConnect: true }); // No attempting to connect to the Redis server here. // Now let's connect to the Redis server redis.get('foo', function () { }); ``` |
+| [options.keyPrefix] | <code>string</code> | <code>&quot;&#x27;&#x27;&quot;</code> | The prefix to prepend to all keys in a command. |
 | [options.retryStrategy] | <code>function</code> |  | See "Quick Start" section |
 | [options.reconnectOnError] | <code>function</code> |  | See "Quick Start" section |
 | [options.readOnly] | <code>boolean</code> | <code>false</code> | Enable READONLY mode for the connection. Only available for cluster mode. |
@@ -221,6 +221,7 @@ Creates a Redis Cluster instance
 | [options.maxRedirections] | <code>number</code> | <code>16</code> | When a MOVED or ASK error is received, client will redirect the command to another node. This option limits the max redirections allowed to send a command. |
 | [options.retryDelayOnFailover] | <code>number</code> | <code>100</code> | When an error is received when sending a command(e.g. "Connection is closed." when the target Redis node is down), |
 | [options.retryDelayOnClusterDown] | <code>number</code> | <code>100</code> | When a CLUSTERDOWN error is received, client will retry if `retryDelayOnClusterDown` is valid delay time. |
+| [options.retryDelayOnTryAgain] | <code>number</code> | <code>100</code> | When a TRYAGAIN error is received, client will retry if `retryDelayOnTryAgain` is valid delay time. |
 | [options.redisOptions] | <code>Object</code> |  | Passed to the constructor of `Redis`. |
 
 <a name="Cluster+connect"></a>

diff --git a/README.md b/README.md
@@ -678,6 +678,8 @@ but a few so that if one is unreachable the client will try the next one, and th
     to insure that no command will fail during a failover.
     * `retryDelayOnClusterDown`: When a cluster is down, all commands will be rejected with the error of `CLUSTERDOWN`. If this option is a number (by default, it is `100`), the client
     will resend the commands after the specified time (in ms).
+    * `retryDelayOnTryAgain`: If this option is a number (by default, it is `100`), the client
+    will resend the commands rejected with `TRYAGAIN` error after the specified time (in ms).
     * `redisOptions`: Default options passed to the constructor of `Redis` when connecting to a node.
 
 ### Read-write splitting

diff --git a/lib/cluster/delay_queue.js b/lib/cluster/delay_queue.js
@@ -0,0 +1,48 @@
+'use strict';
+
+var Deque = require('double-ended-queue');
+var debug = require('debug')('ioredis:delayqueue');
+
+function DelayQueue() {
+  this.queues = {};
+  this.timeouts = {};
+}
+
+DelayQueue.prototype.push = function (bucket, item, options) {
+  var callback = options.callback || process.nextTick;
+  if (!this.queues[bucket]) {
+    this.queues[bucket] = new Deque();
+  }
+
+  var queue = this.queues[bucket];
+  queue.push(item);
+
+  if (!this.timeouts[bucket]) {
+    var _this = this;
+    this.timeouts[bucket] = setTimeout(function () {
+      callback(function () {
+        _this.timeouts[bucket] = null;
+        _this._execute(bucket);
+      });
+    }, options.timeout);
+  }
+};
+
+DelayQueue.prototype._execute = function (bucket) {
+  var queue = this.queues[bucket];
+  if (!queue) {
+    return;
+  }
+  var length = queue.length;
+  if (!length) {
+    return;
+  }
+  debug('send %d commands in %s queue', length, bucket);
+
+  this.queues[bucket] = null;
+  while (queue.length > 0) {
+    queue.shift()();
+  }
+};
+
+module.exports = DelayQueue;
diff --git a/lib/cluster/index.js b/lib/cluster/index.js
@@ -13,6 +13,7 @@ var Commander = require('../commander');
 var Command = require('../command');
 var commands = require('redis-commands');
 var ConnectionPool = require('./connection_pool');
+var DelayQueue = require('./delay_queue');
 
 /**
  * Creates a Redis Cluster instance
@@ -32,6 +33,8 @@ var ConnectionPool = require('./connection_pool');
  * "Connection is closed." when the target Redis node is down),
  * @param {number} [options.retryDelayOnClusterDown=100] - When a CLUSTERDOWN error is received, client will retry
  * if `retryDelayOnClusterDown` is valid delay time.
+ * @param {number} [options.retryDelayOnTryAgain=100] - When a TRYAGAIN error is received, client will retry
+ * if `retryDelayOnTryAgain` is valid delay time.
  * @param {Object} [options.redisOptions] - Passed to the constructor of `Redis`.
  * @extends [EventEmitter](http://nodejs.org/api/events.html#events_class_events_eventemitter)
  * @extends Commander
@@ -70,8 +73,7 @@ function Cluster(startupNodes, options) {
   this.retryAttempts = 0;
 
   this.resetOfflineQueue();
-  this.resetFailoverQueue();
-  this.resetClusterDownQueue();
+  this.delayQueue = new DelayQueue();
 
   this.subscriber = null;
 
@@ -93,7 +95,8 @@ Cluster.defaultOptions = {
   scaleReads: 'master',
   maxRedirections: 16,
   retryDelayOnFailover: 100,
-  retryDelayOnClusterDown: 100
+  retryDelayOnClusterDown: 100,
+  retryDelayOnTryAgain: 100
 };
 
 util.inherits(Cluster, EventEmitter);
@@ -103,14 +106,6 @@ Cluster.prototype.resetOfflineQueue = function () {
   this.offlineQueue = new Deque();
 };
 
-Cluster.prototype.resetFailoverQueue = function () {
-  this.failoverQueue = new Deque();
-};
-
-Cluster.prototype.resetClusterDownQueue = function () {
-  this.clusterDownQueue = new Deque();
-};
-
 /**
  * Connect to a cluster
  *
@@ -365,30 +360,6 @@ Cluster.prototype.executeOfflineCommands = function () {
   }
 };
 
-Cluster.prototype.executeFailoverCommands = function () {
-  if (this.failoverQueue.length) {
-    debug('send %d commands in failover queue', this.failoverQueue.length);
-    var failoverQueue = this.failoverQueue;
-    this.resetFailoverQueue();
-    while (failoverQueue.length > 0) {
-      var item = failoverQueue.shift();
-      item();
-    }
-  }
-};
-
-Cluster.prototype.executeClusterDownCommands = function () {
-  if (this.clusterDownQueue.length) {
-    debug('send %d commands in cluster down queue', this.clusterDownQueue.length);
-    var clusterDownQueue = this.clusterDownQueue;
-    this.resetClusterDownQueue();
-    while (clusterDownQueue.length > 0) {
-      var item = clusterDownQueue.shift();
-      item();
-    }
-  }
-};
-
 Cluster.prototype.sendCommand = function (command, stream, node) {
   if (this.status === 'end') {
     command.reject(new Error('Connection is closed.'));
@@ -427,6 +398,7 @@ Cluster.prototype.sendCommand = function (command, stream, node) {
           debug('command %s is required to ask %s:%s', command.name, key);
           tryConnection(false, key);
         },
+        tryagain: partialTry,
         clusterDown: partialTry,
         connectionClosed: partialTry,
         maxRedirections: function (redirectionError) {
@@ -511,7 +483,6 @@ Cluster.prototype.sendCommand = function (command, stream, node) {
 };
 
 Cluster.prototype.handleError = function (error, ttl, handlers) {
-  var _this = this;
   if (typeof ttl.value === 'undefined') {
     ttl.value = this.options.maxRedirections;
   } else {
@@ -524,26 +495,20 @@ Cluster.prototype.handleError = function (error, ttl, handlers) {
   var errv = error.message.split(' ');
   if (errv[0] === 'MOVED' || errv[0] === 'ASK') {
     handlers[errv[0] === 'MOVED' ? 'moved' : 'ask'](errv[1], errv[2]);
+  } else if (errv[0] === 'TRYAGAIN') {
+    this.delayQueue.push('tryagain', handlers.tryagain, {
+      timeout: this.options.retryDelayOnTryAgain
+    });
   } else if (errv[0] === 'CLUSTERDOWN' && this.options.retryDelayOnClusterDown > 0) {
-    this.clusterDownQueue.push(handlers.clusterDown);
-    if (!this.clusterDownTimeout) {
-      this.clusterDownTimeout = setTimeout(function () {
-        _this.refreshSlotsCache(function () {
-          _this.clusterDownTimeout = null;
-          _this.executeClusterDownCommands();
-        });
-      }, this.options.retryDelayOnClusterDown);
-    }
+    this.delayQueue.push('clusterdown', handlers.connectionClosed, {
+      timeout: this.options.retryDelayOnClusterDown,
+      callback: this.refreshSlotsCache.bind(this)
+    });
   } else if (error.message === 'Connection is closed.' && this.options.retryDelayOnFailover > 0) {
-    this.failoverQueue.push(handlers.connectionClosed);
-    if (!this.failoverTimeout) {
-      this.failoverTimeout = setTimeout(function () {
-        _this.refreshSlotsCache(function () {
-          _this.failoverTimeout = null;
-          _this.executeFailoverCommands();
-        });
-      }, this.options.retryDelayOnFailover);
-    }
+    this.delayQueue.push('failover', handlers.connectionClosed, {
+      timeout: this.options.retryDelayOnFailover,
+      callback: this.refreshSlotsCache.bind(this)
+    });
   } else {
     handlers.defaults();
   }

diff --git a/lib/pipeline.js b/lib/pipeline.js
@@ -102,6 +102,9 @@ Pipeline.prototype.fillResult = function (value, position) {
       if (typeof this.leftRedirections === 'undefined') {
         this.leftRedirections = {};
       }
+      var exec = function () {
+        _this.exec();
+      };
       this.redis.handleError(commonError, this.leftRedirections, {
         moved: function (slot, key) {
           _this.preferKey = key;
@@ -113,12 +116,9 @@ Pipeline.prototype.fillResult = function (value, position) {
           _this.preferKey = key;
           _this.exec();
         },
-        clusterDown: function () {
-          _this.exec();
-        },
-        connectionClosed: function () {
-          _this.exec();
-        },
+        tryagain: exec,
+        clusterDown: exec,
+        connectionClosed: exec,
         maxRedirections: function () {
           matched = false;
         },

diff --git a/lib/redis.js b/lib/redis.js
@@ -60,8 +60,7 @@ var ScanStream = require('./scan_stream');
  * When a new `Redis` instance is created, it will connect to Redis server automatically.
  * If you want to keep disconnected util a command is called, you can pass the `lazyConnect` option to
  * the constructor:
- * @param {string} [options.keyPrefix=''] - The prefix to prepend to all keys in a command.
-
+ *
  * ```javascript
  * var redis = new Redis({ lazyConnect: true });
  * // No attempting to connect to the Redis server here.
@@ -70,6 +69,7 @@ var ScanStream = require('./scan_stream');
  * redis.get('foo', function () {
  * });
  * ```
+ * @param {string} [options.keyPrefix=''] - The prefix to prepend to all keys in a command.
  * @param {function} [options.retryStrategy] - See "Quick Start" section
  * @param {function} [options.reconnectOnError] - See "Quick Start" section
  * @param {boolean} [options.readOnly=false] - Enable READONLY mode for the connection.

diff --git a/test/functional/cluster.js b/test/functional/cluster.js
@@ -331,7 +331,7 @@ describe('cluster', function () {
 
       var cluster = new Redis.Cluster([
         { host: '127.0.0.1', port: '30001' }
-      ], { lazyConnect: false });
+      ]);
       cluster.get('foo', function () {
         cluster.get('foo');
       });
@@ -361,7 +361,7 @@ describe('cluster', function () {
       });
       var cluster = new Redis.Cluster([
         { host: '127.0.0.1', port: '30001' }
-      ], { lazyConnect: false });
+      ], { retryDelayOnFailover: 1 });
       cluster.get('foo', function (err, res) {
         expect(res).to.eql('bar');
         cluster.disconnect();
@@ -453,6 +453,35 @@ describe('cluster', function () {
     });
   });
 
+  describe('TRYAGAIN', function () {
+    it('should retry the command', function (done) {
+      var times = 0;
+      var slotTable = [
+        [0, 16383, ['127.0.0.1', 30001]]
+      ];
+      var server = new MockServer(30001, function (argv) {
+        if (argv[0] === 'cluster' && argv[1] === 'slots') {
+          return slotTable;
+        }
+        if (argv[0] === 'get' && argv[1] === 'foo') {
+          if (times++ === 1) {
+            process.nextTick(function () {
+              cluster.disconnect();
+              disconnect([server], done);
+            });
+          } else {
+            return new Error('TRYAGAIN Multiple keys request during rehashing of slot');
+          }
+        }
+      });
+
+      var cluster = new Redis.Cluster([
+        { host: '127.0.0.1', port: '30001' }
+      ], { retryDelayOnTryAgain: 1 });
+      cluster.get('foo');
+    });
+  });
+
   describe('CLUSTERDOWN', function () {
     it('should redirect the command to a random node', function (done) {
       var slotTable = [
@@ -680,6 +709,33 @@ describe('cluster', function () {
       });
     });
 
+    it('should retry the command on TRYAGAIN', function (done) {
+      var times = 0;
+      var slotTable = [
+        [0, 16383, ['127.0.0.1', 30001]]
+      ];
+      var server = new MockServer(30001, function (argv) {
+        if (argv[0] === 'cluster' && argv[1] === 'slots') {
+          return slotTable;
+        }
+        if (argv[1] === 'foo') {
+          if (times++ < 2) {
+            return new Error('TRYAGAIN Multiple keys request during rehashing of slot');
+          }
+        }
+      });
+
+      var cluster = new Redis.Cluster([
+        { host: '127.0.0.1', port: '30001' }
+      ], { retryDelayOnTryAgain: 1 });
+      cluster.pipeline().get('foo').set('foo', 'bar').exec(function (err, result) {
+        expect(result[0][1]).to.eql('OK');
+        expect(result[1][1]).to.eql('OK');
+        cluster.disconnect();
+        disconnect([server], done);
+      });
+    });
+
     it('should not redirect commands on a non-readonly command is successful', function (done) {
       var slotTable = [
         [0, 12181, ['127.0.0.1', 30001]],

diff --git a/test/unit/cluster.js b/test/unit/cluster.js
@@ -21,38 +21,4 @@ describe('cluster', function () {
     expect(cluster.options).to.have.property('showFriendlyErrorStack', false);
     expect(cluster.options).to.have.property('scaleReads', 'master');
   });
-
-  describe('#executeFailoverCommands', function () {
-    it('should execute the commands', function (done) {
-      var cluster = {
-        resetFailoverQueue: function () {
-          this.failoverQueue = [];
-        },
-        failoverQueue: []
-      };
-
-      cluster.failoverQueue.push(function () {
-        expect(this.failoverQueue).to.have.length(0);
-        done();
-      }.bind(cluster));
-      Cluster.prototype.executeFailoverCommands.call(cluster);
-    });
-  });
-
-  describe('#executeClusterDownCommands', function () {
-    it('should execute the commands', function (done) {
-      var cluster = {
-        resetClusterDownQueue: function () {
-          this.clusterDownQueue = [];
-        },
-        clusterDownQueue: []
-      };
-
-      cluster.clusterDownQueue.push(function () {
-        expect(this.clusterDownQueue).to.have.length(0);
-        done();
-      }.bind(cluster));
-      Cluster.prototype.executeClusterDownCommands.call(cluster);
-    });
-  });
 });