Raft 共识算法与分布式 KV 存储实现¶

// raft/raft.go
package raft

import (
    "sync"
    "time"
)

type NodeRole int

const (
    RoleFollower NodeRole = iota
    RoleCandidate
    RoleLeader
)

type LogEntry struct {
    Term    int         // 该日志条目生成时的任期号
    Index   int         // 全局单调递增的日志索引
    Command interface{} // 业务状态机指令 (例如 SET key value)
}

type RaftNode struct {
    mu        sync.Mutex
    peers     []string // 集群中所有对端节点的网络地址
    me        int      // 当前节点在 peers 中的索引
    currentTerm int    // 当前看到的最新任期号
    votedFor    int    // 当前任期内把票投给了谁 (-1 表示未投)
    log         []LogEntry

    // 所有服务器上的易失性状态
    commitIndex int // 已知已提交的最大日志条目索引
    lastApplied int // 已应用到状态机的最大日志条目索引

    // Leader 独占的易失性状态
    nextIndex  []int // 发送给每个 Follower 的下一条日志索引
    matchIndex []int // 每个 Follower 已成功复制的最大日志索引

    role           NodeRole
    heartbeatTimer *time.Timer
    electionTimer  *time.Timer
}

type RequestVoteArgs struct {
    Term         int // 候选人的任期
    CandidateId  int // 候选人 ID
    LastLogIndex int // 候选人最新一条日志的索引
    LastLogTerm  int // 候选人最新一条日志的任期
}

type RequestVoteReply struct {
    Term        int  // 当前节点的任期，供候选人更新自己
    VoteGranted bool // 是否把票投给候选人
}

func (rf *RaftNode) RequestVote(args *RequestVoteArgs, reply *RequestVoteReply) {
    rf.mu.Lock()
    defer rf.mu.Unlock()

    // 1. 如果请求的任期小于当前任期，直接拒绝
    if args.Term < rf.currentTerm {
        reply.Term = rf.currentTerm
        reply.VoteGranted = false
        return
    }

    // 2. 如果发现更大的任期，立即降级为 Follower
    if args.Term > rf.currentTerm {
        rf.currentTerm = args.Term
        rf.role = RoleFollower
        rf.votedFor = -1
    }

    // 3. 【安全性核心】：检查候选人的日志是否至少与自己一样新 (Log Up-to-date Rule)
    lastLog := rf.log[len(rf.log)-1]
    isUpToDate := false
    if args.LastLogTerm != lastLog.Term {
        isUpToDate = args.LastLogTerm > lastLog.Term
    } else {
        isUpToDate = args.LastLogIndex >= lastLog.Index
    }

    // 4. 判断当前任期是否尚未投票，且对方日志足够新
    if (rf.votedFor == -1 || rf.votedFor == args.CandidateId) && isUpToDate {
        rf.votedFor = args.CandidateId
        reply.VoteGranted = true
        rf.resetElectionTimer() // 投出有效票后重置选举计时器
    } else {
        reply.VoteGranted = false
    }
    reply.Term = rf.currentTerm
}