《后端架构设计·高性能与高可用》7.2 重试放大与退避抖动

amplification = Π (r_i + 1)

cap   = min(maxDelay, base × 2^retry)
delay = random(0, cap)        // full jitter

// file: internal/platform/retry/retry.go
// tier: L1
// verify: go test ./internal/platform/retry/ -count=1
package retry

import (
	"context"
	"errors"
	"math"
	rand "math/rand/v2"
	"time"
)

type Jitter int

const (
	JitterNone Jitter = iota
	JitterFull
)

// Policy describes one retry layer. MaxAttempts counts the first call: 3 means
// one original attempt plus two retries.
type Policy struct {
	MaxAttempts int
	Base        time.Duration
	Max         time.Duration
	Jitter      Jitter
	Rand        func(n int64) int64
}

func (p Policy) random(n int64) int64 {
	if n <= 1 {
		return 0
	}
	if p.Rand != nil {
		return p.Rand(n)
	}
	return rand.Int64N(n)
}

// Delay returns the wait before the given zero-based retry. Full jitter picks
// uniformly from [0, cap], which removes the synchronised retry wave.
func (p Policy) Delay(retry int) time.Duration {
	if retry < 0 {
		retry = 0
	}
	d := p.Base
	if d <= 0 {
		return 0
	}
	for i := 0; i < retry; i++ {
		if p.Max > 0 && d >= p.Max {
			break
		}
		d *= 2
	}
	if p.Max > 0 && d > p.Max {
		d = p.Max
	}
	if p.Jitter == JitterFull {
		return time.Duration(p.random(int64(d)))
	}
	return d
}

type permanentError struct{ err error }

func (e *permanentError) Error() string { return e.err.Error() }
func (e *permanentError) Unwrap() error { return e.err }

func Permanent(err error) error {
	if err == nil {
		return nil
	}
	return &permanentError{err: err}
}

func IsPermanent(err error) bool {
	var target *permanentError
	return errors.As(err, &target)
}

// Do runs fn until success, a permanent error, exhaustion or cancellation.
func Do(ctx context.Context, p Policy, fn func(context.Context, int) error) (int, error) {
	if p.MaxAttempts < 1 {
		p.MaxAttempts = 1
	}
	for attempt := 0; attempt < p.MaxAttempts; attempt++ {
		err := fn(ctx, attempt+1)
		if err == nil {
			return attempt + 1, nil
		}
		if IsPermanent(err) || attempt == p.MaxAttempts-1 {
			return attempt + 1, err
		}
		t := time.NewTimer(p.Delay(attempt))
		select {
		case <-ctx.Done():
			t.Stop()
			return attempt + 1, ctx.Err()
		case <-t.C:
		}
	}
	return p.MaxAttempts, nil
}

// Amplification is the worst-case multiplier when every layer retries every
// failure: the product of (retries + 1).
func Amplification(layers []int) int {
	total := 1
	for _, retries := range layers {
		if retries < 0 {
			retries = 0
		}
		total *= retries + 1
	}
	return total
}

// ExpectedRequests models independent per-layer failure: a layer with r
// retries contributes 1 + p + ... + p^r attempts for one inbound request.
func ExpectedRequests(failureRate float64, layers []int) float64 {
	if failureRate < 0 {
		failureRate = 0
	}
	if failureRate >= 1 {
		failureRate = 1
	}
	total := 1.0
	for _, retries := range layers {
		factor := 0.0
		for i := 0; i <= retries; i++ {
			factor += math.Pow(failureRate, float64(i))
		}
		total *= factor
	}
	return total
}

// file: internal/platform/retry/retry_test.go
// tier: L1
// verify: go test -v ./internal/platform/retry/ -count=1
package retry

import (
	"context"
	"errors"
	"math"
	rand "math/rand/v2"
	"testing"
	"time"
)

func TestBackoffCapsAndJitterBounds(t *testing.T) {
	p := Policy{Base: 100 * time.Millisecond, Max: time.Second}
	for i, want := range []time.Duration{100, 200, 400, 800, 1000, 1000} {
		if got := p.Delay(i); got != want*time.Millisecond {
			t.Fatalf("retry %d: got=%s want=%s", i, got, want*time.Millisecond)
		}
	}
	full := Policy{Base: 100 * time.Millisecond, Max: time.Second, Jitter: JitterFull,
		Rand: func(n int64) int64 { return n - 1 }}
	for retry := 0; retry < 5; retry++ {
		cap := p.Delay(retry)
		if got := full.Delay(retry); got < cap-time.Millisecond || got > cap {
			t.Fatalf("full jitter retry %d: got=%s cap=%s", retry, got, cap)
		}
	}
}

func TestAmplificationAndExpectedLoad(t *testing.T) {
	layers := []int{1, 2, 2}
	want := 1.5 * 1.75 * 1.75
	if got := Amplification(layers); got != 18 {
		t.Fatalf("amplification=%d want 18", got)
	}
	if got := ExpectedRequests(0.5, layers); math.Abs(got-want) > 0.001 {
		t.Fatalf("expected requests=%.5f want %.5f", got, want)
	}
	t.Logf("layers=%v worst_case=%dx expected_at_p50=%.3fx", layers, Amplification(layers), want)
}

func TestPermanentErrorStopsRetries(t *testing.T) {
	calls := 0
	n, err := Do(context.Background(), Policy{MaxAttempts: 4, Base: time.Millisecond},
		func(context.Context, int) error {
			calls++
			return Permanent(errors.New("OUT_OF_STOCK"))
		})
	if !IsPermanent(err) || n != 1 || calls != 1 {
		t.Fatalf("calls=%d n=%d err=%v", calls, n, err)
	}
}

func splitmix64(x uint64) uint64 {
	x += 0x9E3779B97F4A7C15
	x = (x ^ (x >> 30)) * 0xBF58476D1CE4E5B9
	x = (x ^ (x >> 27)) * 0x94D049BB133111EB
	return x ^ (x >> 31)
}

func failed(seed uint64, id, attempt int, rate float64) bool {
	h := splitmix64(seed + uint64(id)*0x9E3779B97F4A7C15 + uint64(attempt+1))
	return float64(h%100000)/100000.0 < rate
}

// simulate advances a virtual clock in 1ms buckets. It counts retry attempts,
// excluding the initial call, so the peak measures the retry wave rather than
// baseline traffic.
func simulate(requests int, rate float64, p Policy, seed uint64) (total, retryPeak int, peakAt int64) {
	buckets := map[int64]int{}
	for id := 0; id < requests; id++ {
		attempt := 0
		at := int64(0)
		for {
			total++
			if attempt > 0 {
				buckets[at]++
				if buckets[at] > retryPeak {
					retryPeak, peakAt = buckets[at], at
				}
			}
			if !failed(seed, id, attempt, rate) || attempt >= p.MaxAttempts-1 {
				break
			}
			at += p.Delay(attempt).Milliseconds()
			attempt++
		}
	}
	return total, retryPeak, peakAt
}

func TestFullJitterReducesRetryPeak(t *testing.T) {
	const requests = 1000
	const rate = 0.5
	none := Policy{MaxAttempts: 3, Base: 100 * time.Millisecond, Max: time.Second}
	r := rand.New(rand.NewPCG(42, 7))
	full := Policy{MaxAttempts: 3, Base: 100 * time.Millisecond, Max: time.Second,
		Jitter: JitterFull, Rand: r.Int64N}
	totalNone, peakNone, atNone := simulate(requests, rate, none, 42)
	totalFull, peakFull, atFull := simulate(requests, rate, full, 42)
	t.Logf("requests=%d failure_rate=%.1f no_jitter: total=%d amp=%.3fx retry_peak=%d@%dms full_jitter: total=%d amp=%.3fx retry_peak=%d@%dms",
		requests, rate, totalNone, float64(totalNone)/requests, peakNone, atNone,
		totalFull, float64(totalFull)/requests, peakFull, atFull)
	if totalNone != totalFull {
		t.Fatalf("failure sequence changed: %d vs %d", totalNone, totalFull)
	}
	if peakFull*5 > peakNone {
		t.Fatalf("full jitter peak=%d did not spread no-jitter peak=%d", peakFull, peakNone)
	}
}

=== RUN   TestAmplificationAndExpectedLoad
    retry_test.go:38: layers=[1 2 2] worst_case=18x expected_at_p50=4.594x
--- PASS: TestAmplificationAndExpectedLoad (0.00s)
=== RUN   TestFullJitterReducesRetryPeak
    retry_test.go:100: requests=1000 failure_rate=0.5 no_jitter: total=1737 amp=1.737x retry_peak=495@100ms full_jitter: total=1737 amp=1.737x retry_peak=11@84ms
--- PASS: TestFullJitterReducesRetryPeak (0.00s)
PASS
ok  	archlab/internal/platform/retry	0.234s