From 5b6bca9d00e05409332d4e9165861d7259d18efa Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Tue, 15 Sep 2026 09:22:48 +0100 Subject: [PATCH 1/3] Tolerate a recently broken replica link when picking hosts for reads A replica is excluded from reads as soon as INFO reports master_link_status:down. That is the right call when the replica has lost a live master, but it is also what every replica of a dead master reports for the whole failover window, while the master's own health bit is never cleared. Every MasterAndSlaves / PreferSlaves read to that shard fails with no_alive_connection until CLUSTER SLOTS names a new master, though the replicas keep serving the freshest data there is. A broken link now excludes a replica only once master_link_down_since_seconds reaches ReplicaLinkDownTolerance (60s by default), which bounds how stale a read can get. Redis reports -1 for a replica that has never synced since it started, and its dataset is then anything from empty to the RDB it booted from, so such a replica stays excluded. MASTERDOWN, the reply of a replica with replica-serve-stale-data set to no, is retried on another host like LOADING: the command was rejected before it ran. --- redis/error.go | 2 + redis/reader.go | 3 ++ redis/reader_test.go | 6 +++ rediscluster/cluster.go | 15 +++++- rediscluster/replica_health_internal_test.go | 33 +++++++++++++ rediscluster/slotrange.go | 50 +++++++++++++++++--- 6 files changed, 102 insertions(+), 7 deletions(-) create mode 100644 rediscluster/replica_health_internal_test.go diff --git a/redis/error.go b/redis/error.go index 8442ce2..c5b71ed 100644 --- a/redis/error.go +++ b/redis/error.go @@ -73,6 +73,8 @@ var ( ErrClusterDown = ErrResult.NewSubtype("clusterdown", ErrTraitNotSent) // ErrLoading - redis didn't finish start ErrLoading = ErrResult.NewSubtype("loading", ErrTraitNotSent) + // ErrMasterDown - MASTERDOWN response: replica lost its master and replica-serve-stale-data is 'no' + ErrMasterDown = ErrResult.NewSubtype("masterdown", ErrTraitNotSent) // ErrExecEmpty - EXEC returns nil (WATCH failed) (it is strange, cause we don't support WATCH) ErrExecEmpty = ErrResult.NewSubtype("exec_empty") // ErrExecAbort - EXEC returns EXECABORT diff --git a/redis/reader.go b/redis/reader.go index ac05ffd..9f9bb01 100644 --- a/redis/reader.go +++ b/redis/reader.go @@ -54,6 +54,9 @@ func ReadResponse(b *bufio.Reader) (interface{}, int) { if strings.HasPrefix(txt, "LOADING") { return ErrLoading.New(txt), len(line) } + if strings.HasPrefix(txt, "MASTERDOWN") { + return ErrMasterDown.New(txt), len(line) + } if strings.HasPrefix(txt, "EXECABORT") { return ErrExecAbort.New(txt), len(line) } diff --git a/redis/reader_test.go b/redis/reader_test.go index 18b1761..87ff387 100644 --- a/redis/reader_test.go +++ b/redis/reader_test.go @@ -145,6 +145,12 @@ func TestReadResponse_Correct(t *testing.T) { assert.Equal(t, "LOADING", err.Message()) } + res = readLines("-MASTERDOWN Link with MASTER is down and replica-serve-stale-data is set to 'no'.\r\n") + if checkErrType(t, res, ErrMasterDown) { + err := res.(*errorx.Error) + assert.True(t, err.HasTrait(ErrTraitNotSent)) + } + for i := -1000; i <= 1000; i++ { res = readLines(fmt.Sprintf(":%d\r\n", i)) assert.Equal(t, int64(i), res) diff --git a/rediscluster/cluster.go b/rediscluster/cluster.go index 8280db2..b00b0c4 100644 --- a/rediscluster/cluster.go +++ b/rediscluster/cluster.go @@ -50,7 +50,11 @@ const ( const ( defaultCheckInterval = 5 * time.Second - defaultWaitToMigrate = 20 * time.Millisecond + // defaultReplicaLinkDownTolerance outlives a failover at cluster-node-timeout of + // up to ~50 seconds, and stays well below the ~160 seconds after which redis itself + // stops considering the replica's data fresh enough for promotion. + defaultReplicaLinkDownTolerance = 60 * time.Second + defaultWaitToMigrate = 20 * time.Millisecond forceInterval = 100 * time.Millisecond @@ -112,6 +116,11 @@ type Opts struct { ForceMinLatencyReplica bool // WeightProvider - enables to explicitly set weights of replicas (has higher priority than LatencyOrientedRR) WeightProvider WeightProvider + // ReplicaLinkDownTolerance - a replica whose link with its master is down keeps receiving + // reads while its master_link_down_since_seconds stays below this value. A replica that + // has never synced since it started is never read regardless of it. + // default: 60 seconds; negative: a replica with a broken link is never read + ReplicaLinkDownTolerance time.Duration // Enable connection with TLS TLSEnabled bool // Config for TLS connection @@ -241,6 +250,10 @@ func NewCluster(ctx context.Context, initAddrs []string, opts Opts) (*Cluster, e cluster.opts.WaitToMigrate = 100 * time.Millisecond } + if cluster.opts.ReplicaLinkDownTolerance == 0 { + cluster.opts.ReplicaLinkDownTolerance = defaultReplicaLinkDownTolerance + } + cluster.latencyAwareness = disabled if cluster.opts.LatencyOrientedRR { cluster.latencyAwareness = enabled diff --git a/rediscluster/replica_health_internal_test.go b/rediscluster/replica_health_internal_test.go new file mode 100644 index 0000000..15e4898 --- /dev/null +++ b/rediscluster/replica_health_internal_test.go @@ -0,0 +1,33 @@ +package rediscluster + +import ( + "strings" + "testing" + "time" +) + +func TestReplicaHealthy(t *testing.T) { + info := func(lines ...string) []byte { + return []byte("# Replication\r\nrole:slave\r\n" + strings.Join(lines, "\r\n") + "\r\n# Persistence\r\nloading:0\r\n") + } + for _, tc := range []struct { + name string + info []byte + tolerance time.Duration + want bool + }{ + {name: "link up", info: info("master_link_status:up"), tolerance: time.Minute, want: true}, + {name: "link down for a while", info: info("master_link_status:down", "master_link_down_since_seconds:3"), tolerance: time.Minute, want: true}, + {name: "link down for the whole tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:60"), tolerance: time.Minute, want: false}, + {name: "never synced since start", info: info("master_link_status:down", "master_link_down_since_seconds:-1"), tolerance: time.Minute, want: false}, + {name: "link down without duration", info: info("master_link_status:down"), tolerance: time.Minute, want: false}, + {name: "negative tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:0"), tolerance: -1, want: false}, + {name: "loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:1\r\n"), tolerance: time.Minute, want: false}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := replicaHealthy(tc.info, tc.tolerance); got != tc.want { + t.Errorf("replicaHealthy(%q, %v) = %v, want %v", tc.info, tc.tolerance, got, tc.want) + } + }) + } +} diff --git a/rediscluster/slotrange.go b/rediscluster/slotrange.go index 359a597..2961e67 100644 --- a/rediscluster/slotrange.go +++ b/rediscluster/slotrange.go @@ -3,6 +3,7 @@ package rediscluster import ( "bytes" "math" + "strconv" "sync/atomic" "time" @@ -142,7 +143,7 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { c.nodeWait.promises = make(map[string]*[]connThen, 1) c.nodeWait.Unlock() - go newConfig.setConnRoles() + go newConfig.setConnRoles(c.opts.ReplicaLinkDownTolerance) var sh uint32 for i := 0; i < redisclusterutil.NumSlots; i++ { @@ -213,7 +214,13 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { }) } -func (s *shard) setReplicaInfo(res interface{}, n uint64) { +func (s *shard) replicaInfoFuture(tolerance time.Duration) redis.FuncFuture { + return func(res interface{}, n uint64) { + s.setReplicaInfo(res, n, tolerance) + } +} + +func (s *shard) setReplicaInfo(res interface{}, n uint64, tolerance time.Duration) { haserr := false if err := redis.AsError(res); err != nil { haserr = true @@ -222,8 +229,8 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64) { haserr = !(ok && str == "OK") } else if buf, ok := res.([]byte); !ok { haserr = true - } else if bytes.Contains(buf, []byte("master_link_status:down")) || bytes.Contains(buf, []byte("loading:1")) { - haserr = true + } else { + haserr = !replicaHealthy(buf, tolerance) } for { oldstate := atomic.LoadUint32(&s.good) @@ -242,7 +249,38 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64) { } } -func (cfg *clusterConfig) setConnRoles() { +// replicaHealthy tells whether INFO output describes a replica worth reading from. +// master_link_down_since_seconds is -1 for a replica that has never synced since it +// started, and its dataset is then anything from empty to the RDB it booted from. +func replicaHealthy(info []byte, tolerance time.Duration) bool { + if bytes.Contains(info, []byte("loading:1")) { + return false + } + if !bytes.Contains(info, []byte("master_link_status:down")) { + return true + } + since, ok := infoInt(info, "master_link_down_since_seconds") + if !ok || since < 0 { + return false + } + return time.Duration(since)*time.Second < tolerance +} + +func infoInt(info []byte, field string) (int64, bool) { + key := []byte("\n" + field + ":") + i := bytes.Index(info, key) + if i < 0 { + return 0, false + } + value := info[i+len(key):] + if end := bytes.IndexAny(value, "\r\n"); end >= 0 { + value = value[:end] + } + v, err := strconv.ParseInt(string(value), 10, 64) + return v, err == nil +} + +func (cfg *clusterConfig) setConnRoles(tolerance time.Duration) { for _, sh := range cfg.shards { for i, addr := range sh.addr { node := cfg.nodes[addr] @@ -254,7 +292,7 @@ func (cfg *clusterConfig) setConnRoles() { conn.Send(Request{"READWRITE", nil}, nil, 0) } else { conn.SendBatch([]Request{{"READONLY", nil}, {"INFO", nil}}, - redis.FuncFuture(sh.setReplicaInfo), uint64(i*2)) + sh.replicaInfoFuture(tolerance), uint64(i*2)) } } } From cc0eba6aa689a50604462751d15198a385c8f949 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Wed, 16 Sep 2026 18:29:47 +0100 Subject: [PATCH 2/3] Name the link-down tolerance parameter and derive the default in its comment --- rediscluster/cluster.go | 8 +++--- rediscluster/replica_health_internal_test.go | 26 ++++++++++---------- rediscluster/slotrange.go | 16 ++++++------ 3 files changed, 26 insertions(+), 24 deletions(-) diff --git a/rediscluster/cluster.go b/rediscluster/cluster.go index b00b0c4..6a86b9f 100644 --- a/rediscluster/cluster.go +++ b/rediscluster/cluster.go @@ -50,9 +50,11 @@ const ( const ( defaultCheckInterval = 5 * time.Second - // defaultReplicaLinkDownTolerance outlives a failover at cluster-node-timeout of - // up to ~50 seconds, and stays well below the ~160 seconds after which redis itself - // stops considering the replica's data fresh enough for promotion. + // A failover takes cluster-node-timeout plus an election of a second or two, and the + // client notices the new master one CheckInterval later; 60 seconds cover the default + // 15-second node timeout several times over and stay well below the ~160 seconds + // (repl-ping-replica-period + node-timeout × cluster-replica-validity-factor) after + // which redis itself stops considering the replica fresh enough for promotion. defaultReplicaLinkDownTolerance = 60 * time.Second defaultWaitToMigrate = 20 * time.Millisecond diff --git a/rediscluster/replica_health_internal_test.go b/rediscluster/replica_health_internal_test.go index 15e4898..0d108a2 100644 --- a/rediscluster/replica_health_internal_test.go +++ b/rediscluster/replica_health_internal_test.go @@ -11,22 +11,22 @@ func TestReplicaHealthy(t *testing.T) { return []byte("# Replication\r\nrole:slave\r\n" + strings.Join(lines, "\r\n") + "\r\n# Persistence\r\nloading:0\r\n") } for _, tc := range []struct { - name string - info []byte - tolerance time.Duration - want bool + name string + info []byte + linkDownTolerance time.Duration + want bool }{ - {name: "link up", info: info("master_link_status:up"), tolerance: time.Minute, want: true}, - {name: "link down for a while", info: info("master_link_status:down", "master_link_down_since_seconds:3"), tolerance: time.Minute, want: true}, - {name: "link down for the whole tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:60"), tolerance: time.Minute, want: false}, - {name: "never synced since start", info: info("master_link_status:down", "master_link_down_since_seconds:-1"), tolerance: time.Minute, want: false}, - {name: "link down without duration", info: info("master_link_status:down"), tolerance: time.Minute, want: false}, - {name: "negative tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:0"), tolerance: -1, want: false}, - {name: "loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:1\r\n"), tolerance: time.Minute, want: false}, + {name: "link up", info: info("master_link_status:up"), linkDownTolerance: time.Minute, want: true}, + {name: "link down for a while", info: info("master_link_status:down", "master_link_down_since_seconds:3"), linkDownTolerance: time.Minute, want: true}, + {name: "link down for the whole linkDownTolerance", info: info("master_link_status:down", "master_link_down_since_seconds:60"), linkDownTolerance: time.Minute, want: false}, + {name: "never synced since start", info: info("master_link_status:down", "master_link_down_since_seconds:-1"), linkDownTolerance: time.Minute, want: false}, + {name: "link down without duration", info: info("master_link_status:down"), linkDownTolerance: time.Minute, want: false}, + {name: "negative tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:0"), linkDownTolerance: -1, want: false}, + {name: "loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:1\r\n"), linkDownTolerance: time.Minute, want: false}, } { t.Run(tc.name, func(t *testing.T) { - if got := replicaHealthy(tc.info, tc.tolerance); got != tc.want { - t.Errorf("replicaHealthy(%q, %v) = %v, want %v", tc.info, tc.tolerance, got, tc.want) + if got := replicaHealthy(tc.info, tc.linkDownTolerance); got != tc.want { + t.Errorf("replicaHealthy(%q, %v) = %v, want %v", tc.info, tc.linkDownTolerance, got, tc.want) } }) } diff --git a/rediscluster/slotrange.go b/rediscluster/slotrange.go index 2961e67..0f83f92 100644 --- a/rediscluster/slotrange.go +++ b/rediscluster/slotrange.go @@ -214,13 +214,13 @@ func (c *Cluster) updateMappings(slotRanges []redisclusterutil.SlotsRange) { }) } -func (s *shard) replicaInfoFuture(tolerance time.Duration) redis.FuncFuture { +func (s *shard) replicaInfoFuture(linkDownTolerance time.Duration) redis.FuncFuture { return func(res interface{}, n uint64) { - s.setReplicaInfo(res, n, tolerance) + s.setReplicaInfo(res, n, linkDownTolerance) } } -func (s *shard) setReplicaInfo(res interface{}, n uint64, tolerance time.Duration) { +func (s *shard) setReplicaInfo(res interface{}, n uint64, linkDownTolerance time.Duration) { haserr := false if err := redis.AsError(res); err != nil { haserr = true @@ -230,7 +230,7 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64, tolerance time.Duratio } else if buf, ok := res.([]byte); !ok { haserr = true } else { - haserr = !replicaHealthy(buf, tolerance) + haserr = !replicaHealthy(buf, linkDownTolerance) } for { oldstate := atomic.LoadUint32(&s.good) @@ -252,7 +252,7 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64, tolerance time.Duratio // replicaHealthy tells whether INFO output describes a replica worth reading from. // master_link_down_since_seconds is -1 for a replica that has never synced since it // started, and its dataset is then anything from empty to the RDB it booted from. -func replicaHealthy(info []byte, tolerance time.Duration) bool { +func replicaHealthy(info []byte, linkDownTolerance time.Duration) bool { if bytes.Contains(info, []byte("loading:1")) { return false } @@ -263,7 +263,7 @@ func replicaHealthy(info []byte, tolerance time.Duration) bool { if !ok || since < 0 { return false } - return time.Duration(since)*time.Second < tolerance + return time.Duration(since)*time.Second < linkDownTolerance } func infoInt(info []byte, field string) (int64, bool) { @@ -280,7 +280,7 @@ func infoInt(info []byte, field string) (int64, bool) { return v, err == nil } -func (cfg *clusterConfig) setConnRoles(tolerance time.Duration) { +func (cfg *clusterConfig) setConnRoles(linkDownTolerance time.Duration) { for _, sh := range cfg.shards { for i, addr := range sh.addr { node := cfg.nodes[addr] @@ -292,7 +292,7 @@ func (cfg *clusterConfig) setConnRoles(tolerance time.Duration) { conn.Send(Request{"READWRITE", nil}, nil, 0) } else { conn.SendBatch([]Request{{"READONLY", nil}, {"INFO", nil}}, - sh.replicaInfoFuture(tolerance), uint64(i*2)) + sh.replicaInfoFuture(linkDownTolerance), uint64(i*2)) } } } From cc922c8ff85e505ba6ba8e526182444a746ea296 Mon Sep 17 00:00:00 2001 From: Sergey Zagursky Date: Wed, 16 Sep 2026 19:13:59 +0100 Subject: [PATCH 3/3] Match INFO fields as whole line prefixes A substring search for loading:1 also matches async_loading:1 on redis 7+, where the replica keeps serving its old dataset while the new one loads. Every field lookup now goes through one parser that anchors the name to the start of a line. --- rediscluster/replica_health_internal_test.go | 22 +++++++++++ rediscluster/slotrange.go | 39 +++++++++++++++----- 2 files changed, 52 insertions(+), 9 deletions(-) diff --git a/rediscluster/replica_health_internal_test.go b/rediscluster/replica_health_internal_test.go index 0d108a2..076fc8b 100644 --- a/rediscluster/replica_health_internal_test.go +++ b/rediscluster/replica_health_internal_test.go @@ -23,6 +23,7 @@ func TestReplicaHealthy(t *testing.T) { {name: "link down without duration", info: info("master_link_status:down"), linkDownTolerance: time.Minute, want: false}, {name: "negative tolerance", info: info("master_link_status:down", "master_link_down_since_seconds:0"), linkDownTolerance: -1, want: false}, {name: "loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:1\r\n"), linkDownTolerance: time.Minute, want: false}, + {name: "async loading is not loading", info: []byte("# Replication\r\nmaster_link_status:up\r\n# Persistence\r\nloading:0\r\nasync_loading:1\r\n"), linkDownTolerance: time.Minute, want: true}, } { t.Run(tc.name, func(t *testing.T) { if got := replicaHealthy(tc.info, tc.linkDownTolerance); got != tc.want { @@ -31,3 +32,24 @@ func TestReplicaHealthy(t *testing.T) { }) } } + +func TestInfoField(t *testing.T) { + info := []byte("loading:0\r\nasync_loading:1\r\nmaster_link_status:down\r\nmaster_link_down_since_seconds:-1\r\n") + for _, tc := range []struct { + field string + want string + ok bool + }{ + {field: "loading", want: "0", ok: true}, + {field: "async_loading", want: "1", ok: true}, + {field: "master_link_status", want: "down", ok: true}, + {field: "master_link_down_since_seconds", want: "-1", ok: true}, + {field: "link_status", ok: false}, + {field: "role", ok: false}, + } { + got, ok := infoField(info, tc.field) + if ok != tc.ok || string(got) != tc.want { + t.Errorf("infoField(%q) = %q, %v; want %q, %v", tc.field, got, ok, tc.want, tc.ok) + } + } +} diff --git a/rediscluster/slotrange.go b/rediscluster/slotrange.go index 0f83f92..b60de2c 100644 --- a/rediscluster/slotrange.go +++ b/rediscluster/slotrange.go @@ -253,10 +253,10 @@ func (s *shard) setReplicaInfo(res interface{}, n uint64, linkDownTolerance time // master_link_down_since_seconds is -1 for a replica that has never synced since it // started, and its dataset is then anything from empty to the RDB it booted from. func replicaHealthy(info []byte, linkDownTolerance time.Duration) bool { - if bytes.Contains(info, []byte("loading:1")) { + if infoHas(info, "loading", "1") { return false } - if !bytes.Contains(info, []byte("master_link_status:down")) { + if !infoHas(info, "master_link_status", "down") { return true } since, ok := infoInt(info, "master_link_down_since_seconds") @@ -266,16 +266,37 @@ func replicaHealthy(info []byte, linkDownTolerance time.Duration) bool { return time.Duration(since)*time.Second < linkDownTolerance } +// infoField returns the value of a named INFO field. The name is matched as a whole +// line prefix: `loading` must not be found inside `async_loading`. +func infoField(info []byte, field string) ([]byte, bool) { + key := []byte(field + ":") + for off := 0; ; { + i := bytes.Index(info[off:], key) + if i < 0 { + return nil, false + } + i += off + if i == 0 || info[i-1] == '\n' { + value := info[i+len(key):] + if end := bytes.IndexAny(value, "\r\n"); end >= 0 { + value = value[:end] + } + return value, true + } + off = i + 1 + } +} + +func infoHas(info []byte, field, value string) bool { + got, ok := infoField(info, field) + return ok && string(got) == value +} + func infoInt(info []byte, field string) (int64, bool) { - key := []byte("\n" + field + ":") - i := bytes.Index(info, key) - if i < 0 { + value, ok := infoField(info, field) + if !ok { return 0, false } - value := info[i+len(key):] - if end := bytes.IndexAny(value, "\r\n"); end >= 0 { - value = value[:end] - } v, err := strconv.ParseInt(string(value), 10, 64) return v, err == nil }