From b2a0dd170254a88a3b0f099fc7328edb49f0801f Mon Sep 17 00:00:00 2001 From: FZambia Date: Sun, 26 Jul 2026 09:51:28 +0300 Subject: [PATCH] fix(sentinel): back off between failed topology refreshes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit refreshRetry was an unbounded 'goto retry' loop with no delay. When every sentinel is unreachable — which is 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, dialing every sentinel in the list on each iteration. Apply equal-jitter exponential backoff between attempts and bail out once Close() has been called, so a client shut down while its sentinels are unreachable does not leak the goroutine. The base grows to 512ms, one shift below the 1024ms cap, so base+jitter always lands under the cap rather than being clamped to it. Capping the sum at the same magnitude as the base would make every sample come out at exactly the cap once the base reached it: the jitter would vanish at steady state and a fleet of clients riding out a long outage would retry in lockstep — the thundering herd the equal-jitter scheme is there to prevent. --- sentinel.go | 57 ++++++++++++++++++++-- sentinel_test.go | 122 +++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 176 insertions(+), 3 deletions(-) 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") + } +}