《后端架构设计·分布式与数据一致性》13.1 订单系统的分布式改造：从单库到分片

// file: internal/platform/sharding/sharding.go
// tier: L2
// verify: go test ./internal/platform/sharding/ -count=1
package sharding

import (
	"errors"
	"fmt"
	"sort"
	"strconv"
	"sync"
)

var (
	ErrClockRegression = errors.New("clock moved backwards")
	ErrTargetWrite     = errors.New("target shard write failed")
	ErrPendingWrites   = errors.New("pending dual-write repairs")
	ErrMismatch        = errors.New("shadow read mismatch")
	ErrCutoverDone     = errors.New("cutover already completed")
	ErrInvalidPhase    = errors.New("invalid migration phase")
)

// Router maps a user to a shard with rendezvous hashing. Adding one shard
// only moves the keys the new shard wins, instead of almost every key.
type Router struct{ Shards int }

// ShardOf returns the shard index for userID. The splitmix finalizer is part
// of the contract: FNV alone is biased when only the node suffix changes.
func (r Router) ShardOf(userID string) int {
	if r.Shards <= 1 {
		return 0
	}
	best, score := 0, mix(fnv1a(userID+"#0"))
	for i := 1; i < r.Shards; i++ {
		if s := mix(fnv1a(userID + "#" + strconv.Itoa(i))); s > score {
			best, score = i, s
		}
	}
	return best
}

func mix(x uint64) uint64 {
	x += 0x9e3779b97f4a7c15
	x = (x ^ (x >> 30)) * 0xbf58476d1ce4e5b9
	x = (x ^ (x >> 27)) * 0x94d049bb133111eb
	return x ^ (x >> 31)
}

func fnv1a(s string) uint64 {
	const offset, prime = 14695981039346656037, 1099511628211
	var h uint64 = offset
	for i := 0; i < len(s); i++ {
		h ^= uint64(s[i])
		h *= prime
	}
	return h
}

// Order is the aggregate stored in a user shard.
type Order struct {
	ID, UserID, Status string
	Amount             int64
	Version            int64
}

// Shard is an in-process, versioned order store.
type Shard struct {
	mu   sync.Mutex
	data map[string]Order
}

func NewShard() *Shard { return &Shard{data: make(map[string]Order)} }

// Put writes only when the incoming version is not older.
func (s *Shard) Put(o Order) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if old, ok := s.data[o.ID]; ok && old.Version > o.Version {
		return
	}
	s.data[o.ID] = o
}

func (s *Shard) All() []Order {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Order, 0, len(s.data))
	for _, o := range s.data {
		out = append(out, o)
	}
	return out
}

// Cluster routes user writes and scatters cross-user reads.
type Cluster struct {
	router Router
	shards []*Shard
}

func NewCluster(r Router) *Cluster {
	if r.Shards < 1 {
		panic("sharding: router needs at least one shard")
	}
	c := &Cluster{router: r, shards: make([]*Shard, r.Shards)}
	for i := range c.shards {
		c.shards[i] = NewShard()
	}
	return c
}

func (c *Cluster) Put(o Order) { c.shards[c.router.ShardOf(o.UserID)].Put(o) }

func (c *Cluster) All() []Order {
	var out []Order
	for _, s := range c.shards {
		out = append(out, s.All()...)
	}
	return out
}

// ByStatus reads every shard and merges the partial results. This is the
// query shape that a user shard key does not make cheap.
func (c *Cluster) ByStatus(status string) []Order {
	var out []Order
	for _, o := range c.All() {
		if o.Status == status {
			out = append(out, o)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
	return out
}

// IDGen is Snowflake-shaped: 41 bits of milliseconds, 10 bits of node and
// 12 bits of sequence. The clock is injected so tests never sleep.
type IDGen struct {
	node, last, seq int64
	now             func() int64
}

const epochMillis int64 = 1700000000000

func NewIDGen(node int64, now func() int64) *IDGen {
	return &IDGen{node: node & 1023, now: now}
}

func (g *IDGen) Next() (int64, error) {
	ms := g.now()
	if ms < g.last {
		return 0, fmt.Errorf("%w: last=%d now=%d", ErrClockRegression, g.last, ms)
	}
	if ms == g.last {
		g.seq++
		if g.seq > 4095 {
			return 0, errors.New("idgen: sequence exhausted in one millisecond")
		}
	} else {
		g.last, g.seq = ms, 0
	}
	return ((ms - epochMillis) << 22) | (g.node << 12) | g.seq, nil
}

type Phase string

const (
	PhaseDualWrite  Phase = "dual_write"
	PhaseBackfill   Phase = "backfill"
	PhaseVerify     Phase = "verify"
	PhaseCutover    Phase = "cutover"
	PhaseRolledBack Phase = "rolled_back"
)

// Migration dual-writes new orders, backfills history and gates cutover on
// an empty repair list and zero shadow-read mismatches.
type Migration struct {
	from, to         Router
	source, target   *Cluster
	phase            Phase
	journal, pending map[string]bool
	mismatches       int
	verified         bool
	failTarget       bool
}

func NewMigration(from, to Router, source *Cluster) *Migration {
	return &Migration{from: from, to: to, source: source, target: NewCluster(to),
		phase: PhaseDualWrite, journal: map[string]bool{}, pending: map[string]bool{}}
}

func (m *Migration) Phase() Phase { return m.phase }

func (m *Migration) StartBackfill() { m.phase = PhaseBackfill }

// Write sends the order to the source and, during dual-write, to the target.
// A target failure is recorded for retry and blocks cutover.
func (m *Migration) Write(o Order) error {
	if m.phase == PhaseCutover || m.phase == PhaseRolledBack {
		return fmt.Errorf("write in phase %s: %w", m.phase, ErrInvalidPhase)
	}
	m.source.Put(o)
	if m.failTarget {
		m.pending[o.ID] = true
		return fmt.Errorf("order %s: %w", o.ID, ErrTargetWrite)
	}
	m.target.Put(o)
	delete(m.pending, o.ID)
	m.journal[o.ID] = true
	return nil
}

// Backfill copies at most limit uncopied source orders and is re-entrant.
func (m *Migration) Backfill(limit int) int {
	copied := 0
	for _, o := range m.source.All() {
		if copied >= limit {
			break
		}
		if m.journal[o.ID] {
			continue
		}
		m.target.Put(o)
		m.journal[o.ID] = true
		copied++
	}
	return copied
}

func (m *Migration) Verify() int {
	target := make(map[string]Order)
	for _, o := range m.target.All() {
		target[o.ID] = o
	}
	m.mismatches, m.verified = 0, true
	for _, src := range m.source.All() {
		dst, ok := target[src.ID]
		if !ok || !m.journal[src.ID] || dst.Version != src.Version || dst.Status != src.Status {
			m.mismatches++
		}
	}
	return m.mismatches
}

// CanCutover enforces all orders copied, no pending repair and no mismatch.
func (m *Migration) CanCutover() error {
	if len(m.pending) > 0 {
		return fmt.Errorf("%d orders: %w", len(m.pending), ErrPendingWrites)
	}
	for _, o := range m.source.All() {
		if !m.journal[o.ID] {
			return fmt.Errorf("order %s not copied: %w", o.ID, ErrPendingWrites)
		}
	}
	if !m.verified {
		return fmt.Errorf("shadow read not run: %w", ErrMismatch)
	}
	if m.mismatches > 0 {
		return fmt.Errorf("%d mismatches: %w", m.mismatches, ErrMismatch)
	}
	return nil
}

func (m *Migration) Cutover() error {
	if err := m.CanCutover(); err != nil {
		return err
	}
	m.phase = PhaseCutover
	return nil
}

// Rollback is allowed only before cutover; afterwards a reverse migration is
// required because the target already accepted writes.
func (m *Migration) Rollback() error {
	if m.phase == PhaseCutover {
		return ErrCutoverDone
	}
	m.phase = PhaseRolledBack
	return nil
}

=== RUN   TestRendezvousMovesFewerKeys
    sharding_test.go:25: 4->5 moved: rendezvous=0.205 modulo=0.799
--- PASS: TestRendezvousMovesFewerKeys (0.01s)
=== RUN   TestClusterQueries
--- PASS: TestClusterQueries (0.00s)
=== RUN   TestIDGenUniqueAndClockGuard
--- PASS: TestIDGenUniqueAndClockGuard (0.00s)
=== RUN   TestMigrationCutoverGate
--- PASS: TestMigrationCutoverGate (0.00s)
PASS
ok  	archlab/internal/platform/sharding	0.281s