《后端架构设计·分布式与数据一致性》4.2 选主、日志复制与提交规则的可运行实现

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

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

// Role is the node role for the current term.
type Role uint8

const (
	Follower Role = iota
	Candidate
	Leader
)

type Entry struct {
	Index uint64
	Term  uint64
	Cmd   string
}

type Kind uint8

const (
	MsgVote Kind = iota
	MsgVoteResp
	MsgAppend
	MsgAppendResp
)

// Message is one in-flight RPC; fields are shared across kinds.
type Message struct {
	From, To            string
	Kind                Kind
	Term                uint64
	LastIndex, LastTerm uint64
	Granted             bool
	PrevIndex, PrevTerm uint64
	Entries             []Entry
	LeaderCommit        uint64
	Success             bool
	MatchIndex          uint64
}

var ErrNotLeader = errors.New("raft: not leader")

// UpToDate compares (lastTerm, lastIndex) lexicographically.
func UpToDate(ct, ci, lt, li uint64) bool {
	if ct != lt {
		return ct > lt
	}
	return ci >= li
}

// Node is driven by logical ticks; it owns no clock and no goroutines.
type Node struct {
	ID     string
	Role   Role
	Term   uint64
	Vote   string
	Leader string
	Log    []Entry
	Commit uint64

	peers      []string
	electionAt int
	next       map[string]uint64
	match      map[string]uint64
	votes      map[string]bool
}

func NewNode(id string, peers []string) *Node {
	n := &Node{ID: id, Log: []Entry{{}}, peers: peers,
		next: map[string]uint64{}, match: map[string]uint64{}, votes: map[string]bool{}}
	for _, p := range peers {
		n.match[p] = 0
	}
	return n
}

func (n *Node) quorum() int            { return (len(n.peers)+1)/2 + 1 }
func (n *Node) lastIndex() uint64      { return n.Log[len(n.Log)-1].Index }
func (n *Node) lastTerm() uint64       { return n.Log[len(n.Log)-1].Term }
func (n *Node) termAt(i uint64) uint64 { return n.Log[i].Term }

// resetElection gives each id and term a different deterministic timeout.
func (n *Node) resetElection(now int) {
	n.electionAt = now + 6 + int((uint64(n.ID[0])+n.Term)%5)
}

func (n *Node) becomeFollower(term uint64, now int, leader string) {
	if term > n.Term {
		n.Term, n.Vote = term, ""
	}
	n.Role, n.Leader = Follower, leader
	n.votes = map[string]bool{}
	n.resetElection(now)
}

func (n *Node) reply(m Message, kind Kind, ok bool) Message {
	return Message{From: n.ID, To: m.From, Kind: kind, Term: n.Term, Granted: ok, Success: ok}
}

// Step processes one message and returns messages queued for the next tick.
func (n *Node) Step(now int, m Message) []Message {
	switch m.Kind {
	case MsgVote:
		if m.Term < n.Term {
			return []Message{n.reply(m, MsgVoteResp, false)}
		}
		if m.Term > n.Term {
			n.becomeFollower(m.Term, now, "")
		}
		ok := (n.Vote == "" || n.Vote == m.From) &&
			UpToDate(m.LastTerm, m.LastIndex, n.lastTerm(), n.lastIndex())
		if ok {
			n.Vote = m.From
			n.resetElection(now)
		}
		return []Message{n.reply(m, MsgVoteResp, ok)}

	case MsgVoteResp:
		if m.Term > n.Term {
			n.becomeFollower(m.Term, now, "")
			return nil
		}
		if n.Role != Candidate || m.Term != n.Term || !m.Granted {
			return nil
		}
		n.votes[m.From] = true
		if len(n.votes) < n.quorum() {
			return nil
		}
		n.Role, n.Leader, n.Vote = Leader, n.ID, n.ID
		for _, p := range n.peers {
			n.next[p], n.match[p] = n.lastIndex()+1, 0
		}
		return n.appendAll()

	case MsgAppend:
		if m.Term < n.Term {
			r := n.reply(m, MsgAppendResp, false)
			r.MatchIndex = n.lastIndex()
			return []Message{r}
		}
		n.becomeFollower(m.Term, now, m.From)
		if m.PrevIndex > n.lastIndex() || n.termAt(m.PrevIndex) != m.PrevTerm {
			r := n.reply(m, MsgAppendResp, false)
			r.MatchIndex = n.lastIndex()
			if m.PrevIndex > 0 && m.PrevIndex <= n.lastIndex() {
				r.MatchIndex = m.PrevIndex - 1
			}
			return []Message{r}
		}
		for _, e := range m.Entries {
			if e.Index <= n.lastIndex() && n.termAt(e.Index) == e.Term {
				continue
			}
			if e.Index <= n.lastIndex() {
				n.Log = n.Log[:e.Index]
			}
			n.Log = append(n.Log, e)
		}
		if m.LeaderCommit > n.Commit {
			n.Commit = min(m.LeaderCommit, n.lastIndex())
		}
		r := n.reply(m, MsgAppendResp, true)
		r.MatchIndex = m.PrevIndex + uint64(len(m.Entries))
		return []Message{r}

	case MsgAppendResp:
		if m.Term > n.Term {
			n.becomeFollower(m.Term, now, "")
			return nil
		}
		if n.Role != Leader || m.Term != n.Term {
			return nil
		}
		if m.Success {
			if m.MatchIndex > n.match[m.From] {
				n.match[m.From] = m.MatchIndex
			}
			n.next[m.From] = n.match[m.From] + 1
			n.advanceCommit()
			return nil
		}
		next := n.next[m.From]
		if m.MatchIndex+1 < next {
			next = m.MatchIndex + 1
		}
		if next < 1 {
			next = 1
		}
		n.next[m.From] = next
		return []Message{n.appendFor(m.From)}
	}
	return nil
}

func (n *Node) appendFor(p string) Message {
	next := n.next[p]
	if next < 1 {
		next = 1
	}
	m := Message{From: n.ID, To: p, Kind: MsgAppend, Term: n.Term,
		PrevIndex: next - 1, PrevTerm: n.termAt(next - 1), LeaderCommit: n.Commit}
	m.Entries = append(m.Entries, n.Log[next:]...)
	return m
}

func (n *Node) appendAll() []Message {
	out := make([]Message, 0, len(n.peers))
	for _, p := range n.peers {
		out = append(out, n.appendFor(p))
	}
	return out
}

// advanceCommit counts the leader's own tail plus every peer's match index.
func (n *Node) advanceCommit() {
	idx := make([]uint64, 0, len(n.peers)+1)
	idx = append(idx, n.lastIndex())
	for _, p := range n.peers {
		idx = append(idx, n.match[p])
	}
	sort.Slice(idx, func(i, j int) bool { return idx[i] < idx[j] })
	c := idx[len(idx)-n.quorum()]
	if c > n.Commit && n.termAt(c) == n.Term {
		n.Commit = c
	}
}

func (n *Node) TickTimeout(now int) []Message {
	if n.Role == Leader || now < n.electionAt {
		return nil
	}
	n.Role, n.Term, n.Vote = Candidate, n.Term+1, n.ID
	n.votes = map[string]bool{n.ID: true}
	n.resetElection(now)
	out := make([]Message, 0, len(n.peers))
	for _, p := range n.peers {
		out = append(out, Message{From: n.ID, To: p, Kind: MsgVote, Term: n.Term,
			LastIndex: n.lastIndex(), LastTerm: n.lastTerm()})
	}
	return out
}

// Propose appends a command on the leader; acceptance is not commit.
func (n *Node) Propose(cmd string) error {
	if n.Role != Leader {
		return fmt.Errorf("raft: %s cannot propose: %w", n.ID, ErrNotLeader)
	}
	n.Log = append(n.Log, Entry{Index: n.lastIndex() + 1, Term: n.Term, Cmd: cmd})
	return nil
}

=== RUN   TestElectionIsSingleAndConverges
--- PASS: TestElectionIsSingleAndConverges (0.00s)
=== RUN   TestReplicationCommitsOnMajority
--- PASS: TestReplicationCommitsOnMajority (0.00s)
=== RUN   TestMinorityLeaderCannotCommitOrOverwrite
--- PASS: TestMinorityLeaderCannotCommitOrOverwrite (0.00s)
=== RUN   TestAppendEntriesConflictRollback
--- PASS: TestAppendEntriesConflictRollback (0.00s)
=== RUN   TestProposeOnFollowerFails
--- PASS: TestProposeOnFollowerFails (0.00s)
PASS
ok  	archlab/internal/platform/consensus/raft	0.272s