《后端架构设计·演进与治理》4.3 网关失败模式：超时预算、连接池与级联故障

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

import (
	"context"
	"errors"
	"fmt"
	"sync/atomic"
	"time"
)

var ErrBudgetExhausted = errors.New("gateway: timeout budget exhausted")

// Budget is an absolute deadline, so a child cannot silently outlive its
// parent.
type Budget struct{ deadline time.Time }

func NewBudget(now time.Time, total time.Duration) (Budget, error) {
	if total <= 0 {
		return Budget{}, fmt.Errorf("gateway: budget must be positive, got %v", total)
	}
	return Budget{deadline: now.Add(total)}, nil
}
func (b Budget) Deadline() time.Time { return b.deadline }
func (b Budget) Remaining(now time.Time) time.Duration {
	if d := b.deadline.Sub(now); d > 0 {
		return d
	}
	return 0
}

// Child divides the remaining budget into parts and keeps reserve for the
// gateway's own response path.
func (b Budget) Child(now time.Time, parts int, reserve time.Duration) (Budget, time.Duration, error) {
	if parts < 1 || reserve < 0 {
		return Budget{}, 0, errors.New("gateway: parts must be positive and reserve non-negative")
	}
	rem := b.Remaining(now)
	if rem <= reserve {
		return Budget{}, 0, ErrBudgetExhausted
	}
	share := (rem - reserve) / time.Duration(parts)
	if share <= 0 {
		return Budget{}, 0, ErrBudgetExhausted
	}
	return Budget{deadline: now.Add(share)}, share, nil
}

// Pool is a fixed-size connection pool. Acquire waits for a slot or for ctx
// cancellation, whichever comes first.
type Pool struct {
	sem   chan struct{}
	inUse atomic.Int64
	peak  atomic.Int64
}

func NewPool(size int) (*Pool, error) {
	if size < 1 {
		return nil, fmt.Errorf("gateway: pool size must be at least 1, got %d", size)
	}
	return &Pool{sem: make(chan struct{}, size)}, nil
}

// Acquire takes one slot. A canceled context returns ctx.Err() without
// consuming a slot.
func (p *Pool) Acquire(ctx context.Context) error {
	select {
	case p.sem <- struct{}{}:
		n := p.inUse.Add(1)
		for {
			peak := p.peak.Load()
			if n <= peak || p.peak.CompareAndSwap(peak, n) {
				return nil
			}
		}
	case <-ctx.Done():
		return ctx.Err()
	}
}

// Release returns one slot. Releasing without Acquire panics so a double
// release cannot quietly create capacity.
func (p *Pool) Release() {
	select {
	case <-p.sem:
		p.inUse.Add(-1)
	default:
		panic("gateway: Release without Acquire")
	}
}
func (p *Pool) Stats() (inUse, peak, size int) {
	return int(p.inUse.Load()), int(p.peak.Load()), cap(p.sem)
}

type CascadeConfig struct {
	Requests      int
	PoolSize      int
	ArrivalEvery  time.Duration
	ServiceTime   time.Duration
	ClientTimeout time.Duration
	MaxAttempts   int
}
type CascadeReport struct {
	Requests  int
	Succeeded int
	Failed    int
	Attempts  int
	TimedOut  int
	MaxWait   time.Duration
	BusyFor   time.Duration
}

// SimulateCascade runs a deterministic model: requests arrive evenly, take the
// earliest free slot, and time out if service cannot finish before their
// attempt deadline. A timed-out service still occupies the slot until it
// finishes, which is what turns a slow dependency into pool exhaustion.
func SimulateCascade(cfg CascadeConfig) (CascadeReport, error) {
	if cfg.Requests < 1 || cfg.PoolSize < 1 || cfg.ArrivalEvery < 0 ||
		cfg.ServiceTime <= 0 || cfg.ClientTimeout <= 0 || cfg.MaxAttempts < 1 {
		return CascadeReport{}, fmt.Errorf("gateway: invalid cascade config %+v", cfg)
	}
	free := make([]time.Duration, cfg.PoolSize)
	rep := CascadeReport{Requests: cfg.Requests}
	for i := 0; i < cfg.Requests; i++ {
		arrival := time.Duration(i) * cfg.ArrivalEvery
		begin := arrival
		succeeded := false
		for attempt := 0; attempt < cfg.MaxAttempts; attempt++ {
			rep.Attempts++
			deadline := begin + cfg.ClientTimeout
			slot := 0
			for j := 1; j < len(free); j++ {
				if free[j] < free[slot] {
					slot = j
				}
			}
			start := free[slot]
			if start < begin {
				start = begin
			}
			if start >= deadline {
				if wait := deadline - begin; wait > rep.MaxWait {
					rep.MaxWait = wait
				}
				rep.TimedOut++
				begin = deadline
				continue
			}
			if wait := start - begin; wait > rep.MaxWait {
				rep.MaxWait = wait
			}
			finish := start + cfg.ServiceTime
			free[slot] = finish
			rep.BusyFor += cfg.ServiceTime
			if finish <= deadline {
				succeeded = true
				break
			}
			rep.TimedOut++
			begin = deadline
		}
		if succeeded {
			rep.Succeeded++
		} else {
			rep.Failed++
		}
	}
	return rep, nil
}

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

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

func TestBudgetPropagation(t *testing.T) {
	now := time.Date(2026, 10, 28, 17, 50, 0, 0, time.UTC)
	b, err := NewBudget(now, 300*time.Millisecond)
	if err != nil {
		t.Fatalf("NewBudget: %v", err)
	}
	if got := b.Remaining(now.Add(40 * time.Millisecond)); got != 260*time.Millisecond {
		t.Fatalf("remaining=%v want 260ms", got)
	}
	child, share, err := b.Child(now.Add(40*time.Millisecond), 3, 20*time.Millisecond)
	if err != nil {
		t.Fatalf("Child: %v", err)
	}
	if share != 80*time.Millisecond || !child.Deadline().Equal(now.Add(120*time.Millisecond)) {
		t.Fatalf("share=%v deadline=%v want 80ms/now+120ms", share, child.Deadline())
	}
	if got := child.Remaining(now.Add(100 * time.Millisecond)); got != 20*time.Millisecond {
		t.Fatalf("child remaining=%v want 20ms", got)
	}
	if _, _, err := child.Child(now.Add(120*time.Millisecond), 1, 0); !errors.Is(err, ErrBudgetExhausted) {
		t.Fatalf("exhausted child err=%v want ErrBudgetExhausted", err)
	}
}

func TestPoolExhaustion(t *testing.T) {
	p, err := NewPool(2)
	if err != nil {
		t.Fatalf("NewPool: %v", err)
	}
	for i := 0; i < 2; i++ {
		if err := p.Acquire(context.Background()); err != nil {
			t.Fatalf("acquire %d: %v", i+1, err)
		}
	}
	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Millisecond)
	defer cancel()
	if err := p.Acquire(ctx); !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("third acquire err=%v want DeadlineExceeded", err)
	}
	if inUse, peak, size := p.Stats(); inUse != 2 || peak != 2 || size != 2 {
		t.Fatalf("stats=%d/%d/%d want 2/2/2", inUse, peak, size)
	}
	p.Release()
	if err := p.Acquire(context.Background()); err != nil {
		t.Fatalf("acquire after release: %v", err)
	}
	p.Release()
	p.Release()
	if inUse, _, _ := p.Stats(); inUse != 0 {
		t.Fatalf("inUse=%d after releases want 0", inUse)
	}
}

func TestCascadeHealthy(t *testing.T) {
	rep, err := SimulateCascade(CascadeConfig{Requests: 20, PoolSize: 8,
		ArrivalEvery: 10 * time.Millisecond, ServiceTime: 20 * time.Millisecond,
		ClientTimeout: 100 * time.Millisecond, MaxAttempts: 1})
	if err != nil {
		t.Fatalf("SimulateCascade: %v", err)
	}
	t.Logf("healthy: succeeded=%d failed=%d attempts=%d timedOut=%d maxWait=%v", rep.Succeeded, rep.Failed, rep.Attempts, rep.TimedOut, rep.MaxWait)
	if rep.Succeeded != 20 || rep.Failed != 0 || rep.TimedOut != 0 || rep.Attempts != 20 {
		t.Fatalf("healthy report=%+v", rep)
	}
}

func TestCascadePoolStarvation(t *testing.T) {
	rep, err := SimulateCascade(CascadeConfig{Requests: 8, PoolSize: 1,
		ArrivalEvery: 5 * time.Millisecond, ServiceTime: 40 * time.Millisecond,
		ClientTimeout: 30 * time.Millisecond, MaxAttempts: 1})
	if err != nil {
		t.Fatalf("SimulateCascade: %v", err)
	}
	t.Logf("starvation: succeeded=%d failed=%d attempts=%d timedOut=%d maxWait=%v", rep.Succeeded, rep.Failed, rep.Attempts, rep.TimedOut, rep.MaxWait)
	if rep.Succeeded != 0 || rep.Failed != 8 || rep.TimedOut != 8 || rep.MaxWait != 30*time.Millisecond {
		t.Fatalf("starvation report=%+v", rep)
	}
}

func TestCascadeRetryAmplifies(t *testing.T) {
	base := CascadeConfig{Requests: 8, PoolSize: 1, ArrivalEvery: 5 * time.Millisecond,
		ServiceTime: 40 * time.Millisecond, ClientTimeout: 30 * time.Millisecond, MaxAttempts: 1}
	single, err := SimulateCascade(base)
	if err != nil {
		t.Fatalf("single attempt: %v", err)
	}
	base.MaxAttempts = 3
	retry, err := SimulateCascade(base)
	if err != nil {
		t.Fatalf("retry attempts: %v", err)
	}
	t.Logf("single: attempts=%d timedOut=%d busy=%v", single.Attempts, single.TimedOut, single.BusyFor)
	t.Logf("retry: attempts=%d timedOut=%d busy=%v", retry.Attempts, retry.TimedOut, retry.BusyFor)
	if retry.Attempts <= single.Attempts || retry.BusyFor < single.BusyFor || retry.Failed != 8 {
		t.Fatalf("single=%+v retry=%+v: retries must add attempts but not failures", single, retry)
	}
}

$ go test ./internal/platform/gateway/ -run 'TestBudget|TestPool|TestCascade' -count=1 -v
=== RUN   TestBudgetPropagation
--- PASS: TestBudgetPropagation (0.00s)
=== RUN   TestPoolExhaustion
--- PASS: TestPoolExhaustion (0.02s)
=== RUN   TestCascadeHealthy
    cascade_test.go:73: healthy: succeeded=20 failed=0 attempts=20 timedOut=0 maxWait=0s
--- PASS: TestCascadeHealthy (0.00s)
=== RUN   TestCascadePoolStarvation
    cascade_test.go:86: starvation: succeeded=0 failed=8 attempts=8 timedOut=8 maxWait=30ms
--- PASS: TestCascadePoolStarvation (0.00s)
=== RUN   TestCascadeRetryAmplifies
    cascade_test.go:104: single: attempts=8 timedOut=8 busy=80ms
    cascade_test.go:105: retry: attempts=24 timedOut=24 busy=160ms
--- PASS: TestCascadeRetryAmplifies (0.00s)
PASS
ok  	archlab/internal/platform/gateway	0.254s