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

Code before:
                    || pr.pending_request_snapshot != INVALID_INDEX
                {
                    return Err(box_err!(
                        "there is a pending snapshot peer {} [{:?}], skip merge",
                        id,
                        pr
                    ));
                }
                if min_m.unwrap_or(u64::MAX) > pr.matched {
                    min_m = Some(pr.matched);
                }
                if min_c.unwrap_or(u64::MAX) > pr.committed_index {
                    min_c = Some(pr.committed_index);
                }
            }
        }
        let (min_m, min_c) = (min_m.unwrap_or(0), min_c.unwrap_or(0));
        if min_m < min_c {
            warn!(
                "min_matched < min_committed, this is likely due to unpersisted raft log loss after restart";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "min_matched" => min_m,
                "min_committed" => min_c,
            );
            // Do NOT override `min_matched` to `min_committed` here because
            // some logs maybe not be persisted after restart when
            // `max_apply_unpersisted_log_limit > 0`.
            // The CatchUpLogs phase should use the real min matched log index
            // to recover raft state.
        }
        Ok((min_m, min_c))
    }

    fn pre_propose_prepare_merge<T: Transport>(
        &mut self,
        ctx: &mut PollContext<EK, ER, T>,
        req: &mut RaftCmdRequest,
    ) -> Result<()> {
        // Check existing prepare_merge_fence.
        let mut passed_merge_fence = false;
        if self.prepare_merge_fence > 0 {
            let applied_index = self.get_store().applied_index();
            if applied_index >= self.prepare_merge_fence {
                // Check passed, clear fence and start proposing pessimistic locks and
                // PrepareMerge.
                self.prepare_merge_fence = 0;
                self.pending_prepare_merge = None;
                passed_merge_fence = true;
            } else {
                self.pending_prepare_merge = Some(mem::take(req));

Review comment (bug):
Can we reject proposing `PrepareMerge` when `min_m < min_c` or even disable early apply temporally during region merge?

AFAIK, the region merge (at least part of it) is built on the assumption `min_m >= min_c`[^1].

Also, there is a minor issue: the entry size check might be wrong, now it should account the size of entries from `min_m`[^2]. 

[^1]: https://github.com/tikv/tikv/pull/9982#discussion_r610990186
[^2]: https://github.com/tikv/tikv/blob/55993bf0a6766bd65680b72f5e5ab3b84d3815c2/components/raftstore/src/store/peer.rs#L4365-L4408

Code after:
                    || pr.pending_request_snapshot != INVALID_INDEX
                {
                    return Err(box_err!(
                        "there is a pending snapshot peer {} [{:?}], skip merge",
                        id,
                        pr
                    ));
                }
                if min_m.unwrap_or(u64::MAX) > pr.matched {
                    min_m = Some(pr.matched);
                }
                if min_c.unwrap_or(u64::MAX) > pr.committed_index {
                    min_c = Some(pr.committed_index);
                }
            }
        }
        let (min_m, min_c) = (min_m.unwrap_or(0), min_c.unwrap_or(0));
        if min_m < min_c {
            warn!(
                "min_matched < min_committed, this is likely caused by disk IO jitters";
                "region_id" => self.region_id,
                "peer_id" => self.peer.get_id(),
                "min_matched" => min_m,
                "min_committed" => min_c,
            );
            // For region leader, matched index can be smaller than committed
            // index if the leader's disk IO is slow.
        }
        Ok((min_m, min_c))
    }

    fn pre_propose_prepare_merge<T: Transport>(
        &mut self,
        ctx: &mut PollContext<EK, ER, T>,
        req: &mut RaftCmdRequest,
    ) -> Result<()> {
        // Check existing prepare_merge_fence.
        let mut passed_merge_fence = false;
        if self.prepare_merge_fence > 0 {
            let applied_index = self.get_store().applied_index();
            if applied_index >= self.prepare_merge_fence {
                // Check passed, clear fence and start proposing pessimistic locks and
                // PrepareMerge.
                self.prepare_merge_fence = 0;
                self.pending_prepare_merge = None;
                passed_merge_fence = true;
            } else {
                self.pending_prepare_merge = Some(mem::take(req));
                info!(
                    "reject PrepareMerge because applied_index has not reached prepare_merge_fence";
                    "region_id" => self.region_id,