《MMO 服务器开发》3.2 重试、熔断与背压

// file: internal/rpc/retry.go
// tier: L1
// verify: go test ./internal/rpc/ -run TestRetry
package rpc

import (
	"context"
	"errors"
	"time"
)

// ErrNonIdempotent 表示调用没有幂等保证，按纪律不允许重试。
var ErrNonIdempotent = errors.New("rpc: non-idempotent call must not be retried")

// RetryPolicy 定义重试预算：最多几次、间隔多久、什么错误才重试。
type RetryPolicy struct {
	MaxAttempts int
	Base        time.Duration
	Retryable   func(error) bool
}

// DefaultRetryable 只重试「可能没送达」的错误（超时 / 连接类）；
// 明确的业务错误（如「道具不足」）绝不重试。
func DefaultRetryable(err error) bool {
	return errors.Is(err, context.DeadlineExceeded)
}

// Call 在幂等前提下按策略重试。idempotent=false 时直接拒绝，
// 因为「超时了到底成没成功」在没有幂等键时根本无法判定。
func (p RetryPolicy) Call(ctx context.Context, idempotent bool, fn func(context.Context) error) error {
	if !idempotent {
		return ErrNonIdempotent
	}
	var err error
	for attempt := 1; attempt <= p.MaxAttempts; attempt++ {
		err = fn(ctx)
		if err == nil {
			return nil
		}
		if p.Retryable != nil && !p.Retryable(err) {
			return err
		}
		if attempt == p.MaxAttempts {
			break
		}
		wait := p.Base << (attempt - 1) // 指数退避：Base * 2^(attempt-1)
		select {
		case <-time.After(wait):
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return err
}

// file: internal/rpc/breaker.go
// tier: L1
// verify: go test ./internal/rpc/ -run TestBreaker
package rpc

import (
	"errors"
	"sync"
	"time"
)

// ErrOpen 表示熔断器已打开，调用被快速失败（不消耗下游资源）。
var ErrOpen = errors.New("rpc: circuit open")

type State int

const (
	StateClosed State = iota
	StateOpen
	StateHalfOpen
)

// Breaker 三态熔断器：closed -> open -> half-open -> closed。
type Breaker struct {
	mu           sync.Mutex
	state        State
	failures     int
	successes    int
	threshold    int           // 连续失败多少次要打开
	openFor      time.Duration // 打开后多久进入 half-open
	halfOpenNeed int           // half-open 下连续成功多少次才闭合
	openedAt     time.Time
	now          func() time.Time
}

func NewBreaker(threshold int, openFor time.Duration, halfOpenNeed int) *Breaker {
	return &Breaker{threshold: threshold, openFor: openFor, halfOpenNeed: halfOpenNeed, now: time.Now}
}

// check 需持锁：冷却期结束就进入 half-open，放一次探测。
func (b *Breaker) check() {
	if b.state == StateOpen && b.now().Sub(b.openedAt) >= b.openFor {
		b.state, b.successes = StateHalfOpen, 0
	}
}

// Allow 返回当前是否放行一次调用。
func (b *Breaker) Allow() error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.check()
	if b.state == StateOpen {
		return ErrOpen
	}
	return nil
}

func (b *Breaker) State() State { b.mu.Lock(); defer b.mu.Unlock(); b.check(); return b.state }

// record 记录一次调用结果。
func (b *Breaker) record(ok bool) {
	b.mu.Lock()
	defer b.mu.Unlock()
	if !ok {
		if b.state == StateHalfOpen { // 半开再失败，立刻退回 open
			b.state, b.openedAt = StateOpen, b.now()
			return
		}
		if b.failures++; b.failures >= b.threshold {
			b.state, b.openedAt = StateOpen, b.now()
		}
		return
	}
	if b.state == StateHalfOpen {
		if b.successes++; b.successes >= b.halfOpenNeed {
			b.state, b.failures = StateClosed, 0
		}
		return
	}
	b.failures = 0
}

func (b *Breaker) OnSuccess() { b.record(true) }
func (b *Breaker) OnFailure() { b.record(false) }

// snippet: internal/rpc/ratelimit.go
// 令牌桶：rate 个/秒，最多攒 burst 个。now/last 可注入，便于确定性测试。
func (t *TokenBucket) Allow() bool {
	t.mu.Lock()
	defer t.mu.Unlock()
	now := t.now()
	t.tokens += now.Sub(t.last).Seconds() * t.rate // 按流逝时间补令牌
	if t.tokens > t.burst {
		t.tokens = t.burst // 攒不过 burst
	}
	t.last = now
	if t.tokens >= 1 {
		t.tokens--
		return true
	}
	return false
}

// 有界队列：队列满时直接拒绝新任务，而不是把内存和延迟一起拖爆。
func (q *BoundedQueue) Submit(task func()) error {
	select {
	case q.ch <- task:
		return nil
	default:
		return ErrRejected // 过载暴露成快速失败
	}
}

// file: internal/rpc/loadtest_test.go
// tier: B
// verify: go test ./internal/rpc/ -run TestBreakerLoad -v
package rpc

import (
	"context"
	"sync/atomic"
	"testing"
	"time"
)

// TestBreakerLoad 实测「熔断打开前后」上游与下游的差异。
// 下游固定慢 40ms，上游每次预算 20ms —— 必然超时。
func TestBreakerLoad(t *testing.T) {
	var downstream atomic.Int64
	addr, stop := startPing(t, func(ctx context.Context, in []byte) ([]byte, error) {
		downstream.Add(1)
		time.Sleep(40 * time.Millisecond)
		f, _ := Decode(in)
		return Frame{Cmd: f.Cmd + 1, Seq: f.Seq, Payload: f.Payload}.Encode()
	})
	defer stop()
	cc := dialPing(t, addr)
	defer cc.Close()
	req, _ := Frame{Cmd: 0x0E01, Seq: 1, Payload: []byte("x")}.Encode()

	const n = 100
	call := func(ctx context.Context) error {
		var out []byte
		return cc.Invoke(ctx, "/aegis.Ping/Call", req, &out)
	}

	// A. 无熔断：每个请求都在 20ms 预算里超时，且全部打到下游
	downstream.Store(0)
	aFail := 0
	t0 := time.Now()
	for i := 0; i < n; i++ {
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
		if call(ctx) != nil {
			aFail++
		}
		cancel()
	}
	da, downA := time.Since(t0), downstream.Load()

	// B. 有熔断：连续失败 5 次即打开，之后快速失败，不再打扰下游
	br := NewBreaker(5, time.Minute, 1)
	downstream.Store(0)
	bFail, bFast := 0, 0
	t1 := time.Now()
	for i := 0; i < n; i++ {
		if br.Allow() != nil {
			bFail++
			bFast++
			continue
		}
		ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
		err := call(ctx)
		cancel()
		if err != nil {
			br.OnFailure()
			bFail++
		} else {
			br.OnSuccess()
		}
	}
	db, downB := time.Since(t1), downstream.Load()

	t.Logf("无熔断: %d 次, 失败 %d, 下游收到 %d, 耗时 %s", n, aFail, downA, da)
	t.Logf("有熔断: %d 次, 失败 %d(快速失败 %d), 下游收到 %d, 耗时 %s", n, bFail, bFast, downB, db)
	if downB >= downA {
		t.Fatalf("熔断未减少下游压力: A=%d B=%d", downA, downB)
	}
}

无熔断: 100 次, 失败 100, 下游收到 100, 耗时 2.182469083s
有熔断: 100 次, 失败 100(快速失败 95), 下游收到 5, 耗时 107.879083ms