Repository: tikv/tikv
File: components/raftstore/src/store/fsm/peer.rs (Rust)

Code before:
                    "msg_type" => %util::MsgType(msg),
                );
                self.ctx.raft_metrics.message_dropped.stale_msg.inc();
                return Ok(Either::Left(key));
            } else {
                panic!(
                    "{} meta corrupted: {:?} != {:?}",
                    self.fsm.peer.tag,
                    meta.regions[&self.region_id()],
                    self.region()
                );
            }
        }

        if meta.atomic_snap_regions.contains_key(&region_id) {
            info!(
                "atomic snapshot is applying, skip";
                "region_id" => self.fsm.region_id(),
                "peer_id" => self.fsm.peer_id(),
            );
            return Ok(Either::Left(key));
        }

        for region in &meta.pending_snapshot_regions {
            if enc_start_key(region) < snap_enc_end_key &&
               enc_end_key(region) > snap_enc_start_key &&
               // Same region can overlap, we will apply the latest version of snapshot.
               region.get_id() != snap_region.get_id()
            {
                info!(
                    "pending region overlapped";
                    "region_id" => self.fsm.region_id(),
                    "peer_id" => self.fsm.peer_id(),
                    "region" => ?region,
                    "snap" => ?snap_region,
                    "msg_type" => %util::MsgType(msg),
                );
                self.ctx.raft_metrics.message_dropped.region_overlap.inc();
                return Ok(Either::Left(key));
            }
        }

        let mut is_overlapped = false;
        let mut regions_to_destroy = vec![];
        // In some extreme cases, it may cause source peer destroyed improperly so that
        // a later CommitMerge may panic because source is already destroyed, so just
        // drop the message:
        // - A new snapshot is received whereas a snapshot is still in applying, and the
        //   snapshot under applying is generated before merge and the new snapshot is
        //   generated after merge. After the applying snapshot is finished, the log may
        //   able to catch up and so a CommitMerge will be applied.

Review comment (suggestion):
This change can avoid the panic, but whether it is sufficient needs more careful thoughts. In some cases, panicking is better than letting the process run and do unexpected things.

Code after:
            return Ok(Either::Left(key));
        }

        if snap_region
            .get_peers()
            .iter()
            .all(|p| p.get_id() != peer_id)
        {
            info!(
                "snapshot doesn't contain to peer, skip";
                "region_id" => self.fsm.region_id(),
                "peer_id" => self.fsm.peer_id(),
                "snap" => ?snap_region,
                "to_peer" => ?msg.get_to_peer(),
                "msg_type" => %util::MsgType(msg),
            );
            self.ctx.raft_metrics.message_dropped.region_no_peer.inc();
            return Ok(Either::Left(key));
        }

        let mut meta = self.ctx.store_meta.lock().unwrap();
        // Check if the region matches the metadata. A mismatch means another
        // peer has replaced the current peer, which can happen during a split: a
        // peer is first created via raft message, then replaced by another peer
        // (of the same region) when the split is applied.
        let region_mismatch = match meta.regions.get(&self.region_id()) {
            Some(region) => *region != *self.region(),
            None => {
                // If the region doesn't exist, treat it as a mismatch. This can
                // happen in rare situations (e.g. #17469).
                warn!(
                    "region not found in meta";
                    "region_id" => self.fsm.region_id(),
                    "peer_id" => self.fsm.peer_id(),
                );
                true
            }
        };
        if region_mismatch {
            if !self.fsm.peer.is_initialized() {
                info!(
                    "stale delegate detected, skip";
                    "region_id" => self.fsm.region_id(),
                    "peer_id" => self.fsm.peer_id(),
                    "msg_type" => %util::MsgType(msg),
                );
                self.ctx.raft_metrics.message_dropped.stale_msg.inc();
                return Ok(Either::Left(key));
            } else {
                panic!(
                    "{} meta corrupted: {:?} != {:?}",