From 6f46080ecaf12ea9330af14f7f39358a88e87e08 Mon Sep 17 00:00:00 2001
From: Leo Ponomarev <levpono@amazon.com>
Date: Tue, 17 May 2022 15:32:43 -0400
Subject: [PATCH] cluster checks are now unconditional and have exponential
 backoff. previously they ran at a fixed interval only when there were no
 commands that could generate a MOVE/ASK error. That wasn't good enough for
 detecting newly added slaves

---
 src/nchan_commands.rb           |  53 +++++++++---
 src/nchan_config_commands.c     |  65 ++++++++++++---
 src/nchan_defs.h                |   5 ++
 src/nchan_setup.c               |  25 ++++--
 src/nchan_types.h               |   2 +-
 src/store/redis/redis_nodeset.c | 143 ++++++++++++++++++--------------
 src/store/redis/redis_nodeset.h |   4 +-
 src/store/redis/store.h         |   1 -
 8 files changed, 205 insertions(+), 93 deletions(-)

diff --git a/src/nchan_commands.rb b/src/nchan_commands.rb
index a6d6c05d9..fa9ebecec 100644
--- a/src/nchan_commands.rb
+++ b/src/nchan_commands.rb
@@ -593,15 +593,6 @@
       default: "4m",
       info: "Send a keepalive command to redis to keep the Nchan redis clients from disconnecting. Set to 0 to disable."
   
-  nchan_redis_cluster_check_interval [:main, :srv, :upstream, :loc],
-      :ngx_conf_set_sec_slot,
-      [:loc_conf, :"redis.cluster_check_interval"],
-      
-      group: "storage",
-      tags: ['redis'],
-      default: "5s",
-      info: "Send a CLUSTER INFO command to each connected Redis node to see if the cluster config epoch has changed. Sent only when in Cluster mode and if any other command that may result in a MOVE error has not been sent in the configured time."
-  
     nchan_redis_load_scripts_unconditionally [:upstream],
       :ngx_conf_set_flag_slot,
       [:srv_conf, "redis.load_scripts_unconditionally"],
@@ -677,7 +668,7 @@
       default: "0.5 (increase delay by 50% each try)",
       info: "Add an exponentially increasing delay to Redis connection retries. `Delay[n] = (Delay[n-1] + jitter) * (nchan_redis_reconnect_delay_backoff + 1)`."
       
-    nchan_redis_reconnect_delay_max [:upstream],
+  nchan_redis_reconnect_delay_max [:upstream],
       :ngx_conf_set_msec_slot,
       [:srv_conf, :"redis.reconnect_delay.max"],
   
@@ -687,6 +678,48 @@
       value: "<time> (0 to disable)",
       info: "Maximum Redis reconnection delay after backoff and jitter."
 
+  nchan_redis_cluster_check_interval_min [:upstream],
+      :ngx_conf_set_msec_slot,
+      [:srv_conf, :"redis.cluster_check_interval.min"],
+  
+      alt: :nchan_redis_cluster_check_interval,
+      group: "storage",
+      tags: ['redis'],
+      value: "<time>",
+      default: "1s (0 to disable)",
+      info: "When connected to a cluster, periodically check the cluster state and layout via a random master node."
+  
+  nchan_redis_cluster_check_interval_jitter [:upstream],
+      :ngx_conf_set_jitter,
+      [:srv_conf, :"redis.cluster_check_interval.jitter_multiplier"],
+  
+      group: "storage",
+      tags: ['redis'],
+      value: "<floating point> >= 0, (0 to disable)",
+      default: "0.2 (20% of inverval value)",
+      info: "Introduce random jitter to Redis cluster chck interval, where the range is `±(cluster_check_interval * nchan_redis_cluster_check_interval_jitter) / 2`."
+  
+  nchan_redis_cluster_check_interval_backoff [:upstream],
+      :ngx_conf_set_exponential_backoff,
+      [:srv_conf, :"redis.cluster_check_interval.backoff_multiplier"],
+  
+      group: "storage",
+      tags: ['redis'],
+      value: "<floating point> >= 0, ratio of current delay",
+      default: "2 (increase delay by 200% each try)",
+      info: "Add an exponentially increasing delay to the Redis cluster check interval. `Delay[n] = (Delay[n-1] + jitter) * (nchan_redis_cluster_check_interval_backoff + 1)`."
+      
+  nchan_redis_cluster_check_interval_max [:upstream],
+      :ngx_conf_set_msec_slot,
+      [:srv_conf, :"redis.cluster_check_interval.max"],
+  
+      group: "storage",
+      tags: ['redis'],
+      value: "<time> (0 to disable)",
+      default: "30s",
+      info: "Maximum Redis cluster check interval after backoff and jitter."
+    
+
   nchan_redis_cluster_recovery_delay [:upstream],
       :ngx_conf_set_msec_slot,
       [:srv_conf, :"redis.cluster_recovery_delay.min"],
diff --git a/src/nchan_config_commands.c b/src/nchan_config_commands.c
index 504e5e889..db2c8e7b3 100644
--- a/src/nchan_config_commands.c
+++ b/src/nchan_config_commands.c
@@ -755,19 +755,6 @@ static ngx_command_t  nchan_commands[] = {
     offsetof(nchan_loc_conf_t, redis.ping_interval),
     NULL } ,
 
-  { ngx_string("nchan_redis_cluster_check_interval"),
-    NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_UPS_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1,
-    ngx_conf_set_sec_slot,
-    NGX_HTTP_LOC_CONF_OFFSET,
-    offsetof(nchan_loc_conf_t, redis.cluster_check_interval),
-    NULL } ,
-  { ngx_string("nchan_redis_cluster_check_interval"), //alt for nchan_redis_cluster_check_interval
-    NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_HTTP_UPS_CONF|NGX_HTTP_LOC_CONF|NGX_CONF_TAKE1,
-    ngx_conf_set_sec_slot,
-    NGX_HTTP_LOC_CONF_OFFSET,
-    offsetof(nchan_loc_conf_t, redis.cluster_check_interval),
-    NULL } ,
-
   { ngx_string("nchan_redis_load_scripts_unconditionally"),
     NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
     ngx_conf_set_flag_slot,
@@ -885,6 +872,58 @@ static ngx_command_t  nchan_commands[] = {
     offsetof(nchan_srv_conf_t, redis.reconnect_delay.max),
     NULL } ,
 
+  { ngx_string("nchan_redis_cluster_check_interval_min"),
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_msec_slot,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.min),
+    NULL } ,
+  { ngx_string("nchan_redis_cluster_check_interval"), //alt for nchan_redis_cluster_check_interval_min
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_msec_slot,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.min),
+    NULL } ,
+
+  { ngx_string("nchan_redis_cluster_check_interval_jitter"),
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_jitter,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.jitter_multiplier),
+    NULL } ,
+  { ngx_string("nchan_redis_cluster_check_interval_jitter"), //alt for nchan_redis_cluster_check_interval_jitter
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_jitter,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.jitter_multiplier),
+    NULL } ,
+
+  { ngx_string("nchan_redis_cluster_check_interval_backoff"),
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_exponential_backoff,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.backoff_multiplier),
+    NULL } ,
+  { ngx_string("nchan_redis_cluster_check_interval_backoff"), //alt for nchan_redis_cluster_check_interval_backoff
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_exponential_backoff,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.backoff_multiplier),
+    NULL } ,
+
+  { ngx_string("nchan_redis_cluster_check_interval_max"),
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_msec_slot,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.max),
+    NULL } ,
+  { ngx_string("nchan_redis_cluster_check_interval_max"), //alt for nchan_redis_cluster_check_interval_max
+    NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
+    ngx_conf_set_msec_slot,
+    NGX_HTTP_SRV_CONF_OFFSET,
+    offsetof(nchan_srv_conf_t, redis.cluster_check_interval.max),
+    NULL } ,
+
   { ngx_string("nchan_redis_cluster_recovery_delay"),
     NGX_HTTP_UPS_CONF|NGX_CONF_TAKE1,
     ngx_conf_set_msec_slot,
diff --git a/src/nchan_defs.h b/src/nchan_defs.h
index 68e3cc32a..a4cc104bc 100644
--- a/src/nchan_defs.h
+++ b/src/nchan_defs.h
@@ -9,6 +9,11 @@
 #define NCHAN_DEFAULT_REDIS_CLUSTER_CONNECT_TIMEOUT_MSEC 15000
 #define NCHAN_DEFAULT_REDIS_CLUSTER_MAX_FAILING_TIME_MSEC 30000
 
+#define NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MIN_MSEC 1000
+#define NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_BACKOFF_MULTIPLIER 2
+#define NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_JITTER_MULTIPLIER 0.2
+#define NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MAX_MSEC 30000
+
 #define NCHAN_DEFAULT_REDIS_RECONNECT_DELAY_MIN_MSEC 500
 #define NCHAN_DEFAULT_REDIS_RECONNECT_DELAY_BACKOFF_MULTIPLIER 0.5
 #define NCHAN_DEFAULT_REDIS_RECONNECT_DELAY_JITTER_MULTIPLIER 0.1
diff --git a/src/nchan_setup.c b/src/nchan_setup.c
index 0c8a5d42e..501d35460 100644
--- a/src/nchan_setup.c
+++ b/src/nchan_setup.c
@@ -146,6 +146,10 @@ static void *nchan_create_srv_conf(ngx_conf_t *cf) {
   scf->redis.cluster_recovery_delay.jitter_multiplier = NGX_CONF_UNSET;
   scf->redis.cluster_recovery_delay.backoff_multiplier = NGX_CONF_UNSET;
   scf->redis.cluster_recovery_delay.max = NGX_CONF_UNSET_MSEC;
+  scf->redis.cluster_check_interval.min = NGX_CONF_UNSET_MSEC;
+  scf->redis.cluster_check_interval.jitter_multiplier = NGX_CONF_UNSET;
+  scf->redis.cluster_check_interval.backoff_multiplier = NGX_CONF_UNSET;
+  scf->redis.cluster_check_interval.max = NGX_CONF_UNSET_MSEC;
   scf->redis.cluster_max_failing_msec = NGX_CONF_UNSET_MSEC;
   scf->redis.load_scripts_unconditionally = NGX_CONF_UNSET;
   scf->redis.optimize_target = NCHAN_REDIS_OPTIMIZE_UNSET;
@@ -197,6 +201,22 @@ static char *nchan_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child) {
   }
   ngx_conf_merge_msec_value(conf->redis.cluster_recovery_delay.max, prev->redis.cluster_recovery_delay.max, NCHAN_DEFAULT_REDIS_CLUSTER_RECOVERY_DELAY_MAX_MSEC);
   
+  
+  ngx_conf_merge_msec_value(conf->redis.cluster_check_interval.min, prev->redis.cluster_check_interval.min, NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MIN_MSEC);
+  if(prev->redis.cluster_check_interval.jitter_multiplier == NGX_CONF_UNSET) {
+    conf->redis.cluster_check_interval.jitter_multiplier = NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_JITTER_MULTIPLIER;
+  }
+  else {
+    conf->redis.cluster_check_interval.jitter_multiplier = prev->redis.cluster_check_interval.jitter_multiplier;
+  }
+  if(prev->redis.cluster_check_interval.backoff_multiplier == NGX_CONF_UNSET) {
+    conf->redis.cluster_check_interval.backoff_multiplier = NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_BACKOFF_MULTIPLIER;
+  }
+  else {
+    conf->redis.cluster_check_interval.backoff_multiplier = prev->redis.cluster_check_interval.backoff_multiplier;
+  }
+  ngx_conf_merge_msec_value(conf->redis.cluster_check_interval.max, prev->redis.cluster_check_interval.max, NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MAX_MSEC);
+  
   ngx_conf_merge_value(conf->redis.load_scripts_unconditionally, prev->redis.load_scripts_unconditionally, 0);
   
   MERGE_UNSET_CONF(conf->redis.optimize_target, prev->redis.optimize_target, NCHAN_REDIS_OPTIMIZE_UNSET, NCHAN_REDIS_OPTIMIZE_CPU);
@@ -294,7 +314,6 @@ static void *nchan_create_loc_conf(ngx_conf_t *cf) {
   ngx_memzero(&lcf->redis, sizeof(lcf->redis));
   lcf->redis.url_enabled=NGX_CONF_UNSET;
   lcf->redis.ping_interval = NGX_CONF_UNSET;
-  lcf->redis.cluster_check_interval=NGX_CONF_UNSET;
   lcf->redis.upstream_inheritable=NGX_CONF_UNSET;
   lcf->redis.storage_mode = REDIS_MODE_CONF_UNSET;
   lcf->redis.nostore_fastpublish = NGX_CONF_UNSET;
@@ -561,10 +580,6 @@ static char * nchan_merge_loc_conf(ngx_conf_t *cf, void *parent, void *child) {
     ngx_conf_merge_value(conf->redis.ping_interval, up->redis.ping_interval, NGX_CONF_UNSET);
   ngx_conf_merge_value(conf->redis.ping_interval, prev->redis.ping_interval, NCHAN_REDIS_DEFAULT_PING_INTERVAL_TIME);
   
-  if(up)
-    ngx_conf_merge_value(conf->redis.cluster_check_interval, up->redis.cluster_check_interval, NGX_CONF_UNSET);
-  ngx_conf_merge_value(conf->redis.cluster_check_interval, prev->redis.cluster_check_interval, NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_TIME);
-  
   if(up)
     ngx_conf_merge_value(conf->redis.nostore_fastpublish, up->redis.nostore_fastpublish, NGX_CONF_UNSET);
   ngx_conf_merge_value(conf->redis.nostore_fastpublish, prev->redis.nostore_fastpublish, 0);
diff --git a/src/nchan_types.h b/src/nchan_types.h
index 3b4af8481..a05f1d092 100644
--- a/src/nchan_types.h
+++ b/src/nchan_types.h
@@ -39,7 +39,6 @@ typedef struct {
   ngx_str_t                     url;
   ngx_flag_t                    url_enabled;
   time_t                        ping_interval;
-  time_t                        cluster_check_interval;
   ngx_str_t                     namespace;
   nchan_redis_storage_mode_t    storage_mode;
   ngx_int_t                     nostore_fastpublish;
@@ -335,6 +334,7 @@ typedef struct {
       ngx_int_t                     load_scripts_unconditionally;
       nchan_backoff_settings_t      reconnect_delay;
       nchan_backoff_settings_t      cluster_recovery_delay;
+      nchan_backoff_settings_t      cluster_check_interval;
       nchan_redis_optimize_t        optimize_target;
       ngx_int_t                     master_weight;
       ngx_int_t                     slave_weight;
diff --git a/src/store/redis/redis_nodeset.c b/src/store/redis/redis_nodeset.c
index 9519193e5..4fa3c3fe9 100644
--- a/src/store/redis/redis_nodeset.c
+++ b/src/store/redis/redis_nodeset.c
@@ -323,7 +323,6 @@ redis_nodeset_t *nodeset_create(nchan_loc_conf_t *lcf) {
   ns->settings.nostore_fastpublish = rcf->nostore_fastpublish;
   
   ns->settings.ping_interval = rcf->ping_interval;
-  ns->settings.cluster_check_interval = rcf->cluster_check_interval;
   
   ns->status = REDIS_NODESET_DISCONNECTED;
   ngx_memzero(&ns->status_check_ev, sizeof(ns->status_check_ev));
@@ -333,7 +332,7 @@ redis_nodeset_t *nodeset_create(nchan_loc_conf_t *lcf) {
   //init cluster stuff
   ns->cluster.enabled = 0;
   ns->cluster.recovering_on_node = NULL;
-  ns->cluster.last_successful_check = 0;
+  ns->cluster.current_check_interval = 0;
   ngx_memzero(&ns->cluster.check_ev, sizeof(ns->cluster.check_ev));
   nchan_init_timer(&ns->cluster.check_ev, nodeset_cluster_check_event, ns);
   ns->cluster.current_epoch = 0;
@@ -367,6 +366,11 @@ redis_nodeset_t *nodeset_create(nchan_loc_conf_t *lcf) {
     ns->settings.cluster_recovery_delay.backoff_multiplier = scf->redis.cluster_recovery_delay.backoff_multiplier == NGX_CONF_UNSET ? NCHAN_DEFAULT_REDIS_CLUSTER_RECOVERY_DELAY_BACKOFF_MULTIPLIER : scf->redis.cluster_recovery_delay.backoff_multiplier;
     ns->settings.cluster_recovery_delay.max = scf->redis.cluster_recovery_delay.max == (ngx_msec_t )NGX_CONF_UNSET ? NCHAN_DEFAULT_REDIS_CLUSTER_RECOVERY_DELAY_MAX_MSEC : scf->redis.cluster_recovery_delay.max;
     
+    ns->settings.cluster_check_interval.min = scf->redis.cluster_check_interval.min == NGX_CONF_UNSET_MSEC ? NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MIN_MSEC : scf->redis.cluster_check_interval.min;
+    ns->settings.cluster_check_interval.jitter_multiplier = scf->redis.cluster_check_interval.jitter_multiplier == NGX_CONF_UNSET ? NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_JITTER_MULTIPLIER : scf->redis.cluster_check_interval.jitter_multiplier;
+    ns->settings.cluster_check_interval.backoff_multiplier = scf->redis.cluster_check_interval.backoff_multiplier == NGX_CONF_UNSET ? NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_BACKOFF_MULTIPLIER : scf->redis.cluster_check_interval.backoff_multiplier;
+    ns->settings.cluster_check_interval.max = scf->redis.cluster_check_interval.max == (ngx_msec_t )NGX_CONF_UNSET ? NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_MAX_MSEC : scf->redis.cluster_check_interval.max;
+    
     ns->settings.node_weight.master = scf->redis.master_weight == NGX_CONF_UNSET ? 1 : scf->redis.master_weight;
     ns->settings.node_weight.slave = scf->redis.slave_weight == NGX_CONF_UNSET ? 1 : scf->redis.slave_weight;
     
@@ -410,6 +414,7 @@ redis_nodeset_t *nodeset_create(nchan_loc_conf_t *lcf) {
   }
   ns->current_reconnect_delay = 0;
   ns->current_cluster_recovery_delay = 0;
+  ns->cluster.current_check_interval = 0;
   DBG("nodeset created");
   
   char buf[1024];
@@ -725,6 +730,7 @@ static void node_ping_event(ngx_event_t *ev) {
 }
 
 static void nodeset_stop_cluster_check_timer(redis_nodeset_t *ns) {
+  ns->cluster.current_check_interval = 0;
   if(ns->cluster.check_ev.timer_set) {
     ngx_del_timer(&ns->cluster.check_ev);
   }
@@ -732,82 +738,102 @@ static void nodeset_stop_cluster_check_timer(redis_nodeset_t *ns) {
 
 static void nodeset_start_cluster_check_timer(redis_nodeset_t *ns) {
   if(ns->cluster.enabled) {
-    if(!ns->cluster.check_ev.timer_set && ns->settings.cluster_check_interval > 0) {
-      ngx_add_timer(&ns->cluster.check_ev, ns->settings.cluster_check_interval * 1000);
+    if(!ns->cluster.check_ev.timer_set && ns->settings.cluster_check_interval.min > 0) {
+      nchan_set_next_backoff(&ns->cluster.current_check_interval, &ns->settings.cluster_check_interval);
+      ngx_add_timer(&ns->cluster.check_ev, ns->cluster.current_check_interval);
     }
   }
 }
 
-static void rearm_cluster_check_event(ngx_event_t *ev, redis_nodeset_t *ns) {
-  time_t              max_interval = ns->settings.cluster_check_interval;
-  time_t              interval_since_last_check = ngx_time() - ns->cluster.last_successful_check;
-  
-  if(interval_since_last_check <= 0 || interval_since_last_check >= max_interval) {
-    ngx_add_timer(ev, max_interval * 1000);
-  }
-  else {
-    ngx_add_timer(ev, (max_interval - interval_since_last_check) * 1000);
-  }
-}
-
 static void nodeset_cluster_check_event_callback(redisAsyncContext *ac, void *rep, void *privdata) {
   redisReply                 *reply = rep;
   redis_node_t               *node = privdata;
   redis_nodeset_t            *ns = node->nodeset;
   ngx_str_t                   epoch_str;
   int                         epoch;
+  const char                 *err = NULL;
+  
   
   if(reply == NULL) {
     //Node should have already disconnected
     return;
   }
   
-  if(!reply_str_ok(reply)) {
-    node_log_error(node, "CLUSTER INFO command failed");
-    rearm_cluster_check_event(&ns->cluster.check_ev, ns);
-    if(node->state >= REDIS_NODE_READY) {
-      node_disconnect(node, REDIS_NODE_FAILED);
-      nodeset_examine(node->nodeset);
-    }
-    return;
+  if(reply->type == REDIS_REPLY_ERROR) {
+    err = reply->str ? reply->str : "error in reply";
+    goto error;
+  }
+  
+  if(reply->type != REDIS_REPLY_ARRAY || reply->elements != 2) {
+    err = "reply not an array of size 2";
+    goto error;
   }
+  redisReply *info = reply->element[0];
+  redisReply *cluster_nodes = reply->element[1];
   
-  if(!nchan_cstr_match_line(reply->str, "cluster_state:ok")) {
+  //parse CLUSTER INFO reply
+  if(!reply_str_ok(info)) {
+    err = info->str ? info->str : "error in CLUSTER INFO reply";
+    goto error;
+  }
+  if(!nchan_cstr_match_line(info->str, "cluster_state:ok")) {
     char errstr[512];
     ngx_snprintf((u_char *)errstr, 512, "cluster_state not ok on node %s.%Z", node_nickname_cstr(node));
     nodeset_set_status(ns, REDIS_NODESET_CLUSTER_FAILING, errstr);
     return;
   }
-  
-  if(!nchan_get_rest_of_line_in_cstr(reply->str, "cluster_current_epoch:", &epoch_str)) {
-    node_log_error(node, "CLUSTER INFO command reply is weird");
-    //why would this happen? dunno, fail node just in case
-    rearm_cluster_check_event(&ns->cluster.check_ev, ns);
-    if(node->state >= REDIS_NODE_READY) {
-      node_disconnect(node, REDIS_NODE_FAILED);
-      nodeset_examine(node->nodeset);
-    }
-    return;
+  if(!nchan_get_rest_of_line_in_cstr(info->str, "cluster_current_epoch:", &epoch_str)) {
+    err = "Cluster check failed: CLUSTER info command is missing 'cluster_current_epoch'";
+    goto error;
   }
   if((epoch = ngx_atoi(epoch_str.data, epoch_str.len)) == NGX_ERROR) {
-    node_log_error(node, "CLUSTER INFO command failed to parse current epoch number");
+    err = "failed to parse current config epoch number";
     //why would this happen? dunno, fail node just in case
-    rearm_cluster_check_event(&ns->cluster.check_ev, ns);
-    if(node->state >= REDIS_NODE_READY) {
-      node_disconnect(node, REDIS_NODE_FAILED);
-      nodeset_examine(node->nodeset);
-    }
-    return;
+    goto error;
   }
-  
   if(ns->cluster.current_epoch < epoch) {
     char errstr[512];
     ngx_snprintf((u_char *)errstr, 512, "config epoch has changed on node %s.%Z", node_nickname_cstr(node));
     nodeset_set_status(ns, REDIS_NODESET_CLUSTER_FAILING, errstr);
+    return;
   }
-  else {
-    node->nodeset->cluster.last_successful_check = ngx_time();
-    rearm_cluster_check_event(&ns->cluster.check_ev, ns);
+  
+  //parse CLUSTER NODES reply
+  if(cluster_nodes->type != REDIS_REPLY_STRING) {
+    err = "CLUSTER NODES reply is not a string";
+    goto error;
+  }
+  size_t                  i, n;
+  cluster_nodes_line_t   *lines;
+  cluster_nodes_line_t   *l;
+  redis_node_t           *peer;
+  
+  if(!(lines = parse_cluster_nodes(node, cluster_nodes->str, &n))) {
+    err = "parsing CLUSTER NODES reply failed";
+    goto error;
+  }
+  
+  for(i=0; i<n; i++) {
+    l = &lines[i];
+    if(l->failed) {
+      continue;
+    }
+    if(node_skip_cluster_peer(node, l, 0)) {
+      continue;
+    }
+    
+    node_discover_cluster_peer(node, l, &peer);
+  }
+  
+  nodeset_start_cluster_check_timer(ns);
+  return;
+
+error:
+  nodeset_log_error(ns, "Cluster check failed on node %s: %s", node_nickname_cstr(node), err ? err : "unkown error");
+  nodeset_start_cluster_check_timer(ns);
+  if(node->state >= REDIS_NODE_READY) {
+    node_disconnect(node, REDIS_NODE_FAILED);
+    nodeset_examine(node->nodeset);
   }
 }
 
@@ -818,24 +844,19 @@ static void nodeset_cluster_check_event(ngx_event_t *ev) {
   redis_nodeset_t    *ns = ev->data;
   ev->timedout = 0;
   
-  time_t              max_interval = ns->settings.cluster_check_interval;
-  time_t              interval_since_last_check = ngx_time() - ns->cluster.last_successful_check;
-   if(interval_since_last_check < max_interval) {
-    nodeset_log_debug(ns, "skipping cluster_check event");
-    rearm_cluster_check_event(ev, ns);
-    return;
-  }
-
   redis_node_t *node = nodeset_random_node(ns, REDIS_NODE_GET_CLUSTERINFO, REDIS_NODE_ROLE_ANY);
   
   if(!node) {
     nodeset_log_error(ns, "no suitable node to run cluster check. when idle, Nchan may not be aware of cluster changes!");
-    rearm_cluster_check_event(ev, ns);
+    nodeset_start_cluster_check_timer(ns);
     return;
   }
   
   nodeset_log_debug(ns, "cluster_check event on node %s", node_nickname_cstr(node));
-  redisAsyncCommand(node->ctx.cmd, nodeset_cluster_check_event_callback, node, "CLUSTER INFO");
+  redisAsyncCommand(node->ctx.cmd, NULL, NULL, "MULTI");
+  redisAsyncCommand(node->ctx.cmd, NULL, NULL, "CLUSTER INFO");
+  redisAsyncCommand(node->ctx.cmd, NULL, NULL, "CLUSTER NODES");
+  redisAsyncCommand(node->ctx.cmd, nodeset_cluster_check_event_callback, node, "EXEC");
 }
 
 static int nodeset_reset_cluster_node_info(redis_nodeset_t *ns) {
@@ -943,8 +964,8 @@ static void nodeset_recover_cluster_handler(redisAsyncContext *ac, void *rep, vo
   }
   
   redisReply        *cluster_nodes_reply = reply->element[1];
-  if(cluster_info_reply->type != REDIS_REPLY_STRING) {
-    ngx_snprintf(errbuf, 1024, "CLUSTER INFO reply is not a string%Z");
+  if(cluster_nodes_reply->type != REDIS_REPLY_STRING) {
+    ngx_snprintf(errbuf, 1024, "CLUSTER NODES reply is not a string%Z");
     goto fail;
   }
   //parse CLUSTER NODES
@@ -1328,6 +1349,9 @@ static int node_discover_cluster_peer(redis_node_t *node, cluster_nodes_line_t *
   nodeset_log_notice(node->nodeset, "Discovering cluster %s %s", (l->master ? "master" : "slave"), rcp_cstr(&rcp));
   peer = nodeset_node_create_with_connect_params(node->nodeset, &rcp);
   peer->discovered = 1;
+  if(!l->master && l->master_id.len > 0) {
+    nchan_strcpy(&peer->cluster.master_id, &l->master_id, MAX_CLUSTER_ID_LENGTH);
+  }
   nchan_strcpy(&peer->cluster.id, &l->id, MAX_CLUSTER_ID_LENGTH);
   node_set_role(peer, l->master ? REDIS_NODE_ROLE_MASTER : REDIS_NODE_ROLE_SLAVE);
   //ignore all the other things for now
@@ -1851,9 +1875,6 @@ int nodeset_node_reply_keyslot_ok(redis_node_t *node, redisReply *reply) {
       return 0;
     }
   }
-  if(node->cluster.enabled) {
-    node->nodeset->cluster.last_successful_check = ngx_time();
-  }
   return 1;
 }
 
diff --git a/src/store/redis/redis_nodeset.h b/src/store/redis/redis_nodeset.h
index dbbe64acc..8523315f3 100644
--- a/src/store/redis/redis_nodeset.h
+++ b/src/store/redis/redis_nodeset.h
@@ -88,8 +88,8 @@ typedef struct { //redis_nodeset_cluster_t
   rbtree_seed_t               keyslots; //cluster rbtree seed
   redis_node_t               *recovering_on_node;
   int                         current_epoch;
+  ngx_msec_t                  current_check_interval;
   ngx_event_t                 check_ev;
-  time_t                      last_successful_check;
 } redis_nodeset_cluster_t;
 
 typedef struct {
@@ -161,7 +161,6 @@ struct redis_nodeset_s {
       ngx_int_t                   slave;
     }                           node_weight;
     time_t                      ping_interval;
-    time_t                      cluster_check_interval;
     ngx_str_t                  *namespace;
     nchan_redis_optimize_t      optimize_target;
     ngx_msec_t                  node_connect_timeout;
@@ -169,6 +168,7 @@ struct redis_nodeset_s {
     
     nchan_backoff_settings_t    reconnect_delay;
     nchan_backoff_settings_t    cluster_recovery_delay;
+    nchan_backoff_settings_t    cluster_check_interval;
     
     ngx_msec_t                  cluster_max_failing_msec;
     ngx_int_t                   load_scripts_unconditionally;
diff --git a/src/store/redis/store.h b/src/store/redis/store.h
index 6bfa30ce9..973bbcf15 100755
--- a/src/store/redis/store.h
+++ b/src/store/redis/store.h
@@ -2,7 +2,6 @@
 #define NCHAN_REDIS_STORE_H
 
 #define NCHAN_REDIS_DEFAULT_PING_INTERVAL_TIME 4*60
-#define NCHAN_REDIS_DEFAULT_CLUSTER_CHECK_INTERVAL_TIME 5
 #define NCHAN_REDIS_DEFAULT_PUBSUB_MESSAGE_MSGKEY_SIZE 1024*5
 
 extern nchan_store_t  nchan_store_redis;