《后端架构设计·高性能与高可用》11.3 读写分离的延迟预算与一致性兜底

// file: internal/perf/readsplit/router.go
// tier: L1
// verify: go test ./internal/perf/readsplit/ -count=1 -v
package readsplit

import (
	"errors"
	"math"
	"sort"
	"time"
)

type Role int

const (
	Primary Role = iota
	Replica
)

type Node struct {
	Name           string
	Role           Role
	Healthy        bool
	AppliedVersion uint64
	Lag            time.Duration
	InFlight       int
	Capacity       int
}

func (n *Node) Overloaded() bool { return n.Capacity > 0 && n.InFlight >= n.Capacity }

type ReadKind int

const (
	// ReadYourWrites must observe the session's own last write.
	ReadYourWrites ReadKind = iota
	// ReadFresh cannot tolerate any stale replica, so it uses the primary.
	ReadFresh
	// ReadStaleOK accepts bounded staleness, e.g. dashboards.
	ReadStaleOK
)

type Request struct {
	Kind           ReadKind
	SessionVersion uint64
}

type Decision struct {
	Node     *Node
	Reason   string
	Degraded bool
}

var (
	ErrNoPrimary        = errors.New("readsplit: no healthy primary")
	ErrOverloaded       = errors.New("readsplit: primary overloaded")
	ErrStaleBeyondBound = errors.New("readsplit: no replica within hard lag bound")
)

type Router struct {
	Primary    *Node
	Replicas   []*Node
	MaxLag     time.Duration
	HardMaxLag time.Duration
}

// Route never sends a read-your-writes request to a replica that has not
// applied the session version, even when that replica is much less loaded.
func (r *Router) Route(req Request) (Decision, error) {
	if req.Kind == ReadStaleOK {
		if n := r.bestReplica(r.HardMaxLag); n != nil {
			return Decision{Node: n, Reason: "stale-ok replica", Degraded: n.Lag > r.MaxLag}, nil
		}
		if r.Primary != nil && r.Primary.Healthy && !r.Primary.Overloaded() {
			return Decision{Node: r.Primary, Reason: "no replica within hard bound"}, nil
		}
		return Decision{}, ErrStaleBeyondBound
	}
	if r.Primary == nil || !r.Primary.Healthy {
		return Decision{}, ErrNoPrimary
	}
	if req.Kind == ReadYourWrites {
		if n := r.caughtUpReplica(req.SessionVersion); n != nil {
			return Decision{Node: n, Reason: "replica caught up to session"}, nil
		}
		if r.Primary.Overloaded() {
			return Decision{}, ErrOverloaded
		}
		return Decision{Node: r.Primary, Reason: "session write not replicated"}, nil
	}
	if r.Primary.Overloaded() {
		return Decision{}, ErrOverloaded
	}
	return Decision{Node: r.Primary, Reason: "fresh read requires primary"}, nil
}

func (r *Router) bestReplica(maxLag time.Duration) *Node {
	var best *Node
	for _, n := range r.Replicas {
		if !n.Healthy || n.Overloaded() || n.Lag > maxLag {
			continue
		}
		if best == nil || n.InFlight < best.InFlight {
			best = n
		}
	}
	return best
}

func (r *Router) caughtUpReplica(sessionVersion uint64) *Node {
	var best *Node
	for _, n := range r.Replicas {
		if !n.Healthy || n.Overloaded() || n.AppliedVersion < sessionVersion {
			continue
		}
		if best == nil || n.InFlight < best.InFlight {
			best = n
		}
	}
	return best
}

// LagTracker keeps the last size samples and reports a nearest-rank quantile.
type LagTracker struct {
	samples []time.Duration
	next    int
	count   int
}

func NewLagTracker(size int) *LagTracker {
	if size < 1 {
		size = 1
	}
	return &LagTracker{samples: make([]time.Duration, size)}
}

func (t *LagTracker) Observe(d time.Duration) {
	t.samples[t.next] = d
	t.next = (t.next + 1) % len(t.samples)
	if t.count < len(t.samples) {
		t.count++
	}
}

func (t *LagTracker) Quantile(q float64) (time.Duration, bool) {
	if t.count == 0 {
		return 0, false
	}
	buf := append([]time.Duration(nil), t.samples[:t.count]...)
	sort.Slice(buf, func(i, j int) bool { return buf[i] < buf[j] })
	q = math.Max(0, math.Min(1, q))
	idx := int(math.Ceil(q*float64(len(buf)))) - 1
	if idx < 0 {
		idx = 0
	}
	return buf[idx], true
}

// file: internal/perf/readsplit/simulate.go
// tier: L1
// verify: go test ./internal/perf/readsplit/ -count=1 -v
package readsplit

import (
	"math/rand"
	"time"
)

// Result separates the read-your-writes question from stale-tolerant traffic.
// A read is in RYWPrimary when the replica would not have applied the session
// write, or when replication lag exceeds the freshness budget.
type Result struct {
	Reads         int
	RYWPrimary    int
	StaleReplica  int
	StaleDegraded int
	StaleBeyond   int
}

// Simulate walks paired lag and read-delay samples. The model is deliberately
// simple: a read arriving delay after the write sees it when lag <= delay.
func Simulate(lags, delays []time.Duration, budget, hardMax time.Duration) Result {
	n := len(lags)
	if len(delays) < n {
		n = len(delays)
	}
	res := Result{Reads: n}
	for i := 0; i < n; i++ {
		if lags[i] > delays[i] || lags[i] > budget {
			res.RYWPrimary++
		}
		switch {
		case lags[i] > hardMax:
			res.StaleBeyond++
		default:
			res.StaleReplica++
			if lags[i] > budget {
				res.StaleDegraded++
			}
		}
	}
	return res
}

// LagSamples is a deterministic mixture: 90% low lag, 9% medium, 1% long.
func LagSamples(n int, seed int64) []time.Duration {
	r := rand.New(rand.NewSource(seed))
	out := make([]time.Duration, n)
	for i := range out {
		switch p := r.Float64(); {
		case p < 0.90:
			out[i] = time.Duration(5+r.Intn(35)) * time.Millisecond
		case p < 0.99:
			out[i] = time.Duration(40+r.Intn(260)) * time.Millisecond
		default:
			out[i] = time.Duration(300+r.Intn(1700)) * time.Millisecond
		}
	}
	return out
}

// DelaySamples models 1-200ms between the session's write and its next read.
func DelaySamples(n int, seed int64) []time.Duration {
	r := rand.New(rand.NewSource(seed))
	out := make([]time.Duration, n)
	for i := range out {
		out[i] = time.Duration(1+r.Intn(200)) * time.Millisecond
	}
	return out
}

func Percent(part, total int) float64 {
	if total == 0 {
		return 0
	}
	return 100 * float64(part) / float64(total)
}

// file: internal/perf/readsplit/router_test.go
// tier: L1
// verify: go test ./internal/perf/readsplit/ -count=1 -v
package readsplit

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

func TestRouting(t *testing.T) {
	p := &Node{Name: "primary", Role: Primary, Healthy: true, Capacity: 10}
	rep := &Node{Name: "replica", Role: Replica, Healthy: true, AppliedVersion: 9, Capacity: 10}
	r := Router{Primary: p, Replicas: []*Node{rep}, MaxLag: 100 * time.Millisecond, HardMaxLag: 500 * time.Millisecond}

	d, err := r.Route(Request{Kind: ReadYourWrites, SessionVersion: 10})
	if err != nil || d.Node != p || d.Reason != "session write not replicated" {
		t.Fatalf("ryw lagging: %+v %v", d, err)
	}
	rep.AppliedVersion = 10
	if d, err = r.Route(Request{Kind: ReadYourWrites, SessionVersion: 10}); err != nil || d.Node != rep {
		t.Fatalf("ryw caught up: %+v %v", d, err)
	}
	rep.Healthy = false
	if d, err = r.Route(Request{Kind: ReadYourWrites, SessionVersion: 10}); err != nil || d.Node != p {
		t.Fatalf("replica down: %+v %v", d, err)
	}
	rep.Healthy, rep.Lag = true, 150*time.Millisecond
	if d, err = r.Route(Request{Kind: ReadStaleOK}); err != nil || !d.Degraded {
		t.Fatalf("degraded: %+v %v", d, err)
	}
	p.InFlight = p.Capacity
	if _, err = r.Route(Request{Kind: ReadFresh}); !errors.Is(err, ErrOverloaded) {
		t.Fatalf("overload: %v", err)
	}
}

func TestFallbackRate(t *testing.T) {
	lags := LagSamples(10_000, 7)
	delays := DelaySamples(10_000, 11)
	a := Simulate(lags, delays, 50*time.Millisecond, 500*time.Millisecond)
	b := Simulate(lags, delays, 200*time.Millisecond, 500*time.Millisecond)
	if b.RYWPrimary > a.RYWPrimary {
		t.Fatalf("larger budget increased fallback: %d > %d", b.RYWPrimary, a.RYWPrimary)
	}
	t.Logf("reads=%d fallback_50ms=%d fallback_200ms=%d", a.Reads, a.RYWPrimary, b.RYWPrimary)
}