Tolerate a recently broken replica link when picking hosts for reads

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

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)
 		}

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)

rediscluster/cluster.go

@@ -50,7 +50,13 @@ const (
 
 const (
 	defaultCheckInterval = 5 * time.Second
-	defaultWaitToMigrate = 20 * time.Millisecond
+	// 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
 
 	forceInterval = 100 * time.Millisecond
 
@@ -112,6 +118,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 +252,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

rediscluster/replica_health_internal_test.go

@@ -0,0 +1,55 @@
+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
+		linkDownTolerance time.Duration
+		want              bool
+	}{
+		{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},
+		{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 {
+				t.Errorf("replicaHealthy(%q, %v) = %v, want %v", tc.info, tc.linkDownTolerance, got, tc.want)
+			}
+		})
+	}
+}
+
+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)
+		}
+	}
+}

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(linkDownTolerance time.Duration) redis.FuncFuture {
+	return func(res interface{}, n uint64) {
+		s.setReplicaInfo(res, n, linkDownTolerance)
+	}
+}
+
+func (s *shard) setReplicaInfo(res interface{}, n uint64, linkDownTolerance 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, linkDownTolerance)
 	}
 	for {
 		oldstate := atomic.LoadUint32(&s.good)
@@ -242,7 +249,59 @@ 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, linkDownTolerance time.Duration) bool {
+	if infoHas(info, "loading", "1") {
+		return false
+	}
+	if !infoHas(info, "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 < 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) {
+	value, ok := infoField(info, field)
+	if !ok {
+		return 0, false
+	}
+	v, err := strconv.ParseInt(string(value), 10, 64)
+	return v, err == nil
+}
+
+func (cfg *clusterConfig) setConnRoles(linkDownTolerance time.Duration) {
 	for _, sh := range cfg.shards {
 		for i, addr := range sh.addr {
 			node := cfg.nodes[addr]
@@ -254,7 +313,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(linkDownTolerance), uint64(i*2))
 				}
 			}
 		}