《游戏服务端架构·品类、状态与拓扑》6.1 客户端预测与服务端裁决

客户端                         服务端
  |-- Input{seq=7,tick=120} ----->|
  |   本地预测：位置 += move      |  校验 tick/seq，按规则裁决
  |<-- Ack{tick=120,state} -------|
  |   丢弃 tick<=120 的待确认输入
  |   以 Ack.state 为基准，重放 tick>120 的输入
  |   记录本次偏差，超过阈值计为硬纠正

// file: internal/authority/authority.go
// tier: L1
// verify: go test ./internal/authority/
package authority

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

// Vec is a position in integer world units. Integers keep replay stable:
// floating point addition is not associative, so replay order could drift.
type Vec struct{ X, Y int }

func (v Vec) Add(o Vec) Vec  { return Vec{X: v.X + o.X, Y: v.Y + o.Y} }
func (v Vec) Sub(o Vec) Vec  { return Vec{X: v.X - o.X, Y: v.Y - o.Y} }
func (v Vec) Manhattan() int { return abs(v.X) + abs(v.Y) }

func abs(v int) int {
	if v < 0 {
		return -v
	}
	return v
}

// Input is one client command. Seq is for deduplication; Tick orders it.
type Input struct {
	Seq  uint32
	Tick uint64
	Move Vec
}

var (
	ErrStaleTick = errors.New("authority: tick not newer than last applied")
	ErrStaleSeq  = errors.New("authority: sequence not newer than last applied")
)

// Ack carries the full server state at Tick, never a delta, so a client that
// lost a packet can resynchronise without replaying server history.
type Ack struct {
	Tick  uint64
	Seq   uint32
	State Vec
}

// Server owns the only writable copy of the character state.
type Server struct {
	state    Vec
	tick     uint64
	seq      uint32
	applied  int
	rejected int

	// Limit clamps one axis of a submitted move; zero means unlimited. It
	// stands in for any server rule that overrides client intent.
	Limit int
}

func NewServer(start Vec) *Server { return &Server{state: start} }

func (s *Server) Submit(in Input) (Ack, error) {
	if in.Tick <= s.tick {
		s.rejected++
		return Ack{}, ErrStaleTick
	}
	if in.Seq <= s.seq {
		s.rejected++
		return Ack{}, ErrStaleSeq
	}
	move := in.Move
	if s.Limit > 0 {
		move = Vec{X: clamp(move.X, s.Limit), Y: clamp(move.Y, s.Limit)}
	}
	s.tick = in.Tick
	s.seq = in.Seq
	s.state = s.state.Add(move)
	s.applied++
	return Ack{Tick: in.Tick, Seq: in.Seq, State: s.state}, nil
}

func (s *Server) State() Vec    { return s.state }
func (s *Server) Applied() int  { return s.applied }
func (s *Server) Rejected() int { return s.rejected }

// Aggression controls how far ahead of the last acknowledgement a client may
// predict. Window is measured in ticks.
type Aggression struct{ Window int }

// Tune lowers prediction aggressiveness as RTT grows. The thresholds are
// deployment policy, not protocol constants.
func Tune(rtt time.Duration) Aggression {
	switch {
	case rtt <= 60*time.Millisecond:
		return Aggression{Window: 8}
	case rtt <= 120*time.Millisecond:
		return Aggression{Window: 4}
	default:
		return Aggression{Window: 1}
	}
}

type Stats struct {
	Acks            int
	Replays         int
	Corrections     int
	HardCorrections int
	MaxDeviation    int
}

// Replay describes what happened when one acknowledgement was applied.
type Replay struct {
	Before        Vec
	Authoritative Vec
	Deviation     int
	Corrected     bool
	Replayed      []Input
}

// Client keeps a speculative copy plus per-tick prediction history. History is
// what makes deviation meaningful: comparing the current prediction with an old
// ack would charge the client for honest future movement.
type Client struct {
	predicted Vec
	ackedTick uint64
	buffer    map[uint32]Input
	history   map[uint64]Vec
	agg       Aggression
	tolerance int
	stats     Stats
}

func NewClient(start Vec, tolerance int) *Client {
	if tolerance < 1 {
		tolerance = 1
	}
	return &Client{
		predicted: start,
		buffer:    make(map[uint32]Input),
		history:   make(map[uint64]Vec),
		agg:       Tune(0),
		tolerance: tolerance,
	}
}

func (c *Client) SetAggression(a Aggression) { c.agg = a }

// Predict buffers an input and applies it locally only inside the prediction
// window. Out-of-window inputs wait for an ack and are applied during replay.
func (c *Client) Predict(in Input) (Vec, bool) {
	if _, ok := c.buffer[in.Seq]; !ok {
		c.buffer[in.Seq] = in
	}
	if in.Tick <= c.ackedTick || in.Tick > c.ackedTick+uint64(c.agg.Window) {
		return c.predicted, false
	}
	c.predicted = c.predicted.Add(in.Move)
	c.history[in.Tick] = c.predicted
	return c.predicted, true
}

// Reconcile compares the stored prediction at the acknowledged tick with the
// server state, snaps to the server, then replays every still-pending input.
func (c *Client) Reconcile(ack Ack) Replay {
	base, ok := c.history[ack.Tick]
	if !ok {
		base = ack.State
	}
	dev := base.Sub(ack.State).Manhattan()
	c.stats.Acks++
	if dev > c.stats.MaxDeviation {
		c.stats.MaxDeviation = dev
	}

	c.ackedTick = ack.Tick
	for seq, in := range c.buffer {
		if in.Tick <= ack.Tick {
			delete(c.buffer, seq)
		}
	}
	for tick := range c.history {
		if tick <= ack.Tick {
			delete(c.history, tick)
		}
	}

	pending := c.pending()
	c.predicted = ack.State
	for _, in := range pending {
		c.predicted = c.predicted.Add(in.Move)
		c.history[in.Tick] = c.predicted
	}

	r := Replay{Before: base, Authoritative: ack.State, Deviation: dev, Replayed: pending}
	if dev > 0 {
		c.stats.Corrections++
		r.Corrected = true
		if dev > c.tolerance {
			c.stats.HardCorrections++
		}
	}
	c.stats.Replays += len(pending)
	return r
}

func (c *Client) pending() []Input {
	out := make([]Input, 0, len(c.buffer))
	for _, in := range c.buffer {
		out = append(out, in)
	}
	sort.Slice(out, func(i, j int) bool {
		if out[i].Tick != out[j].Tick {
			return out[i].Tick < out[j].Tick
		}
		return out[i].Seq < out[j].Seq
	})
	return out
}

func (c *Client) Predicted() Vec { return c.predicted }
func (c *Client) Pending() int   { return len(c.buffer) }
func (c *Client) Stats() Stats   { return c.stats }

func clamp(v, limit int) int {
	if v > limit {
		return limit
	}
	if v < -limit {
		return -limit
	}
	return v
}

$ go test -count=1 -v ./internal/authority/
--- PASS: TestServerRejectsStaleInput (0.00s)
--- PASS: TestReconcileDropsAckedAndReplaysFuture (0.00s)
--- PASS: TestReconcileDeviationTable (0.00s)
--- PASS: TestPendingReplayOrder (0.00s)
--- PASS: TestAggressionWindowLimitsPrediction (0.00s)
--- PASS: TestTuneLadder (0.00s)
PASS
ok  	gsarchlab/internal/authority	0.287s