《后端架构设计·分布式与数据一致性》4.3 成员变更、快照与线性一致读

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

import (
	"errors"
	"fmt"
	"sort"
)

var (
	ErrBadChange   = errors.New("raftx: membership change must add or remove exactly one voter")
	ErrBrokenLog   = errors.New("raftx: broken log")
	ErrNoQuorum    = errors.New("raftx: read index needs a current-term quorum")
	ErrNotApplied  = errors.New("raftx: applied index behind read index")
	ErrSnapshotGap = errors.New("raftx: snapshot index outside compactable range")
)

// Config is one voting configuration; it is copied, never mutated in place.
type Config struct {
	Voters map[string]bool
}

func NewConfig(ids ...string) Config {
	c := Config{Voters: make(map[string]bool, len(ids))}
	for _, id := range ids {
		c.Voters[id] = true
	}
	return c
}

func (c Config) Quorum() int        { return len(c.Voters)/2 + 1 }
func (c Config) Has(id string) bool { return c.Voters[id] }

func (c Config) Clone() Config {
	out := Config{Voters: make(map[string]bool, len(c.Voters))}
	for id := range c.Voters {
		out.Voters[id] = true
	}
	return out
}

// CheckSingleChange accepts configurations that differ by exactly one voter.
// One-at-a-time changes make every old majority intersect every new majority.
func CheckSingleChange(old, next Config) error {
	added, removed := 0, 0
	for id := range next.Voters {
		if !old.Voters[id] {
			added++
		}
	}
	for id := range old.Voters {
		if !next.Voters[id] {
			removed++
		}
	}
	if added+removed != 1 {
		return fmt.Errorf("raftx: added=%d removed=%d: %w", added, removed, ErrBadChange)
	}
	if len(next.Voters) == 0 {
		return fmt.Errorf("raftx: empty configuration: %w", ErrBadChange)
	}
	return nil
}

// AdvanceCommit counts a quorum of the current config; an old-term entry is
// never committed directly, only carried by a current-term entry.
func AdvanceCommit(cfg Config, match map[string]uint64, commit uint64, termAt func(uint64) uint64, currentTerm uint64) uint64 {
	indexes := make([]uint64, 0, len(cfg.Voters))
	for id := range cfg.Voters {
		indexes = append(indexes, match[id])
	}
	sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] })
	candidate := indexes[len(indexes)-cfg.Quorum()]
	if candidate <= commit {
		return commit
	}
	if termAt(candidate) != currentTerm {
		return commit
	}
	return candidate
}

// Entry is a replicated command; index 1 is the first real entry.
type Entry struct {
	Index uint64
	Term  uint64
	Cmd   string
}

// Log is compactable: Entries[0] is the boundary at Base. An index below Base
// is covered by a snapshot instead of a log entry.
type Log struct {
	Entries  []Entry
	Base     uint64
	BaseTerm uint64
}

func NewLog() *Log {
	return &Log{Entries: []Entry{{Index: 0, Term: 0}}}
}

func (l *Log) LastIndex() uint64 { return l.Entries[len(l.Entries)-1].Index }
func (l *Log) LastTerm() uint64  { return l.Entries[len(l.Entries)-1].Term }

func (l *Log) TermAt(index uint64) (uint64, bool) {
	if index < l.Base || index > l.LastIndex() {
		return 0, false
	}
	return l.Entries[index-l.Base].Term, true
}

func (l *Log) Append(e Entry) error {
	if e.Index != l.LastIndex()+1 {
		return fmt.Errorf("raftx: append %d after %d: %w", e.Index, l.LastIndex(), ErrBrokenLog)
	}
	if e.Term < l.LastTerm() {
		return fmt.Errorf("raftx: term %d below %d: %w", e.Term, l.LastTerm(), ErrBrokenLog)
	}
	l.Entries = append(l.Entries, e)
	return nil
}

// Compact drops the prefix through index and keeps index as the new boundary.
func (l *Log) Compact(index, applied uint64) error {
	if index < l.Base || index > l.LastIndex() || index > applied {
		return fmt.Errorf("raftx: compact %d base=%d last=%d applied=%d: %w",
			index, l.Base, l.LastIndex(), applied, ErrSnapshotGap)
	}
	l.Entries = append([]Entry(nil), l.Entries[index-l.Base:]...)
	l.Base = index
	l.BaseTerm = l.Entries[0].Term
	return nil
}

// Snapshot captures the boundary entry, the config and the state-machine image.
type Snapshot struct {
	LastIndex uint64
	LastTerm  uint64
	Config    Config
	Data      []byte
}

func (l *Log) Snapshot(index, applied uint64, cfg Config, data []byte) (Snapshot, error) {
	term, ok := l.TermAt(index)
	if !ok || index > applied {
		return Snapshot{}, fmt.Errorf("raftx: snapshot %d applied=%d: %w", index, applied, ErrSnapshotGap)
	}
	return Snapshot{LastIndex: index, LastTerm: term, Config: cfg.Clone(), Data: data}, nil
}

// Store is the follower-side state touched by InstallSnapshot.
type Store struct {
	Log     *Log
	Commit  uint64
	Applied uint64
	Config  Config
}

// InstallSnapshot keeps a matching local suffix and replaces a conflicting one.
func (s *Store) InstallSnapshot(snap Snapshot) error {
	if snap.LastIndex <= s.Log.Base {
		return nil
	}
	keep := false
	if snap.LastIndex <= s.Log.LastIndex() {
		term, ok := s.Log.TermAt(snap.LastIndex)
		keep = ok && term == snap.LastTerm
	}
	if keep {
		s.Log.Entries = append([]Entry(nil), s.Log.Entries[snap.LastIndex-s.Log.Base:]...)
	} else {
		s.Log.Entries = []Entry{{Index: snap.LastIndex, Term: snap.LastTerm}}
	}
	s.Log.Base, s.Log.BaseTerm = snap.LastIndex, snap.LastTerm
	if s.Commit < snap.LastIndex {
		s.Commit = snap.LastIndex
	}
	if s.Applied < snap.LastIndex {
		s.Applied = snap.LastIndex
	}
	s.Config = snap.Config.Clone()
	return nil
}

// ReadIndex returns a safe read index once a quorum acks the leader's term.
// A stale leader cannot collect current-term acks and fails closed.
func ReadIndex(term, commit uint64, cfg Config, ackedTerm map[string]uint64) (uint64, error) {
	acks := 0
	for id := range cfg.Voters {
		if ackedTerm[id] == term {
			acks++
		}
	}
	if acks < cfg.Quorum() {
		return 0, fmt.Errorf("raftx: %d/%d current-term acks: %w", acks, cfg.Quorum(), ErrNoQuorum)
	}
	return commit, nil
}

// CanServe rejects a read until the state machine has applied the read index.
func CanServe(applied, readIndex uint64) error {
	if applied < readIndex {
		return fmt.Errorf("raftx: applied %d < read %d: %w", applied, readIndex, ErrNotApplied)
	}
	return nil
}

=== RUN   TestSingleChangeKeepsQuorumsOverlapping
--- PASS: TestSingleChangeKeepsQuorumsOverlapping (0.00s)
=== RUN   TestTwoNodeChangeIsRejected
--- PASS: TestTwoNodeChangeIsRejected (0.00s)
=== RUN   TestAdvanceCommitUsesCurrentConfig
--- PASS: TestAdvanceCommitUsesCurrentConfig (0.00s)
=== RUN   TestCompactAndInstallSnapshot
--- PASS: TestCompactAndInstallSnapshot (0.00s)
=== RUN   TestReadIndexNeedsCurrentTermQuorum
--- PASS: TestReadIndexNeedsCurrentTermQuorum (0.00s)
PASS
ok  	archlab/internal/platform/consensus/raftx	0.263s