diff --git a/sentinel.go b/sentinel.go index b3f605d4..ee032131 100644 --- a/sentinel.go +++ b/sentinel.go @@ -530,10 +530,61 @@ func (c *sentinelClient) _switchTarget(addr string, isMaster bool) (err error) { return nil } +const ( + // refreshMaxRetryShift caps the exponent, not the number of retries: the + // delay stops doubling once attempts reach it, but refreshRetry keeps going + // until it succeeds or the client is closed. It only bounds how fast the + // client retries, never whether it does — the client always heals itself. + refreshMaxRetryShift = 9 + refreshMaxRetryDelay = time.Second +) + +// refreshRetryDelay is the backoff applied between failed refresh attempts. +// It mirrors defaultRetryDelayFn's "equal jitter" scheme, but on a millisecond +// rather than a microsecond base: a refresh is a multi-round-trip operation +// against every known sentinel, so retrying it hundreds of thousands of times +// per second is never useful. It settles at ~512ms-1s. +func refreshRetryDelay(attempts int) time.Duration { + base := 1 << min(refreshMaxRetryShift, attempts) + jitter := util.FastRand(base) + return min(refreshMaxRetryDelay, time.Duration(base+jitter)*time.Millisecond) +} + +// refreshRetry keeps refreshing the topology until it succeeds or the client is +// closed. +// +// This used to be an unbounded `goto retry` loop with no delay at all. When +// every sentinel is unreachable — precisely what happens during a failover +// where the primary and a co-located sentinel go down together — refresh() +// fails immediately and the loop spins as fast as the CPU allows. Each +// iteration also dials every sentinel in the list, so one client can saturate a +// core and flood the surviving sentinels with connection attempts exactly while +// they are running the election. +// +// refreshRetry is additionally re-entered from listWatch's error handler, so +// failures compound. func (c *sentinelClient) refreshRetry() { -retry: - if err := c.refresh(); err != nil { - goto retry + // Stop once Close() has been called, so a client shut down while its + // sentinels are unreachable does not leak this goroutine. The check is at + // the loop head so a Close() during the backoff wait exits without a further + // refresh. + for attempts := 0; atomic.LoadUint32(&c.stop) == 0; attempts++ { + if err := c.refresh(); err == nil { + return + } + c.waitBeforeRetry(refreshRetryDelay(attempts)) + } +} + +// waitBeforeRetry sleeps for d, but returns early once Close() has been called +// so shutdown does not have to wait out a full backoff interval. There is no +// close channel to select on here, so it polls c.stop in short steps. +func (c *sentinelClient) waitBeforeRetry(d time.Duration) { + const step = 20 * time.Millisecond + for d > 0 && atomic.LoadUint32(&c.stop) == 0 { + s := min(step, d) + time.Sleep(s) + d -= s } } diff --git a/sentinel_test.go b/sentinel_test.go index 68f46cc8..753abf3a 100644 --- a/sentinel_test.go +++ b/sentinel_test.go @@ -3570,3 +3570,125 @@ func TestSentinelSendToReplicasClientPubSub(t *testing.T) { time.Sleep(time.Millisecond * 100) } } + +func TestRefreshRetryDelay(t *testing.T) { + // Must never be zero: a zero delay reintroduces the hot spin this backoff + // exists to prevent. + for attempts := 0; attempts < 64; attempts++ { + d := refreshRetryDelay(attempts) + if d <= 0 { + t.Fatalf("attempts %d: delay must be positive, got %v", attempts, d) + } + if d > refreshMaxRetryDelay { + t.Fatalf("attempts %d: delay %v exceeds cap %v", attempts, d, refreshMaxRetryDelay) + } + } + // And it must actually grow, otherwise a long outage still hammers the + // sentinels at the initial rate. + if refreshRetryDelay(0) >= refreshRetryDelay(refreshMaxRetryShift) { + t.Fatalf("delay should increase with attempts: first=%v capped=%v", + refreshRetryDelay(0), refreshRetryDelay(refreshMaxRetryShift)) + } + // The jitter has to survive once the base stops growing, which is where a + // long outage spends all of its time. Capping the jittered sum at the same + // magnitude as the base clamps every sample onto the cap, and a fleet of + // clients then retries in lockstep — exactly the thundering herd the jitter + // exists to prevent. + for _, attempts := range []int{refreshMaxRetryShift, refreshMaxRetryShift + 1, 64} { + seen := make(map[time.Duration]struct{}) + for range 2000 { + seen[refreshRetryDelay(attempts)] = struct{}{} + } + if len(seen) < 2 { + t.Fatalf("attempts %d: every retry lands on the same delay %v, the jitter is being clamped away", + attempts, refreshRetryDelay(attempts)) + } + } +} + +// TestRefreshRetryWaitCancelsOnStop pins that the backoff wait is interruptible: +// a Close() mid-wait must return promptly instead of sleeping out the interval. +// It waits an hour, so a plain time.Sleep would hang the test. +func TestRefreshRetryWaitCancelsOnStop(t *testing.T) { + c := &sentinelClient{} + done := make(chan struct{}) + go func() { + c.waitBeforeRetry(time.Hour) + close(done) + }() + time.Sleep(30 * time.Millisecond) // let the goroutine enter the wait + atomic.StoreUint32(&c.stop, 1) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("waitBeforeRetry did not return promptly after stop was set") + } +} + +// TestSentinelRefreshRetryBackoff pins the two properties that matter when every +// sentinel is unreachable: refreshRetry must not spin, and it must exit once the +// client is closed rather than leaking a goroutine that retries forever. +func TestSentinelRefreshRetryBackoff(t *testing.T) { + defer ShouldNotLeak(SetupLeakDetection()) + + var failing atomic.Bool + var attempts atomic.Int64 + + s0 := &mockConn{ + DoFn: func(cmd Completed) RedisResult { return RedisResult{} }, + DoMultiFn: func(multi ...Completed) *redisresults { + if failing.Load() { + attempts.Add(1) + return &redisresults{s: []RedisResult{ + NewErrorResult(ErrClosing), NewErrorResult(ErrClosing), + }} + } + return &redisresults{s: []RedisResult{ + {val: slicemsg('*', []RedisMessage{})}, + {val: slicemsg('*', []RedisMessage{strmsg('+', ""), strmsg('+', "1")})}, + }} + }, + } + m := &mockConn{ + DoFn: func(cmd Completed) RedisResult { + return RedisResult{val: slicemsg('*', []RedisMessage{strmsg('+', "master")})} + }, + } + client, err := newSentinelClient( + &ClientOption{InitAddress: []string{":0"}}, + func(dst string, opt *ClientOption) conn { + switch dst { + case ":0": + return s0 + case ":1": + return m + } + return nil + }, + newRetryer(defaultRetryDelayFn), + ) + if err != nil { + t.Fatalf("unexpected err %v", err) + } + + failing.Store(true) + done := make(chan struct{}) + go func() { + client.refreshRetry() + close(done) + }() + + time.Sleep(200 * time.Millisecond) + // Without backoff this loop managed hundreds of thousands of iterations in + // this window; with it, a few dozen at most. + if n := attempts.Load(); n > 500 { + t.Fatalf("refreshRetry spun %d times in 200ms — backoff not applied", n) + } + + client.Close() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("refreshRetry did not return after Close — goroutine leaked") + } +}