《后端架构设计·分布式与数据一致性》5.3 请求路由、元数据与扩容期间的错误预算

控制面版本 42 ──写入──> 元数据存储 ──watch/轮询──> 网关缓存 ──本地刷新──> 请求
       │                     │                      │
       └── 版本单调递增 ──────┴── 拒绝旧版本 ─────────┘

projected_failures = total_requests × stale_rate × (1 - resolve_rate)

// file: internal/platform/routing/router.go
// tier: L2
// verify: go test ./internal/platform/routing/ -count=1
package routing

import (
	"errors"
	"fmt"
	"math"
	"sync"
)

// Shard identifies one storage shard.
type Shard string

// ErrRouteUnavailable means the router exhausted its resolver attempts.
var ErrRouteUnavailable = errors.New("route unavailable")

// ErrMoved means the contacted shard no longer owns the key. The caller should
// invalidate its cached route and retry.
var ErrMoved = errors.New("key moved")

// Directory is a versioned local route cache. Apply rejects versions that are
// not newer, so delayed metadata cannot overwrite a fresher table.
type Directory struct {
	mu      sync.RWMutex
	version int64
	entries map[string]Shard
}

// NewDirectory returns an empty cache at version 0.
func NewDirectory() *Directory {
	return &Directory{entries: make(map[string]Shard)}
}

// Apply atomically replaces the table when version is newer than the current
// version.
func (d *Directory) Apply(version int64, entries map[string]Shard) error {
	d.mu.Lock()
	defer d.mu.Unlock()
	if version <= d.version {
		return fmt.Errorf("routing: version %d not newer than %d", version, d.version)
	}
	next := make(map[string]Shard, len(entries))
	for k, v := range entries {
		next[k] = v
	}
	d.entries = next
	d.version = version
	return nil
}

// Put caches one resolved route.
func (d *Directory) Put(key string, shard Shard) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.entries[key] = shard
}

// Invalidate drops one cached route after a moved error.
func (d *Directory) Invalidate(key string) {
	d.mu.Lock()
	defer d.mu.Unlock()
	delete(d.entries, key)
}

// Lookup returns the cached route for key.
func (d *Directory) Lookup(key string) (Shard, bool) {
	d.mu.RLock()
	defer d.mu.RUnlock()
	shard, ok := d.entries[key]
	return shard, ok
}

// Version returns the applied metadata version.
func (d *Directory) Version() int64 {
	d.mu.RLock()
	defer d.mu.RUnlock()
	return d.version
}

// Resolver returns the authoritative shard for a key.
type Resolver interface {
	Resolve(key string) (Shard, error)
}

// ResolverFunc adapts a function to Resolver.
type ResolverFunc func(string) (Shard, error)

// Resolve calls f.
func (f ResolverFunc) Resolve(key string) (Shard, error) {
	return f(key)
}

// Metrics counts routing outcomes.
type Metrics struct {
	CacheHits    int64
	ResolveCalls int64
	StaleHits    int64
	Failures     int64
}

// Router resolves keys with a local cache and a bounded resolver.
type Router struct {
	dir        *Directory
	resolver   Resolver
	maxResolve int

	mu      sync.Mutex
	metrics Metrics
}

// NewRouter returns a router. maxResolve is the number of authoritative lookups
// allowed for one request, including retries after ErrMoved.
func NewRouter(dir *Directory, resolver Resolver, maxResolve int) (*Router, error) {
	if dir == nil {
		return nil, fmt.Errorf("routing: nil directory")
	}
	if resolver == nil {
		return nil, fmt.Errorf("routing: nil resolver")
	}
	if maxResolve <= 0 {
		return nil, fmt.Errorf("routing: maxResolve must be positive, got %d", maxResolve)
	}
	return &Router{dir: dir, resolver: resolver, maxResolve: maxResolve}, nil
}

// GetFunc performs the actual read against a shard. It returns ErrMoved when
// the shard does not own the key.
type GetFunc func(shard Shard) (string, error)

// Fetch routes key and executes get. A moved error invalidates the cache entry
// and triggers one authoritative resolve, so a stale route costs one extra hop
// but converges.
func (r *Router) Fetch(key string, get GetFunc) (string, error) {
	for attempt := 0; attempt <= r.maxResolve; attempt++ {
		shard, ok := r.dir.Lookup(key)
		if ok {
			r.inc(&r.metrics.CacheHits)
		} else {
			if attempt == r.maxResolve {
				break
			}
			r.inc(&r.metrics.ResolveCalls)
			resolved, err := r.resolver.Resolve(key)
			if err != nil || resolved == "" {
				continue
			}
			shard = resolved
			r.dir.Put(key, shard)
		}

		value, err := get(shard)
		if err == nil {
			return value, nil
		}
		if errors.Is(err, ErrMoved) {
			r.inc(&r.metrics.StaleHits)
			r.dir.Invalidate(key)
			continue
		}
		r.inc(&r.metrics.Failures)
		return "", fmt.Errorf("routing: fetch %q from %s: %w", key, shard, err)
	}
	r.inc(&r.metrics.Failures)
	return "", fmt.Errorf("routing: fetch %q after %d resolves: %w", key, r.maxResolve, ErrRouteUnavailable)
}

func (r *Router) inc(counter *int64) {
	r.mu.Lock()
	*counter++
	r.mu.Unlock()
}

// Metrics returns a snapshot.
func (r *Router) Metrics() Metrics {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.metrics
}

// Budget expresses the error budget available during one expansion window.
type Budget struct {
	TotalRequests int64
	SLOPercent    float64
	StaleRate     float64
	ResolveRate   float64
}

// Validate checks ranges so a typo cannot silently produce a meaningless
// budget.
func (b Budget) Validate() error {
	if b.TotalRequests <= 0 {
		return fmt.Errorf("budget: total requests must be positive, got %d", b.TotalRequests)
	}
	if b.SLOPercent <= 0 || b.SLOPercent >= 100 {
		return fmt.Errorf("budget: SLO percent must be in (0,100), got %v", b.SLOPercent)
	}
	if b.StaleRate < 0 || b.StaleRate > 1 {
		return fmt.Errorf("budget: stale rate must be in [0,1], got %v", b.StaleRate)
	}
	if b.ResolveRate < 0 || b.ResolveRate > 1 {
		return fmt.Errorf("budget: resolve rate must be in [0,1], got %v", b.ResolveRate)
	}
	return nil
}

// sloBasisPoints converts a percentage to hundredths of a percent. It keeps
// the budget arithmetic in integers so 99.95% yields exactly 5 basis points of
// allowed error rather than a float near 4999.99 failures.
func (b Budget) sloBasisPoints() int64 {
	return int64(math.Round(b.SLOPercent * 100))
}

// AllowedFailures is the error budget for the window.
func (b Budget) AllowedFailures() int64 {
	if b.TotalRequests <= 0 || b.SLOPercent <= 0 || b.SLOPercent >= 100 {
		return 0
	}
	return b.TotalRequests * (10000 - b.sloBasisPoints()) / 10000
}

// ProjectedFailures assumes every stale request needs one authoritative
// resolve and only failed resolves become errors.
func (b Budget) ProjectedFailures() int64 {
	stale := math.Round(float64(b.TotalRequests) * b.StaleRate)
	return int64(math.Round(stale * (1 - b.ResolveRate)))
}

// WithinBudget reports whether the plan fits the window.
func (b Budget) WithinBudget() bool {
	return b.ProjectedFailures() <= b.AllowedFailures()
}

// BreakerThreshold is the failure rate that consumes the whole budget. A
// circuit breaker should trip before the observed rate reaches it.
func (b Budget) BreakerThreshold() float64 {
	return float64(10000-b.sloBasisPoints()) / 10000
}

$ go test ./internal/platform/routing/ -v -count=1
=== RUN   TestFetchRetriesAfterMoved
    routing_test.go:39: value=payload-1001 resolves=1 cache_hits=1 stale_hits=1 failures=0
--- PASS: TestFetchRetriesAfterMoved (0.00s)
=== RUN   TestFetchGivesUpAfterMaxResolve
    routing_test.go:63: resolve_calls=2 failures=1
--- PASS: TestFetchGivesUpAfterMaxResolve (0.00s)
=== RUN   TestDirectoryRejectsStaleVersion
    routing_test.go:89: version guard kept=shard-a final_version=8
--- PASS: TestDirectoryRejectsStaleVersion (0.00s)
=== RUN   TestFetchConcurrent
    routing_test.go:119: concurrent fetches=64 cache_hits=0 resolves=64
--- PASS: TestFetchConcurrent (0.00s)
=== RUN   TestBudgetArithmetic
    routing_test.go:129: allowed_failures=5000 projected_failures=20 within=true breaker=0.000500
    routing_test.go:142: bad_allowed=5000 bad_projected=25000 within=false
--- PASS: TestBudgetArithmetic (0.00s)
PASS
ok  	archlab/internal/platform/routing	0.277s