《MMO 服务器开发》5.1 配置版本化与差量分发

// file: internal/config/store.go
// tier: L2
// verify: go test ./internal/config/ -run TestEtcd -v
package config

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strconv"
	"strings"

	clientv3 "go.etcd.io/etcd/client/v3"
)

// Version 是配置快照的单调版本号。每次发布自增；回滚也是「发一个新版本，
// 内容等于旧版本」，版本号只前进不倒退，历史始终可查。
type Version uint64

// Item 是一个配置项：键、值，以及它被写入时的版本号。
type Item struct {
	Key   string  `json:"key"`
	Value string  `json:"value"`
	Ver   Version `json:"ver"`
}

// Snapshot 是某个版本下的全量配置视图。
type Snapshot struct {
	Version Version         `json:"version"`
	Items   map[string]Item `json:"items"`
}

// Delta 是从一个版本到另一个版本之间「真正变化的部分」。
// 只包含值变化的键与被删除的键——没动的键不进 Delta，这是差量分发的核心。
type Delta struct {
	From    Version  `json:"from"`
	To      Version  `json:"to"`
	Changed []Item   `json:"changed"`
	Deleted []string `json:"deleted"`
}

// Empty 表示这次发布没有任何键变化（幂等发布）。
func (d Delta) Empty() bool { return len(d.Changed) == 0 && len(d.Deleted) == 0 }

// Size 是差量涉及的键数，用于校验「只推变更的键」。
func (d Delta) Size() int { return len(d.Changed) + len(d.Deleted) }

// Diff 计算 from → to 的差量：只收集值变化或被删除的键。
func Diff(from, to Snapshot) Delta {
	d := Delta{From: from.Version, To: to.Version}
	for k, nv := range to.Items {
		if ov, ok := from.Items[k]; !ok || ov.Value != nv.Value {
			d.Changed = append(d.Changed, nv)
		}
	}
	for k := range from.Items {
		if _, ok := to.Items[k]; !ok {
			d.Deleted = append(d.Deleted, k)
		}
	}
	sort.Slice(d.Changed, func(i, j int) bool { return d.Changed[i].Key < d.Changed[j].Key })
	sort.Strings(d.Deleted)
	return d
}

// Store 是挂在 etcd 上的配置中心。键布局：
//
//	<prefix>kv/<key>       → 当前值（JSON 编码的 Item）
//	<prefix>version        → 当前版本号（十进制）
//	<prefix>history/<ver>  → 该版本的完整快照，供回滚与审计
type Store struct {
	cli    *clientv3.Client
	prefix string
}

// NewStore 创建配置中心；prefix 形如 "aegis:test:ch05:"。
func NewStore(cli *clientv3.Client, prefix string) *Store {
	if !strings.HasSuffix(prefix, ":") && !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}
	return &Store{cli: cli, prefix: prefix}
}

func (s *Store) kvPrefix() string   { return s.prefix + "kv/" }
func (s *Store) versionKey() string { return s.prefix + "version" }
func (s *Store) histKey(v Version) string {
	return s.prefix + "history/" + strconv.FormatUint(uint64(v), 10)
}

// Snapshot 读取当前全量配置（强一致读）。
func (s *Store) Snapshot(ctx context.Context) (Snapshot, error) {
	snap, _, err := s.load(ctx)
	return snap, err
}

// load 读版本号与全部 kv，并返回本次读到的 etcd revision（供 Watch 播种）。
func (s *Store) load(ctx context.Context) (Snapshot, int64, error) {
	snap := Snapshot{Items: map[string]Item{}}
	vresp, err := s.cli.Get(ctx, s.versionKey())
	if err != nil {
		return snap, 0, fmt.Errorf("config: read version: %w", err)
	}
	if len(vresp.Kvs) > 0 {
		if n, perr := strconv.ParseUint(string(vresp.Kvs[0].Value), 10, 64); perr == nil {
			snap.Version = Version(n)
		}
	}
	kresp, err := s.cli.Get(ctx, s.kvPrefix(), clientv3.WithPrefix())
	if err != nil {
		return snap, 0, fmt.Errorf("config: read kv: %w", err)
	}
	for _, kv := range kresp.Kvs {
		var it Item
		if err := json.Unmarshal(kv.Value, &it); err != nil {
			return snap, 0, fmt.Errorf("config: decode %q: %w", kv.Key, err)
		}
		snap.Items[it.Key] = it
	}
	return snap, kresp.Header.Revision, nil
}

// Publish 把 desired 发布为新版本：与当前快照求差量，只写变化的键，
// 并用一个事务原子地写入版本号与历史快照。返回本次真正下发的差量。
func (s *Store) Publish(ctx context.Context, desired map[string]string) (Delta, error) {
	cur, _, err := s.load(ctx)
	if err != nil {
		return Delta{}, err
	}
	next := Snapshot{Version: cur.Version + 1, Items: make(map[string]Item, len(desired))}
	for k, v := range desired {
		next.Items[k] = Item{Key: k, Value: v, Ver: next.Version}
	}
	delta := Diff(cur, next)

	ops := make([]clientv3.Op, 0, delta.Size()+2)
	for _, it := range delta.Changed {
		b, _ := json.Marshal(it)
		ops = append(ops, clientv3.OpPut(s.kvPrefix()+it.Key, string(b)))
	}
	for _, k := range delta.Deleted {
		ops = append(ops, clientv3.OpDelete(s.kvPrefix()+k))
	}
	hb, _ := json.Marshal(next)
	ops = append(ops,
		clientv3.OpPut(s.histKey(next.Version), string(hb)),
		clientv3.OpPut(s.versionKey(), strconv.FormatUint(uint64(next.Version), 10)),
	)
	resp, err := s.cli.Txn(ctx).Then(ops...).Commit()
	if err != nil {
		return Delta{}, fmt.Errorf("config: publish txn: %w", err)
	}
	if !resp.Succeeded {
		return Delta{}, fmt.Errorf("config: publish txn not committed")
	}
	return delta, nil
}

// Cleanup 删除本 prefix 下的全部键，供测试收尾。
func (s *Store) Cleanup(ctx context.Context) error {
	_, err := s.cli.Delete(ctx, s.prefix, clientv3.WithPrefix())
	return err
}

// snippet: internal/config/gray.go
// GrayByPlayer 按玩家百分比灰度。用 FNV-1a 对玩家 ID 做确定性散列，
// 而不是随机数：同一玩家在多次判定中结果稳定，否则玩家会在放量与收回之间反复横跳。
func GrayByPlayer(playerID uint64, percent int) bool {
	if percent <= 0 {
		return false
	}
	if percent >= 100 {
		return true
	}
	var buf [8]byte
	binary.BigEndian.PutUint64(buf[:], playerID)
	h := fnv.New64a()
	_, _ = h.Write(buf[:])
	return int(h.Sum64()%100) < percent
}

// PushMessage 是服务端下发给节点的配置推送，命令号 CmdConfigPush。
type PushMessage struct {
	Cmd     uint16  `json:"cmd"`
	Version Version `json:"version"`
	Delta   Delta   `json:"delta"`
}

// AckMessage 是节点应用差量后的回执，命令号 CmdConfigAck。
type AckMessage struct {
	Cmd     uint16  `json:"cmd"`
	Node    string  `json:"node"`
	Version Version `json:"version"`
	Applied int     `json:"applied"`
}

// snippet: internal/config/subscriber.go
// Run 先读一次全量播种、记下 revision，再从 rev+1 起 Watch，
// 既不漏事件也不重放历史。每应用一批变更就回调一次 ack。ctx 取消时返回。
func (s *Subscriber) Run(ctx context.Context, onAck func(AckMessage)) error {
	st := NewStore(s.cli, s.prefix)
	cur, rev, err := st.load(ctx)
	if err != nil {
		return err
	}
	s.mu.Lock()
	s.snap = cur
	s.mu.Unlock()

	ch := s.cli.Watch(ctx, s.prefix+"kv/", clientv3.WithPrefix(), clientv3.WithRev(rev+1))
	for resp := range ch {
		if err := resp.Err(); err != nil {
			return err
		}
		n := s.applyEvents(resp.Events)
		if n == 0 {
			continue
		}
		ver := s.readVersion(ctx)
		s.mu.Lock()
		s.snap.Version = ver
		s.mu.Unlock()
		if onAck != nil {
			onAck(AckMessage{Cmd: CmdConfigAck, Node: s.node, Version: ver, Applied: n})
		}
	}
	return ctx.Err()
}

=== RUN   TestEtcdConfigDeltaPushAndAck
--- PASS: TestEtcdConfigDeltaPushAndAck (0.07s)
PASS
ok  	aegis/internal/config	0.727s