TinyKV Multi-Raft 实现与复习

Project2（单 Raft）：
[key: 0 ～ 100] → 一个 Raft Group → 串行处理所有请求

Project3（Multi-Raft）：
[key: 0 ～ 50]  → Raft Group A → 并行处理
[key: 50 ～ 100] → Raft Group B → 并行处理

┌─────────────┐
│   Client    │  (读写请求)
└──────┬──────┘
       │ gRPC
┌──────▼──────────────────────────────────────────┐
│                  TinyKV Server                  │
│  ┌──────────────────────────────────────────┐   │
│  │              RaftStore                   │   │
│  │  ┌──────────┐  ┌──────────┐  ┌────────┐ │   │
│  │  │ Region 1 │  │ Region 2 │  │ ...    │ │   │
│  │  │ RaftGroup│  │ RaftGroup│  │        │ │   │
│  │  └──────────┘  └──────────┘  └────────┘ │   │
│  └──────────────────────────────────────────┘   │
│  ┌─────────────────┐  ┌──────────────────────┐  │
│  │    raftDB       │  │       kvDB           │  │
│  │  (Raft 元数据)  │  │   (KV 业务数据)      │  │
│  └─────────────────┘  └──────────────────────┘  │
└──────────┬──────────────────────────────────────┘
           │ 心跳/调度指令
┌──────────▼──────┐
│  TinyScheduler  │  (对应 TiDB PD，全局调度)
│  (Project3C)    │
└─────────────────┘

物理层：
Store（物理机器/进程）
├── 运行多个 Peer（每个 Peer 属于不同 Region 的副本）
├── 共享 raftDB 和 kvDB
└── 运行 storeWorker、raftWorker 等

逻辑层：
Region（逻辑分片）
├── 管理一段 key 范围 [StartKey, EndKey)
├── 由 3 个（或更多）Peer 组成（分布在不同 Store 上）
└── 每个 Region 有独立的 Raft Group

对应关系：
Region 1: Peer{StoreId=1, PeerId=1}, Peer{StoreId=2, PeerId=2}, Peer{StoreId=3, PeerId=3}
Region 2: Peer{StoreId=1, PeerId=4}, Peer{StoreId=2, PeerId=5}, Peer{StoreId=3, PeerId=6}

同一 Store 上可以有多个 Region 的 Peer：
Store 1: [Peer 1 (Region 1 副本), Peer 4 (Region 2 副本), ...]

message RegionEpoch {
    uint64 conf_ver = 1;  // ConfChange 版本：AddNode/RemoveNode 时递增
    uint64 version  = 2;  // Split/Merge 版本：Region 分裂或合并时递增
}

// router routes a message to a peer.
type router struct {
    peers       sync.Map         // regionID -> peerState
    peerSender  chan message.Msg  // 发给 peer 的消息队列（容量 40960）
    storeSender chan<- message.Msg // 发给 store 的消息队列
}

raftWorker.run() 循环：
    ├── 从 router.peerSender 批量取消息
    ├── 对每个消息，找到对应 peer 的 peerMsgHandler
    ├── 调用 handler.HandleMsg() 处理消息
    └── 调用 handler.HandleRaftReady() 处理 Ready

// 批量取消息
var msgs []message.Msg
for {
    msgs = append(msgs, msg)
    if len(msgs) >= RaftWorkerMaxRecvMsgCnt {
        break
    }
    // 尝试继续取，取不到就停止
    select {
    case msg = <-rw.receiver:
    default:
        goto DONE
    }
}

func (d *storeWorker) onRaftMessage(msg *rspb.RaftMessage) error {
    // ...各种检查...
    created, err := d.maybeCreatePeer(regionID, msg)
    // ...
}

func IsInitialMsg(msg *eraftpb.Message) bool {
    return msg.MsgType == eraftpb.MessageType_MsgRequestVote ||
        // MsgHeartbeat 且 Commit == 0（RaftInvalidIndex）才触发创建
        (msg.MsgType == eraftpb.MessageType_MsgHeartbeat && msg.Commit == RaftInvalidIndex)
}

type GlobalContext struct {
    cfg                *config.Config
    engine             *engine_util.Engines
    store              *metapb.Store
    storeMeta          *storeMeta      // 全局 Region 元数据
    snapManager        *snap.SnapManager
    router             *router
    trans              Transport
    schedulerTaskSender chan<- worker.Task
    regionTaskSender   chan<- worker.Task
    // ...
}

type storeMeta struct {
    sync.RWMutex
    regions      map[uint64]*metapb.Region  // regionID -> Region 元数据
    regionRanges *btree.BTree               // 按 key 范围索引的 Region（用于路由）
}

Region 发送 RegionHeartbeatRequest
    └── TinyScheduler 接收
            ├── processRegionHeartbeat：更新本地 Region 信息
            └── 检查是否有 pending operator
                    ├── 有：通过 RegionHeartbeatResponse 下发（ChangePeer / TransferLeader）
                    └── 无：检查是否需要 balance，生成新 operator

type Raft struct {
    // ...
    leadTransferee  uint64  // 当前 Leader Transfer 的目标节点 ID（0 表示没有进行中的 Transfer）
    transferElapsed int     // Transfer 已经持续的 tick 数（超过 electionTimeout 则放弃）
}

func (rn *RawNode) TransferLeader(transferee uint64) {
    _ = rn.Raft.Step(pb.Message{
        MsgType: pb.MessageType_MsgTransferLeader,
        From:    transferee,  // From 字段存储转移目标
    })
}

func (r *Raft) handleTransferLeader(m pb.Message) {
    // 1. 检查转移目标是否在集群中
    if _, ok := r.Prs[m.From]; !ok {
        return
    }
    // 2. 如果转移目标就是自己，忽略
    if m.From == r.id {
        return
    }
    // 3. 处理正在进行中的 Transfer
    if r.leadTransferee != None {
        if r.leadTransferee == m.From {
            return  // 相同目标，忽略重复请求
        }
        r.leadTransferee = None  // 不同目标，强制覆盖
    }
    // 4. 设置转移目标，开始计时
    r.leadTransferee = m.From
    r.transferElapsed = 0
    // 5. 判断目标日志是否足够新
    if r.Prs[m.From].Match == r.RaftLog.LastIndex() {
        // 日志已同步，直接发 TimeoutNow
        r.sendTimeoutNow(m.From)
    } else {
        // 日志落后，先帮目标同步日志
        r.sendAppend(m.From)
    }
}

// 在处理 MsgAppendResponse 时
if r.leadTransferee != None && r.leadTransferee == m.From {
    if r.Prs[m.From].Match == r.RaftLog.LastIndex() {
        r.sendTimeoutNow(m.From)  // 同步完成，发 TimeoutNow
    }
}

func (r *Raft) handleTimeoutNow(m pb.Message) {
    // 检查自己是否在集群中（可能已被 RemoveNode）
    if _, ok := r.Prs[r.id]; !ok {
        return
    }
    // 立即发起选举，不等待 electionTimeout
    r.Step(pb.Message{MsgType: pb.MessageType_MsgHup})
}

// 在 tickHeartbeat() 中
if r.leadTransferee != None {
    r.transferElapsed++
    if r.transferElapsed >= r.electionTimeout {
        // 超过一个选举超时，放弃 Transfer
        r.leadTransferee = None
    }
}

func (r *Raft) handlePropose(m pb.Message) {
    if r.leadTransferee != None {
        return  // 正在 Transfer，拒绝新 propose
    }
    // ...正常处理
}

// Follower 收到 MsgTransferLeader
case pb.MessageType_MsgTransferLeader:
    if r.Lead != None {
        m.To = r.Lead
        r.msgs = append(r.msgs, m)  // 转发给当前 Leader
    }

type Raft struct {
    // ...
    PendingConfIndex uint64  // 上一个 ConfChange Entry 的 Index（未 apply 前不允许新 ConfChange）
}

func (rn *RawNode) ProposeConfChange(cc pb.ConfChange) error {
    data, err := cc.Marshal()
    if err != nil {
        return err
    }
    ent := pb.Entry{
        EntryType: pb.EntryType_EntryConfChange,  // 特殊 Entry 类型
        Data:      data,
    }
    return rn.Raft.Step(pb.Message{
        MsgType: pb.MessageType_MsgPropose,
        Entries: []*pb.Entry{&ent},
    })
}

// 在 handlePropose 中，当 entry.EntryType == EntryConfChange 时
r.PendingConfIndex = r.RaftLog.LastIndex() + uint64(i) + 1

func (r *Raft) addNode(id uint64) {
    if _, ok := r.Prs[id]; !ok {
        r.Prs[id] = &Progress{
            Next:  r.RaftLog.LastIndex() + 1,
            Match: 0,
        }
    }
    r.PendingConfIndex = None  // 清除 PendingConfIndex
}

func (r *Raft) removeNode(id uint64) {
    delete(r.Prs, id)
    r.PendingConfIndex = None  // 清除 PendingConfIndex

    // 关键：删除节点后 quorum 变小，可能有新的 entry 可以 commit
    if r.State == StateLeader {
        r.maybeCommit()
        // 需要广播 append，通知其他节点更新 commit
        r.bcastAppend()
    }

    // 如果被删的正是 leadTransferee，终止 Transfer
    if r.leadTransferee == id {
        r.leadTransferee = None
    }
}

初始状态：3 节点集群 [1, 2, 3]，节点 1 是 Leader
1. Leader 发送 entry3 给节点 2 和 3
2. 节点 2 返回了 AppendResponse（Match 更新为 3）
3. 节点 3 在返回 AppendResponse 之前被 removeNode 了
4. 此时集群只剩 [1, 2]，quorum = 2
5. entry3 已经在节点 1 和 2 上，满足多数派条件，应该 commit
6. 但 Leader 还没有收到节点 3 的 AppendResponse，不会主动计算 commit
7. 需要 removeNode 后立即重新计算 commit，推进 committedIndex

func (d *peerMsgHandler) proposeRaftCommand(msg *raft_cmdpb.RaftCmdRequest, cb *message.Callback) {
    if msg.AdminRequest != nil {
        switch msg.AdminRequest.CmdType {
        case raft_cmdpb.AdminCmdType_CompactLog:     // 2C 已实现，propose 到 Raft
        case raft_cmdpb.AdminCmdType_TransferLeader: // 3B 新增，不 propose，直接执行
        case raft_cmdpb.AdminCmdType_ChangePeer:     // 3B 新增，ProposeConfChange
        case raft_cmdpb.AdminCmdType_Split:          // 3B 新增，普通 Propose
        }
    } else {
        // 普通 KV 请求，直接 Propose
    }
}

case raft_cmdpb.AdminCmdType_TransferLeader:
    // 直接调用 TransferLeader，不走 Propose
    d.RaftGroup.TransferLeader(msg.AdminRequest.TransferLeader.Peer.Id)
    // 立即返回 Response
    adminResp := &raft_cmdpb.AdminResponse{
        CmdType:        raft_cmdpb.AdminCmdType_TransferLeader,
        TransferLeader: &raft_cmdpb.TransferLeaderResponse{},
    }
    cb.Done(&raft_cmdpb.RaftCmdResponse{
        Header:        &raft_cmdpb.RaftResponseHeader{},
        AdminResponse: adminResp,
    })

case raft_cmdpb.AdminCmdType_ChangePeer:
    // 1. 单步变更保证：只有上一个 ConfChange 已 apply 才能发起新的
    if d.peerStorage.AppliedIndex() >= d.RaftGroup.Raft.PendingConfIndex {
        // 2. 特殊情况：两节点且要 Remove Leader，先 Transfer
        if len(d.Region().Peers) == 2 &&
            msg.AdminRequest.ChangePeer.ChangeType == pb.ConfChangeType_RemoveNode &&
            msg.AdminRequest.ChangePeer.Peer.Id == d.PeerId() {
            for _, p := range d.Region().Peers {
                if p.Id != d.PeerId() {
                    d.RaftGroup.TransferLeader(p.Id)
                    break
                }
            }
            // 注意：这里不 return，让 client 重试（或者 return 也可以）
        }
        // 3. 创建 proposal
        d.proposals = append(d.proposals, &proposal{
            index: d.nextProposalIndex(),
            term:  d.Term(),
            cb:    cb,
        })
        // 4. 将 RaftCmdRequest 序列化后放入 ConfChange.Context
        context, _ := msg.Marshal()
        d.RaftGroup.ProposeConfChange(pb.ConfChange{
            ChangeType: msg.AdminRequest.ChangePeer.ChangeType,
            NodeId:     msg.AdminRequest.ChangePeer.Peer.Id,
            Context:    context,
        })
    }

// Propose 产生普通 Entry
func (rn *RawNode) Propose(data []byte) error {
    ent := pb.Entry{EntryType: pb.EntryType_EntryNormal, Data: data}
    // ...
}

// ProposeConfChange 产生 ConfChange Entry
func (rn *RawNode) ProposeConfChange(cc pb.ConfChange) error {
    data, _ := cc.Marshal()
    ent := pb.Entry{EntryType: pb.EntryType_EntryConfChange, Data: data}
    // ...
}

func (d *peerMsgHandler) processCommittedEntry(entry *pb.Entry, kvWB *engine_util.WriteBatch) *engine_util.WriteBatch {
    if entry.EntryType == pb.EntryType_EntryConfChange {
        var cc pb.ConfChange
        cc.Unmarshal(entry.Data)
        return d.processConfChange(entry, &cc, kvWB)
    }
    // ...普通 Entry 处理
}

func (d *peerMsgHandler) processConfChange(entry *pb.Entry, cc *pb.ConfChange, kvWB *engine_util.WriteBatch) *engine_util.WriteBatch {
    // 1. 反序列化 Context 获取原始 RaftCmdRequest
    msg := &raft_cmdpb.RaftCmdRequest{}
    msg.Unmarshal(cc.Context)

    region := d.Region()

    // 2. 检查 RegionEpoch 是否过期（防止重复执行同一 ConfChange）
    // 测试会多次提交同一 ConfChange 直到 apply，RegionEpoch 检查能过滤重复
    if err := util.CheckRegionEpoch(msg, region, true); err != nil {
        if errEpochNotMatch, ok := err.(*util.ErrEpochNotMatch); ok {
            d.handleProposal(entry, ErrResp(errEpochNotMatch))
            return kvWB
        }
    }

    switch cc.ChangeType {
    case pb.ConfChangeType_AddNode:
        // 3a. 检查待添加节点是否已存在
        if d.searchPeerWithId(cc.NodeId) == len(region.Peers) {
            // 不存在，执行添加
            region.Peers = append(region.Peers, msg.AdminRequest.ChangePeer.Peer)
            region.RegionEpoch.ConfVer++
            // 写入 kvDB
            meta.WriteRegionState(kvWB, region, rspb.PeerState_Normal)
            // 更新 storeMeta（需加锁）
            d.updateStoreMeta(region)
            // 更新 peerCache（消息发送时需要知道对方的 StoreId）
            d.insertPeerCache(msg.AdminRequest.ChangePeer.Peer)
        }

    case pb.ConfChangeType_RemoveNode:
        // 3b. 如果删除目标是自己，直接销毁
        if cc.NodeId == d.PeerId() {
            d.destroyPeer()
            return kvWB  // 必须 return，后面不能再执行任何操作
        }
        // 不是自己，检查待删节点是否存在
        n := d.searchPeerWithId(cc.NodeId)
        if n != len(region.Peers) {
            region.Peers = append(region.Peers[:n], region.Peers[n+1:]...)
            region.RegionEpoch.ConfVer++
            meta.WriteRegionState(kvWB, region, rspb.PeerState_Normal)
            d.updateStoreMeta(region)
            d.removePeerCache(cc.NodeId)
        }
    }

    // 4. 更新 Raft 层的成员配置（调用 3A 实现的 addNode/removeNode）
    d.RaftGroup.ApplyConfChange(*cc)

    // 5. 处理 proposal 回调
    d.handleProposal(entry, &raft_cmdpb.RaftCmdResponse{
        Header: &raft_cmdpb.RaftResponseHeader{},
        AdminResponse: &raft_cmdpb.AdminResponse{
            CmdType:    raft_cmdpb.AdminCmdType_ChangePeer,
            ChangePeer: &raft_cmdpb.ChangePeerResponse{Region: region},
        },
    })

    // 6. 如果是 Leader，通知 Scheduler 更新 Region 缓存
    if d.IsLeader() {
        d.notifyHeartbeatScheduler(region, d.peer)
    }

    return kvWB
}

func (d *peerMsgHandler) notifyHeartbeatScheduler(region *metapb.Region, peer *peer) {
    clonedRegion := new(metapb.Region)
    util.CloneMsg(region, clonedRegion)
    d.ctx.schedulerTaskSender <- &runner.SchedulerRegionHeartbeatTask{
        Region:          clonedRegion,
        Peer:            peer.Meta,
        PendingPeers:    peer.CollectPendingPeers(),
        ApproximateSize: peer.ApproximateSize,
    }
}

1. 测试/调度器 调用 schedulerClient.RemovePeer(regionID, peer)
   └── 设置 operators[regionID] = RemovePeerOperator

2. 定期心跳：某 Region 的 Leader 发送 RegionHeartbeatRequest
   └── MockSchedulerClient.RegionHeartbeat() 处理
       ├── 发现 operators[regionID] 有待执行的 RemoveNode
       └── 调用 makeRegionHeartbeatResponse 生成响应（ChangePeer: RemoveNode）

3. heartbeatResponseHandler 处理响应
   └── onRegionHeartbeatResponse() 调用 sendAdminRequest
       └── 封装 AdminCmdType_ChangePeer 请求发给 Leader

4. Leader 的 peerMsgHandler 收到 HandleMsg
   └── proposeRaftCommand → ProposeConfChange
       └── Raft 层写入 EntryConfChange Entry

5. Entry 复制到多数节点后 commit
   └── HandleRaftReady → processCommittedEntry → processConfChange
       ├── 所有节点执行：更新 region.Peers，ConfVer++，持久化
       └── 被删节点：调用 destroyPeer()，从集群中移除自己

1. 调度器调用 AddPeer → 通过心跳响应下发 AdminCmdType_ChangePeer(AddNode)

2. Leader proposeRaftCommand → ProposeConfChange
   └── 所有节点 apply：region.Peers 中添加新节点 F，ConfVer++，持久化

3. 此时 F 还不存在，Leader 的 Prs 中有 F，会给 F 发心跳
   └── 心跳进入 storeWorker.onRaftMessage()
       └── F 不存在 → maybeCreatePeer()
           ├── 条件：IsInitialMsg（心跳 + Commit == RaftInvalidIndex）
           └── 成立 → replicatePeer() 创建空的 Peer F
               └── 注册到 router，发送 MsgTypeStart 启动

4. F 启动后，Leader 发现其日志落后（LastIndex=0）
   └── 直接发 Snapshot 给 F

5. F 收到 Snapshot，通过 HandleRaftReady → ApplySnapshot
   └── 更新 F.Region() 信息，同步 storeMeta
   └── F 正式加入集群，可以服务请求

分裂前：Region A [0, 100) 在 Store 1 上
kvDB: {key1: v1, key2: v2, ..., key100: v100}（全在一个 badger 中）

分裂后：
Region A [0, 50)  → 负责 key1~key49
Region B [50, 100) → 负责 key50~key100
共用同一 kvDB，没有数据搬移

Step 1: peer_msg_handler.go onTick()
    └── onSplitRegionCheckTick()
        └── 生成 SplitCheckTask，发给 split_checker.go

Step 2: split_checker.go Handle()
    └── 遍历 kvDB 中该 Region 的 key，按字典序找中位 key（SplitKey）
    └── 如果 Region 大小超过阈值（cfg.RegionMaxSize），发送 MsgTypeSplitRegion

Step 3: peer_msg_handler.go HandleMsg()
    └── 收到 MsgTypeSplitRegion
        └── onPrepareSplitRegion()
            └── 发送 SchedulerAskSplitTask 给 scheduler_task.go
                └── 向 Scheduler 申请新的 RegionId 和 PeerIds（保证全局唯一）

Step 4: scheduler_task.go onAskSplit()
    └── 收到 Scheduler 分配的 NewRegionId 和 NewPeerIds
        └── 发起 AdminCmdType_Split 的 RaftCmdRequest 给 Leader

Step 5: 正常 propose → Raft 同步 → apply 流程

case raft_cmdpb.AdminCmdType_Split:
    // Step 1: 检查请求合法性
    if requests.Header.RegionId != d.regionId {
        // ErrRegionNotFound
        d.handleProposal(entry, ErrResp(&util.ErrRegionNotFound{...}))
        return kvWB
    }
    if err := util.CheckRegionEpoch(requests, d.Region(), true); err != nil {
        // RegionEpoch 不匹配（Region 已经分裂过了）
        d.handleProposal(entry, ErrResp(err))
        return kvWB
    }
    if err := util.CheckKeyInRegion(adminReq.Split.SplitKey, d.Region()); err != nil {
        // SplitKey 不在当前 Region 范围内
        d.handleProposal(entry, ErrResp(err))
        return kvWB
    }
    if len(d.Region().Peers) != len(adminReq.Split.NewPeerIds) {
        // NewPeerIds 数量不一致，拒绝（Scheduler 信息过时）
        d.handleProposal(entry, ErrRespStaleCommand(d.Term()))
        return kvWB
    }

    // Step 2: 创建新 Region
    oldRegion := d.Region()
    split := adminReq.Split

    // 为新 Region 创建 Peers（复制 oldRegion 的 Peers，修改 PeerId）
    newPeers := make([]*metapb.Peer, len(oldRegion.Peers))
    for i, peer := range oldRegion.Peers {
        newPeers[i] = &metapb.Peer{
            Id:      split.NewPeerIds[i],
            StoreId: peer.StoreId,
        }
    }
    newRegion := &metapb.Region{
        Id:       split.NewRegionId,
        StartKey: split.SplitKey,           // 新 Region 从 SplitKey 开始
        EndKey:   oldRegion.EndKey,          // 到原 Region 结束
        RegionEpoch: &metapb.RegionEpoch{
            ConfVer: 1,
            Version: 1,
        },
        Peers: newPeers,
    }

    // Step 3: 更新 RegionEpoch
    oldRegion.RegionEpoch.Version++
    newRegion.RegionEpoch.Version = oldRegion.RegionEpoch.Version  // 同步版本号

    // Step 4: 更新 storeMeta（必须加锁！）
    storeMeta := d.ctx.storeMeta
    storeMeta.Lock()
    storeMeta.regionRanges.Delete(&regionItem{region: oldRegion})       // 先删旧范围
    oldRegion.EndKey = split.SplitKey                                    // 修改 oldRegion 范围
    storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: oldRegion}) // 重新插入
    storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: newRegion}) // 插入新 Region
    storeMeta.regions[newRegion.Id] = newRegion
    storeMeta.Unlock()

    // Step 5: 持久化
    meta.WriteRegionState(kvWB, oldRegion, rspb.PeerState_Normal)
    meta.WriteRegionState(kvWB, newRegion, rspb.PeerState_Normal)

    // Step 6: 创建新 Region 的 Peer，注册到 router 并启动
    newPeer, err := createPeer(d.storeID(), d.ctx.cfg, d.ctx.schedulerTaskSender, d.ctx.engine, newRegion)
    d.ctx.router.register(newPeer)
    d.ctx.router.send(newRegion.Id, message.Msg{Type: message.MsgTypeStart})

    // Step 7: 处理 proposal 回调
    d.handleProposal(entry, &raft_cmdpb.RaftCmdResponse{
        Header: &raft_cmdpb.RaftResponseHeader{},
        AdminResponse: &raft_cmdpb.AdminResponse{
            CmdType: raft_cmdpb.AdminCmdType_Split,
            Split: &raft_cmdpb.SplitResponse{
                Regions: []*metapb.Region{newRegion, oldRegion},
            },
        },
    })

    // Step 8: 通知 Scheduler 更新缓存（防止 no region 问题）
    if d.IsLeader() {
        d.notifyHeartbeatScheduler(oldRegion, d.peer)
        d.notifyHeartbeatScheduler(newRegion, newPeer)  // 也要通知新 Region！
    }

// 在处理 entry 之前
if err := util.CheckRegionEpoch(req, d.Region(), true); err != nil {
    d.handleProposal(entry, ErrResp(err))
    return kvWB
}

case raft_cmdpb.CmdType_Put:
    // Put 时增加 SizeDiffHint
    d.SizeDiffHint += uint64(len(req.Put.Key) + len(req.Put.Value))
    // 执行写入...

case raft_cmdpb.CmdType_Delete:
    // Delete 时减少 SizeDiffHint
    if len(req.Delete.Key) > d.SizeDiffHint {
        d.SizeDiffHint = 0
    } else {
        d.SizeDiffHint -= uint64(len(req.Delete.Key))
    }
    // 执行删除...

// 处理完 ConfChange Entry 后
d.processConfChange(entry, cc, kvWB)

// 关键：检查是否因为 RemoveNode 自己而停止
if d.stopped {
    return
}

// 继续处理下一个 entry...

// RegionInfo 包含心跳中的所有信息
type RegionInfo struct {
    meta            *metapb.Region   // Region 元数据（ID, 范围, Peers, Epoch）
    learners        []*metapb.Peer
    voters          []*metapb.Peer
    leader          *metapb.Peer     // 当前 Leader
    pendingPeers    []*metapb.Peer   // 待同步的 Peer
    approximateSize int64            // Region 大小估算
}

func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
    // ...
}

origin := c.core.GetRegion(region.GetID())
if origin != nil {
    // 比较 RegionEpoch：新 epoch 的 Version 或 ConfVer 必须 >= 旧的
    if util.IsEpochStale(region.GetRegionEpoch(), origin.GetRegionEpoch()) {
        return ErrRegionIsStale(region.GetMeta(), origin.GetMeta())
    }
}

if origin == nil {
    // 扫描所有与该 Region key 范围有 overlap 的 Region
    overlaps := c.core.ScanRegions(region.GetStartKey(), region.GetEndKey(), -1)
    for _, r := range overlaps {
        // 新 Region 必须比所有 overlap Region 都要新
        if util.IsEpochStale(region.GetRegionEpoch(), r.GetRegionEpoch()) {
            return ErrRegionIsStale(region.GetMeta(), r.GetMeta())
        }
    }
}

1. Version 或 ConfVer 大于原来的（Region 发生了 split 或 confchange）
2. Leader 发生了变化
3. 新的或原来的有 pending peers（说明有成员正在追赶日志）
4. ApproximateSize 发生了变化（大小变了，可能触发新的调度）

// 通过检查后，执行更新
c.core.PutRegion(region)                   // 更新 Region 树
c.core.UpdateStoreStatus(region.GetLeaderStoreId())  // 更新 Store 的负载状态

func (s *balanceRegionScheduler) Schedule(cluster opt.Cluster) *operator.Operator {
    // Step 1: 筛选并排序 suitableStore
    stores := cluster.GetStores()
    var suitableStores []*core.StoreInfo
    for _, store := range stores {
        if store.IsUp() && store.DownTime() < cluster.GetMaxStoreDownTime() {
            suitableStores = append(suitableStores, store)
        }
    }
    if len(suitableStores) < 2 {
        return nil
    }
    // 按 regionSize 降序排列
    sort.Slice(suitableStores, func(i, j int) bool {
        return suitableStores[i].GetRegionSize() > suitableStores[j].GetRegionSize()
    })

    // Step 2: 从 regionSize 最大的 Store 中找待迁移 Region
    var sourceRegion *core.RegionInfo
    var sourceStore *core.StoreInfo
    for _, store := range suitableStores {
        // 按优先级依次查找：pending > follower > leader
        cluster.GetPendingRegionsWithLock(store.GetID(), func(container core.RegionsContainer) {
            sourceRegion = container.RandomRegion(nil, nil)
        })
        if sourceRegion != nil {
            sourceStore = store
            break
        }
        cluster.GetFollowersWithLock(store.GetID(), func(container core.RegionsContainer) {
            sourceRegion = container.RandomRegion(nil, nil)
        })
        if sourceRegion != nil {
            sourceStore = store
            break
        }
        cluster.GetLeadersWithLock(store.GetID(), func(container core.RegionsContainer) {
            sourceRegion = container.RandomRegion(nil, nil)
        })
        if sourceRegion != nil {
            sourceStore = store
            break
        }
    }
    if sourceRegion == nil {
        return nil
    }

    // Step 3: 检查 Region 副本数是否满足最小要求
    if len(sourceRegion.GetPeers()) < cluster.GetMaxReplicas() {
        return nil  // 副本数不足，不能再减少
    }

    // Step 4: 找 regionSize 最小的目标 Store（不能在源 Region 中）
    var targetStore *core.StoreInfo
    sourceRegionStoreIDs := sourceRegion.GetStoreIds()
    for i := len(suitableStores) - 1; i >= 0; i-- {
        store := suitableStores[i]
        if _, ok := sourceRegionStoreIDs[store.GetID()]; !ok {
            targetStore = store
            break
        }
    }
    if targetStore == nil {
        return nil
    }

    // Step 5: 判断迁移是否有价值
    // 差值必须大于 Region 大小的两倍，避免来回迁移
    if sourceStore.GetRegionSize()-targetStore.GetRegionSize() <= 2*sourceRegion.GetApproximateSize() {
        return nil
    }

    // Step 6: 创建 MovePeer Operator
    newPeer, err := cluster.AllocPeer(targetStore.GetID())
    if err != nil {
        return nil
    }
    return operator.CreateMovePeerOperator(
        "balance-region",
        cluster,
        sourceRegion,
        operator.OpBalance,
        sourceStore.GetID(),
        targetStore.GetID(),
        newPeer.GetId(),
    )
}

Client 发起 Put(key, value) 请求
    ↓
TinyKV Server 收到 gRPC 请求
    ↓
根据 key 路由到对应 Region（查 storeMeta.regionRanges 或询问 Scheduler）
    ↓
将请求发给该 Region 的 Leader（通过 router.send）
    ↓
Leader 的 peerMsgHandler.HandleMsg(MsgTypeRaftCmd) 处理
    ↓
proposeRaftCommand → RaftGroup.Propose(data)
    ↓
Raft 层：Leader 将 Entry 写入 RaftLog，广播 MsgAppend 给 Followers
    ↓
Followers 收到 MsgAppend，持久化 Entry，回复 MsgAppendResponse
    ↓
Leader 收到多数 AppendResponse，推进 committedIndex
    ↓
raftWorker 触发 HandleRaftReady
    ↓
apply CommittedEntries：执行 kvWB.SetCF(key, value)，写入 kvDB
    ↓
处理 proposal 回调：cb.Done(response)
    ↓
响应 Client

[触发阶段]
1. onTick() → onSplitRegionCheckTick()
   └── 检查 Region 大小（通过 SizeDiffHint 估算）
       └── 超过阈值 → 发送 SplitCheckTask

2. split_checker.go Handle(SplitCheckTask)
   └── 扫描 kvDB，按字典序找 SplitKey（中位 key）
       └── 发送 MsgTypeSplitRegion(splitKey)

3. HandleMsg(MsgTypeSplitRegion)
   └── onPrepareSplitRegion()
       └── 发送 SchedulerAskSplitTask（申请新 RegionId 和 PeerIds）

4. scheduler_task.go onAskSplit()
   └── Scheduler 分配 NewRegionId 和 NewPeerIds
       └── 发起 AdminCmdType_Split 的 RaftCmdRequest

[Propose 阶段]
5. proposeRaftCommand(AdminCmdType_Split)
   └── 检查 RegionEpoch 和 SplitKey 合法性
       └── RaftGroup.Propose(splitRequest)

[Raft 同步阶段]
6. Leader 广播 Entry，多数节点 commit

[Apply 阶段]
7. HandleRaftReady → processCommittedEntry → processSplit
   ├── 创建 newRegion（StartKey=SplitKey, EndKey=oldRegion.EndKey）
   ├── 更新 oldRegion.EndKey = SplitKey
   ├── 两个 Region 的 Version++
   ├── 持久化（WriteRegionState）
   ├── 更新 storeMeta.regionRanges（加锁）
   ├── createPeer(newRegion) → router.register → MsgTypeStart
   └── notifyHeartbeatScheduler（两次，分别通知新旧 Region）

[新 Region 选 Leader]
8. 新 Region 的 Peer 启动后，开始 Raft 选举
   └── 选出 Leader 后，开始正常服务请求

[调度器下发]
1. Schedule() 决定向某 Store 添加一个 Region 的 Peer
   └── 生成 AddPeer Operator
       └── 通过下一次 Region 心跳响应下发 ChangePeer(AddNode)

[Propose 到 Apply]
2. Leader 收到 AdminCmdType_ChangePeer(AddNode)
   └── 检查 AppliedIndex >= PendingConfIndex
       └── ProposeConfChange(AddNode, newPeerID)

3. Raft 同步后 commit，所有节点 Apply
   └── 每个节点：region.Peers 中添加新 Peer，ConfVer++，持久化
       └── ApplyConfChange → addNode（更新 Prs）

[新节点创建]
4. Leader 向新 PeerID 发送心跳（msg.Commit = RaftInvalidIndex）
   └── storeWorker.onRaftMessage → maybeCreatePeer
       └── IsInitialMsg 判断通过 → replicatePeer() 创建空 Peer
           └── router.register → MsgTypeStart 启动

[数据同步]
5. Leader 发现新节点 Match=0，LastIndex 落后
   └── 直接发 Snapshot 给新节点（跳过逐条 AppendEntry）

6. 新节点 HandleRaftReady 处理 Snapshot
   └── ApplySnapshot → 更新 Region 信息、storeMeta
       └── 新节点正式加入集群，开始参与 Raft 投票

1. Schedule() 决定移除某 Store 上的 Region Peer
   └── 生成 RemovePeer Operator → 心跳响应下发 ChangePeer(RemoveNode)

2. Leader proposeRaftCommand(ChangePeer, RemoveNode)
   └── 特殊检查：若 2 节点且 Remove 的是 Leader
       └── 先 TransferLeader，等 Leader 迁移后再 Remove

3. Entry commit，所有节点 Apply processConfChange
   ├── 非目标节点：region.Peers 中删除该 Peer，ConfVer++，removePeerCache
   └── 目标节点（被删除的）：destroyPeer()
       ├── 清理 raftDB 中的 Region 信息
       ├── 清理 storeMeta（regions, regionRanges）
       ├── 关闭 router 中的 Peer
       └── d.stopped = true

4. destroyPeer 后不再处理任何 entry（检查 d.stopped）

1. 调度器通过心跳响应下发 TransferLeader 指令

2. Leader 收到 AdminCmdType_TransferLeader
   └── 直接调用 RaftGroup.TransferLeader(targetPeerID)
       └── Step(MsgTransferLeader{From: targetPeerID})

3. Raft 层处理 MsgTransferLeader
   ├── 若目标日志已同步：直接发 MsgTimeoutNow
   └── 若目标日志落后：发 MsgAppend，等 AppendResponse 后再发 TimeoutNow

4. 目标节点收到 MsgTimeoutNow
   └── 立即 Step(MsgHup)，发起选举

5. 新的 Leader 选举成功（term+1，日志最新）
   └── 原 Leader 收到更高 term 的消息，退化为 Follower

6. 新 Leader 就位，上层收到 AdminCmdType_TransferLeader 的 resp

[调度决策]
1. Scheduler 定期调用 balanceRegionScheduler.Schedule()
   └── 选出负载最重的 Store（sourceStore）中的一个 Region
       └── 选出负载最轻的 Store（targetStore）

2. 生成 MovePeer Operator：[AddPeer, TransferLeader?, RemovePeer]

[执行 AddPeer]
3. 通过心跳响应下发 ChangePeer(AddNode) 给源 Region 的 Leader
   └── ... 执行 ConfChange AddNode 流程（见 6.3）
       └── 新 Peer 在 targetStore 上创建并同步数据

[执行 TransferLeader（如需）]
4. 若源 Store 是 Leader，下发 TransferLeader 到 targetStore 上的新 Peer
   └── ... 执行 Leader Transfer 流程（见 6.5）

[执行 RemovePeer]
5. 下发 ChangePeer(RemoveNode) 删除 sourceStore 上的 Peer
   └── ... 执行 ConfChange RemoveNode 流程（见 6.4）

[完成]
6. Region 成功从 sourceStore 迁移到 targetStore
   └── 数据通过 Snapshot 完成同步

func IsEpochStale(epoch *metapb.RegionEpoch, checkEpoch *metapb.RegionEpoch) bool {
    return epoch.GetVersion() < checkEpoch.GetVersion() ||
        epoch.GetConfVer() < checkEpoch.GetConfVer()
}

type storeMeta struct {
    sync.RWMutex
    regions      map[uint64]*metapb.Region
    regionRanges *btree.BTree
}

Operator = [Step1: AddPeer, Step2: TransferLeader, Step3: RemovePeer]

case pb.MessageType_MsgTransferLeader:
    if r.State != StateLeader {
        if r.Lead != None {
            m.To = r.Lead
            r.msgs = append(r.msgs, m)
        }
        return nil
    }
    // Leader 处理 Transfer...

func (r *Raft) addNode(id uint64) {
    if _, ok := r.Prs[id]; ok {
        return  // 已存在，忽略
    }
    r.Prs[id] = &Progress{Next: r.RaftLog.LastIndex() + 1}
    r.PendingConfIndex = None
}

func (r *Raft) removeNode(id uint64) {
    delete(r.Prs, id)
    r.PendingConfIndex = None
    if r.State == StateLeader {
        if r.maybeCommit() {
            r.bcastAppend()
        }
    }
}

storeMeta.regionRanges.Delete(&regionItem{region: oldRegion})  // 先删（此时 EndKey 还是旧值）
oldRegion.EndKey = split.SplitKey                               // 再修改
storeMeta.regionRanges.ReplaceOrInsert(&regionItem{region: oldRegion})  // 再插入

// 在 maybeCreatePeer 中，或者 addNode apply 后
meta.Lock()
meta.regionRanges.ReplaceOrInsert(&regionItem{region: peer.Region()})
meta.Unlock()

if len(d.Region().Peers) == 2 &&
    cc.ChangeType == pb.ConfChangeType_RemoveNode &&
    cc.Peer.Id == d.PeerId() {
    // 先转移 Leader，让 client 重试
    d.RaftGroup.TransferLeader(otherPeerID)
    return  // 不 propose，让 client 重试
}

func (d *peerMsgHandler) startToDestroyPeer() {
    if len(d.Region().Peers) == 2 && d.IsLeader() {
        // 找到另一个节点
        for _, peer := range d.Region().Peers {
            if peer.Id != d.PeerId() {
                heartbeat := pb.Message{
                    To:      peer.Id,
                    MsgType: pb.MessageType_MsgHeartbeat,
                    Commit:  d.peerStorage.raftState.HardState.Commit,
                }
                for i := 0; i < 10; i++ {
                    d.Send(d.ctx.trans, []pb.Message{heartbeat})
                    time.Sleep(100 * time.Millisecond)
                }
                break
            }
        }
    }
    d.destroyPeer()
}

for _, entry := range committedEntries {
    kvWB = d.processCommittedEntry(&entry, kvWB)
    if d.stopped {
        return  // 已销毁，不再处理
    }
}

// ConfChange apply 后
if d.IsLeader() {
    d.notifyHeartbeatScheduler(region, d.peer)
}

// Split apply 后
if d.IsLeader() {
    d.notifyHeartbeatScheduler(oldRegion, d.peer)
    d.notifyHeartbeatScheduler(newRegion, newPeer)  // 新 Region 也要通知！
}

if len(d.Region().Peers) != len(adminReq.Split.NewPeerIds) {
    d.handleProposal(entry, ErrRespStaleCommand(d.Term()))
    return kvWB
}

if err := util.CheckRegionEpoch(msg, region, true); err != nil {
    d.handleProposal(entry, ErrResp(err))
    return kvWB
}

// 在 sendHeartbeat 中
if pr.Match == 0 {
    // 对方还没有数据，Commit 设为 RaftInvalidIndex
    m.Commit = 0
}

if len(sourceRegion.GetPeers()) < cluster.GetMaxReplicas() {
    return nil
}

diff = sourceStore.regionSize - targetStore.regionSize
如果 diff > 2 * region.ApproximateSize，才执行迁移

迁移 Region R（大小 S）之后：
sourceStore 减少 S，targetStore 增加 S
新的 diff = (originalDiff - 2S)

如果 originalDiff > 2S，则新 diff > 0（source 仍然大于 target），不会反向迁移

// Region 元信息
message Region {
    uint64 id = 1;
    bytes start_key = 2;
    bytes end_key = 3;
    RegionEpoch region_epoch = 4;
    repeated Peer peers = 5;
}

// Region 版本信息
message RegionEpoch {
    uint64 conf_ver = 1;  // ConfChange 版本
    uint64 version = 2;   // Split 版本
}

// 成员变更
message ConfChange {
    ConfChangeType change_type = 1;  // AddNode / RemoveNode
    uint64 node_id = 2;
    bytes context = 3;  // 序列化的 RaftCmdRequest
}