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

Code before:
    EK: KvEngine,
    R: CasualRouter<EK> + Send + Clone + 'static,
    T: PdClient + 'static,
{
    fn on_timeout(&mut self) {
        self.handle_pending_applies(true);
        self.clean_stale_tick += 1;
        if self.clean_stale_tick >= self.clean_stale_ranges_tick {
            self.clean_stale_ranges();
            self.clean_stale_tick = 0;
        }
    }

    fn get_interval(&self) -> Duration {
        self.clean_stale_check_interval
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use std::{
        io,
        sync::{atomic::AtomicUsize, mpsc, Arc},
        thread,
        time::Duration,
    };

    use engine_test::{
        ctor::CfOptions,
        kv::{KvTestEngine, KvTestSnapshot},
    };
    use engine_traits::{
        CompactExt, FlowControlFactorsExt, KvEngine, MiscExt, Mutable, Peekable,
        RaftEngineReadOnly, SyncMutable, WriteBatch, WriteBatchExt, CF_DEFAULT, CF_WRITE,
    };
    use keys::data_key;
    use kvproto::raft_serverpb::{PeerState, RaftApplyState, RaftSnapshotData, RegionLocalState};
    use pd_client::RpcClient;
    use protobuf::Message;
    use tempfile::Builder;
    use tikv_util::{
        config::{ReadableDuration, ReadableSize},
        worker::{LazyWorker, Worker},
    };

    use super::*;
    use crate::{
        coprocessor::{
            ApplySnapshotObserver, BoxApplySnapshotObserver, Coprocessor, CoprocessorHost,
            ObserverContext,
        },

Review comment (suggestion):
It looks like they are executed asynchronously, which may cause problems with accidentally deleting the just applied snapshot.

Code after:
                    let _ = self.pre_apply_snapshot(&task);
                }
                SNAP_COUNTER.apply.all.inc();
                // to makes sure applying snapshots in order.
                self.pending_applies.push_back(task);
                self.mgr.set_pending_apply_count(self.pending_applies.len());
                self.handle_pending_applies(false);
                if !self.pending_applies.is_empty() {
                    // delay the apply and retry later
                    SNAP_COUNTER.apply.delay.inc()
                }
            }
            Task::Destroy {
                region_id,
                start_key,
                end_key,
            } => {
                let region_cleaner = self.region_cleaner.clone();
                self.region_cleanup_pool
                    .spawn(async move {
                        fail_point!("on_region_worker_destroy", region_id == 1000, |_| {});
                        let mut region_cleaner = region_cleaner.lock().unwrap();
                        // try to delay the range deletion because
                        // there might be a coprocessor request related to this range
                        region_cleaner.insert_pending_delete_range(region_id, start_key, end_key);
                        region_cleaner.clean_stale_ranges();
                    })
                    .unwrap_or_else(|e| {
                        error!("failed to destroy region"; "region_id" => region_id, "err" => ?e);
                    });
            }
        }
    }
}

impl<EK, R, T> RunnableWithTimer for Runner<EK, R, T>
where
    EK: KvEngine,
    R: CasualRouter<EK> + Send + Clone + 'static,
    T: PdClient + 'static,
{
    fn on_timeout(&mut self) {
        self.handle_pending_applies(true);
        self.clean_stale_tick += 1;
        if self.clean_stale_tick >= self.clean_stale_ranges_tick {
            self.region_cleaner.lock().unwrap().clean_stale_ranges();
            self.clean_stale_tick = 0;
        }
    }

    fn get_interval(&self) -> Duration {