【编程难度第二名】一致性协议(Raft / Paxos) - 行业里最少人能写出来的代码

type Node struct {
    id          string
    currentTerm int64
    votedFor    string
    state       NodeState  // Follower, Candidate, Leader
    votes       map[string]bool

    // 日志相关
    log          []LogEntry
    lastLogIndex int64
    lastLogTerm  int64

    // 通信
    peers       []Peer
    electionWon chan bool
}

type NodeState int

const (
    Follower NodeState = iota
    Candidate
    Leader
)

func (n *Node) StartElection() {
    // 1. 增加任期
    n.currentTerm++

    // 2. 转为候选人,给自己投票
    n.state = Candidate
    n.votedFor = n.id
    n.votes = map[string]bool{n.id: true}

    log.Info("starting election",
        "nodeID", n.id,
        "term", n.currentTerm)

    // 3. 向所有其他节点请求投票
    for _, peer := range n.peers {
        go n.requestVote(peer, n.currentTerm)
    }

    // 4. 设置选举超时(随机 150-300ms)
    timeout := randomTimeout(150, 300)
    timer := time.NewTimer(timeout)

    select {
    case <-timer.C:
        // 超时,重新选举
        log.Warn("election timeout, restarting",
            "nodeID", n.id,
            "term", n.currentTerm)
        n.StartElection()

    case <-n.electionWon:
        // 赢得选举,成为 Leader
        log.Info("election won",
            "nodeID", n.id,
            "term", n.currentTerm,
            "votes", len(n.votes))
        n.becomeLeader()
    }
}

func (n *Node) requestVote(peer Peer, term int64) {
    resp, err := peer.RequestVote(RequestVoteArgs{
        Term:         term,
        CandidateId:  n.id,
        LastLogIndex: n.lastLogIndex,
        LastLogTerm:  n.lastLogTerm,
    })

    if err != nil {
        log.Error("request vote failed",
            "peer", peer.id,
            "err", err)
        return
    }

    // 处理投票响应
    if resp.VoteGranted {
        n.mu.Lock()
        n.votes[peer.id] = true
        voteCount := len(n.votes)
        n.mu.Unlock()

        log.Info("received vote",
            "from", peer.id,
            "totalVotes", voteCount)

        // 检查是否获得多数票
        if voteCount > len(n.peers)/2 {
            select {
            case n.electionWon <- true:
            default:
            }
        }
    } else {
        log.Info("vote denied",
            "from", peer.id,
            "theirTerm", resp.Term)

        // 如果对方的 term 更高,更新自己的 term
        if resp.Term > n.currentTerm {
            n.currentTerm = resp.Term
            n.state = Follower
            n.votedFor = ""
        }
    }
}

func (n *Node) HandleRequestVote(args RequestVoteArgs) RequestVoteReply {
    n.mu.Lock()
    defer n.mu.Unlock()

    log.Info("received vote request",
        "from", args.CandidateId,
        "term", args.Term,
        "myTerm", n.currentTerm)

    // 1. 如果对方的 term 小于自己,拒绝
    if args.Term < n.currentTerm {
        return RequestVoteReply{
            Term:        n.currentTerm,
            VoteGranted: false,
        }
    }

    // 2. 如果对方的 term 大于自己,更新自己的 term
    if args.Term > n.currentTerm {
        n.currentTerm = args.Term
        n.state = Follower
        n.votedFor = ""
    }

    // 3. 检查是否已经投票
    if n.votedFor != "" && n.votedFor != args.CandidateId {
        log.Info("already voted for another candidate",
            "votedFor", n.votedFor)
        return RequestVoteReply{
            Term:        n.currentTerm,
            VoteGranted: false,
        }
    }

    // 4. 检查候选人的日志是否至少和自己一样新
    // (这是保证一致性的关键!)
    logOk := args.LastLogTerm > n.lastLogTerm ||
             (args.LastLogTerm == n.lastLogTerm &&
              args.LastLogIndex >= n.lastLogIndex)

    if !logOk {
        log.Info("candidate log not up-to-date",
            "candidateLastTerm", args.LastLogTerm,
            "myLastTerm", n.lastLogTerm)
        return RequestVoteReply{
            Term:        n.currentTerm,
            VoteGranted: false,
        }
    }

    // 5. 投票
    n.votedFor = args.CandidateId
    n.resetElectionTimeout()

    log.Info("granted vote",
        "to", args.CandidateId,
        "term", n.currentTerm)

    return RequestVoteReply{
        Term:        n.currentTerm,
        VoteGranted: true,
    }
}

type LogEntry struct {
    Index   int64
    Term    int64
    Command interface{}
}

type Node struct {
    // ... 其他字段

    // 日志
    log          []LogEntry
    lastLogIndex int64
    commitIndex  int64
    lastApplied  int64

    // Leader 维护的状态
    nextIndex    map[string]int64  // 每个 Follower 的下一条日志索引
    matchIndex   map[string]int64  // 每个 Follower 已复制的最高日志索引
}

func (n *Node) AppendEntry(ctx context.Context, cmd interface{}) error {
    if n.state != Leader {
        return errors.New("not leader")
    }

    n.mu.Lock()

    // 1. 写入本地日志
    entry := LogEntry{
        Index:   n.lastLogIndex + 1,
        Term:    n.currentTerm,
        Command: cmd,
    }
    n.log = append(n.log, entry)
    n.lastLogIndex++

    log.Info("appended entry",
        "index", entry.Index,
        "term", entry.Term)

    n.mu.Unlock()

    // 2. 并发向所有 Follower 发送
    successCount := 1  // Leader 自己
    var mu sync.Mutex
    var wg sync.WaitGroup

    for _, peer := range n.peers {
        wg.Add(1)
        go func(p Peer) {
            defer wg.Done()

            n.mu.RLock()
            prevLogIndex := n.nextIndex[p.id] - 1
            prevLogTerm := n.log[prevLogIndex].Term
            n.mu.RUnlock()

            // 发送 AppendEntries RPC
            resp, err := p.AppendEntries(ctx, AppendEntriesArgs{
                Term:         n.currentTerm,
                LeaderId:     n.id,
                PrevLogIndex: prevLogIndex,
                PrevLogTerm:  prevLogTerm,
                Entries:      []LogEntry{entry},
                LeaderCommit: n.commitIndex,
            })

            if err != nil {
                log.Error("append entries failed",
                    "peer", p.id,
                    "err", err)
                return
            }

            if resp.Success {
                mu.Lock()
                successCount++
                n.matchIndex[p.id] = entry.Index
                n.nextIndex[p.id] = entry.Index + 1
                mu.Unlock()

                log.Info("append entries success",
                    "peer", p.id,
                    "index", entry.Index)
            } else {
                // 日志不匹配,需要回退
                n.mu.Lock()
                n.nextIndex[p.id]--
                n.mu.Unlock()

                log.Warn("append entries conflict",
                    "peer", p.id,
                    "prevIndex", prevLogIndex)
            }
        }(peer)
    }

    // 等待所有响应
    wg.Wait()

    // 3. 检查是否达到多数派
    if successCount > len(n.peers)/2 {
        // 提交日志
        n.mu.Lock()
        n.commitIndex = entry.Index
        n.mu.Unlock()

        log.Info("entry committed",
            "index", entry.Index,
            "replicas", successCount)

        // 应用到状态机
        n.applyLog(entry)

        return nil
    }

    log.Error("failed to reach quorum",
        "successCount", successCount,
        "required", len(n.peers)/2+1)

    return errors.New("failed to reach quorum")
}

func (n *Node) HandleAppendEntries(args AppendEntriesArgs) AppendEntriesReply {
    n.mu.Lock()
    defer n.mu.Unlock()

    log.Info("received append entries",
        "from", args.LeaderId,
        "term", args.Term,
        "prevIndex", args.PrevLogIndex,
        "entries", len(args.Entries))

    // 1. 检查 term
    if args.Term < n.currentTerm {
        log.Info("rejecting outdated append entries",
            "theirTerm", args.Term,
            "myTerm", n.currentTerm)
        return AppendEntriesReply{
            Term:    n.currentTerm,
            Success: false,
        }
    }

    // 2. 更新 term,转为 Follower
    if args.Term > n.currentTerm {
        n.currentTerm = args.Term
        n.state = Follower
        n.votedFor = ""
    }

    // 3. 重置选举超时(收到 Leader 心跳)
    n.resetElectionTimeout()

    // 4. 检查日志一致性
    if args.PrevLogIndex > 0 {
        if n.lastLogIndex < args.PrevLogIndex {
            // 日志太短
            log.Warn("log too short",
                "myLastIndex", n.lastLogIndex,
                "requiredIndex", args.PrevLogIndex)
            return AppendEntriesReply{
                Term:    n.currentTerm,
                Success: false,
            }
        }

        prevEntry := n.log[args.PrevLogIndex]
        if prevEntry.Term != args.PrevLogTerm {
            // 日志不匹配,删除冲突的日志
            log.Warn("log conflict detected",
                "index", args.PrevLogIndex,
                "myTerm", prevEntry.Term,
                "expectedTerm", args.PrevLogTerm)

            n.log = n.log[:args.PrevLogIndex]
            n.lastLogIndex = args.PrevLogIndex - 1

            return AppendEntriesReply{
                Term:    n.currentTerm,
                Success: false,
            }
        }
    }

    // 5. 追加新日志
    for _, entry := range args.Entries {
        n.log = append(n.log, entry)
        n.lastLogIndex++

        log.Info("appended entry",
            "index", entry.Index,
            "term", entry.Term)
    }

    // 6. 更新 commitIndex
    if args.LeaderCommit > n.commitIndex {
        oldCommitIndex := n.commitIndex
        n.commitIndex = min(args.LeaderCommit, n.lastLogIndex)

        log.Info("updated commit index",
            "old", oldCommitIndex,
            "new", n.commitIndex)

        // 应用已提交的日志到状态机
        for i := n.lastApplied + 1; i <= n.commitIndex; i++ {
            n.applyLog(n.log[i])
        }
        n.lastApplied = n.commitIndex
    }

    return AppendEntriesReply{
        Term:    n.currentTerm,
        Success: true,
    }
}

func (n *Node) applyLog(entry LogEntry) {
    log.Info("applying log to state machine",
        "index", entry.Index,
        "command", entry.Command)

    // 应用到状态机
    n.stateMachine.Apply(entry.Command)
}

type RaftLogger struct {
    file *os.File
}

func (l *RaftLogger) LogStateChange(
    nodeID string,
    oldState, newState NodeState,
    term int64,
) {
    entry := LogEntry{
        Timestamp: time.Now(),
        NodeID:    nodeID,
        Event:     "StateChange",
        Details: map[string]interface{}{
            "oldState": oldState,
            "newState": newState,
            "term":     term,
        },
    }

    // 写入日志文件
    json.NewEncoder(l.file).Encode(entry)
}

func (l *RaftLogger) LogRPC(
    from, to string,
    rpcType string,
    args, reply interface{},
) {
    entry := LogEntry{
        Timestamp: time.Now(),
        From:      from,
        To:        to,
        RPC:       rpcType,
        Args:      args,
        Reply:     reply,
    }

    json.NewEncoder(l.file).Encode(entry)
}

type FaultyNetwork struct {
    // 模拟网络延迟
    DelayRange [2]time.Duration  // [min, max]

    // 模拟丢包率
    DropRate float64

    // 模拟网络分区
    Partitions [][]string
}

func (n *FaultyNetwork) Send(from, to string, msg Message) error {
    // 1. 检查是否被分区隔离
    if n.isPartitioned(from, to) {
        return errors.New("network partitioned")
    }

    // 2. 模拟丢包
    if rand.Float64() < n.DropRate {
        return errors.New("packet dropped")
    }

    // 3. 模拟延迟
    delay := n.randomDelay()
    time.Sleep(delay)

    // 4. 发送
    return n.realSend(from, to, msg)
}

type Snapshot struct {
    LastIncludedIndex int64
    LastIncludedTerm  int64
    Data              []byte
}

func (n *Node) CreateSnapshot() error {
    n.mu.Lock()
    defer n.mu.Unlock()

    // 检查是否需要做快照
    if n.lastApplied - n.lastSnapshotIndex < snapshotThreshold {
        return nil
    }

    log.Info("creating snapshot",
        "lastApplied", n.lastApplied,
        "lastSnapshot", n.lastSnapshotIndex)

    // 1. 获取状态机快照
    snapshotData := n.stateMachine.Snapshot()

    // 2. 创建快照对象
    snapshot := Snapshot{
        LastIncludedIndex: n.lastApplied,
        LastIncludedTerm:  n.log[n.lastApplied].Term,
        Data:              snapshotData,
    }

    // 3. 保存快照到磁盘
    err := n.persister.SaveSnapshot(snapshot)
    if err != nil {
        return err
    }

    // 4. 删除旧日志
    oldLogLen := len(n.log)
    n.log = n.log[n.lastApplied-n.lastSnapshotIndex:]
    n.lastSnapshotIndex = n.lastApplied

    log.Info("snapshot created",
        "index", snapshot.LastIncludedIndex,
        "oldLogLen", oldLogLen,
        "newLogLen", len(n.log))

    return nil
}

func (n *Node) InstallSnapshot(snapshot Snapshot) error {
    n.mu.Lock()
    defer n.mu.Unlock()

    log.Info("installing snapshot",
        "lastIncludedIndex", snapshot.LastIncludedIndex)

    // 1. 检查快照是否更新
    if snapshot.LastIncludedIndex <= n.lastSnapshotIndex {
        return nil  // 旧快照,忽略
    }

    // 2. 应用快照到状态机
    err := n.stateMachine.Restore(snapshot.Data)
    if err != nil {
        return err
    }

    // 3. 更新日志
    n.lastSnapshotIndex = snapshot.LastIncludedIndex
    n.lastApplied = snapshot.LastIncludedIndex
    n.commitIndex = snapshot.LastIncludedIndex

    // 删除快照之前的日志
    if snapshot.LastIncludedIndex < n.lastLogIndex {
        n.log = n.log[snapshot.LastIncludedIndex-n.lastSnapshotIndex:]
    } else {
        n.log = []LogEntry{}
        n.lastLogIndex = snapshot.LastIncludedIndex
    }

    return nil
}

type Configuration struct {
    Old []string  // 旧配置
    New []string  // 新配置
}

func (n *Node) ChangeConfiguration(newMembers []string) error {
    // 阶段 1: 进入联合配置 C(old,new)
    jointConfig := Configuration{
        Old: n.config.Old,
        New: newMembers,
    }

    log.Info("entering joint consensus",
        "old", jointConfig.Old,
        "new", jointConfig.New)

    // 提交联合配置
    err := n.AppendEntry(context.Background(), jointConfig)
    if err != nil {
        return err
    }

    // 联合配置生效:
    // - 日志复制需要 old 的多数 AND new 的多数
    // - Leader 选举需要 old 的多数 AND new 的多数

    // 阶段 2: 提交新配置 C(new)
    time.Sleep(100 * time.Millisecond)  // 等待联合配置生效

    finalConfig := Configuration{
        Old: newMembers,
        New: nil,
    }

    log.Info("entering new configuration",
        "config", finalConfig.Old)

    err = n.AppendEntry(context.Background(), finalConfig)
    if err != nil {
        return err
    }

    // 新配置生效
    n.config = finalConfig
    n.peers = buildPeers(newMembers)

    return nil
}

func (n *Node) isQuorum(voters map[string]bool) bool {
    if n.config.New == nil {
        // 普通配置
        count := 0
        for _, member := range n.config.Old {
            if voters[member] {
                count++
            }
        }
        return count > len(n.config.Old)/2
    }

    // 联合配置:需要同时满足 old 和 new 的多数
    oldCount := 0
    for _, member := range n.config.Old {
        if voters[member] {
            oldCount++
        }
    }

    newCount := 0
    for _, member := range n.config.New {
        if voters[member] {
            newCount++
        }
    }

    return oldCount > len(n.config.Old)/2 &&
           newCount > len(n.config.New)/2
}

func (n *Node) Read(ctx context.Context, key string) (string, error) {
    if n.state != Leader {
        return "", errors.New("not leader")
    }

    // 1. 记录当前的 commitIndex(Read Index)
    readIndex := n.commitIndex

    log.Info("read request",
        "key", key,
        "readIndex", readIndex)

    // 2. 发送心跳到多数派(确认自己还是 Leader)
    if !n.checkQuorum(ctx) {
        return "", errors.New("lost leadership")
    }

    // 3. 等待 applyIndex >= readIndex
    for n.lastApplied < readIndex {
        time.Sleep(10 * time.Millisecond)
    }

    // 4. 读取状态机
    value := n.stateMachine.Get(key)

    log.Info("read completed",
        "key", key,
        "value", value)

    return value, nil
}

func (n *Node) checkQuorum(ctx context.Context) bool {
    responseCount := 1  // Leader 自己
    var mu sync.Mutex
    var wg sync.WaitGroup

    for _, peer := range n.peers {
        wg.Add(1)
        go func(p Peer) {
            defer wg.Done()

            // 发送心跳
            resp, err := p.AppendEntries(ctx, AppendEntriesArgs{
                Term:         n.currentTerm,
                LeaderId:     n.id,
                PrevLogIndex: n.lastLogIndex,
                PrevLogTerm:  n.log[n.lastLogIndex].Term,
                Entries:      nil,  // 空心跳
                LeaderCommit: n.commitIndex,
            })

            if err == nil && resp.Success {
                mu.Lock()
                responseCount++
                mu.Unlock()
            }
        }(peer)
    }

    wg.Wait()

    // 检查是否达到多数派
    return responseCount > len(n.peers)/2
}

// 更激进的优化:Leader 维护一个租约
type Node struct {
    // ... 其他字段
    leaseExpire time.Time
}

func (n *Node) renewLease() {
    // 发送心跳成功后,续约 lease
    n.leaseExpire = time.Now().Add(leaseTimeout)
}

func (n *Node) LeaseRead(key string) (string, error) {
    if n.state != Leader {
        return "", errors.New("not leader")
    }

    // 检查 lease 是否过期
    if time.Now().After(n.leaseExpire) {
        return "", errors.New("lease expired")
    }

    // 直接读取,无需等待
    return n.stateMachine.Get(key), nil
}

import "go.etcd.io/etcd/raft/v3"

type KVStore struct {
    node   raft.Node
    store  map[string]string
    mu     sync.RWMutex
}

func (kv *KVStore) Start() {
    c := &raft.Config{
        ID:              0x01,
        ElectionTick:    10,
        HeartbeatTick:   1,
        Storage:         raft.NewMemoryStorage(),
        MaxSizePerMsg:   4096,
        MaxInflightMsgs: 256,
    }

    kv.node = raft.StartNode(c, []raft.Peer{{ID: 0x01}})

    go kv.run()
}

func (kv *KVStore) run() {
    for {
        select {
        case <-kv.node.Tick():
            // 定时器触发

        case rd := <-kv.node.Ready():
            // 处理 Ready

            // 1. 保存状态
            if !raft.IsEmptySnap(rd.Snapshot) {
                kv.processSnapshot(rd.Snapshot)
            }
            kv.storage.Append(rd.Entries)

            // 2. 发送消息
            for _, msg := range rd.Messages {
                kv.send(msg)
            }

            // 3. 应用已提交的日志
            for _, entry := range rd.CommittedEntries {
                kv.apply(entry)
            }

            // 4. 通知 raft 已处理
            kv.node.Advance()
        }
    }
}

func (kv *KVStore) Set(key, value string) error {
    cmd := Command{
        Type:  "SET",
        Key:   key,
        Value: value,
    }

    data, _ := json.Marshal(cmd)

    return kv.node.Propose(context.Background(), data)
}

func (kv *KVStore) apply(entry raft.Entry) {
    var cmd Command
    json.Unmarshal(entry.Data, &cmd)

    kv.mu.Lock()
    defer kv.mu.Unlock()

    switch cmd.Type {
    case "SET":
        kv.store[cmd.Key] = cmd.Value
    case "DELETE":
        delete(kv.store, cmd.Key)
    }
}

func TestElection(t *testing.T) {
    // 创建 5 个节点
    nodes := createTestCluster(5)

    // 启动所有节点
    for _, node := range nodes {
        node.Start()
    }

    // 等待选举
    time.Sleep(1 * time.Second)

    // 检查只有一个 Leader
    leaders := countLeaders(nodes)
    assert.Equal(t, 1, leaders)
}

func TestLogReplication(t *testing.T) {
    nodes := createTestCluster(5)

    // 找到 Leader
    leader := findLeader(nodes)

    // 写入数据
    leader.Set("x", "1")

    // 等待复制
    time.Sleep(100 * time.Millisecond)

    // 检查所有节点
    for _, node := range nodes {
        value := node.Get("x")
        assert.Equal(t, "1", value)
    }
}

func BenchmarkWrite(b *testing.B) {
    cluster := createTestCluster(5)
    leader := findLeader(cluster)

    b.ResetTimer()

    for i := 0; i < b.N; i++ {
        leader.Set(fmt.Sprintf("key-%d", i), "value")
    }
}

func BenchmarkRead(b *testing.B) {
    cluster := createTestCluster(5)
    leader := findLeader(cluster)

    // 预写数据
    leader.Set("x", "1")

    b.ResetTimer()

    for i := 0; i < b.N; i++ {
        leader.Get("x")
    }
}