Repository: tikv/tikv (language: rust)

Issue:
more than one of tikv panic when injection pd leader timeoffset (ahead 5mins)
## Bug Report

<!-- Thanks for your bug report! Don't worry if you can't fill out all the sections. -->

### What version of TiKV are you using?
./tikv-server -V
 TiKV 
Release Version:   8.5.0-alpha
Edition:           Community
Git Commit Hash:   678edbe8fa3404c2d9f703031ca621792bf61cb7
Git Commit Branch: HEAD
UTC Build Time:    2024-12-23 07:24:00
Rust Version:      rustc 1.77.0-nightly (89e2160c4 2023-12-27)
Enable Features:   memory-engine pprof-fp jemalloc mem-profiling portable sse test-engine-kv-rocksdb test-engine-raft-raft-engine trace-async-tasks openssl-vendored
Profile:           dist_release
2024-12-24T16:32:05.343+0800

### What operating system and CPU are you using?
8c/32g

### Steps to reproduce
1、run tpcc
2、inject pd leader timeoffset （ahead 5mins）

### What did you expect?
no panic

### What did happened?
more than one of tikv panic 

`[2024/12/24 20:48:29.849 +08:00] [FATAL] [lib.rs:480] ["invalid max_ts update: 454831586436448257 exceeds the limit 454831522761670656, source=prepare_snap_ctx-get-454831586436448257"] [backtrace="   0: tikv_util::set_panic_hook::{{closure}}\n             at workspace/source/tikv/components/tikv_util/src/lib.rs:479:18\n   1: <alloc::boxed::Box<F,A> as core::ops::function::Fn<Args>>::call\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2029:9\n      std::panicking::rust_panic_with_hook\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:783:13\n   2: std::panicking::begin_panic_handler::{{closure}}\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:657:13\n   3: std::sys_common::backtrace::__rust_end_short_backtrace\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:171:18\n   4: rust_begin_unwind\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:645:5\n   5: core::panicking::panic_fmt\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panicking.rs:72:14\n   6: concurrency_manager::ConcurrencyManager::report_error\n   7: concurrency_manager::ConcurrencyManager::update_max_ts\n      tikv::storage::prepare_snap_ctx\n             at workspace/source/tikv/src/storage/mod.rs:3358:9\n   8: tikv::storage::Storage<E,L,F>::get::{{closure}}\n             at workspace/source/tikv/src/storage/mod.rs:666:32\n      <resource_metering::InTags<T> as core::future::future::Future>::poll\n             at workspace/source/tikv/components/resource_metering/src/lib.rs:274:9\n      <futures_util::future::future::map::Map<Fut,F> as core::future::future::Future>::poll\n             at workspace/.cargo/registry/src/mirrors.tuna.tsinghua.edu.cn-df7c3c540f42cdbd/futures-util-0.3.31/src/future/future/map.rs:55:37\n      <futures_util::future::future::Map<Fut,F> as core::future::future::Future>::poll\n             at workspace/.cargo/registry/src/mirrors.tuna.tsinghua.edu.cn-df7c3c540f42cdbd/futures-util-0.3.31/src/lib.rs:86:13\n   9: <futures_util::future::future::map::Map<Fut,F> as core::future::future::Future>::poll\n             at workspace/.cargo/registry/src/mirrors.tuna.tsinghua.edu.cn-df7c3c540f42cdbd/futures-util-0.3.31/src/future/future/map.rs:55:37\n      <futures_util::future::future::Map<Fut,F> as core::future::future::Future>::poll\n             at workspace/.cargo/registry/src/mirrors.tuna.tsinghua.edu.cn-df7c3c540f42cdbd/futures-util-0.3.31/src/lib.rs:86:13\n      <resource_control::future::ControlledFuture<F> as core::future::future::Future>::poll\n             at workspace/source/tikv/components/resource_control/src/future.rs:48:19\n  10: resource_control::future::with_resource_limiter::{{closure}}\n             at workspace/source/tikv/components/resource_control/src/future.rs:204:11\n      <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll::{{closure}}\n             at workspace/source/tikv/components/tracker/src/tls.rs:64:23\n      std::thread::local::LocalKey<T>::try_with\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:270:16\n      std::thread::local::LocalKey<T>::with\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/local.rs:246:9\n      <tracker::tls::TrackedFuture<F> as core::future::future::Future>::poll\n             at workspace/source/tikv/components/tracker/src/tls.rs:62:27\n      yatp::task::future::RawTask<F>::poll\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:59:9\n  11: yatp::task::future::TaskCell::poll\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:103:9\n      <yatp::task::future::Runner as yatp::pool::runner::Runner>::handle\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/task/future.rs:387:20\n  12: <tikv_util::yatp_pool::YatpPoolRunner<T> as yatp::pool::runner::Runner>::handle\n             at workspace/source/tikv/components/tikv_util/src/yatp_pool/mod.rs:199:24\n      <yatp::queue::multilevel::TrackedRunner<R> as yatp::pool::runner::Runner>::handle\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/queue/multilevel.rs:285:19\n  13: yatp::pool::worker::WorkerThread<T,R>::run\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/worker.rs:48:13\n      yatp::pool::builder::LazyBuilder<T>::build::{{closure}}\n             at workspace/.cargo/git/checkouts/yatp-e704b73c3ee279b6/793be4d/src/pool/builder.rs:114:25\n      std::sys_common::backtrace::__rust_begin_short_backtrace\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys_common/backtrace.rs:155:18\n  14: std::thread::Builder::spawn_unchecked_::{{closure}}::{{closure}}\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:529:17\n      <core::panic::unwind_safe::AssertUnwindSafe<F> as core::ops::function::FnOnce<()>>::call_once\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/panic/unwind_safe.rs:272:9\n      std::panicking::try::do_call\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:552:40\n      std::panicking::try\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panicking.rs:516:19\n      std::panic::catch_unwind\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/panic.rs:142:14\n      std::thread::Builder::spawn_unchecked_::{{closure}}\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/thread/mod.rs:528:30\n      core::ops::function::FnOnce::call_once{{vtable.shim}}\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/core/src/ops/function.rs:250:5\n  15: <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9\n      <alloc::boxed::Box<F,A> as core::ops::function::FnOnce<Args>>::call_once\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/alloc/src/boxed.rs:2015:9\n      std::sys::unix::thread::Thread::new::thread_start\n             at root/.rustup/toolchains/nightly-2023-12-28-x86_64-unknown-linux-gnu/lib/rustlib/src/rust/library/std/src/sys/unix/thread.rs:108:17\n  16: start_thread\n  17: __GI___clone\n"] [location=/workspace/source/tikv/components/concurrency_manager/src/lib.rs:224] [thread_name=unified-read-pool-5] [thread_id=37]`


Discussion / hints:
/type bug
/severity critical 
/assign glorv
reassign to @ekexium  as it's a txn related issue.
It is caused by an intentional max_ts validity check https://github.com/tikv/tikv/pull/17917. The injected clock offset far exceeds the allowance config (by default 60s)

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index f4ef3094d93..16bc3274aaf 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1563,6 +1563,7 @@ dependencies = [
  "lazy_static",
  "online_config",
  "parking_lot 0.12.1",
+ "pd_client",
  "prometheus",
  "rand 0.8.5",
  "serde",
diff --git a/components/concurrency_manager/Cargo.toml b/components/concurrency_manager/Cargo.toml
index dc9cf1b558a..b98e721b593 100644
--- a/components/concurrency_manager/Cargo.toml
+++ b/components/concurrency_manager/Cargo.toml
@@ -13,6 +13,7 @@ kvproto = { workspace = true }
 lazy_static = "1.4.0"
 online_config = { workspace = true }
 parking_lot = "0.12"
+pd_client = { workspace = true }
 prometheus = "0.13"
 serde = { version = "1.0.194", features = ["derive"] }
 slog = { workspace = true }
diff --git a/components/concurrency_manager/benches/update_max_ts.rs b/components/concurrency_manager/benches/update_max_ts.rs
index 2c8dfd58120..df2e61aeb42 100644
--- a/components/concurrency_manager/benches/update_max_ts.rs
+++ b/components/concurrency_manager/benches/update_max_ts.rs
@@ -13,6 +13,8 @@ fn benchmark_update_max_ts(c: &mut Criterion) {
         latest_ts,
         limit_valid_time,
         ActionOnInvalidMaxTs::Error,
+        None,
+        Duration::ZERO,
     );
 
     cm.set_max_ts_limit(TimeStamp::new(4000));
diff --git a/components/concurrency_manager/src/lib.rs b/components/concurrency_manager/src/lib.rs
index 20b2b39ba22..31c57693baa 100644
--- a/components/concurrency_manager/src/lib.rs
+++ b/components/concurrency_manager/src/lib.rs
@@ -29,9 +29,10 @@ use std::{
 
 use crossbeam::atomic::AtomicCell;
 use lazy_static::lazy_static;
+use pd_client::{PdClient, PdFuture};
 use prometheus::{register_int_gauge, IntGauge};
 use thiserror::Error;
-use tikv_util::{error, time::Instant};
+use tikv_util::{error, future::block_on_timeout, time::Instant};
 use txn_types::{Key, Lock, TimeStamp};
 
 pub use self::{
@@ -58,12 +59,24 @@ const DEFAULT_LIMIT_VALID_DURATION: Duration = Duration::from_secs(60);
 //    between TiKV and PD.
 pub const LIMIT_VALID_TIME_MULTIPLIER: u32 = 3;
 
+const TSO_TIMEOUT: Duration = Duration::from_secs(5);
+
 #[derive(Copy, Clone, PartialEq, Eq)]
 struct MaxTsLimit {
     limit: TimeStamp,
     update_time: Instant,
 }
 
+pub trait TSOProvider: Send + Sync {
+    fn get_tso(&self) -> PdFuture<TimeStamp>;
+}
+
+impl<T: PdClient> TSOProvider for T {
+    fn get_tso(&self) -> PdFuture<TimeStamp> {
+        PdClient::get_tso(self)
+    }
+}
+
 // Pay attention that the async functions of ConcurrencyManager should not hold
 // the mutex.
 #[derive(Clone)]
@@ -82,6 +95,10 @@ pub struct ConcurrencyManager {
     limit_valid_duration: Duration,
     action_on_invalid_max_ts: Arc<AtomicActionOnInvalidMaxTs>,
 
+    max_ts_drift_allowance_ms: Arc<AtomicU64>,
+
+    tso: Option<Arc<dyn TSOProvider>>,
+
     time_provider: Arc<dyn TimeProvider>,
 }
 
@@ -91,6 +108,8 @@ impl ConcurrencyManager {
             latest_ts,
             DEFAULT_LIMIT_VALID_DURATION,
             ActionOnInvalidMaxTs::Panic,
+            None,
+            Duration::ZERO,
         )
     }
 
@@ -98,6 +117,8 @@ impl ConcurrencyManager {
         latest_ts: TimeStamp,
         limit_valid_duration: Duration,
         action_on_invalid_max_ts: ActionOnInvalidMaxTs,
+        tso: Option<Arc<dyn TSOProvider>>,
+        max_ts_drift_allowance: Duration,
     ) -> Self {
         let initial_limit = MaxTsLimit {
             limit: TimeStamp::new(0),
@@ -113,6 +134,10 @@ impl ConcurrencyManager {
             )),
             limit_valid_duration,
             time_provider: Arc::new(CoarseInstantTimeProvider),
+            tso,
+            max_ts_drift_allowance_ms: Arc::new(AtomicU64::new(
+                max_ts_drift_allowance.as_millis() as u64
+            )),
         }
     }
 
@@ -122,6 +147,8 @@ impl ConcurrencyManager {
         limit_valid_duration: Duration,
         action_on_invalid_max_ts: ActionOnInvalidMaxTs,
         time_provider: Arc<dyn TimeProvider>,
+        tso: Option<Arc<dyn TSOProvider>>,
+        max_ts_drift_allowance: Duration,
     ) -> Self {
         let initial_limit = MaxTsLimit {
             limit: TimeStamp::new(0),
@@ -136,6 +163,10 @@ impl ConcurrencyManager {
             )),
             limit_valid_duration,
             time_provider,
+            tso,
+            max_ts_drift_allowance_ms: Arc::new(AtomicU64::new(
+                max_ts_drift_allowance.as_millis() as u64
+            )),
         }
     }
 
@@ -180,7 +211,7 @@ impl ConcurrencyManager {
             if duration_to_last_limit_update < self.limit_valid_duration {
                 // limit is valid
                 let source = source.into_error_source();
-                self.report_error(new_ts, limit.limit, source, false)?;
+                self.double_check(new_ts, limit.limit, source, false)?;
             } else {
                 // limit is stale
                 // use an approximate limit to avoid false alerts caused by failed limit updates
@@ -192,7 +223,7 @@ impl ConcurrencyManager {
 
                 if new_ts > approximate_limit {
                     let source = source.into_error_source();
-                    self.report_error(new_ts, approximate_limit, source, true)?;
+                    self.double_check(new_ts, approximate_limit, source, true)?;
                 }
             }
         }
@@ -205,6 +236,42 @@ impl ConcurrencyManager {
         Ok(())
     }
 
+    // new_ts is greater than limit, or the approximate limit.
+    // To avoid false positive and guarantee TiKV availability, we need to
+    // double-check the new_ts with PD TSO.
+    fn double_check(
+        &self,
+        new_ts: TimeStamp,
+        limit: TimeStamp,
+        source: impl slog::Value + Display,
+        using_approximate: bool,
+    ) -> Result<(), crate::InvalidMaxTsUpdate> {
+        error!("possible invalid max-ts update; double checking";
+            "attempted_ts" => new_ts,
+            "max_allowed" => limit.into_inner(),
+            "source" => &source,
+            "using_approximate" => using_approximate,
+            "TSO_TIMEOUT" => ?TSO_TIMEOUT,
+        );
+        if let Some(tso) = &self.tso {
+            match block_on_timeout(tso.get_tso(), TSO_TIMEOUT) {
+                Ok(Ok(ts)) => {
+                    self.set_max_ts_limit(ts);
+                }
+                Ok(Err(e)) => {
+                    error!("failed to fetch from TSO for double checking"; "err" => ?e);
+                }
+                Err(()) => {
+                    error!("timeout when fetching from TSO for double checking"; "timeout" => ?TSO_TIMEOUT);
+                }
+            }
+        }
+        if new_ts > self.max_ts_limit.load().limit {
+            self.report_error(new_ts, limit, source, using_approximate)?;
+        }
+        Ok(())
+    }
+
     fn report_error(
         &self,
         new_ts: TimeStamp,
@@ -246,12 +313,17 @@ impl ConcurrencyManager {
     /// # Note
     /// If the new limit is smaller than the current limit, this operation will
     /// have no effect and return silently.
-    pub fn set_max_ts_limit(&self, limit: TimeStamp) {
-        if limit.is_max() {
+    pub fn set_max_ts_limit(&self, ts_from_tso: TimeStamp) {
+        if ts_from_tso.is_max() {
             error!("max_ts_limit cannot be set to u64::max");
             return;
         }
 
+        let limit = TimeStamp::compose(
+            ts_from_tso.physical() + self.max_ts_drift_allowance_ms.load(Ordering::SeqCst),
+            ts_from_tso.logical(),
+        );
+
         loop {
             let current = self.max_ts_limit.load();
 
@@ -363,6 +435,11 @@ impl ConcurrencyManager {
     pub fn set_action_on_invalid_max_ts(&self, action: ActionOnInvalidMaxTs) {
         self.action_on_invalid_max_ts.store(action);
     }
+
+    pub fn set_max_ts_drift_allowance(&self, allowance: Duration) {
+        self.max_ts_drift_allowance_ms
+            .store(allowance.as_millis() as u64, Ordering::SeqCst);
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq)]
@@ -512,8 +589,9 @@ impl TimeProvider for CoarseInstantTimeProvider {
 
 #[cfg(test)]
 mod tests {
-    use std::sync::Mutex;
+    use std::{future::ready, sync::Mutex};
 
+    use futures::FutureExt;
     use txn_types::LockType;
 
     use super::*;
@@ -629,6 +707,8 @@ mod tests {
             TimeStamp::new(100),
             DEFAULT_LIMIT_VALID_DURATION,
             ActionOnInvalidMaxTs::Error,
+            None,
+            Duration::ZERO,
         );
 
         // Initially limit should be 0
@@ -671,6 +751,8 @@ mod tests {
             TimeStamp::new(100),
             DEFAULT_LIMIT_VALID_DURATION,
             ActionOnInvalidMaxTs::Error,
+            None,
+            Duration::ZERO,
         );
 
         // Set limit to 200
@@ -703,6 +785,8 @@ mod tests {
             Duration::from_secs(60),
             ActionOnInvalidMaxTs::Error,
             time_provider.clone(),
+            None,
+            Duration::ZERO,
         );
 
         cm.set_max_ts_limit(TimeStamp::new(200));
@@ -726,6 +810,8 @@ mod tests {
             Duration::from_secs(60),
             ActionOnInvalidMaxTs::Error,
             time_provider.clone(),
+            None,
+            Duration::ZERO,
         );
 
         cm.set_max_ts_limit(TimeStamp::new(200));
@@ -758,4 +844,40 @@ mod tests {
             .unwrap();
         assert_eq!(cm.max_ts().into_inner(), 500);
     }
+
+    struct MockPD {
+        tso: AtomicU64,
+    }
+
+    impl MockPD {
+        fn new(ts: u64) -> Self {
+            Self {
+                tso: AtomicU64::new(ts),
+            }
+        }
+    }
+
+    impl TSOProvider for MockPD {
+        fn get_tso(&self) -> PdFuture<TimeStamp> {
+            ready(Ok(TimeStamp::new(self.tso.fetch_add(1, Ordering::SeqCst)))).boxed()
+        }
+    }
+
+    #[test]
+    fn test_pd_tso_jump_not_panic() {
+        let mock_pd = Arc::new(MockPD::new(100));
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Panic,
+            Some(mock_pd.clone()),
+            Duration::ZERO,
+        );
+
+        cm.set_max_ts_limit(TimeStamp::new(200));
+        // PD TSO jumps from 100 to 300
+        mock_pd.tso.store(300, Ordering::SeqCst);
+        cm.update_max_ts(TimeStamp::new(300), "test_source".to_string())
+            .unwrap();
+    }
 }
diff --git a/components/server/src/server.rs b/components/server/src/server.rs
index 0f69c09ce73..b8ba53641ab 100644
--- a/components/server/src/server.rs
+++ b/components/server/src/server.rs
@@ -114,7 +114,7 @@ use tikv::{
         config::EngineType,
         config_manager::StorageConfigManger,
         kv::LocalTablets,
-        mvcc::{MvccConsistencyCheckObserver, TimeStamp},
+        mvcc::MvccConsistencyCheckObserver,
         txn::{
             flow_controller::{EngineFlowController, FlowController},
             txn_status_cache::TxnStatusCache,
@@ -411,6 +411,8 @@ where
                 .as_str()
                 .try_into()
                 .unwrap(),
+            Some(pd_client.clone()),
+            config.storage.max_ts_drift_allowance.0,
         );
 
         // use different quota for front-end and back-end requests
@@ -1170,19 +1172,16 @@ where
         let pd_client = self.pd_client.clone();
 
         let max_ts_sync_interval = self.core.config.storage.max_ts_sync_interval.into();
-        let cfg_controller = self.cfg_controller.as_ref().unwrap().clone();
         self.core
             .background_worker
             .spawn_interval_async_task(max_ts_sync_interval, move || {
                 let cm = cm.clone();
                 let pd_client = pd_client.clone();
-                let allowance_ms =
-                    cfg_controller.get_current().storage.max_ts_drift_allowance.as_millis();
 
                 async move {
                     let pd_tso = pd_client.get_tso().await;
                     if let Ok(ts) = pd_tso {
-                        cm.set_max_ts_limit(TimeStamp::compose(ts.physical() + allowance_ms, 0));
+                        cm.set_max_ts_limit(ts);
                     } else {
                         warn!("failed to get tso from pd in background, the max_ts validity check could be skipped");
                     }
diff --git a/components/server/src/server2.rs b/components/server/src/server2.rs
index d2732dd42eb..5381a365de8 100644
--- a/components/server/src/server2.rs
+++ b/components/server/src/server2.rs
@@ -101,7 +101,7 @@ use tikv::{
         config::EngineType,
         config_manager::StorageConfigManger,
         kv::LocalTablets,
-        mvcc::{MvccConsistencyCheckObserver, TimeStamp},
+        mvcc::MvccConsistencyCheckObserver,
         txn::{
             flow_controller::{FlowController, TabletFlowController},
             txn_status_cache::TxnStatusCache,
@@ -335,6 +335,8 @@ where
                 .as_str()
                 .try_into()
                 .unwrap(),
+            Some(pd_client.clone()),
+            config.storage.max_ts_drift_allowance.0,
         );
 
         // use different quota for front-end and back-end requests
@@ -963,22 +965,16 @@ where
         let pd_client = self.pd_client.clone();
 
         let max_ts_sync_interval = self.core.config.storage.max_ts_sync_interval.into();
-        let cfg_controller = self.cfg_controller.as_ref().unwrap().clone();
         self.core
             .background_worker
             .spawn_interval_async_task(max_ts_sync_interval, move || {
                 let cm = cm.clone();
                 let pd_client = pd_client.clone();
-                let allowance_ms = cfg_controller
-                    .get_current()
-                    .storage
-                    .max_ts_drift_allowance
-                    .as_millis();
 
                 async move {
                     let pd_tso = pd_client.get_tso().await;
                     if let Ok(ts) = pd_tso {
-                        cm.set_max_ts_limit(TimeStamp::compose(ts.physical() + allowance_ms, 0));
+                        cm.set_max_ts_limit(ts);
                     } else {
                         warn!("failed to get tso from pd in background");
                     }
diff --git a/src/storage/config_manager.rs b/src/storage/config_manager.rs
index fac10a40161..c1676f96b04 100644
--- a/src/storage/config_manager.rs
+++ b/src/storage/config_manager.rs
@@ -116,6 +116,10 @@ impl<EK: Engine, K: ConfigurableDb, L: LockManager> ConfigManager
             self.concurrency_manager
                 .set_action_on_invalid_max_ts(action);
         }
+        if let Some(v) = change.remove("max_ts_drift_allowance") {
+            let dur_v: ReadableDuration = v.into();
+            self.concurrency_manager.set_max_ts_drift_allowance(dur_v.0);
+        }
         Ok(())
     }
 }


Test patch: