Repository: tikv/tikv
File: src/server/gc_worker/gc_worker.rs (Rust)

Code before:
    /// Check need gc without getting snapshot.
    /// If this is not supported or any error happens, returns true to do
    /// further check after getting snapshot.
    fn need_gc(&self, start_key: &[u8], end_key: &[u8], safe_point: TimeStamp) -> bool {
        // todo(SpadeA): multi-rocks db version should handle with this differently
        // which will be reflected in the imlementation of the v2's RaftKv.
        let props = match self
            .engine
            .get_mvcc_properties_cf(CF_WRITE, safe_point, start_key, end_key)
        {
            Some(c) => c,
            None => return true,
        };
        check_need_gc(safe_point, self.cfg.ratio_threshold, &props)
    }

    /// Cleans up outdated data.
    fn gc_key(
        &mut self,
        safe_point: TimeStamp,
        key: &Key,
        gc_info: &mut GcInfo,
        txn: &mut MvccTxn,
        reader: &mut MvccReader<E::Snap>,
    ) -> Result<()> {
        let _guard = set_panic_context! {"key" => key};
        let next_gc_info = gc(txn, reader, key.clone(), safe_point).map_err(TxnError::from_mvcc)?;
        gc_info.found_versions += next_gc_info.found_versions;
        gc_info.deleted_versions += next_gc_info.deleted_versions;
        gc_info.is_completed = next_gc_info.is_completed;
        let stats = mem::take(&mut reader.statistics);
        self.mut_stats(GcKeyMode::txn).add(&stats);
        Ok(())
    }

    fn new_txn() -> MvccTxn {
        // TODO txn only used for GC, but this is hacky, maybe need an Option?
        let concurrency_manager = ConcurrencyManager::new(1.into());
        MvccTxn::new(TimeStamp::zero(), concurrency_manager)
    }

    fn flush_txn(
        limiter: &Limiter,
        engine: &E,
        txns: &mut HashMap<u64, MvccTxn>, // region id -> MvccTxn for this region
    ) -> Result<()> {
        let mut modifies = HashMap::default();
        let mut write_size = 0;
        for (id, txn) in txns.drain() {
            write_size += txn.write_size();
            modifies.insert(id, txn.into_modifies());

Review comment (bug):
This should help locate the bad key that caused the GC worker to panic (`self.0 > 0`) in https://github.com/pingcap/tidb/issues/57980. cc @zhangjinpeng87

Example panic log:
```
[2025/02/14 08:44:39.142 +08:00] [FATAL] [lib.rs:544] ["self.0 > 0"] [context="(src/server/gc_worker/gc_worker.rs:468) region=>264, (src/server/gc_worker/gc_worker.rs:361) key=>7480000000000000FF6E5F698000000000FF0000010380000000FF0000001A03800000FF0000000033000000FC"] [backtrace="   0: tikv_util::set_panic_hook::{{closure}}\n   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call\n   2: std::panicking::rust_panic_with_hook\n   3: std::panicking::begin_panic_handler::{{closure}}\n   4: std::sys_common::backtrace::__rust_end_short_backtrace\n   5: rust_begin_unwind\n   6: core::panicking::panic_fmt\n   7: tikv::storage::txn::actions::gc::Gc<S>::next_write\n   8: tikv::storage::txn::actions::gc::Gc<S>::run\n   9: tikv::storage::txn::actions::gc::gc\n  10: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::gc_key\n  11: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::gc_keys\n  12: tikv::server::gc_worker::gc_worker::GcRunnerCore<E>::run\n  13: <tikv::server::gc_worker::gc_worker::GcRunner<E> as tikv_util::worker::pool::Runnable>::run::{{closure}}\n  14: yatp::task::future::RawTask<F>::poll\n  15: yatp::task::future::TaskCell::poll\n  16: <yatp::task::future::Runner as yatp::pool::runner::Runner>::handle\n  17: <tikv_util::yatp_pool::YatpPoolRunner<T> as yatp::pool::runner::Runner>::handle\n  18: yatp::pool::worker::WorkerThread<T,R>::run\n  19: yatp::pool::builder::LazyBuilder<T>::build::{{closure}}\n  20: std::sys_common::backtrace::__rust_begin_short_backtrace\n  21: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}\n  22: <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once\n  23: std::panicking::try::do_call\n  24: __rust_try\n  25: std::panicking::try\n  26: std::panic::catch_unwind\n  27: std::thread::Builder::spawn_unchecked_::{{closure}}\n  28: core::ops::function::FnOnce::call_once{{vtable.shim}}\n  29: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n  30: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n  31: std::sys::unix::thread::Thread::new::thread_start\n  32: start_thread\n  33: __clone\n"] [location=/home/bishengh/Desktop/tikv/src/storage/txn/actions/gc.rs:61] [thread_name=gc-worker-0] [thread_id=28]
```

There was another panic location (`!last_change_ts.is_zero()`) in the txn scheduler but adding a panic context for it is difficult, so I'll leave it for now. 



Code after:
    /// Check need gc without getting snapshot.
    /// If this is not supported or any error happens, returns true to do
    /// further check after getting snapshot.
    fn need_gc(&self, start_key: &[u8], end_key: &[u8], safe_point: TimeStamp) -> bool {
        // todo(SpadeA): multi-rocks db version should handle with this differently
        // which will be reflected in the imlementation of the v2's RaftKv.
        let props = match self
            .engine
            .get_mvcc_properties_cf(CF_WRITE, safe_point, start_key, end_key)
        {
            Some(c) => c,
            None => return true,
        };
        check_need_gc(safe_point, self.cfg.ratio_threshold, &props)
    }

    /// Cleans up outdated data.
    fn gc_key(
        &mut self,
        safe_point: TimeStamp,
        key: &Key,
        gc_info: &mut GcInfo,
        txn: &mut MvccTxn,
        reader: &mut MvccReader<E::Snap>,
    ) -> Result<()> {
        let _guard = set_panic_context! {"key" => key};
        let next_gc_info = gc(txn, reader, key.clone(), safe_point).map_err(TxnError::from_mvcc)?;
        gc_info.found_versions += next_gc_info.found_versions;
        gc_info.deleted_versions += next_gc_info.deleted_versions;
        gc_info.is_completed = next_gc_info.is_completed;
        let stats = mem::take(&mut reader.statistics);
        self.mut_stats(GcKeyMode::txn).add(&stats);
        Ok(())
    }

    /// Creates a `MvccTxn` with a dummy concurrency manager, do not use it
    /// beside the `GCRunner`. It's better to abstract `MvccTxn` for
    /// recording modifies only.
    fn mvcc_txn_with_dummy_cm() -> MvccTxn {
        // TODO txn only used for GC, but this is hacky, maybe need an Option?
        let concurrency_manager = ConcurrencyManager::new_dummy();
        MvccTxn::new(TimeStamp::zero(), concurrency_manager)
    }

    fn flush_txn(
        limiter: &Limiter,
        engine: &E,
        txns: &mut HashMap<u64, MvccTxn>, // region id -> MvccTxn for this region
    ) -> Result<()> {
        let mut modifies = HashMap::default();
        let mut write_size = 0;