From e5fb547e9558cca049b99e1c8ee6b5cfd3d83ac0 Mon Sep 17 00:00:00 2001 From: Mohamed Chorfa Date: Fri, 29 May 2026 23:54:54 -0400 Subject: [PATCH] wdclient, dailyrun: add equal jitter to retry backoff (#9737) * wdclient, dailyrun: add equal jitter to retry backoff Prevents thundering-herd retries when many clients recover from a transient failure at the same instant (e.g., filer restart, network partition healing). Uses equal jitter: wait in [d/2, d) instead of deterministic d. This bounds the maximum wait while still desynchronizing clients. Files: - weed/wdclient/filer_client.go (LookupVolumeIds retry loop) - weed/s3api/s3lifecycle/dailyrun/dispatch.go (dispatchWithRetry) Tests added for bounds, zero/negative inputs, and distribution sanity. Closes #9735 * wdclient: honor ctx cancellation during LookupVolumeIds backoff --------- Co-authored-by: Mohamed Chorfa Co-authored-by: Chris Lu --- weed/s3api/s3lifecycle/dailyrun/dispatch.go | 16 ++++- .../s3api/s3lifecycle/dailyrun/jitter_test.go | 44 +++++++++++++ weed/wdclient/filer_client.go | 25 ++++++- weed/wdclient/jitter_test.go | 65 +++++++++++++++++++ 4 files changed, 147 insertions(+), 3 deletions(-) create mode 100644 weed/s3api/s3lifecycle/dailyrun/jitter_test.go create mode 100644 weed/wdclient/jitter_test.go diff --git a/weed/s3api/s3lifecycle/dailyrun/dispatch.go b/weed/s3api/s3lifecycle/dailyrun/dispatch.go index dba3656b7..542a1d73b 100644 --- a/weed/s3api/s3lifecycle/dailyrun/dispatch.go +++ b/weed/s3api/s3lifecycle/dailyrun/dispatch.go @@ -3,6 +3,7 @@ package dailyrun import ( "context" "errors" + "math/rand" "time" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" @@ -52,7 +53,7 @@ func dispatchWithRetry(ctx context.Context, client LifecycleClient, m router.Mat select { case <-ctx.Done(): return s3_lifecycle_pb.LifecycleDeleteOutcome_LIFECYCLE_DELETE_OUTCOME_UNSPECIFIED, ctx.Err() - case <-time.After(backoff): + case <-time.After(jitter(backoff)): } backoff *= 2 if backoff > transportRetryMax { @@ -62,6 +63,19 @@ func dispatchWithRetry(ctx context.Context, client LifecycleClient, m router.Mat return s3_lifecycle_pb.LifecycleDeleteOutcome_LIFECYCLE_DELETE_OUTCOME_UNSPECIFIED, lastErr } +// jitter returns a duration in the range [d/2, d) using equal jitter. +// Prevents thundering herds when many daily-run workers retry simultaneously. +func jitter(d time.Duration) time.Duration { + if d <= 0 { + return 0 + } + half := d / 2 + if half <= 0 { + return d + } + return half + time.Duration(rand.Int63n(int64(half))) +} + // buildDeleteRequest constructs the LifecycleDelete RPC payload for a // router Match. Mirrors dispatcher.dispatchOne's request shape — both // targets the same server-side handler and the proto encoding must diff --git a/weed/s3api/s3lifecycle/dailyrun/jitter_test.go b/weed/s3api/s3lifecycle/dailyrun/jitter_test.go new file mode 100644 index 000000000..30fbf1c35 --- /dev/null +++ b/weed/s3api/s3lifecycle/dailyrun/jitter_test.go @@ -0,0 +1,44 @@ +package dailyrun + +import ( + "testing" + "time" +) + +func TestJitterBounds(t *testing.T) { + cases := []time.Duration{ + 200 * time.Millisecond, + 1 * time.Second, + 5 * time.Second, + } + + for _, d := range cases { + for i := 0; i < 100; i++ { + j := jitter(d) + if j < d/2 { + t.Errorf("jitter(%v) = %v, below lower bound %v", d, j, d/2) + } + if j >= d { + t.Errorf("jitter(%v) = %v, at or above upper bound %v", d, j, d) + } + } + } +} + +func TestJitterZeroAndNegative(t *testing.T) { + if j := jitter(0); j != 0 { + t.Errorf("jitter(0) = %v, want 0", j) + } + if j := jitter(-1 * time.Second); j != 0 { + t.Errorf("jitter(-1s) = %v, want 0", j) + } +} + +func TestJitterTinyDuration(t *testing.T) { + // When d < 2, half == 0 and rand.Int63n(0) panics. + // We should return d unmodified in that case. + j := jitter(1 * time.Nanosecond) + if j != 1*time.Nanosecond { + t.Errorf("jitter(1ns) = %v, want 1ns", j) + } +} diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go index bff17ed0b..5c3b35089 100644 --- a/weed/wdclient/filer_client.go +++ b/weed/wdclient/filer_client.go @@ -527,6 +527,20 @@ func isRetryableGrpcError(err error) bool { strings.Contains(errStr, "unavailable") } +// jitter returns a duration in the range [d/2, d) using equal jitter. +// This prevents thundering herds when many clients retry simultaneously +// after a transient failure (e.g., network partition healing). +func jitter(d time.Duration) time.Duration { + if d <= 0 { + return 0 + } + half := d / 2 + if half <= 0 { + return d + } + return half + time.Duration(rand.Int63n(int64(half))) +} + // shouldSkipUnhealthyFiler checks if we should skip a filer based on recent failures // Circuit breaker pattern: skip filers with multiple recent consecutive failures // shouldSkipUnhealthyFilerWithHealth checks if a filer should be skipped based on health @@ -694,9 +708,16 @@ func (p *filerVolumeProvider) LookupVolumeIds(ctx context.Context, volumeIds []s // Transient error - retry if we have attempts left if retry < maxRetries-1 { + jitteredWait := jitter(waitTime) glog.V(1).Infof("FilerClient: all %d filer(s) failed with retryable error (attempt %d/%d), retrying in %v: %v", - n, retry+1, maxRetries, waitTime, lastErr) - time.Sleep(waitTime) + n, retry+1, maxRetries, jitteredWait, lastErr) + timer := time.NewTimer(jitteredWait) + select { + case <-ctx.Done(): + timer.Stop() + return nil, ctx.Err() + case <-timer.C: + } waitTime = time.Duration(float64(waitTime) * fc.retryBackoffFactor) } } diff --git a/weed/wdclient/jitter_test.go b/weed/wdclient/jitter_test.go new file mode 100644 index 000000000..139857a18 --- /dev/null +++ b/weed/wdclient/jitter_test.go @@ -0,0 +1,65 @@ +package wdclient + +import ( + "testing" + "time" +) + +func TestJitterBounds(t *testing.T) { + cases := []time.Duration{ + 1 * time.Millisecond, + 100 * time.Millisecond, + 1 * time.Second, + 5 * time.Second, + } + + for _, d := range cases { + for i := 0; i < 100; i++ { + j := jitter(d) + if j < d/2 { + t.Errorf("jitter(%v) = %v, below lower bound %v", d, j, d/2) + } + if j >= d { + t.Errorf("jitter(%v) = %v, at or above upper bound %v", d, j, d) + } + } + } +} + +func TestJitterZeroAndNegative(t *testing.T) { + if j := jitter(0); j != 0 { + t.Errorf("jitter(0) = %v, want 0", j) + } + if j := jitter(-1 * time.Second); j != 0 { + t.Errorf("jitter(-1s) = %v, want 0", j) + } +} + +func TestJitterTinyDuration(t *testing.T) { + // When d < 2, half == 0 and rand.Int63n(0) panics. + // We should return d unmodified in that case. + j := jitter(1 * time.Nanosecond) + if j != 1*time.Nanosecond { + t.Errorf("jitter(1ns) = %v, want 1ns", j) + } +} + +func TestJitterDistribution(t *testing.T) { + const iterations = 10000 + const base = 100 * time.Millisecond + var sum time.Duration + + for i := 0; i < iterations; i++ { + j := jitter(base) + sum += j + } + + avg := sum / iterations + // Equal jitter average should be around 75% of base (midpoint of [50%, 100%)) + expected := base * 3 / 4 + tolerance := base / 10 // ±10% + + if avg < expected-tolerance || avg > expected+tolerance { + t.Errorf("average jitter %v deviated from expected %v", avg, expected) + } +}