《后端架构设计·高性能与高可用》5.3 慢调用与故障扩散

// file: internal/platform/slowcall/guard.go
// tier: L1
// verify: go test ./internal/platform/slowcall/
package slowcall

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

var (
	ErrTimeout  = errors.New("slowcall: deadline exceeded")
	ErrCanceled = errors.New("slowcall: canceled by caller")
)

type Config struct {
	Budget        time.Duration
	SlowThreshold time.Duration
	MaxSamples    int
}

type Stats struct {
	Calls, Completed, Timeouts, Canceled, Slow int64
	InFlight, MaxInFlight                      int64
}

// Guard bounds one downstream call by the caller's deadline and classifies
// timeout, cancellation, and slow-but-successful calls separately.
type Guard struct {
	cfg         Config
	mu          sync.Mutex
	samples     []time.Duration
	next        int
	calls       atomic.Int64
	completed   atomic.Int64
	timeouts    atomic.Int64
	canceled    atomic.Int64
	slow        atomic.Int64
	inFlight    atomic.Int64
	maxInFlight atomic.Int64
}

func New(cfg Config) *Guard {
	if cfg.Budget <= 0 {
		cfg.Budget = time.Second
	}
	if cfg.SlowThreshold <= 0 {
		cfg.SlowThreshold = cfg.Budget / 2
	}
	if cfg.MaxSamples < 1 {
		cfg.MaxSamples = 1024
	}
	return &Guard{cfg: cfg}
}

// Do can only tighten the parent: WithTimeout keeps the earlier deadline.
func (g *Guard) Do(parent context.Context, fn func(context.Context) error) error {
	ctx, cancel := context.WithTimeout(parent, g.cfg.Budget)
	defer cancel()
	g.calls.Add(1)
	bump(&g.maxInFlight, g.inFlight.Add(1))
	start := time.Now()
	err := fn(ctx)
	g.inFlight.Add(-1)
	g.record(time.Since(start))

	switch {
	case parent.Err() == context.Canceled:
		g.canceled.Add(1)
		return ErrCanceled
	case ctx.Err() == context.DeadlineExceeded:
		g.timeouts.Add(1)
		return ErrTimeout
	case err != nil:
		return err
	}
	g.completed.Add(1)
	return nil
}

// A bounded ring keeps memory flat; P99 is a window, not all history.
func (g *Guard) record(d time.Duration) {
	if d >= g.cfg.SlowThreshold {
		g.slow.Add(1)
	}
	g.mu.Lock()
	if len(g.samples) < g.cfg.MaxSamples {
		g.samples = append(g.samples, d)
	} else {
		g.samples[g.next] = d
		g.next = (g.next + 1) % len(g.samples)
	}
	g.mu.Unlock()
}

func (g *Guard) Stats() Stats {
	return Stats{
		Calls: g.calls.Load(), Completed: g.completed.Load(),
		Timeouts: g.timeouts.Load(), Canceled: g.canceled.Load(),
		Slow: g.slow.Load(), InFlight: g.inFlight.Load(),
		MaxInFlight: g.maxInFlight.Load(),
	}
}

// P99 is per-instance and per-window; never average it across instances.
func (g *Guard) P99() time.Duration {
	g.mu.Lock()
	values := append([]time.Duration(nil), g.samples...)
	g.mu.Unlock()
	if len(values) == 0 {
		return 0
	}
	sort.Slice(values, func(i, j int) bool { return values[i] < values[j] })
	return values[len(values)*99/100]
}

func bump(dst *atomic.Int64, v int64) {
	for cur := dst.Load(); v > cur; cur = dst.Load() {
		if dst.CompareAndSwap(cur, v) {
			return
		}
	}
}

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

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

func TestGuardClassifies(t *testing.T) {
	g := New(Config{Budget: 80 * time.Millisecond, SlowThreshold: 50 * time.Millisecond})
	_ = g.Do(context.Background(), func(context.Context) error { return nil })
	timeout := func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }
	if err := g.Do(context.Background(), timeout); !errors.Is(err, ErrTimeout) {
		t.Fatal(err)
	}
	parent, cancel := context.WithCancel(context.Background())
	cancel()
	if err := g.Do(parent, timeout); !errors.Is(err, ErrCanceled) {
		t.Fatal(err)
	}
	if s := g.Stats(); s.Calls != 3 || s.Completed != 1 || s.Timeouts != 1 || s.Canceled != 1 || s.Slow != 1 {
		t.Fatalf("stats = %+v", s)
	}
	t.Logf("calls=3 completed=1 timeouts=1 canceled=1 slow=1 p99=%s", g.P99().Round(time.Millisecond))
}

type downstream struct {
	latency  time.Duration
	active   atomic.Int64
	peak     atomic.Int64
	calls    atomic.Int64
	canceled atomic.Int64
}

func (d *downstream) Call(ctx context.Context) error {
	d.calls.Add(1)
	bump(&d.peak, d.active.Add(1))
	defer d.active.Add(-1)
	select {
	case <-time.After(d.latency):
		return nil
	case <-ctx.Done():
		d.canceled.Add(1)
		return ctx.Err()
	}
}

type request struct {
	ctx  context.Context
	done chan error
}

type service struct {
	queue      chan *request
	down       *downstream
	guard      *Guard
	inFlight   atomic.Int64
	peakFlight atomic.Int64
	peakQueue  atomic.Int64
	blocked    atomic.Int64
}

func newService(workers, queue int, d *downstream, g *Guard) *service {
	s := &service{queue: make(chan *request, queue), down: d, guard: g}
	for i := 0; i < workers; i++ {
		go func() {
			for req := range s.queue {
				bump(&s.peakFlight, s.inFlight.Add(1))
				var err error
				if s.guard != nil {
					err = s.guard.Do(req.ctx, s.down.Call)
				} else {
					err = s.down.Call(context.Background())
				}
				s.inFlight.Add(-1)
				req.done <- err
			}
		}()
	}
	return s
}

func (s *service) handle(ctx context.Context) error {
	req := &request{ctx: ctx, done: make(chan error, 1)}
	select {
	case s.queue <- req:
		bump(&s.peakQueue, int64(len(s.queue)))
	case <-ctx.Done():
		s.blocked.Add(1)
		return ctx.Err()
	}
	select {
	case err := <-req.done:
		return err
	case <-ctx.Done():
		return ctx.Err()
	}
}

type report struct {
	svcTimeout, gateway, peakFlight, peakQueue, blocked int64
	downPeak, downCanceled                              int64
}

func runVariant(t *testing.T, fixed bool, budget time.Duration) report {
	t.Helper()
	d := &downstream{latency: 300 * time.Millisecond}
	var g *Guard
	if fixed {
		g = New(Config{Budget: budget, SlowThreshold: budget})
	}
	s := newService(16, 32, d, g)

	var svcTimeout, gateway atomic.Int64
	var wg sync.WaitGroup
	for i := 0; i < 240; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
			defer cancel()
			switch err := s.handle(ctx); {
			case errors.Is(err, ErrTimeout):
				svcTimeout.Add(1)
			case errors.Is(err, context.DeadlineExceeded):
				gateway.Add(1)
			}
		}()
		time.Sleep(7 * time.Millisecond)
	}
	wg.Wait()
	time.Sleep(50 * time.Millisecond)
	return report{
		svcTimeout: svcTimeout.Load(), gateway: gateway.Load(),
		peakFlight: s.peakFlight.Load(), peakQueue: s.peakQueue.Load(), blocked: s.blocked.Load(),
		downPeak: d.peak.Load(), downCanceled: d.canceled.Load(),
	}
}

func TestFailureSpreadBeforeAfter(t *testing.T) {
	before := runVariant(t, false, 0)
	after := runVariant(t, true, 70*time.Millisecond)

	t.Logf("before svc_timeout=%d gateway_timeout=%d peak_flight=%d peak_queue=%d submit_blocked=%d down_peak=%d down_canceled=%d",
		before.svcTimeout, before.gateway, before.peakFlight, before.peakQueue,
		before.blocked, before.downPeak, before.downCanceled)
	t.Logf("after  svc_timeout=%d gateway_timeout=%d peak_flight=%d peak_queue=%d submit_blocked=%d down_peak=%d down_canceled=%d",
		after.svcTimeout, after.gateway, after.peakFlight, after.peakQueue,
		after.blocked, after.downPeak, after.downCanceled)

	if before.gateway < 50 || before.blocked < 50 || before.downCanceled != 0 {
		t.Fatalf("before did not reproduce the spread: %+v", before)
	}
	if after.gateway != 0 || after.blocked != 0 || after.downCanceled < 230 {
		t.Fatalf("after did not stop the spread: %+v", after)
	}
	if after.peakFlight > 16 || after.downPeak > 16 {
		t.Fatalf("after exceeded worker bound: flight=%d down=%d", after.peakFlight, after.downPeak)
	}
}

$ go test ./internal/platform/slowcall/ -count=1 -v
before svc_timeout=0   gateway_timeout=240 peak_flight=16 peak_queue=32 submit_blocked=96 down_peak=16 down_canceled=0
after  svc_timeout=240 gateway_timeout=0   peak_flight=10 peak_queue=1  submit_blocked=0  down_peak=10 down_canceled=240