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