《游戏服务端架构·平台化与架构治理》2.2 热更新与灰度

// file: internal/hotupdategray/gray.go
// tier: L1
// verify: go test ./internal/hotupdategray/
package hotupdategray

import (
	"errors"
	"fmt"
	"hash/fnv"
	"sort"
	"sync"
)

// FullPercent is 100 percent expressed in basis points. Basis points keep
// integer arithmetic exact and avoid a float comparison at the bucket edge.
const FullPercent uint32 = 10_000

// Target is the set of facts a rollout selector may use. PlayerID is the only
// stable identity for percentage bucketing; server and channel are coarse
// dimensions that can move when a player migrates.
type Target struct {
	PlayerID string
	ServerID string
	Channel  string
}

// Selector is one rollout scope. Empty server or channel lists mean "all".
// PercentBP is applied only after the coarse dimensions match.
type Selector struct {
	Servers   []string
	Channels  []string
	PercentBP uint32
	Salt      string
}

func (s Selector) Validate() error {
	if s.PercentBP > FullPercent {
		return fmt.Errorf("hotupdategray: percent %d exceeds %d basis points", s.PercentBP, FullPercent)
	}
	for _, group := range [][]string{s.Servers, s.Channels} {
		seen := make(map[string]bool, len(group))
		for _, v := range group {
			if v == "" {
				return errors.New("hotupdategray: selector contains an empty value")
			}
			if seen[v] {
				return fmt.Errorf("hotupdategray: selector repeats %q", v)
			}
			seen[v] = true
		}
	}
	return nil
}

// Match answers whether one target is inside the scope. A percentage below
// full requires a player identity; falling back to server or channel would
// make every player on a node flip together, which is the opposite of a
// per-player canary.
func (s Selector) Match(t Target) bool {
	if !contains(s.Servers, t.ServerID) || !contains(s.Channels, t.Channel) {
		return false
	}
	if s.PercentBP >= FullPercent {
		return true
	}
	if t.PlayerID == "" {
		return false
	}
	return Bucket(t.PlayerID, s.Salt) < s.PercentBP
}

func contains(list []string, value string) bool {
	if len(list) == 0 {
		return true
	}
	for _, v := range list {
		if v == value {
			return true
		}
	}
	return false
}

// Bucket maps a player and a salt to a stable point in [0, FullPercent).
// Changing the salt reshuffles membership, so a new experiment must use a new
// salt rather than reusing the one from the previous rollout.
func Bucket(key, salt string) uint32 {
	h := fnv.New32a()
	h.Write([]byte(salt))
	h.Write([]byte{0})
	h.Write([]byte(key))
	return h.Sum32() % FullPercent
}

// Version is one immutable configuration revision. Revoked is a separate
// flag rather than a new version so that the audit trail keeps the fact that
// the revision existed and was withdrawn.
type Version struct {
	ID       int64
	Name     string
	Selector Selector
	Revoked  bool
}

// Registry holds rollout revisions in ascending ID order. Resolve walks from
// the newest downwards, so a rollback is just "mark the newest revoked" and
// the previous matching revision becomes visible again.
type Registry struct {
	mu       sync.RWMutex
	versions []Version
	seen     map[int64]bool
}

func NewRegistry() *Registry {
	return &Registry{seen: make(map[int64]bool)}
}

func (r *Registry) Add(v Version) error {
	if v.ID <= 0 {
		return errors.New("hotupdategray: version id must be positive")
	}
	if err := v.Selector.Validate(); err != nil {
		return err
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.seen[v.ID] {
		return fmt.Errorf("hotupdategray: version %d already exists", v.ID)
	}
	r.seen[v.ID] = true
	r.versions = append(r.versions, v)
	sort.Slice(r.versions, func(i, j int) bool { return r.versions[i].ID < r.versions[j].ID })
	return nil
}

func (r *Registry) Revoke(id int64) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	for i := range r.versions {
		if r.versions[i].ID == id {
			r.versions[i].Revoked = true
			return nil
		}
	}
	return fmt.Errorf("hotupdategray: version %d not found", id)
}

func (r *Registry) Resolve(t Target) (Version, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	for i := len(r.versions) - 1; i >= 0; i-- {
		v := r.versions[i]
		if v.Revoked {
			continue
		}
		if v.Selector.Match(t) {
			return v, true
		}
	}
	return Version{}, false
}

func (r *Registry) Versions() []Version {
	r.mu.RLock()
	defer r.mu.RUnlock()
	out := make([]Version, len(r.versions))
	copy(out, r.versions)
	return out
}

// Fleet tracks the highest version each node reports as applied. The minimum
// is the watermark: a version is not globally visible until the slowest live
// node has reached it. This is deliberately weaker than "the registry says
// so", because a node that missed a push is invisible from the registry.
type Fleet struct {
	mu      sync.RWMutex
	applied map[string]int64
}

func NewFleet() *Fleet {
	return &Fleet{applied: make(map[string]int64)}
}

// Join is idempotent. A node that restarts keeps its last reported version
// until it reports again, which is why the operator must watch staleness too.
func (f *Fleet) Join(node string) error {
	if node == "" {
		return errors.New("hotupdategray: node name is empty")
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	if _, ok := f.applied[node]; !ok {
		f.applied[node] = 0
	}
	return nil
}

// Report accepts a lower version as well as a higher one. Rollback is a
// legitimate reason for the applied version to move backwards.
func (f *Fleet) Report(node string, version int64) error {
	if version < 0 {
		return errors.New("hotupdategray: applied version must not be negative")
	}
	f.mu.Lock()
	defer f.mu.Unlock()
	if _, ok := f.applied[node]; !ok {
		return fmt.Errorf("hotupdategray: unknown node %q", node)
	}
	f.applied[node] = version
	return nil
}

// Watermark returns the lowest applied version and whether any node exists.
// An empty fleet has no watermark; treating it as zero would claim a rollout
// succeeded before a single node reported.
func (f *Fleet) Watermark() (int64, bool) {
	f.mu.RLock()
	defer f.mu.RUnlock()
	first := true
	var min int64
	for _, v := range f.applied {
		if first || v < min {
			min = v
			first = false
		}
	}
	return min, !first
}

func (f *Fleet) IsEffective(version int64) bool {
	min, ok := f.Watermark()
	return ok && min >= version
}

// Pending lists nodes below a version in stable order. The list is the
// operator's action queue; an aggregate count is not enough to chase them.
func (f *Fleet) Pending(version int64) []string {
	f.mu.RLock()
	defer f.mu.RUnlock()
	var out []string
	for node, applied := range f.applied {
		if applied < version {
			out = append(out, node)
		}
	}
	sort.Strings(out)
	return out
}

func (f *Fleet) Snapshot() map[string]int64 {
	f.mu.RLock()
	defer f.mu.RUnlock()
	out := make(map[string]int64, len(f.applied))
	for k, v := range f.applied {
		out[k] = v
	}
	return out
}