TinyKV Raft Review

┌─────────────────────────────────────────────┐
│              Client                         │
└─────────────────┬───────────────────────────┘
                  │ RaftCmdRequest (gRPC)
┌─────────────────▼───────────────────────────┐
│          RaftStore / PeerMsgHandler         │  ← peer 层（Project2B）
│  proposeRaftCommand()  HandleRaftReady()    │
└─────────────────┬───────────────────────────┘
                  │ Propose / Ready
┌─────────────────▼───────────────────────────┐
│              RawNode                        │  ← raft 层接口（Project2A）
│    Tick()  Step()  Ready()  Advance()       │
└─────────────────┬───────────────────────────┘
                  │
┌─────────────────▼───────────────────────────┐
│            Raft 状态机                      │  ← raft 核心（Project2A）
│  选举  日志复制  角色转换  消息处理          │
└─────────────────┬───────────────────────────┘
                  │
┌─────────────────▼───────────────────────────┐
│             RaftLog                         │  ← 日志内存缓存（Project2A）
│  entries[]  committed  applied  stabled     │
└─────────────────┬───────────────────────────┘
                  │
┌─────────────────▼───────────────────────────┐
│    PeerStorage（Storage 接口实现）           │  ← 持久化层（Project2B）
│         raftDB（badger）                    │
│         kvDB（badger）                      │
└─────────────────────────────────────────────┘

// raft.go - becomeFollower
func (r *Raft) becomeFollower(term uint64, lead uint64) {
    r.State = StateFollower
    r.Term = term
    r.Vote = lead
    r.Lead = lead
    r.electionElapsed = 0
    r.heartbeatElapsed = 0
    r.leadTransferee = None
}

message Entry {
    EntryType entry_type = 1;
    uint64 term = 2;     // 生成该日志时 Leader 的任期
    uint64 index = 3;    // 日志在序列中的位置（全局单调递增）
    bytes data = 4;      // 实际数据（序列化的命令）
}

// util.go
func randElectionTimeout(timeout int) int {
    n, _ := rand.Int(rand.Reader, big.NewInt(int64(timeout)))
    return int(n.Int64()) + timeout
}

// raft.go - becomeCandidate
func (r *Raft) becomeCandidate() {
    r.State = StateCandidate
    r.Term += 1
    r.Vote = r.id      // 给自己投票
    r.Lead = None
    r.votes = make(map[uint64]bool)
    r.votes[r.id] = true

    if len(r.Prs) == 1 {
        r.becomeLeader()  // 单节点直接成为 Leader
        return
    }
}

// raft.go - handleRequestVote
} else if r.Vote == None {
    lastIndex := r.RaftLog.LastIndex()
    lastTerm, _ := r.RaftLog.Term(lastIndex)
    if m.LogTerm > lastTerm || m.LogTerm == lastTerm && m.Index >= lastIndex {
        reject = false
        r.Vote = m.From
    }
}

// raft.go - becomeLeader
func (r *Raft) becomeLeader() {
    r.State = StateLeader
    r.Lead = r.id

    // 初始化所有节点的进度
    for id := range r.Prs {
        if id == r.id {
            r.Prs[id].Match = r.RaftLog.LastIndex()
        } else {
            r.Prs[id].Match = 0
        }
        r.Prs[id].Next = r.RaftLog.LastIndex() + 1
    }

    r.startPropose()  // 提交一条空日志（noop entry）
}

MsgAppend 字段：
- Index:   prevLogIndex（要追加日志的前一条日志的 index）
- LogTerm: prevLogTerm（要追加日志的前一条日志的 term）
- Entries: 要追加的日志条目
- Commit:  Leader 的 commitIndex

// log.go - Append
func (l *RaftLog) Append(preIndex, preTerm uint64, entries []*pb.Entry) bool {
    if term, err := l.Term(preIndex); err != nil || term != preTerm {
        return false
    }
    // 去除重复项（跳过已经一致的条目）
    offset := l.entries[0].Index
    for len(entries) > 0 {
        firstIndex := entries[0].Index
        if firstIndex > l.LastIndex() || l.entries[firstIndex-offset].Term != entries[0].Term {
            break
        }
        entries = entries[1:]
    }
    // 截断冲突条目并追加
    if len(entries) > 0 {
        l.entries = l.entries[:entries[0].Index-offset]
        l.stabled = min(l.stabled, l.LastIndex())
        for len(entries) > 0 {
            l.entries = append(l.entries, *entries[0])
            entries = entries[1:]
        }
    }
    return true
}

// raft.go - handleAppendResponse
if m.Reject {
    r.Prs[m.From].Next = m.Commit + 1
    r.Prs[m.From].Match = m.Commit
    r.sendAppend(m.From)
}

// raft.go - checkCommit
func (r *Raft) checkCommit() {
    if _, exist := r.Prs[r.id]; !exist {
        return
    }
    arr := make([]uint64, r.Prs[r.id].Next-r.RaftLog.committed)
    for _, p := range r.Prs {
        if p.Match >= r.RaftLog.committed {
            arr[p.Match-r.RaftLog.committed]++
        }
    }
    for i, n := len(arr)-1, 0; i > 0; i-- {
        n += int(arr[i])
        if n > len(r.Prs)/2 {
            r.RaftLog.committed = r.RaftLog.committed + uint64(i)
            r.sendAllAppend()
            break
        }
    }
}

时刻 a: S1 是 Leader，index=2 写入 term=2 的日志，复制到 S2
时刻 b: S1 崩溃，S5 成为 Leader(term=3)，在 index=2 写入 term=3 的日志
时刻 c: S5 崩溃，S1 重新成为 Leader(term=4)，将 term=2 的日志复制到多数节点
时刻 d1: S1 崩溃，S5 成为 Leader，覆盖了多数节点的 index=2 日志
时刻 d2: S1 不崩溃，通过 term=4 的日志间接提交了 index=2 的日志

// rawnode.go
type Ready struct {
    *SoftState              // 易失性状态（Lead, RaftState），仅用于通知上层状态变化
    pb.HardState            // 持久性状态（Term, Vote, Commit），需要持久化
    Entries []pb.Entry      // 需要持久化的日志条目（unstable entries）
    Snapshot pb.Snapshot    // 需要应用的快照
    CommittedEntries []pb.Entry  // 可以应用到状态机的日志
    Messages []pb.Message   // 需要发送给其他节点的消息
}

// storage.go - Storage 接口
type Storage interface {
    InitialState() (pb.HardState, pb.ConfState, error)
    Entries(lo, hi uint64) ([]pb.Entry, error)
    Term(i uint64) (uint64, error)
    LastIndex() (uint64, error)
    FirstIndex() (uint64, error)
    Snapshot() (pb.Snapshot, error)
}

┌─────────────────────────────────────┐
│  unstable（raft 库内存中）           │
│  - 尚未持久化的日志                 │
│  - 待应用的快照                     │
└───────────────┬─────────────────────┘
                │ 持久化后
┌───────────────▼─────────────────────┐
│  Storage（应用层管理的持久化数据）   │
│  - 已持久化但未压缩的日志            │
│  - 最新的快照                       │
└─────────────────────────────────────┘

// log.go
type RaftLog struct {
    storage Storage       // 已持久化数据的视图（只读）
    committed uint64      // 已被多数节点确认的最高 index
    applied   uint64      // 已应用到状态机的最高 index
    stabled   uint64      // 已持久化到 storage 的最高 index
    entries   []pb.Entry  // 所有未被 compact 的日志（含持久化和非持久化）
    pendingSnapshot *pb.Snapshot  // 待处理的快照（2C 使用）
}

snapshot.Index <= firstIndex - 1 <= applied <= committed <= stabled <= lastIndex

// log.go - newLog
func newLog(storage Storage) *RaftLog {
    firstIndex, _ := storage.FirstIndex()
    lastIndex, _ := storage.LastIndex()
    term, _ := storage.Term(firstIndex - 1)  // 获取 truncatedTerm

    entries := make([]pb.Entry, 1)
    entries[0].Index = firstIndex - 1  // 哨兵节点的 Index = truncatedIndex
    entries[0].Term = term             // 哨兵节点的 Term = truncatedTerm
    ents, _ := storage.Entries(firstIndex, lastIndex+1)
    entries = append(entries, ents...)

    return &RaftLog{
        storage:   storage,
        stabled:   lastIndex,
        entries:   entries,
        committed: firstIndex - 1,
        applied:   firstIndex - 1,
    }
}

func (l *RaftLog) LastIndex() uint64 {
    return l.entries[0].Index + uint64(len(l.entries)) - 1
    // 哨兵的 Index + 总条数 - 1（哨兵本身）
}

func (l *RaftLog) FirstIndex() uint64 {
    return l.entries[0].Index + 1  // 哨兵 Index + 1
}

func (l *RaftLog) Term(i uint64) (uint64, error) {
    offset := l.entries[0].Index
    if i < offset {
        return 0, ErrCompacted
    }
    if int(i-offset) >= len(l.entries) {
        return 0, ErrUnavailable
    }
    return l.entries[i-offset].Term, nil
}

func (l *RaftLog) unstableEntries() []pb.Entry {
    offset := l.entries[0].Index
    return l.entries[l.stabled+1-offset:]  // stabled+1 之后的都是未持久化的
}

func (l *RaftLog) nextEnts() (ents []pb.Entry) {
    offset := l.entries[0].Index
    lo := l.applied + 1 - offset
    hi := l.committed + 1 - offset
    return l.entries[lo:hi]
}

// log.go - maybeCompact
func (l *RaftLog) maybeCompact() {
    if FirstIndex, _ := l.storage.FirstIndex(); l.FirstIndex() < FirstIndex {
        l.entries = l.entries[FirstIndex-l.FirstIndex():]
        l.pendingSnapshot = nil
    }
}

// log.go - ApplySnapshot
func (l *RaftLog) ApplySnapshot(snap *pb.Snapshot) {
    l.pendingSnapshot = snap
    l.entries = make([]pb.Entry, 1)
    l.entries[0].Index = snap.Metadata.Index  // 新的哨兵
    l.entries[0].Term = snap.Metadata.Term
    l.applied = snap.Metadata.Index
    l.committed = snap.Metadata.Index
    l.stabled = snap.Metadata.Index
}

type Raft struct {
    id   uint64
    Term uint64        // 当前任期（需持久化）
    Vote uint64        // 当前任期投票给谁（需持久化）

    RaftLog *RaftLog   // 日志模块
    Prs  map[uint64]*Progress  // 每个节点的复制进度（仅 Leader 使用）
    State StateType    // 当前角色
    votes map[uint64]bool  // 选票记录（Candidate 使用）
    msgs  []pb.Message     // 待发送的消息
    Lead  uint64           // 当前 Leader ID

    heartbeatTimeout int   // 心跳超时（固定，来自 Config.HeartbeatTick）
    electionTimeout  int   // 选举超时（随机化）
    heartbeatElapsed int   // 心跳计时
    electionElapsed  int   // 选举计时

    leadTransferee  uint64  // Leader 迁移目标（3A）
    PendingConfIndex uint64 // 待应用的配置变更（3A）
    active  map[uint64]bool // 存活节点集合（用于孤岛检测）
    electionTick int        // ElectionTick 基准值（用于随机化）
}

type Progress struct {
    Match, Next uint64
}

// raft.go - tick
func (r *Raft) tick() {
    r.electionElapsed++
    r.RaftLog.maybeCompact()  // 顺便检查日志压缩

    switch r.State {
    case StateFollower:
        if r.electionElapsed >= r.electionTimeout {
            r.startElection()
        }
    case StateCandidate:
        if r.electionElapsed >= r.electionTimeout {
            r.startElection()  // 超时后重新选举
        }
    case StateLeader:
        if r.electionElapsed >= r.electionTimeout {
            r.electionElapsed = 0
            r.leadTransferee = None
            // 检查是否收到足够心跳响应（孤岛检测）
            activeNum := len(r.active)
            if activeNum*2 <= len(r.Prs) {
                r.becomeFollower(r.Term, None)
                r.RaftLog.pendingSnapshot = nil
                return
            }
            r.active = make(map[uint64]bool)
            r.active[r.id] = true
        }
        r.heartbeatElapsed++
        if r.heartbeatElapsed >= r.heartbeatTimeout {
            r.startBeat()
        }
    }
}

// raft.go - Step
func (r *Raft) Step(m pb.Message) error {
    switch r.State {
    case StateFollower:
        r.FollowerStep(m)
    case StateCandidate:
        r.CandidateStep(m)
    case StateLeader:
        r.LeaderStep(m)
    }
    return nil
}

case pb.MessageType_MsgHup:
    r.becomeCandidate()
    r.sendRequestVote()  // 广播投票请求

case pb.MessageType_MsgBeat:
    r.sendHeartbeat()  // 向所有 Follower 发送心跳

// raft.go - handlePropose
func (r *Raft) handlePropose(m pb.Message) error {
    if r.leadTransferee != None {
        return ErrProposalDropped  // 正在迁移 Leader，拒绝
    }
    // 设置 entry 的 index 和 term
    NextIndex := r.Prs[r.id].Next
    for i, entry := range m.Entries {
        entry.Index = NextIndex + uint64(i)
        entry.Term = r.Term
    }
    // 追加到本地日志
    lastIndex := r.RaftLog.LastIndex()
    term, _ := r.RaftLog.Term(lastIndex)
    r.RaftLog.Append(lastIndex, term, m.Entries)
    // 更新自身进度
    r.Prs[r.id].Next = r.RaftLog.LastIndex() + 1
    r.Prs[r.id].Match = r.RaftLog.LastIndex()
    // 单节点直接提交
    if len(r.Prs) == 1 {
        r.RaftLog.committed = r.RaftLog.LastIndex()
    }
    // 广播 AppendEntries
    r.sendAllAppend()
    return nil
}

// raft.go - sendAppend
func (r *Raft) sendAppend(to uint64) bool {
    index := r.Prs[to].Next - 1
    term, err := r.RaftLog.Term(index)
    if err == ErrCompacted {
        r.sendSnapshot(to)  // 需要的日志已被压缩，发快照
    } else {
        entries, _ := r.RaftLog.Entries(r.Prs[to].Next, r.RaftLog.LastIndex()+1)
        r.msgs = append(r.msgs, pb.Message{
            MsgType: pb.MessageType_MsgAppend,
            Term:    r.Term, From: r.id, To: to,
            Commit:  r.RaftLog.committed,
            Index:   index, LogTerm: term,
            Entries: entries,
        })
    }
    return true
}

// raft.go - handleAppendEntries
func (r *Raft) handleAppendEntries(m pb.Message) {
    // 旧 Leader 的消息，拒绝
    if m.Term < r.Term || r.State == StateLeader && m.Term == r.Term {
        r.sendAppendResponse(m.From, true)
        return
    }
    r.becomeFollower(m.Term, m.From)  // 更新 term，确认 Leader
    reject := !r.RaftLog.Append(m.Index, m.LogTerm, m.Entries)
    if !reject {
        // 更新 committed
        if len(m.Entries) > 0 {
            r.RaftLog.committed = max(r.RaftLog.committed,
                min(m.Commit, m.Entries[len(m.Entries)-1].Index))
        } else {
            r.RaftLog.committed = max(r.RaftLog.committed, min(m.Commit, m.Index))
        }
    }
    r.sendAppendResponse(m.From, reject)
}

// raft.go - handleAppendResponse
func (r *Raft) handleAppendResponse(m pb.Message) {
    if m.Term > r.Term {
        r.becomeFollower(m.Term, None)
        return
    }
    r.active[m.From] = true  // 记录该节点活跃

    if m.Reject {
        r.Prs[m.From].Next = m.Commit + 1  // 回退到对方 committed 之后
        r.Prs[m.From].Match = m.Commit
        r.sendAppend(m.From)               // 重新发送
    } else {
        r.Prs[m.From].Next = m.Index + 1
        r.Prs[m.From].Match = r.Prs[m.From].Next - 1
        // 检查是否可以推进 commit
        if term, _ := r.RaftLog.Term(m.Index); term == r.Term && m.Index > r.RaftLog.committed {
            r.checkCommit()
        }
        // Leader Transfer 完成检查
        if r.leadTransferee == m.From && r.Prs[m.From].Match == r.RaftLog.LastIndex() {
            r.sendTimeoutNow(m.From)
        }
    }
}

// raft.go - handleRequestVote
func (r *Raft) handleRequestVote(m pb.Message) {
    if m.Term > r.Term {
        r.becomeFollower(m.Term, None)
    }
    reject := true
    if m.Term < r.Term || r.State != StateFollower && m.Term == r.Term {
        reject = true  // term 小或非 Follower 状态
    } else if r.Vote == m.From {
        reject = false  // 已经投给他了
    } else if r.Vote == None {
        // 检查日志新旧
        lastIndex := r.RaftLog.LastIndex()
        lastTerm, _ := r.RaftLog.Term(lastIndex)
        if m.LogTerm > lastTerm || m.LogTerm == lastTerm && m.Index >= lastIndex {
            reject = false
            r.Vote = m.From
        }
    }
    r.sendRequestVoteResponse(m.From, reject)
}

// raft.go - handleRequestVoteResponse
func (r *Raft) handleRequestVoteResponse(m pb.Message) {
    if m.Term > r.Term {
        r.becomeFollower(m.Term, None)
        return
    }
    if _, exists := r.votes[m.From]; exists {
        return  // 已处理过该节点的响应
    }
    r.votes[m.From] = !m.Reject

    if len(r.votes)*2 <= len(r.Prs) {
        return  // 票数还不够，继续等待
    }
    // 统计票数
    argNum, denNum := 0, 0
    for _, v := range r.votes {
        if v { argNum++ } else { denNum++ }
    }
    if argNum*2 > len(r.Prs) {
        r.becomeLeader()
    }
    if denNum*2 > len(r.Prs) {
        r.becomeFollower(m.Term, None)
    }
}

// raft.go - handleHeartbeat
func (r *Raft) handleHeartbeat(m pb.Message) {
    if m.Term < r.Term || r.State == StateLeader && m.Term == r.Term {
        r.sendHeartbeatResponse(m.From)
        return
    }
    r.becomeFollower(m.Term, m.From)
    r.sendHeartbeatResponse(m.From)
}

// raft.go - handleHeartbeatResponse
func (r *Raft) handleHeartbeatResponse(m pb.Message) {
    if m.Term > r.Term {
        r.becomeFollower(m.Term, None)
        return
    }
    r.active[m.From] = true  // 记录活跃（用于孤岛检测）
    // 如果 Follower 落后，触发日志同步
    if r.RaftLog.committed > m.Commit {
        r.sendAppend(m.From)
    }
}

// raft.go - sendSnapshot
func (r *Raft) sendSnapshot(to uint64) {
    if r.RaftLog.pendingSnapshot == nil {
        snapshot, err := r.RaftLog.storage.Snapshot()
        if err != nil {
            return  // 快照尚未生成好，等待下次
        }
        r.RaftLog.pendingSnapshot = &snapshot
    }
    r.msgs = append(r.msgs, pb.Message{
        MsgType:  pb.MessageType_MsgSnapshot,
        Term:     r.Term, From: r.id, To: to,
        Snapshot: r.RaftLog.pendingSnapshot,
    })
}

// raft.go - handleSnapshot
func (r *Raft) handleSnapshot(m pb.Message) {
    if m.Term < r.Term || r.State == StateLeader && m.Term == r.Term {
        r.sendAppendResponse(m.From, true)
        return
    }
    r.becomeFollower(m.Term, m.From)
    reject := false
    if m.Snapshot == nil || m.Snapshot.Metadata.Index <= r.RaftLog.committed {
        reject = true  // 快照比自己旧
    } else {
        // 更新 Prs（集群配置可能变化）
        if m.Snapshot.Metadata.ConfState != nil {
            r.Prs = make(map[uint64]*Progress, len(m.Snapshot.Metadata.ConfState.Nodes))
            for _, id := range m.Snapshot.Metadata.ConfState.Nodes {
                r.Prs[id] = &Progress{}
            }
        }
        r.RaftLog.ApplySnapshot(m.Snapshot)
    }
    r.sendAppendResponse(m.From, reject)
}

func (r *Raft) handleTimeoutNow() {
    if _, ok := r.Prs[r.id]; !ok {
        return  // 自己已被移出集群
    }
    r.startElection()
}

func (r *Raft) sendAllAppend() {
    r.heartbeatElapsed = 0  // 重置心跳计时（AppendEntries 兼有心跳作用）
    for id := range r.Prs {
        if id == r.id {
            continue
        }
        r.sendAppend(id)
    }
}

type RawNode struct {
    Raft      *Raft
    softState SoftState      // 上次生成 Ready 时的 SoftState 快照
    hardState pb.HardState   // 上次生成 Ready 时的 HardState 快照
}

func (rn *RawNode) HasReady() bool {
    return len(rn.Raft.msgs) > 0 ||
        rn.Raft.RaftLog.stabled != rn.Raft.RaftLog.LastIndex() ||
        rn.Raft.RaftLog.applied != rn.Raft.RaftLog.committed ||
        rn.softState != rn.Raft.getSoftState() ||
        !isHardStateEqual(rn.hardState, rn.Raft.GetHardState()) ||
        rn.Raft.State != StateLeader && rn.Raft.RaftLog.pendingSnapshot != nil
}

func (rn *RawNode) Ready() Ready {
    ready := Ready{
        Entries:          rn.Raft.RaftLog.unstableEntries(),
        CommittedEntries: rn.Raft.RaftLog.nextEnts(),
        Messages:         rn.Raft.msgs,
    }
    if softState := rn.Raft.getSoftState(); rn.softState != softState {
        ready.SoftState = &softState
    }
    if hardState := rn.Raft.GetHardState(); !isHardStateEqual(rn.hardState, hardState) {
        ready.HardState = hardState
    }
    if rn.Raft.State != StateLeader && rn.Raft.RaftLog.pendingSnapshot != nil {
        ready.Snapshot = *rn.Raft.RaftLog.pendingSnapshot
    }
    return ready
}

func (rn *RawNode) Advance(rd Ready) {
    rn.Raft.msgs = nil  // 清空已发送的消息
    if len(rd.Entries) > 0 {
        rn.Raft.RaftLog.stabled = rd.Entries[len(rd.Entries)-1].Index
    }
    if len(rd.CommittedEntries) > 0 {
        rn.Raft.RaftLog.applied = rd.CommittedEntries[len(rd.CommittedEntries)-1].Index
    }
    if rd.SoftState != nil {
        rn.softState = *rd.SoftState
    }
    if !IsEmptyHardState(rd.HardState) {
        rn.hardState = rd.HardState
    }
    if !IsEmptySnap(&rd.Snapshot) {
        rn.Raft.RaftLog.pendingSnapshot = nil
    }
}

┌─────────────────────────────────────────────────────────┐
│                     RaftStore（节点）                    │
│  ┌──────────────────────┐  ┌──────────────────────────┐ │
│  │  Region A (Peer A1)  │  │  Region B (Peer B1)      │ │
│  │  [key_start, split)  │  │  [split, key_end)        │ │
│  └──────────────────────┘  └──────────────────────────┘ │
│          共用同一个 badger 实例                          │
└─────────────────────────────────────────────────────────┘

外部 RPC → RaftStore → router → Peer 的 mailbox → raftWorker
                                                       ↓
                                               HandleMsg(msg)
                                                       ↓
                                               HandleRaftReady()

// raft_worker.go - run
func (rw *raftWorker) run(closeCh <-chan struct{}, wg *sync.WaitGroup) {
    defer wg.Done()
    var msgs []message.Msg
    for {
        msgs = msgs[:0]
        select {
        case <-closeCh:
            return
        case msg := <-rw.raftCh:  // 阻塞等待消息
            msgs = append(msgs, msg)
        }
        // 批量取出所有待处理消息
        pending := len(rw.raftCh)
        for i := 0; i < pending; i++ {
            msgs = append(msgs, <-rw.raftCh)
        }
        // 分发处理
        peerStateMap := make(map[uint64]*peerState)
        for _, msg := range msgs {
            peerState := rw.getPeerState(peerStateMap, msg.RegionID)
            if peerState == nil {
                continue
            }
            newPeerMsgHandler(peerState.peer, rw.ctx).HandleMsg(msg)
        }
        // 处理所有涉及到的 Peer 的 Ready
        for _, peerState := range peerStateMap {
            newPeerMsgHandler(peerState.peer, rw.ctx).HandleRaftReady()
        }
    }
}

// peer.go
type proposal struct {
    index uint64          // 该请求在 Raft 日志中的 index
    term  uint64          // 请求时的 Leader term
    cb    *message.Callback  // 回调函数，用于通知客户端
}

// peer_msg_handler.go
func (d *peerMsgHandler) proposeRaftCommand(msg *raft_cmdpb.RaftCmdRequest, cb *message.Callback) {
    // 前置检查：是否是 Leader，term 是否正确，epoch 是否匹配
    err := d.preProposeRaftCommand(msg)
    if err != nil {
        if cb != nil { cb.Done(ErrResp(err)) }
        return
    }
    // 序列化请求
    data, err := msg.Marshal()
    if err != nil {
        if cb != nil { cb.Done(ErrResp(err)) }
        return
    }
    // 提交给 Raft
    err = d.RaftGroup.Propose(data)
    if err != nil {
        if cb != nil { cb.Done(ErrResp(err)) }
        return
    }
    // 记录 proposal（等待回调）
    if cb != nil {
        d.proposals = append(d.proposals, &proposal{
            index: d.RaftGroup.Raft.RaftLog.LastIndex(),
            term:  d.Term(),
            cb:    cb,
        })
    }
}

// peer_msg_handler.go
func (d *peerMsgHandler) HandleRaftReady() {
    if d.stopped { return }
    if !d.RaftGroup.HasReady() { return }

    ready := d.RaftGroup.Ready()
    d.Send(d.ctx.trans, ready.Messages)  // 先发消息

    // Apply CommittedEntries（含回调处理）
    if d.apply(&ready) {
        return  // 自身被删除，停止处理
    }

    // 持久化 Ready 中的数据
    if res, err := d.peerStorage.SaveReadyState(&ready); err != nil {
        log.Errorf(...)
    } else if res != nil {
        d.updateMeta(res.Region)  // 快照应用后更新 Region 元数据
    }

    d.RaftGroup.Advance(ready)  // 推进 RawNode
}

// peer_msg_handler.go - apply
func (d *peerMsgHandler) apply(ready *raft.Ready) bool {
    if len(ready.CommittedEntries) > 0 {
        for _, entry := range ready.CommittedEntries {
            if stop, resp := d.newCmdResp(entry); stop {
                return true
            } else {
                d.doResp(resp, &entry)
            }
        }
    }
    return false
}

// peer_msg_handler.go - doResp
func (d *peerMsgHandler) doResp(resp *raft_cmdpb.RaftCmdResponse, entry *eraftpb.Entry) {
    for len(d.proposals) > 0 {
        p := d.proposals[0]
        if entry.Index < p.index {
            return  // 当前 entry 还没到这个 proposal 的 index，跳过
        }
        if entry.Index > p.index {
            p.cb.Done(ErrRespStaleCommand(p.term))  // proposal 对应的 entry 已经被覆盖
            d.proposals = d.proposals[1:]
            continue
        }
        // entry.Index == p.index
        if entry.Term < p.term {
            return  // term 不匹配，等待
        }
        if entry.Term > p.term {
            p.cb.Done(ErrRespStaleCommand(p.term))  // term 不匹配，stale
            d.proposals = d.proposals[1:]
            continue
        }
        // index 和 term 都匹配，正常回调
        d.proposals = d.proposals[1:]
        p.cb.Txn = d.ctx.engine.Kv.NewTransaction(false)
        p.cb.Done(resp)
    }
}

// peer_msg_handler.go - HandleMsg
func (d *peerMsgHandler) HandleMsg(msg message.Msg) {
    switch msg.Type {
    case message.MsgTypeRaftMessage:
        // 外部 Raft 消息（其他节点发来的）
        d.onRaftMsg(msg.Data.(*rspb.RaftMessage))
    case message.MsgTypeRaftCmd:
        // 客户端请求（需要 propose）
        raftCMD := msg.Data.(*message.MsgRaftCmd)
        d.proposeRaftCommand(raftCMD.Request, raftCMD.Callback)
    case message.MsgTypeTick:
        d.onTick()  // 触发 RawNode.Tick()
    case message.MsgTypeSplitRegion:
        // Region Split（Project3B）
    case message.MsgTypeGcSnap:
        d.onGCSnap(...)  // 清理已应用的快照文件
    case message.MsgTypeStart:
        d.startTicker()  // 启动定时器（新 Peer 创建时）
    }
}

// peer_storage.go - SaveReadyState
func (ps *PeerStorage) SaveReadyState(ready *raft.Ready) (*ApplySnapResult, error) {
    kvWB := new(engine_util.WriteBatch)
    raftWB := new(engine_util.WriteBatch)
    var result *ApplySnapResult

    // 1. 更新 HardState（如果有变化）
    if !raft.IsEmptyHardState(ready.HardState) {
        ps.raftState.HardState = &ready.HardState
    }

    // 2. 更新 AppliedIndex
    if len(ready.CommittedEntries) > 0 {
        ps.applyState.AppliedIndex = ready.CommittedEntries[len(ready.CommittedEntries)-1].Index
    }

    // 3. 持久化日志条目
    ps.Append(ready.Entries, raftWB)

    // 4. 处理快照（如果有）
    if !raft.IsEmptySnap(&ready.Snapshot) {
        result, err = ps.ApplySnapshot(&ready.Snapshot, kvWB, raftWB)
    } else {
        // 写入 RaftLocalState 和 ApplyState
        raftWB.SetMeta(meta.RaftStateKey(ps.region.Id), ps.raftState)
        kvWB.SetMeta(meta.ApplyStateKey(ps.region.Id), ps.applyState)
        meta.WriteRegionState(kvWB, ps.region, rspb.PeerState_Normal)
    }

    // 5. 批量写入磁盘
    raftWB.WriteToDB(ps.Engines.Raft)
    kvWB.WriteToDB(ps.Engines.Kv)
    return result, nil
}

// peer_storage.go - Append
func (ps *PeerStorage) Append(entries []eraftpb.Entry, raftWB *engine_util.WriteBatch) error {
    if len(entries) == 0 {
        return nil
    }
    lastIndex := ps.raftState.LastIndex
    // 删除可能冲突的旧条目（index 在新条目范围内的旧条目）
    if entries[0].Index > lastIndex+1 {
        return errors.Errorf("missing log entry")
    }
    for i := entries[0].Index; i <= lastIndex; i++ {
        raftWB.DeleteMeta(meta.RaftLogKey(ps.region.Id, i))
    }
    // 写入新条目
    for _, entry := range entries {
        raftWB.SetMeta(meta.RaftLogKey(ps.region.Id, entry.Index), &entry)
    }
    // 更新 RaftLocalState
    ps.raftState.LastIndex = entries[len(entries)-1].Index
    ps.raftState.LastTerm = entries[len(entries)-1].Term
    return nil
}

// peer_storage.go - InitialState
func (ps *PeerStorage) InitialState() (eraftpb.HardState, eraftpb.ConfState, error) {
    raftState := ps.raftState
    if raft.IsEmptyHardState(*raftState.HardState) {
        return eraftpb.HardState{}, eraftpb.ConfState{}, nil
    }
    return *raftState.HardState, util.ConfStateFromRegion(ps.region), nil
}

func (ps *PeerStorage) FirstIndex() (uint64, error) {
    return ps.truncatedIndex() + 1, nil
}

func (ps *PeerStorage) LastIndex() (uint64, error) {
    return ps.raftState.LastIndex, nil
}

func (ps *PeerStorage) Term(idx uint64) (uint64, error) {
    if idx == ps.truncatedIndex() {
        return ps.truncatedTerm(), nil
    }
    // ... 从 raftDB 读取
}

触发条件：appliedIdx - firstIdx >= RaftLogGcCountLimit
           ↓
      onRaftGCLogTick()
           ↓
  proposeRaftCommand(CompactLogRequest)
           ↓
     Raft 集群同步该请求
           ↓
  HandleRaftReady -> apply CompactLogRequest
           ↓
  ScheduleCompactLog()（异步删除 raftDB 日志）
           ↓
  truncatedIndex 更新，FirstIndex 增大
           ↓
  Leader sendAppend 时发现 nextIndex < firstIndex
           ↓
  storage.Snapshot() 生成快照（异步）
           ↓
  Leader 发送 MsgSnapshot 给落后 Follower
           ↓
  Follower handleSnapshot → ApplySnapshot（RaftLog 层）
           ↓
  HandleRaftReady -> SaveReadyState -> ApplySnapshot（PeerStorage 层）
           ↓
  RegionTaskApply → 异步写入 kvDB

// peer_msg_handler.go - onRaftGCLogTick
func (d *peerMsgHandler) onRaftGCLogTick() {
    d.ticker.schedule(PeerTickRaftLogGC)
    if !d.IsLeader() { return }

    appliedIdx := d.peerStorage.AppliedIndex()
    firstIdx, _ := d.peerStorage.FirstIndex()
    var compactIdx uint64
    if appliedIdx > firstIdx && appliedIdx-firstIdx >= d.ctx.cfg.RaftLogGcCountLimit {
        compactIdx = appliedIdx
    } else {
        return
    }

    compactIdx -= 1  // 保留最后一条 applied 日志
    if compactIdx < firstIdx { return }

    term, err := d.RaftGroup.Raft.RaftLog.Term(compactIdx)
    // 构造 CompactLogRequest 并提交
    request := newCompactLogRequest(regionID, d.Meta, compactIdx, term)
    d.proposeRaftCommand(request, nil)  // callback 为 nil（系统内部请求）
}

// peer_msg_handler.go - handleAdminReq
case raft_cmdpb.AdminCmdType_CompactLog:
    compactLogReq := req.CompactLog
    if compactLogReq.CompactIndex > d.LastCompactedIdx &&
       compactLogReq.CompactIndex > d.peerStorage.truncatedIndex() {
        d.ScheduleCompactLog(compactLogReq.CompactIndex)  // 异步删除旧日志
        d.peerStorage.applyState.TruncatedState.Index = compactLogReq.CompactIndex
        d.peerStorage.applyState.TruncatedState.Term = compactLogReq.CompactTerm
    }
    resp.CompactLog = &raft_cmdpb.CompactLogResponse{}

// peer_storage.go - Snapshot
func (ps *PeerStorage) Snapshot() (eraftpb.Snapshot, error) {
    var snapshot eraftpb.Snapshot
    if ps.snapState.StateType == snap.SnapState_Generating {
        // 检查是否生成完成
        select {
        case s := <-ps.snapState.Receiver:
            if s != nil { snapshot = *s }
        default:
            return snapshot, raft.ErrSnapshotTemporarilyUnavailable
        }
        // 验证快照有效性
        // ...
        return snapshot, nil
    }
    // 发起异步生成任务
    ch := make(chan *eraftpb.Snapshot, 1)
    ps.snapState = snap.SnapState{
        StateType: snap.SnapState_Generating,
        Receiver:  ch,
    }
    ps.regionSched <- &runner.RegionTaskGen{
        RegionId: ps.region.GetId(),
        Notifier: ch,
    }
    return snapshot, raft.ErrSnapshotTemporarilyUnavailable  // 第一次调用总是返回此错误
}

// raft.go - sendSnapshot
func (r *Raft) sendSnapshot(to uint64) {
    if r.RaftLog.pendingSnapshot == nil {
        snapshot, err := r.RaftLog.storage.Snapshot()
        if err != nil {
            return  // 快照未就绪，等待下次
        }
        r.RaftLog.pendingSnapshot = &snapshot
    }
    // 发送快照消息
    r.msgs = append(r.msgs, pb.Message{...Snapshot: r.RaftLog.pendingSnapshot})
}