《游戏服务端架构·规模化、全球化与成本》4.3 再均衡期间的读写

source --advance--> dual --advance--> target
source <--rollback-- dual <--rollback-- target

// file: internal/rebalance/machine.go
// tier: L1
// verify: go test ./internal/rebalance/
package rebalance

import (
	"errors"
	"fmt"
)

// Phase is the routing contract during a migration. Every phase names exactly
// one authoritative writer; "dual" is the only phase with two writes, and even
// there only the source is read as the truth.
type Phase string

const (
	PhaseSource Phase = "source"
	PhaseDual   Phase = "dual"
	PhaseTarget Phase = "target"
)

var (
	ErrEmptyTarget = errors.New("rebalance: source and target must be non-empty")
	ErrSameTarget  = errors.New("rebalance: source and target are the same")
	ErrTransition  = errors.New("rebalance: transition not allowed")
	ErrVersion     = errors.New("rebalance: version must not be negative")
	ErrConflict    = errors.New("rebalance: equal versions with different values")
	ErrRequestID   = errors.New("rebalance: empty request id")
)

// Machine moves source -> dual -> target and can roll back one step at a time.
// It owns no data; it only decides which operations the router may allow.
type Machine struct {
	source string
	target string
	phase  Phase
}

func NewMachine(source, target string) (*Machine, error) {
	if source == "" || target == "" {
		return nil, ErrEmptyTarget
	}
	if source == target {
		return nil, ErrSameTarget
	}
	return &Machine{source: source, target: target, phase: PhaseSource}, nil
}

func (m *Machine) Phase() Phase   { return m.phase }
func (m *Machine) Source() string { return m.source }
func (m *Machine) Target() string { return m.target }

// Advance moves one step forward. Cutover is deliberate: there is no method
// that jumps from source to target and skips the dual-write window.
func (m *Machine) Advance() error {
	switch m.phase {
	case PhaseSource:
		m.phase = PhaseDual
	case PhaseDual:
		m.phase = PhaseTarget
	default:
		return fmt.Errorf("%w: %s -> forward", ErrTransition, m.phase)
	}
	return nil
}

// Rollback moves one step back. A failed cutover returns to dual write, not to
// source-only: the target may already hold writes that the source has not seen.
func (m *Machine) Rollback() error {
	switch m.phase {
	case PhaseDual:
		m.phase = PhaseSource
	case PhaseTarget:
		m.phase = PhaseDual
	default:
		return fmt.Errorf("%w: %s -> backward", ErrTransition, m.phase)
	}
	return nil
}

// Route is the read/write plan for one logical key. Mirror is empty unless the
// machine is in dual-write phase and the caller is performing a write.
type Route struct {
	Read   string
	Write  string
	Mirror string
}

func (m *Machine) Route(write bool) Route {
	switch m.phase {
	case PhaseDual:
		if write {
			return Route{Read: m.source, Write: m.source, Mirror: m.target}
		}
		return Route{Read: m.source}
	case PhaseTarget:
		if write {
			return Route{Read: m.target, Write: m.target}
		}
		return Route{Read: m.target}
	default:
		if write {
			return Route{Read: m.source, Write: m.source}
		}
		return Route{Read: m.source}
	}
}

// Record is one versioned value. Versions are logical, assigned by the owning
// service, not by wall-clock time: two replicas must not invent their own.
type Record struct {
	Version int64
	Value   string
}

// Decision says which side wins and whether the loser needs read repair.
type Decision struct {
	Winner Record
	Repair bool
}

// Resolve applies last-writer-wins by version. Equal versions with different
// values are a conflict, not a tie to break silently; that is the point where
// a retry may have produced two legitimate writes.
func Resolve(source, target Record) (Decision, error) {
	if source.Version < 0 || target.Version < 0 {
		return Decision{}, ErrVersion
	}
	switch {
	case source.Version > target.Version:
		return Decision{Winner: source, Repair: true}, nil
	case target.Version > source.Version:
		return Decision{Winner: target, Repair: true}, nil
	case source.Value != target.Value:
		return Decision{}, fmt.Errorf("%w: version %d", ErrConflict, source.Version)
	default:
		return Decision{Winner: source}, nil
	}
}

// ApplyResult distinguishes a fresh write, a retry and an older write. The
// caller must not retry a duplicate into a different shard without this bit.
type ApplyResult struct {
	Applied   bool
	Duplicate bool
	Stale     bool
	Current   Record
}

// Ledger is the idempotency table. It must be shared by every node that can
// accept the same request, or a retry reaching the target will be applied
// twice no matter how careful the router is.
type Ledger struct {
	byRequest map[string]Record
	current   Record
}

func NewLedger() *Ledger {
	return &Ledger{byRequest: make(map[string]Record)}
}

func (l *Ledger) Current() Record { return l.current }

func (l *Ledger) Apply(requestID string, rec Record) (ApplyResult, error) {
	if requestID == "" {
		return ApplyResult{}, ErrRequestID
	}
	if rec.Version < 0 {
		return ApplyResult{}, ErrVersion
	}
	if prior, ok := l.byRequest[requestID]; ok {
		return ApplyResult{Duplicate: true, Current: prior}, nil
	}
	switch {
	case rec.Version < l.current.Version:
		l.byRequest[requestID] = l.current
		return ApplyResult{Stale: true, Current: l.current}, nil
	case rec.Version == l.current.Version && rec.Value != l.current.Value:
		return ApplyResult{}, fmt.Errorf("%w: version %d", ErrConflict, rec.Version)
	}
	l.byRequest[requestID] = rec
	l.current = rec
	return ApplyResult{Applied: true, Current: rec}, nil
}