《互联网产品的服务端架构·产品、流量与基础能力》9.3 重复、丢失与补偿的用户感知

// file: internal/async/retry.go
// tier: L1
// verify: go test ./internal/async/

package async

import (
	"fmt"
	"sort"
	"time"
)

// Failure 是一条投递失败记录。Attempts 是已经尝试过的次数。
type Failure struct {
	MsgID    string
	Attempts int
}

// Policy 是重试上限和退避参数。Base 是第一次重试的等待时间。
type Policy struct {
	MaxAttempts int
	Base        time.Duration
	Max         time.Duration
}

// Next 计算下一次重试前的等待时间。
// 指数退避以 Base 起步，翻倍增长但不超过 Max。
// 已经达到 MaxAttempts 时返回 shouldRetry=false，由调用方转入死信。
func Next(f Failure, p Policy) (time.Duration, bool, error) {
	if p.Base <= 0 {
		return 0, false, fmt.Errorf("async: retry base must be positive")
	}
	if p.Max < p.Base {
		return 0, false, fmt.Errorf("async: retry max %s below base %s", p.Max, p.Base)
	}
	if f.Attempts >= p.MaxAttempts {
		return 0, false, nil
	}
	delay := p.Base
	for i := 1; i < f.Attempts; i++ {
		if delay >= p.Max/2 {
			delay = p.Max
			break
		}
		delay *= 2
	}
	if delay > p.Max {
		delay = p.Max
	}
	return delay, true, nil
}

// Reconcile 对账：返回 expected 中存在、actual 中缺失的消息 ID，按升序排列。
// 重复的期望值只算一次；输入不会被修改。
func Reconcile(expected, actual []string) []string {
	seen := make(map[string]bool, len(actual))
	for _, id := range actual {
		seen[id] = true
	}
	missing := make(map[string]bool)
	for _, id := range expected {
		if !seen[id] {
			missing[id] = true
		}
	}
	out := make([]string, 0, len(missing))
	for id := range missing {
		out = append(out, id)
	}
	sort.Strings(out)
	return out
}

// file: internal/async/retry_test.go
// tier: L1
// verify: go test ./internal/async/

package async

import (
	"reflect"
	"testing"
	"time"
)

func TestNextBacksOffExponentially(t *testing.T) {
	p := Policy{MaxAttempts: 5, Base: 100 * time.Millisecond, Max: time.Second}
	want := []time.Duration{
		100 * time.Millisecond,
		200 * time.Millisecond,
		400 * time.Millisecond,
		800 * time.Millisecond,
	}
	for i, w := range want {
		got, retry, err := Next(Failure{MsgID: "m-1", Attempts: i + 1}, p)
		if err != nil {
			t.Fatal(err)
		}
		if !retry {
			t.Fatalf("attempt %d: retry = false", i+1)
		}
		if got != w {
			t.Fatalf("attempt %d: delay = %s, want %s", i+1, got, w)
		}
	}
}

func TestNextCapsAtMax(t *testing.T) {
	p := Policy{MaxAttempts: 10, Base: 100 * time.Millisecond, Max: 250 * time.Millisecond}
	want := []time.Duration{
		100 * time.Millisecond,
		200 * time.Millisecond,
		250 * time.Millisecond,
		250 * time.Millisecond,
	}
	for i, w := range want {
		got, retry, err := Next(Failure{MsgID: "m-2", Attempts: i + 1}, p)
		if err != nil || !retry {
			t.Fatalf("attempt %d: retry=%v err=%v", i+1, retry, err)
		}
		if got != w {
			t.Fatalf("attempt %d: delay = %s, want %s", i+1, got, w)
		}
	}
}

func TestNextStopsAtMaxAttempts(t *testing.T) {
	p := Policy{MaxAttempts: 3, Base: time.Second, Max: time.Minute}
	delay, retry, err := Next(Failure{MsgID: "m-3", Attempts: 3}, p)
	if err != nil {
		t.Fatal(err)
	}
	if retry {
		t.Fatal("exhausted failure must not retry")
	}
	if delay != 0 {
		t.Fatalf("delay = %s, want 0", delay)
	}
}

func TestNextRejectsInvalidPolicy(t *testing.T) {
	if _, _, err := Next(Failure{}, Policy{MaxAttempts: 3, Base: 0, Max: time.Second}); err == nil {
		t.Fatal("want error for zero base")
	}
	if _, _, err := Next(Failure{}, Policy{MaxAttempts: 3, Base: -time.Second, Max: time.Second}); err == nil {
		t.Fatal("want error for negative base")
	}
	if _, _, err := Next(Failure{}, Policy{MaxAttempts: 3, Base: time.Second, Max: 500 * time.Millisecond}); err == nil {
		t.Fatal("want error when max is below base")
	}
}

func TestReconcileReturnsSortedMissing(t *testing.T) {
	expected := []string{"c", "a", "b", "a"}
	actual := []string{"b", "d"}
	want := []string{"a", "c"}

	got := Reconcile(expected, actual)
	if !reflect.DeepEqual(got, want) {
		t.Fatalf("missing = %v, want %v", got, want)
	}
	if !reflect.DeepEqual(expected, []string{"c", "a", "b", "a"}) {
		t.Fatalf("expected was mutated: %v", expected)
	}
	if !reflect.DeepEqual(actual, []string{"b", "d"}) {
		t.Fatalf("actual was mutated: %v", actual)
	}
}

func TestReconcileEmptyWhenNothingMissing(t *testing.T) {
	got := Reconcile([]string{"a", "b"}, []string{"b", "a", "c"})
	if len(got) != 0 {
		t.Fatalf("missing = %v, want empty", got)
	}
	if got == nil {
		t.Fatal("want a non-nil empty slice")
	}
}

cd .local/checks/webapp-lab && go test ./internal/async/ -count=1 -run 'TestNext|TestReconcile' -v

=== RUN   TestNextBacksOffExponentially
--- PASS: TestNextBacksOffExponentially (0.00s)
=== RUN   TestNextCapsAtMax
--- PASS: TestNextCapsAtMax (0.00s)
=== RUN   TestNextStopsAtMaxAttempts
--- PASS: TestNextStopsAtMaxAttempts (0.00s)
=== RUN   TestNextRejectsInvalidPolicy
--- PASS: TestNextRejectsInvalidPolicy (0.00s)
=== RUN   TestReconcileReturnsSortedMissing
--- PASS: TestReconcileReturnsSortedMissing (0.00s)
=== RUN   TestReconcileEmptyWhenNothingMissing
--- PASS: TestReconcileEmptyWhenNothingMissing (0.00s)
PASS
ok  	webapp/internal/async	0.232s