Changelog.md

@@ -1,3 +1,10 @@
+# [4.24.0](https://github.com/luin/ioredis/compare/v4.23.1...v4.24.0) (2021-03-14)
+
+
+### Features
+
+* **cluster:** support retrying MOVED with a delay ([#1254](https://github.com/luin/ioredis/issues/1254)) ([8599981](https://github.com/luin/ioredis/commit/8599981141e8357f5ae2706fffb55010490bf002))
+
 ## [4.23.1](https://github.com/luin/ioredis/compare/v4.23.0...v4.23.1) (2021-03-14)
 
 

README.md

@@ -889,9 +889,12 @@ cluster.get("foo", (err, res) => {
       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).
+    - `retryDelayOnMoved`: By default, this value is `0` (in ms), which means when a `MOVED` error is received, the client will resend
+      the command instantly to the node returned together with the `MOVED` error. However, sometimes it takes time for a cluster to become
+      state stabilized after a failover, so adding a delay before resending can prevent a ping pong effect.
     - `redisOptions`: Default options passed to the constructor of `Redis` when connecting to a node.
-    - `slotsRefreshTimeout`: Milliseconds before a timeout occurs while refreshing slots from the cluster (default `1000`)
-    - `slotsRefreshInterval`: Milliseconds between every automatic slots refresh (default `5000`)
+    - `slotsRefreshTimeout`: Milliseconds before a timeout occurs while refreshing slots from the cluster (default `1000`).
+    - `slotsRefreshInterval`: Milliseconds between every automatic slots refresh (default `5000`).
 
 ### Read-write splitting
 

lib/cluster/ClusterOptions.ts

@@ -94,6 +94,17 @@ export interface IClusterOptions {
    */
   retryDelayOnTryAgain?: number;
 
+  /**
+   * By default, this value is 0, which means when a `MOVED` error is received,
+   * the client will resend the command instantly to the node returned together with
+   * the `MOVED` error. However, sometimes it takes time for a cluster to become
+   * state stabilized after a failover, so adding a delay before resending can
+   * prevent a ping pong effect.
+   *
+   * @default 0
+   */
+  retryDelayOnMoved?: number;
+
   /**
    * The milliseconds before a timeout occurs while refreshing
    * slots from the cluster.
@@ -184,6 +195,7 @@ export const DEFAULT_CLUSTER_OPTIONS: IClusterOptions = {
   enableReadyCheck: true,
   scaleReads: "master",
   maxRedirections: 16,
+  retryDelayOnMoved: 0,
   retryDelayOnFailover: 100,
   retryDelayOnClusterDown: 100,
   retryDelayOnTryAgain: 100,

lib/cluster/index.ts

@@ -381,7 +381,7 @@ class Cluster extends EventEmitter {
 
     const Promise = PromiseContainer.get();
     if (status === "wait") {
-      const ret = asCallback(Promise.resolve("OK"), callback);
+      const ret = asCallback(Promise.resolve<"OK">("OK"), callback);
 
       // use setImmediate to make sure "close" event
       // being emitted after quit() is returned
@@ -745,8 +745,19 @@ class Cluster extends EventEmitter {
       return;
     }
     const errv = error.message.split(" ");
-    if (errv[0] === "MOVED" || errv[0] === "ASK") {
-      handlers[errv[0] === "MOVED" ? "moved" : "ask"](errv[1], errv[2]);
+    if (errv[0] === "MOVED") {
+      const timeout = this.options.retryDelayOnMoved;
+      if (timeout && typeof timeout === "number") {
+        this.delayQueue.push(
+          "moved",
+          handlers.moved.bind(null, errv[1], errv[2]),
+          { timeout }
+        );
+      } else {
+        handlers.moved(errv[1], errv[2]);
+      }
+    } else if (errv[0] === "ASK") {
+      handlers.ask(errv[1], errv[2]);
     } else if (errv[0] === "TRYAGAIN") {
       this.delayQueue.push("tryagain", handlers.tryagain, {
         timeout: this.options.retryDelayOnTryAgain,

lib/command.ts

@@ -382,10 +382,7 @@ const msetArgumentTransformer = function (args) {
   return args;
 };
 
-Command.setArgumentTransformer("mset", msetArgumentTransformer);
-Command.setArgumentTransformer("msetnx", msetArgumentTransformer);
-
-Command.setArgumentTransformer("hmset", function (args) {
+const hsetArgumentTransformer = function (args) {
   if (args.length === 2) {
     if (typeof Map !== "undefined" && args[1] instanceof Map) {
       return [args[0]].concat(convertMapToArray(args[1]));
@@ -395,7 +392,13 @@ Command.setArgumentTransformer("hmset", function (args) {
     }
   }
   return args;
-});
+}
+
+Command.setArgumentTransformer("mset", msetArgumentTransformer);
+Command.setArgumentTransformer("msetnx", msetArgumentTransformer);
+
+Command.setArgumentTransformer("hset", hsetArgumentTransformer);
+Command.setArgumentTransformer("hmset", hsetArgumentTransformer);
 
 Command.setReplyTransformer("hgetall", function (result) {
   if (Array.isArray(result)) {
@@ -408,18 +411,6 @@ Command.setReplyTransformer("hgetall", function (result) {
   return result;
 });
 
-Command.setArgumentTransformer("hset", function (args) {
-  if (args.length === 2) {
-    if (typeof Map !== "undefined" && args[1] instanceof Map) {
-      return [args[0]].concat(convertMapToArray(args[1]));
-    }
-    if (typeof args[1] === "object" && args[1] !== null) {
-      return [args[0]].concat(convertObjectToArray(args[1]));
-    }
-  }
-  return args;
-});
-
 class MixedBuffers {
   length = 0;
   items = [];

lib/redis/index.ts

@@ -17,6 +17,7 @@ import {
   ReconnectOnError,
   DEFAULT_REDIS_OPTIONS,
 } from "./RedisOptions";
+import { NetStream } from "../types";
 
 const debug = Debug("redis");
 
@@ -317,7 +318,7 @@ Redis.prototype.connect = function (callback) {
     asCallback(
       this.connector.connect(function (type, err) {
         _this.silentEmit(type, err);
-      }),
+      }) as Promise<NetStream>,
       function (err, stream) {
         if (err) {
           _this.flushQueue(err);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ioredis",
-  "version": "4.23.1",
+  "version": "4.24.0",
   "description": "A robust, performance-focused and full-featured Redis client for Node.js.",
   "main": "built/index.js",
   "files": [
@@ -34,15 +34,15 @@
   },
   "dependencies": {
     "cluster-key-slot": "^1.1.0",
-    "debug": "^4.1.1",
+    "debug": "^4.3.1",
     "denque": "^1.1.0",
     "lodash.defaults": "^4.2.0",
     "lodash.flatten": "^4.4.0",
     "p-map": "^2.1.0",
     "redis-commands": "1.7.0",
     "redis-errors": "^1.2.0",
     "redis-parser": "^3.0.0",
-    "standard-as-callback": "^2.0.1"
+    "standard-as-callback": "^2.1.0"
   },
   "devDependencies": {
     "@semantic-release/changelog": "^5.0.1",

test/functional/cluster/moved.ts

@@ -2,6 +2,7 @@ import * as calculateSlot from "cluster-key-slot";
 import MockServer from "../../helpers/mock_server";
 import { expect } from "chai";
 import { Cluster } from "../../../lib";
+import * as sinon from "sinon";
 
 describe("cluster:MOVED", function () {
   it("should auto redirect the command to the correct nodes", function (done) {
@@ -117,4 +118,43 @@ describe("cluster:MOVED", function () {
       cluster.get("foo");
     });
   });
+
+  it("should supports retryDelayOnMoved", (done) => {
+    let cluster = undefined;
+    const slotTable = [[0, 16383, ["127.0.0.1", 30001]]];
+    new MockServer(30001, function (argv) {
+      if (argv[0] === "cluster" && argv[1] === "slots") {
+        return slotTable;
+      }
+      if (argv[0] === "get" && argv[1] === "foo") {
+        return new Error("MOVED " + calculateSlot("foo") + " 127.0.0.1:30002");
+      }
+    });
+
+    new MockServer(30002, function (argv) {
+      if (argv[0] === "cluster" && argv[1] === "slots") {
+        return slotTable;
+      }
+      if (argv[0] === "get" && argv[1] === "foo") {
+        cluster.disconnect();
+        done();
+      }
+    });
+
+    const retryDelayOnMoved = 789;
+    cluster = new Cluster([{ host: "127.0.0.1", port: "30001" }], {
+      retryDelayOnMoved,
+    });
+    cluster.on("ready", function () {
+      sinon.stub(global, "setTimeout").callsFake((body, ms) => {
+        if (ms === retryDelayOnMoved) {
+          process.nextTick(() => {
+            body();
+          });
+        }
+      });
+
+      cluster.get("foo");
+    });
+  });
 });