fn check_msg(&mut self, msg: &RaftMessage) -> Result<CheckMsgStatus> {
        let region_id = msg.get_region_id();
        let from_epoch = msg.get_region_epoch();
        let msg_type = msg.get_message().get_msg_type();
        let from_store_id = msg.get_from_peer().get_store_id();
        let to_peer_id = msg.get_to_peer().get_id();

        // Check if the target peer is tombstone.
        let state_key = keys::region_state_key(region_id);
        let local_state: RegionLocalState =
            match self.ctx.engines.kv.get_msg_cf(CF_RAFT, &state_key)? {
                Some(state) => state,
                None => return Ok(CheckMsgStatus::NewPeerFirst),
            };

        if local_state.get_state() != PeerState::Tombstone {
            // Maybe split, but not registered yet.
            if !util::is_first_vote_msg(msg.get_message()) {
                self.ctx.raft_metrics.message_dropped.region_nonexistent += 1;
                return Err(box_err!(
                    "[region {}] region not exist but not tombstone: {:?}",
                    region_id,
                    local_state
                ));
            }
            info!(
                "region doesn't exist yet, wait for it to be split";
                "region_id" => region_id
            );
            return Ok(CheckMsgStatus::FirstRequestVote);
        }
        debug!(
            "region is in tombstone state";
            "region_id" => region_id,
            "region_local_state" => ?local_state,
        );
        let region = local_state.get_region();
        let region_epoch = region.get_region_epoch();
        if local_state.has_merge_state() {
            info!(
                "merged peer receives a stale message";
                "region_id" => region_id,
                "current_region_epoch" => ?region_epoch,
                "msg_type" => ?msg_type,
            );

            let merge_target = if let Some(peer) = util::find_peer(region, from_store_id) {
                // Maybe the target is promoted from learner to voter, but the follower
                // doesn't know it. So we only compare peer id.
                assert_eq!(peer.get_id(), msg.get_from_peer().get_id());
                // Let stale peer decides whether it should wait for merging or just remove
                // itself.
                Some(local_state.get_merge_state().get_target().to_owned())
            } else {
                // If a peer is isolated before prepare_merge and conf remove, it should just
                // remove itself.
                None
            };
            self.ctx
                .handle_stale_msg(msg, region_epoch.clone(), true, merge_target);
            return Ok(CheckMsgStatus::DropMsg);
        }
        // The region in this peer is already destroyed
        if util::is_epoch_stale(from_epoch, region_epoch) {
            self.ctx.raft_metrics.message_dropped.region_tombstone_peer += 1;
            info!(
                "tombstone peer receives a stale message";
                "region_id" => region_id,
                "from_region_epoch" => ?from_epoch,
                "current_region_epoch" => ?region_epoch,
                "msg_type" => ?msg_type,
            );
            let mut need_gc_msg = util::is_vote_msg(msg.get_message());
            if msg.has_extra_msg() {
                // A learner can't vote so it sends the check-stale-peer msg to others to find out whether
                // it is removed due to conf change or merge.
                need_gc_msg |=
                    msg.get_extra_msg().get_type() == ExtraMessageType::MsgCheckStalePeer;
                // For backward compatibility
                need_gc_msg |= msg.get_extra_msg().get_type() == ExtraMessageType::MsgRegionWakeUp;
            }
            let not_exist = util::find_peer(region, from_store_id).is_none();
            self.ctx
                .handle_stale_msg(msg, region_epoch.clone(), need_gc_msg && not_exist, None);

            if need_gc_msg && !not_exist {
                let mut send_msg = RaftMessage::default();
                send_msg.set_region_id(region_id);
                send_msg.set_from_peer(msg.get_to_peer().clone());
                send_msg.set_to_peer(msg.get_from_peer().clone());
                send_msg.set_region_epoch(region_epoch.clone());
                let extra_msg = send_msg.mut_extra_msg();
                extra_msg.set_type(ExtraMessageType::MsgCheckStalePeerResponse);
                extra_msg.set_check_peers(region.get_peers().into());
                if let Err(e) = self.ctx.trans.send(send_msg) {
                    error!(
                        "send check stale peer response message failed";
                        "region_id" => region_id,
                        "error_code" => %e.error_code(),
                        "err" => ?e
                    );
                }
                self.ctx.need_flush_trans = true;
            }

            return Ok(CheckMsgStatus::DropMsg);
        }
        // A tombstone peer may not apply the conf change log which removes itself.
        // In this case, the local epoch is stale and the local peer can be found from region.
        // We can compare the local peer id with to_peer_id to verify whether it is correct to create a new peer.
        if let Some(local_peer_id) =
            util::find_peer(region, self.ctx.store_id()).map(|r| r.get_id())
        {
            if to_peer_id <= local_peer_id {
                self.ctx.raft_metrics.message_dropped.region_tombstone_peer += 1;
                info!(
                    "tombstone peer receives a stale message, local_peer_id >= to_peer_id in msg";
                    "region_id" => region_id,
                    "local_peer_id" => local_peer_id,
                    "to_peer_id" => to_peer_id,
                    "msg_type" => ?msg_type
                );
                return Ok(CheckMsgStatus::DropMsg);
            }
        }
        Ok(CheckMsgStatus::NewPeer)
    }
/// Checks if the message is targeting a stale peer.
Checks if the message is targeting a stale peer.
Checks if the message is targeting a stale peer .
fn check_msg(&mut self, msg: &RaftMessage) -> Result<CheckMsgStatus> {
        let region_id = msg.get_region_id();
        let from_epoch = msg.get_region_epoch();
        let msg_type = msg.get_message().get_msg_type();
        let from_store_id = msg.get_from_peer().get_store_id();
        let to_peer_id = msg.get_to_peer().get_id();
        // Check if the target peer is tombstone.
        let state_key = keys::region_state_key(region_id);
        let local_state: RegionLocalState =
            match self.ctx.engines.kv.get_msg_cf(CF_RAFT, &state_key)? {
                Some(state) => state,
                None => return Ok(CheckMsgStatus::NewPeerFirst),
            };
        if local_state.get_state() != PeerState::Tombstone {
            // Maybe split, but not registered yet.
            if !util::is_first_vote_msg(msg.get_message()) {
                self.ctx.raft_metrics.message_dropped.region_nonexistent += 1;
                return Err(box_err!(
                    "[region {}] region not exist but not tombstone: {:?}",
                    region_id,
                    local_state
                ));
            }
            info!(
                "region doesn't exist yet, wait for it to be split";
                "region_id" => region_id
            );
            return Ok(CheckMsgStatus::FirstRequestVote);
        }
        debug!(
            "region is in tombstone state";
            "region_id" => region_id,
            "region_local_state" => ?local_state,
        );
        let region = local_state.get_region();
        let region_epoch = region.get_region_epoch();
        if local_state.has_merge_state() {
            info!(
                "merged peer receives a stale message";
                "region_id" => region_id,
                "current_region_epoch" => ?region_epoch,
                "msg_type" => ?msg_type,
            );
            let merge_target = if let Some(peer) = util::find_peer(region, from_store_id) {
                // Maybe the target is promoted from learner to voter, but the follower
                // doesn't know it. So we only compare peer id.
                assert_eq!(peer.get_id(), msg.get_from_peer().get_id());
                // Let stale peer decides whether it should wait for merging or just remove
                // itself.
                Some(local_state.get_merge_state().get_target().to_owned())
            } else {
                // If a peer is isolated before prepare_merge and conf remove, it should just
                // remove itself.
                None
            };
            self.ctx
                .handle_stale_msg(msg, region_epoch.clone(), true, merge_target);
            return Ok(CheckMsgStatus::DropMsg);
        }
        // The region in this peer is already destroyed
        if util::is_epoch_stale(from_epoch, region_epoch) {
            self.ctx.raft_metrics.message_dropped.region_tombstone_peer += 1;
            info!(
                "tombstone peer receives a stale message";
                "region_id" => region_id,
                "from_region_epoch" => ?from_epoch,
                "current_region_epoch" => ?region_epoch,
                "msg_type" => ?msg_type,
            );
            let mut need_gc_msg = util::is_vote_msg(msg.get_message());
            if msg.has_extra_msg() {
                // A learner can't vote so it sends the check-stale-peer msg to others to find out whether
                // it is removed due to conf change or merge.
                need_gc_msg |=
                    msg.get_extra_msg().get_type() == ExtraMessageType::MsgCheckStalePeer;
                // For backward compatibility
                need_gc_msg |= msg.get_extra_msg().get_type() == ExtraMessageType::MsgRegionWakeUp;
            }
            let not_exist = util::find_peer(region, from_store_id).is_none();
            self.ctx
                .handle_stale_msg(msg, region_epoch.clone(), need_gc_msg && not_exist, None);
            if need_gc_msg && !not_exist {
                let mut send_msg = RaftMessage::default();
                send_msg.set_region_id(region_id);
                send_msg.set_from_peer(msg.get_to_peer().clone());
                send_msg.set_to_peer(msg.get_from_peer().clone());
                send_msg.set_region_epoch(region_epoch.clone());
                let extra_msg = send_msg.mut_extra_msg();
                extra_msg.set_type(ExtraMessageType::MsgCheckStalePeerResponse);
                extra_msg.set_check_peers(region.get_peers().into());
                if let Err(e) = self.ctx.trans.send(send_msg) {
                    error!(
                        "send check stale peer response message failed";
                        "region_id" => region_id,
                        "error_code" => %e.error_code(),
                        "err" => ?e
                    );
                }
                self.ctx.need_flush_trans = true;
            }
            return Ok(CheckMsgStatus::DropMsg);
        }
        // A tombstone peer may not apply the conf change log which removes itself.
        // In this case, the local epoch is stale and the local peer can be found from region.
        // We can compare the local peer id with to_peer_id to verify whether it is correct to create a new peer.
        if let Some(local_peer_id) =
            util::find_peer(region, self.ctx.store_id()).map(|r| r.get_id())
        {
            if to_peer_id <= local_peer_id {
                self.ctx.raft_metrics.message_dropped.region_tombstone_peer += 1;
                info!(
                    "tombstone peer receives a stale message, local_peer_id >= to_peer_id in msg";
                    "region_id" => region_id,
                    "local_peer_id" => local_peer_id,
                    "to_peer_id" => to_peer_id,
                    "msg_type" => ?msg_type
                );
                return Ok(CheckMsgStatus::DropMsg);
            }
        }
        Ok(CheckMsgStatus::NewPeer)
    }
fn check_msg ( & mut self , msg : & RaftMessage ) -> Result < CheckMsgStatus > { let region_id = msg . get_region_id ( ) ; let from_epoch = msg . get_region_epoch ( ) ; let msg_type = msg . get_message ( ) . get_msg_type ( ) ; let from_store_id = msg . get_from_peer ( ) . get_store_id ( ) ; let to_peer_id = msg . get_to_peer ( ) . get_id ( ) ; // Check if the target peer is tombstone. let state_key = keys :: region_state_key ( region_id ) ; let local_state : RegionLocalState = match self . ctx . engines . kv . get_msg_cf ( CF_RAFT , & state_key ) ? { Some ( state ) => state , None => return Ok ( CheckMsgStatus :: NewPeerFirst ) , } ; if local_state . get_state ( ) != PeerState :: Tombstone { // Maybe split, but not registered yet. if ! util :: is_first_vote_msg ( msg . get_message ( ) ) { self . ctx . raft_metrics . message_dropped . region_nonexistent += 1 ; return Err ( box_err ! ( " " region_id local_state ) ) ; } info ! ( " " " " region_id ) ; return Ok ( CheckMsgStatus :: FirstRequestVote ) ; } debug ! ( " " " " region_id " " local_state ) ; let region = local_state . get_region ( ) ; let region_epoch = region . get_region_epoch ( ) ; if local_state . has_merge_state ( ) { info ! ( " " " " region_id " " region_epoch " " msg_type ) ; let merge_target = if let Some ( peer ) = util :: find_peer ( region , from_store_id ) { // Maybe the target is promoted from learner to voter, but the follower // doesn't know it. So we only compare peer id. assert_eq ! ( peer get_id ( ) msg get_from_peer ( ) get_id ( ) ) ; // Let stale peer decides whether it should wait for merging or just remove // itself. Some ( local_state . get_merge_state ( ) . get_target ( ) . to_owned ( ) ) } else { // If a peer is isolated before prepare_merge and conf remove, it should just // remove itself. None } ; self . ctx . handle_stale_msg ( msg , region_epoch . clone ( ) , true , merge_target ) ; return Ok ( CheckMsgStatus :: DropMsg ) ; } // The region in this peer is already destroyed if util :: is_epoch_stale ( from_epoch , region_epoch ) { self . ctx . raft_metrics . message_dropped . region_tombstone_peer += 1 ; info ! ( " " " " region_id " " from_epoch " " region_epoch " " msg_type ) ; let mut need_gc_msg = util :: is_vote_msg ( msg . get_message ( ) ) ; if msg . has_extra_msg ( ) { // A learner can't vote so it sends the check-stale-peer msg to others to find out whether // it is removed due to conf change or merge. need_gc_msg |= msg . get_extra_msg ( ) . get_type ( ) == ExtraMessageType :: MsgCheckStalePeer ; // For backward compatibility need_gc_msg |= msg . get_extra_msg ( ) . get_type ( ) == ExtraMessageType :: MsgRegionWakeUp ; } let not_exist = util :: find_peer ( region , from_store_id ) . is_none ( ) ; self . ctx . handle_stale_msg ( msg , region_epoch . clone ( ) , need_gc_msg && not_exist , None ) ; if need_gc_msg && ! not_exist { let mut send_msg = RaftMessage :: default ( ) ; send_msg . set_region_id ( region_id ) ; send_msg . set_from_peer ( msg . get_to_peer ( ) . clone ( ) ) ; send_msg . set_to_peer ( msg . get_from_peer ( ) . clone ( ) ) ; send_msg . set_region_epoch ( region_epoch . clone ( ) ) ; let extra_msg = send_msg . mut_extra_msg ( ) ; extra_msg . set_type ( ExtraMessageType :: MsgCheckStalePeerResponse ) ; extra_msg . set_check_peers ( region . get_peers ( ) . into ( ) ) ; if let Err ( e ) = self . ctx . trans . send ( send_msg ) { error ! ( " " " " region_id " " e error_code ( ) " " e ) ; } self . ctx . need_flush_trans = true ; } return Ok ( CheckMsgStatus :: DropMsg ) ; } // A tombstone peer may not apply the conf change log which removes itself. // In this case, the local epoch is stale and the local peer can be found from region. // We can compare the local peer id with to_peer_id to verify whether it is correct to create a new peer. if let Some ( local_peer_id ) = util :: find_peer ( region , self . ctx . store_id ( ) ) . map ( | r | r . get_id ( ) ) { if to_peer_id <= local_peer_id { self . ctx . raft_metrics . message_dropped . region_tombstone_peer += 1 ; info ! ( " " " " region_id " " local_peer_id " " to_peer_id " " msg_type ) ; return Ok ( CheckMsgStatus :: DropMsg ) ; } } Ok ( CheckMsgStatus :: NewPeer ) }
Checks if the message is targeting a stale peer.
Checks if the message is targeting a stale peer .
&mut self
msg
&RaftMessage