《后端架构设计·分布式与数据一致性》4.1 共识问题与 Raft 的角色、任期、日志

// file: internal/platform/consensus/state.go
// tier: L1
// verify: go test ./internal/platform/consensus/ -run Test -count=1
package consensus

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

type Role string

const (
	RoleFollower  Role = "follower"
	RoleCandidate Role = "candidate"
	RoleLeader    Role = "leader"
)

var (
	ErrBrokenLog      = errors.New("consensus: broken log")
	ErrElectionSafety = errors.New("consensus: election safety violated")
)

// Entry is one replicated command; index 0 is a sentinel in every Log.
type Entry struct {
	Index uint64
	Term  uint64
	Cmd   string
}

type Log struct {
	entries []Entry
}

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) EntryAt(index uint64) (Entry, bool) {
	if index > l.LastIndex() {
		return Entry{}, false
	}
	return l.entries[index], true
}

// Append enforces contiguous indexes and non-decreasing terms.
func (l *Log) Append(e Entry) error {
	if e.Index != l.LastIndex()+1 {
		return fmt.Errorf("consensus: append index %d after %d: %w", e.Index, l.LastIndex(), ErrBrokenLog)
	}
	if e.Term < l.LastTerm() {
		return fmt.Errorf("consensus: term %d below last %d: %w", e.Term, l.LastTerm(), ErrBrokenLog)
	}
	l.entries = append(l.entries, e)
	return nil
}

// TruncateFrom removes index and everything after it on a conflicting follower.
func (l *Log) TruncateFrom(index uint64) error {
	if index == 0 || index > l.LastIndex() {
		return fmt.Errorf("consensus: truncate %d outside (0,%d]: %w", index, l.LastIndex(), ErrBrokenLog)
	}
	l.entries = l.entries[:index]
	return nil
}

// Node is the crash-persistent state: term, vote, log, plus two watermarks.
type Node struct {
	ID       string
	Role     Role
	Term     uint64
	VotedFor string
	Log      *Log
	Commit   uint64
	Applied  uint64
	Cluster  int
}

func NewNode(id string, cluster int) *Node {
	return &Node{ID: id, Role: RoleFollower, Log: NewLog(), Cluster: cluster}
}

func (n *Node) Quorum() int { return n.Cluster/2 + 1 }

// BecomeFollower raises the term and clears the vote when the term advances.
func (n *Node) BecomeFollower(term uint64) {
	if term > n.Term {
		n.Term = term
		n.VotedFor = ""
	}
	n.Role = RoleFollower
}

// UpToDate compares (lastTerm, lastIndex) lexicographically.
func UpToDate(candTerm, candIndex, localTerm, localIndex uint64) bool {
	if candTerm != localTerm {
		return candTerm > localTerm
	}
	return candIndex >= localIndex
}

type VoteRequest struct {
	Term        uint64
	CandidateID string
	LastIndex   uint64
	LastTerm    uint64
}

type VoteResponse struct {
	Term    uint64
	Granted bool
}

// HandleVoteRequest grants at most one vote per term and only to an up-to-date log.
func (n *Node) HandleVoteRequest(req VoteRequest) VoteResponse {
	if req.Term < n.Term {
		return VoteResponse{Term: n.Term}
	}
	if req.Term > n.Term {
		n.BecomeFollower(req.Term)
	}
	if n.VotedFor != "" && n.VotedFor != req.CandidateID {
		return VoteResponse{Term: n.Term}
	}
	if !UpToDate(req.LastTerm, req.LastIndex, n.Log.LastTerm(), n.Log.LastIndex()) {
		return VoteResponse{Term: n.Term}
	}
	n.VotedFor = req.CandidateID
	return VoteResponse{Term: n.Term, Granted: true}
}

// AdvanceCommit needs a majority match and an entry from the leader's own term.
// match must contain every other voter, including nodes known to have nothing.
func (n *Node) AdvanceCommit(match map[string]uint64) uint64 {
	if n.Role != RoleLeader {
		return n.Commit
	}
	indexes := make([]uint64, 0, len(match)+1)
	indexes = append(indexes, n.Log.LastIndex())
	for _, idx := range match {
		indexes = append(indexes, idx)
	}
	sort.Slice(indexes, func(i, j int) bool { return indexes[i] < indexes[j] })
	candidate := indexes[len(indexes)-n.Quorum()]
	if candidate <= n.Commit {
		return n.Commit
	}
	entry, ok := n.Log.EntryAt(candidate)
	if !ok || entry.Term != n.Term {
		return n.Commit
	}
	n.Commit = candidate
	return n.Commit
}

// Check verifies the local invariants that do not need a second node.
func (n *Node) Check() error {
	if n.Commit > n.Log.LastIndex() {
		return fmt.Errorf("consensus: commit %d beyond last %d: %w", n.Commit, n.Log.LastIndex(), ErrBrokenLog)
	}
	if n.Applied > n.Commit {
		return fmt.Errorf("consensus: applied %d beyond commit %d: %w", n.Applied, n.Commit, ErrBrokenLog)
	}
	for i := 1; i < len(n.Log.entries); i++ {
		prev, cur := n.Log.entries[i-1], n.Log.entries[i]
		if cur.Index != prev.Index+1 || cur.Term < prev.Term {
			return fmt.Errorf("consensus: entries %d/%d break contiguity or term order: %w", prev.Index, cur.Index, ErrBrokenLog)
		}
	}
	return nil
}

// ValidateCluster checks one leader per term and agreement on committed prefixes.
func ValidateCluster(nodes []*Node) error {
	leaders := make(map[uint64][]string)
	for _, n := range nodes {
		if err := n.Check(); err != nil {
			return err
		}
		if n.Role == RoleLeader {
			leaders[n.Term] = append(leaders[n.Term], n.ID)
		}
	}
	for term, ids := range leaders {
		if len(ids) > 1 {
			return fmt.Errorf("consensus: term %d has leaders %v: %w", term, ids, ErrElectionSafety)
		}
	}
	for i := 0; i < len(nodes); i++ {
		for j := i + 1; j < len(nodes); j++ {
			limit := nodes[i].Commit
			if nodes[j].Commit < limit {
				limit = nodes[j].Commit
			}
			for idx := uint64(1); idx <= limit; idx++ {
				a, aok := nodes[i].Log.EntryAt(idx)
				b, bok := nodes[j].Log.EntryAt(idx)
				if !aok || !bok || a.Term != b.Term || a.Cmd != b.Cmd {
					return fmt.Errorf("consensus: committed index %d differs on %s/%s: %w", idx, nodes[i].ID, nodes[j].ID, ErrBrokenLog)
				}
			}
		}
	}
	return nil
}

=== RUN   TestVoteDecisionTable
--- PASS: TestVoteDecisionTable (0.00s)
=== RUN   TestTermOnlyMovesForward
--- PASS: TestTermOnlyMovesForward (0.00s)
=== RUN   TestAdvanceCommitNeedsMajorityAndCurrentTerm
--- PASS: TestAdvanceCommitNeedsMajorityAndCurrentTerm (0.00s)
=== RUN   TestValidateClusterCatchesSplitBrainAndDivergence
--- PASS: TestValidateClusterCatchesSplitBrainAndDivergence (0.00s)
PASS
ok  	archlab/internal/platform/consensus	0.314s