《游戏服务端架构·品类、状态与拓扑》7.1 MOBA 的房间、tick 与帧同步

当前 tick = 100，窗口 = 3
可接受：100、101、102、103
迟到：   99 及更早  -> 拒绝，late++
超窗：   104 及更远  -> 拒绝，reject++，通常是客户端作弊或时钟错乱
重复：   同一 tick 同一玩家第二次 -> 拒绝，dupe++，第一份生效

// file: internal/moba/aggregator.go
// tier: L1
// verify: go test ./internal/moba/
package moba

import (
	"encoding/binary"
	"errors"
	"fmt"
	"hash/crc32"
	"sort"
)

var (
	ErrStaleInput  = errors.New("moba: input tick already committed")
	ErrFutureInput = errors.New("moba: input tick beyond buffer window")
	ErrDuplicate   = errors.New("moba: duplicate input for player in tick")
)

// Input is one player's command for one simulation tick. The fields are small
// integers on purpose: frame-sync packets must be cheap to serialize, and a
// fixed layout keeps the digest reproducible across languages.
type Input struct {
	Player uint32
	MoveX  int16
	MoveY  int16
	Action uint16
}

// Frame is the committed input set for a single tick. Inputs are sorted by
// Player so two servers that received the same packets in different orders
// produce byte-identical frames and the same digest.
type Frame struct {
	Tick   uint64
	Inputs []Input
	Digest uint32
}

// Aggregator buffers per-tick inputs and commits them one tick at a time.
// Window is how many ticks ahead a client may submit. Anything further is a
// protocol violation, not a slow network: accepting it would let one client
// grow the server's buffer without bound.
type Aggregator struct {
	window  uint64
	current uint64
	pending map[uint64]map[uint32]Input
	roll    uint32
	frames  uint64
	missing uint64
	late    uint64
	dupes   uint64
	rejects uint64
}

// New returns an aggregator whose future window is measured in ticks.
func New(window uint64) *Aggregator {
	if window == 0 {
		window = 1
	}
	return &Aggregator{
		window:  window,
		pending: make(map[uint64]map[uint32]Input),
	}
}

// Submit buffers one input. tick is the simulation tick the input belongs to,
// not the wall-clock time the packet arrived. Late inputs are counted and
// rejected because the frame has already been committed and sent to clients.
func (a *Aggregator) Submit(tick uint64, in Input) error {
	if tick < a.current {
		a.late++
		return fmt.Errorf("%w: tick=%d current=%d", ErrStaleInput, tick, a.current)
	}
	if tick > a.current+a.window {
		a.rejects++
		return fmt.Errorf("%w: tick=%d current=%d window=%d", ErrFutureInput, tick, a.current, a.window)
	}
	bucket := a.pending[tick]
	if bucket == nil {
		bucket = make(map[uint32]Input)
		a.pending[tick] = bucket
	}
	if _, exists := bucket[in.Player]; exists {
		a.dupes++
		return fmt.Errorf("%w: tick=%d player=%d", ErrDuplicate, tick, in.Player)
	}
	bucket[in.Player] = in
	return nil
}

// Advance commits the current tick and returns its frame. An empty frame means
// no input arrived for that tick: the simulation still advances, and Missing
// counts the gap. Never stall the world waiting for a packet that may never
// arrive.
func (a *Aggregator) Advance() Frame {
	tick := a.current
	bucket := a.pending[tick]
	delete(a.pending, tick)

	inputs := make([]Input, 0, len(bucket))
	for _, in := range bucket {
		inputs = append(inputs, in)
	}
	sort.Slice(inputs, func(i, j int) bool { return inputs[i].Player < inputs[j].Player })
	if len(inputs) == 0 {
		a.missing++
	}

	a.current++
	a.frames++
	a.roll = chain(a.roll, tick, inputs)
	return Frame{Tick: tick, Inputs: inputs, Digest: a.roll}
}

// chain folds one committed frame into the rolling digest. Sort order is part
// of the contract: unsorted input would make the hash depend on map iteration
// and turn a determinism bug into an intermittent one.
func chain(prev uint32, tick uint64, inputs []Input) uint32 {
	h := crc32.NewIEEE()
	var b [8]byte
	binary.BigEndian.PutUint32(b[:4], prev)
	h.Write(b[:4])
	binary.BigEndian.PutUint64(b[:], tick)
	h.Write(b[:])
	for _, in := range inputs {
		binary.BigEndian.PutUint32(b[:4], in.Player)
		h.Write(b[:4])
		binary.BigEndian.PutUint16(b[:2], uint16(in.MoveX))
		h.Write(b[:2])
		binary.BigEndian.PutUint16(b[:2], uint16(in.MoveY))
		h.Write(b[:2])
		binary.BigEndian.PutUint16(b[:2], in.Action)
		h.Write(b[:2])
	}
	return h.Sum32()
}

// Digest is the rolling hash over every committed frame in order. Two rooms
// that received the same inputs, possibly in different packet order, must end
// with the same digest. A mismatch is a determinism bug, not a network bug.
func (a *Aggregator) Digest() uint32 { return a.roll }

// Tick is the next tick that Advance will commit.
func (a *Aggregator) Tick() uint64 { return a.current }

// Frames is the number of committed ticks, including empty ones.
func (a *Aggregator) Frames() uint64 { return a.frames }

// Stats is a fixed-width snapshot for metrics and replay headers.
type Stats struct {
	Frames  uint64
	Missing uint64
	Late    uint64
	Dupe    uint64
	Reject  uint64
	Digest  uint32
}

// Stats reports counters without exposing the pending buffer.
func (a *Aggregator) Stats() Stats {
	return Stats{
		Frames:  a.frames,
		Missing: a.missing,
		Late:    a.late,
		Dupe:    a.dupes,
		Reject:  a.rejects,
		Digest:  a.roll,
	}
}