Repository: tikv/tikv (language: rust)

Issue:
Enhancement: Add timestamp safety boundary to prevent unreasonable max_ts updates
Subtask of 
- https://github.com/pingcap/tidb/issues/57786
## Enhancement
Currently, `max_ts` can be updated to any larger value (**exceeding current PD TSO**). This poses a potential risk where bugs or unsafe usage could set `max_ts` to an unreasonably large value, causing critical bugs and data corruption.

Proposed Solution:
Add a configurable safety boundary (`max_ts_limit`) that:
- Enforces a maximum allowed deviation of `max_ts` from PD timestamp
- Regularly syncs with PD to maintain a reasonable limit
- Prevents updates to unreasonable future timestamps
- Timestamps that are known to be from PD TSO can bypass the limit.


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 8a75c70f8cf..3c617ffb14f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1092,12 +1092,21 @@ name = "concurrency_manager"
 version = "0.0.1"
 dependencies = [
  "criterion",
+ "crossbeam",
  "crossbeam-skiplist",
  "fail",
  "futures 0.3.15",
  "kvproto",
+ "lazy_static",
+ "online_config",
  "parking_lot 0.12.1",
+ "pd_client",
+ "prometheus",
  "rand 0.8.5",
+ "serde",
+ "slog",
+ "slog-global",
+ "thiserror",
  "tikv_alloc",
  "tikv_util",
  "tokio",
@@ -4045,9 +4054,9 @@ checksum = "ac74c624d6b2d21f425f752262f42188365d7b8ff1aff74c82e45136510a4857"
 
 [[package]]
 name = "prettyplease"
-version = "0.1.21"
+version = "0.1.25"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c142c0e46b57171fe0c528bee8c5b7569e80f0c17e377cd0e30ea57dbc11bb51"
+checksum = "6c8646e95016a7a6c4adea95bafa8a16baab64b583356217f2c85db4a39d9a86"
 dependencies = [
  "proc-macro2",
  "syn 1.0.103",
@@ -4183,7 +4192,7 @@ dependencies = [
  "log",
  "multimap",
  "petgraph",
- "prettyplease 0.1.21",
+ "prettyplease 0.1.25",
  "prost",
  "prost-types",
  "regex",
@@ -7104,7 +7113,7 @@ version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "48c6fd7c2581e36d63388a9e04c350c21beb7a8b059580b2e93993c526899ddc"
 dependencies = [
- "prettyplease 0.1.21",
+ "prettyplease 0.1.25",
  "proc-macro2",
  "prost-build",
  "quote",
diff --git a/components/backup-stream/src/endpoint.rs b/components/backup-stream/src/endpoint.rs
index 45ad4305abb..2c1a9e0c5da 100644
--- a/components/backup-stream/src/endpoint.rs
+++ b/components/backup-stream/src/endpoint.rs
@@ -764,7 +764,7 @@ where
                 .await
                 .map_err(|err| Error::from(err).report("failed to get tso from pd"))
                 .unwrap_or_default();
-            cm.update_max_ts(pd_tso);
+            cm.update_max_ts(pd_tso, "backup-stream").unwrap();
             let min_ts = cm.global_min_lock_ts().unwrap_or(TimeStamp::max());
             Ord::min(pd_tso, min_ts)
         }
diff --git a/components/backup/src/endpoint.rs b/components/backup/src/endpoint.rs
index 4ceb538bb56..5860cbde111 100644
--- a/components/backup/src/endpoint.rs
+++ b/components/backup/src/endpoint.rs
@@ -346,7 +346,9 @@ impl BackupRange {
             snap_ctx.key_ranges = vec![key_range];
         } else {
             // Update max_ts and check the in-memory lock table before getting the snapshot
-            concurrency_manager.update_max_ts(backup_ts);
+            concurrency_manager
+                .update_max_ts(backup_ts, "backup_range")
+                .map_err(TxnError::from)?;
             concurrency_manager
                 .read_range_check(
                     self.start_key.as_ref(),
diff --git a/components/cdc/src/endpoint.rs b/components/cdc/src/endpoint.rs
index 1b9f5bb1a52..ef2b79e64e3 100644
--- a/components/cdc/src/endpoint.rs
+++ b/components/cdc/src/endpoint.rs
@@ -1171,7 +1171,7 @@ impl<T: 'static + CdcHandle<E>, E: KvEngine, S: StoreRegionMeta> Endpoint<T, E,
             // optimizations like async commit is enabled.
             // Note: This step must be done before scheduling `Task::MinTs` task, and the
             // resolver must be checked in or after `Task::MinTs`' execution.
-            cm.update_max_ts(min_ts);
+            cm.update_max_ts(min_ts, "cdc").unwrap();
             if let Some(min_mem_lock_ts) = cm.global_min_lock_ts() {
                 if min_mem_lock_ts < min_ts {
                     min_ts = min_mem_lock_ts;
diff --git a/components/concurrency_manager/Cargo.toml b/components/concurrency_manager/Cargo.toml
index 846f140dc46..9cf0c644479 100644
--- a/components/concurrency_manager/Cargo.toml
+++ b/components/concurrency_manager/Cargo.toml
@@ -5,10 +5,19 @@ publish = false
 version = "0.0.1"
 
 [dependencies]
+crossbeam = "0.8"
 crossbeam-skiplist = "0.1"
 fail = "0.5"
 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 }
+slog-global = { workspace = true }
+thiserror = "1.0"
 tikv_util = { workspace = true }
 tokio = { version = "1.5", features = ["macros", "sync", "time"] }
 txn_types = { workspace = true }
@@ -23,3 +32,8 @@ tikv_alloc = { workspace = true, features = ["jemalloc"] }
 name = "lock_table"
 path = "benches/lock_table.rs"
 harness = false
+
+[[bench]]
+name = "update_max_ts"
+path = "benches/update_max_ts.rs"
+harness = false
diff --git a/components/concurrency_manager/benches/update_max_ts.rs b/components/concurrency_manager/benches/update_max_ts.rs
new file mode 100644
index 00000000000..df2e61aeb42
--- /dev/null
+++ b/components/concurrency_manager/benches/update_max_ts.rs
@@ -0,0 +1,33 @@
+// Copyright 2024 TiKV Project Authors. Licensed under Apache-2.0.
+
+use std::time::Duration;
+
+use concurrency_manager::{ActionOnInvalidMaxTs, ConcurrencyManager};
+use criterion::{black_box, criterion_group, criterion_main, Criterion};
+use txn_types::TimeStamp;
+
+fn benchmark_update_max_ts(c: &mut Criterion) {
+    let latest_ts = TimeStamp::new(1000);
+    let limit_valid_time = Duration::from_secs(20);
+    let cm = ConcurrencyManager::new_with_config(
+        latest_ts,
+        limit_valid_time,
+        ActionOnInvalidMaxTs::Error,
+        None,
+        Duration::ZERO,
+    );
+
+    cm.set_max_ts_limit(TimeStamp::new(4000));
+
+    let new_ts = TimeStamp::new(3000);
+
+    c.bench_function("update_max_ts", |b| {
+        b.iter(|| {
+            cm.update_max_ts(black_box(new_ts), || format!("benchmark-{}", new_ts))
+                .unwrap();
+        })
+    });
+}
+
+criterion_group!(benches, benchmark_update_max_ts);
+criterion_main!(benches);
diff --git a/components/concurrency_manager/src/lib.rs b/components/concurrency_manager/src/lib.rs
index 1c6bdb8dbf1..622803a50b8 100644
--- a/components/concurrency_manager/src/lib.rs
+++ b/components/concurrency_manager/src/lib.rs
@@ -16,13 +16,23 @@ mod key_handle;
 mod lock_table;
 
 use std::{
+    error::Error,
+    fmt,
+    fmt::Display,
     mem::MaybeUninit,
     sync::{
-        atomic::{AtomicU64, Ordering},
+        atomic::{AtomicU64, AtomicUsize, Ordering},
         Arc,
     },
+    time::Duration,
 };
 
+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, future::block_on_timeout, time::Instant, warn};
 use txn_types::{Key, Lock, TimeStamp};
 
 pub use self::{
@@ -30,19 +40,143 @@ pub use self::{
     lock_table::LockTable,
 };
 
+lazy_static! {
+    static ref MAX_TS_LIMIT_GAUGE: IntGauge = register_int_gauge!(
+        "tikv_concurrency_manager_max_ts_limit",
+        "Current value of max_ts_limit"
+    )
+    .unwrap();
+    static ref MAX_TS_GAUGE: IntGauge =
+        register_int_gauge!("tikv_concurrency_manager_max_ts", "Current value of max_ts").unwrap();
+}
+
+const DEFAULT_LIMIT_VALID_DURATION: Duration = Duration::from_secs(45);
+
+// It is suggested that limit_valid_duration = sync_interval *
+// LIMIT_VALID_TIME_MULTIPLIER, to balance between
+// 1. tolerate temporary issues in updating the limit.
+// 2. avoid long-term blocking of max_ts update caused by network partition
+//    between TiKV and PD.
+pub const LIMIT_VALID_TIME_MULTIPLIER: u32 = 3;
+
+pub const DEFAULT_MAX_TS_SYNC_INTERVAL: Duration = Duration::from_secs(15);
+pub const DEFAULT_MAX_TS_DRIFT_ALLOWANCE: Duration = Duration::from_secs(60);
+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)]
 pub struct ConcurrencyManager {
     max_ts: Arc<AtomicU64>,
     lock_table: LockTable,
+
+    // max_ts_limit and its update time.
+    //
+    // max_ts_limit is an assertion: max_ts should not be updated to a value greater than this
+    // limit.
+    //
+    // When the limit is not updated for a long time(exceeding the threshold), we use an
+    // approximate limit.
+    max_ts_limit: Arc<AtomicCell<MaxTsLimit>>,
+    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>,
 }
 
 impl ConcurrencyManager {
     pub fn new(latest_ts: TimeStamp) -> Self {
+        Self::new_with_config(
+            latest_ts,
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Panic,
+            None,
+            Duration::ZERO,
+        )
+    }
+
+    pub fn new_with_config(
+        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),
+            update_time: Instant::now(),
+        };
+
+        if limit_valid_duration >= max_ts_drift_allowance {
+            error!("improper setting: limit_valid_duration >= max_ts_drift_allowance; \
+                consider increasing max-ts-drift-allowance or decreasing max-ts-sync-interval";
+                "limit_valid_duration" => ?limit_valid_duration,
+                "max_ts_drift_allowance" => ?max_ts_drift_allowance,
+            );
+        }
+
         ConcurrencyManager {
             max_ts: Arc::new(AtomicU64::new(latest_ts.into_inner())),
+            max_ts_limit: Arc::new(AtomicCell::new(initial_limit)),
             lock_table: LockTable::default(),
+            action_on_invalid_max_ts: Arc::new(AtomicActionOnInvalidMaxTs::new(
+                action_on_invalid_max_ts,
+            )),
+            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
+            )),
+        }
+    }
+
+    #[cfg(test)]
+    fn new_with_time_provider(
+        latest_ts: TimeStamp,
+        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),
+            update_time: time_provider.now(),
+        };
+        ConcurrencyManager {
+            max_ts: Arc::new(AtomicU64::new(latest_ts.into_inner())),
+            max_ts_limit: Arc::new(AtomicCell::new(initial_limit)),
+            lock_table: LockTable::default(),
+            action_on_invalid_max_ts: Arc::new(AtomicActionOnInvalidMaxTs::new(
+                action_on_invalid_max_ts,
+            )),
+            limit_valid_duration,
+            time_provider,
+            tso,
+            max_ts_drift_allowance_ms: Arc::new(AtomicU64::new(
+                max_ts_drift_allowance.as_millis() as u64
+            )),
         }
     }
 
@@ -52,9 +186,183 @@ impl ConcurrencyManager {
 
     /// Updates max_ts with the given new_ts. It has no effect if
     /// max_ts >= new_ts or new_ts is TimeStamp::max().
-    pub fn update_max_ts(&self, new_ts: TimeStamp) {
-        if new_ts != TimeStamp::max() {
-            self.max_ts.fetch_max(new_ts.into_inner(), Ordering::SeqCst);
+    ///
+    /// To avoid invalid ts breaking the invariants, the new_ts should be
+    /// less than or equal to the max_ts_limit.
+    ///
+    /// # Returns
+    /// - Ok(()): If the update is successful or has no effect
+    /// - Err(limit): If new_ts is greater than the max_ts_limit, returns the
+    ///   current limit value
+
+    pub fn update_max_ts(
+        &self,
+        new_ts: TimeStamp,
+        source: impl IntoErrorSource,
+    ) -> Result<(), InvalidMaxTsUpdate> {
+        if new_ts.is_max() {
+            return Ok(());
+        }
+        let limit = self.max_ts_limit.load();
+
+        // check that new_ts is less than or equal to the limit
+        if !limit.limit.is_zero() && new_ts > limit.limit {
+            let last_update = limit.update_time;
+            let now = self.time_provider.now();
+            assert!(now >= last_update);
+            let duration_to_last_limit_update = now - last_update;
+
+            if duration_to_last_limit_update < self.limit_valid_duration {
+                // limit is valid
+                let source = source.into_error_source();
+                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
+
+                let approximate_limit = TimeStamp::compose(
+                    limit.limit.physical() + duration_to_last_limit_update.as_millis() as u64,
+                    limit.limit.logical(),
+                );
+
+                if new_ts > approximate_limit {
+                    let source = source.into_error_source();
+                    self.double_check(new_ts, approximate_limit, source, true)?;
+                }
+            }
+        }
+
+        MAX_TS_GAUGE.set(
+            self.max_ts
+                .fetch_max(new_ts.into_inner(), Ordering::SeqCst)
+                .max(new_ts.into_inner()) as i64,
+        );
+        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> {
+        warn!("possible invalid max_ts update; double checking";
+            "attempted_ts" => new_ts,
+            "limit" => limit.into_inner(),
+            "source" => &source,
+            "using_approximate" => using_approximate,
+            "TSO_TIMEOUT" => ?TSO_TIMEOUT,
+        );
+        let mut tso_confirmed = false;
+        if let Some(tso) = &self.tso {
+            match block_on_timeout(tso.get_tso(), TSO_TIMEOUT) {
+                Ok(Ok(ts)) => {
+                    self.set_max_ts_limit(ts);
+                    tso_confirmed = true;
+                }
+                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);
+                }
+            }
+        }
+        let new_limit = self.max_ts_limit.load();
+        if new_ts > new_limit.limit {
+            self.report_error(new_ts, new_limit, source, using_approximate, tso_confirmed)?;
+        }
+        Ok(())
+    }
+
+    fn report_error(
+        &self,
+        new_ts: TimeStamp,
+        limit: MaxTsLimit,
+        source: impl slog::Value + Display,
+        using_approximate: bool,
+        tso_confirmed: bool,
+    ) -> Result<(), InvalidMaxTsUpdate> {
+        if tso_confirmed {
+            error!("invalid max_ts update";
+                "attempted_ts" => new_ts,
+                "limit" => limit.limit.into_inner(),
+                "limit_update_time" => ?limit.update_time,
+                "source" => &source,
+                "using_approximate" => using_approximate,
+            );
+        } else {
+            warn!("possible invalid max_ts update";
+                "attempted_ts" => new_ts,
+                "limit" => limit.limit.into_inner(),
+                "limit_update_time" => ?limit.update_time,
+                "source" => &source,
+                "using_approximate" => using_approximate,
+            );
+        }
+
+        match self.action_on_invalid_max_ts.load() {
+            ActionOnInvalidMaxTs::Panic if tso_confirmed => {
+                panic!(
+                    "invalid max_ts update: {} exceeds the limit {}, source={}",
+                    new_ts,
+                    limit.limit.into_inner(),
+                    source
+                );
+            }
+            ActionOnInvalidMaxTs::Error if tso_confirmed => Err(InvalidMaxTsUpdate {
+                attempted_ts: new_ts,
+                limit: limit.limit,
+            }),
+            _ => Ok(()),
+        }
+    }
+
+    /// Set the maximum allowed value for max_ts updates, except for the updates
+    /// from PD TSO. The limit must be updated regularly to prevent the
+    /// blocking of max_ts. It prevents max_ts from being updated to an
+    /// unreasonable value, which is usually caused by bugs or unsafe
+    /// usages.
+    ///
+    /// # 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, 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();
+
+            if limit.into_inner() <= current.limit.into_inner() {
+                break;
+            }
+
+            let new_state = MaxTsLimit {
+                limit,
+                update_time: self.time_provider.now(),
+            };
+
+            match self.max_ts_limit.compare_exchange(current, new_state) {
+                Ok(_) => {
+                    MAX_TS_LIMIT_GAUGE.set(limit.into_inner() as i64);
+                    break;
+                }
+                Err(_) => {
+                    continue;
+                }
+            }
         }
     }
 
@@ -141,14 +449,230 @@ impl ConcurrencyManager {
         });
         min_lock
     }
+
+    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)]
+pub enum ActionOnInvalidMaxTs {
+    Panic,
+    Error,
+    Log,
+}
+
+#[derive(Debug)]
+pub struct ParseActionError(String);
+
+impl fmt::Display for ParseActionError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+impl Error for ParseActionError {}
+
+impl TryFrom<&str> for ActionOnInvalidMaxTs {
+    type Error = ParseActionError;
+
+    fn try_from(value: &str) -> Result<Self, ParseActionError> {
+        match value.to_lowercase().as_str() {
+            "panic" => Ok(Self::Panic),
+            "error" => Ok(Self::Error),
+            "log" => Ok(Self::Log),
+            _ => Err(ParseActionError(format!("invalid action value: {}", value))),
+        }
+    }
+}
+
+impl TryFrom<String> for ActionOnInvalidMaxTs {
+    type Error = ParseActionError;
+
+    fn try_from(value: String) -> Result<Self, ParseActionError> {
+        Self::try_from(value.as_str())
+    }
+}
+
+pub struct AtomicActionOnInvalidMaxTs {
+    inner: AtomicUsize,
+}
+
+impl AtomicActionOnInvalidMaxTs {
+    pub fn new(initial: ActionOnInvalidMaxTs) -> Self {
+        Self {
+            inner: AtomicUsize::new(match initial {
+                ActionOnInvalidMaxTs::Panic => 0,
+                ActionOnInvalidMaxTs::Error => 1,
+                ActionOnInvalidMaxTs::Log => 2,
+            }),
+        }
+    }
+
+    pub fn store(&self, value: ActionOnInvalidMaxTs) {
+        self.inner.store(
+            match value {
+                ActionOnInvalidMaxTs::Panic => 0,
+                ActionOnInvalidMaxTs::Error => 1,
+                ActionOnInvalidMaxTs::Log => 2,
+            },
+            Ordering::SeqCst,
+        );
+    }
+
+    pub fn load(&self) -> ActionOnInvalidMaxTs {
+        match self.inner.load(Ordering::SeqCst) {
+            0 => ActionOnInvalidMaxTs::Panic,
+            1 => ActionOnInvalidMaxTs::Error,
+            2 => ActionOnInvalidMaxTs::Log,
+            _ => unreachable!("Invalid atomic state"),
+        }
+    }
+}
+
+#[derive(Debug, Error, Clone)]
+#[error("invalid max_ts update: {attempted_ts} exceeds the limit {limit}")]
+pub struct InvalidMaxTsUpdate {
+    pub attempted_ts: TimeStamp,
+    pub limit: TimeStamp,
+}
+
+pub trait ValueDisplay: slog::Value + Display {}
+impl ValueDisplay for String {}
+impl ValueDisplay for &str {}
+
+mod sealed {
+    pub trait Sealed {}
+}
+
+pub trait IntoErrorSource: sealed::Sealed {
+    type Output: ValueDisplay;
+    fn into_error_source(self) -> Self::Output;
+}
+
+// &str impl
+impl<'a> sealed::Sealed for &'a str {}
+impl<'a> IntoErrorSource for &'a str {
+    type Output = &'a str;
+    fn into_error_source(self) -> Self::Output {
+        self
+    }
+}
+
+// String impl
+impl sealed::Sealed for String {}
+impl IntoErrorSource for String {
+    type Output = String;
+    fn into_error_source(self) -> Self::Output {
+        self
+    }
+}
+
+// Closure impl
+impl<F, T> sealed::Sealed for F
+where
+    F: FnOnce() -> T,
+    T: ValueDisplay,
+{
+}
+impl<F, T> IntoErrorSource for F
+where
+    F: FnOnce() -> T,
+    T: ValueDisplay,
+{
+    type Output = T;
+    fn into_error_source(self) -> T {
+        self()
+    }
+}
+
+/// Trait to abstract time-related functionality, for a monotonic clock
+trait TimeProvider: Send + Sync {
+    /// Returns the current instant.
+    fn now(&self) -> Instant;
+}
+
+struct CoarseInstantTimeProvider;
+
+impl TimeProvider for CoarseInstantTimeProvider {
+    fn now(&self) -> Instant {
+        Instant::now_coarse()
+    }
 }
 
 #[cfg(test)]
 mod tests {
+    use std::{future::ready, sync::Mutex};
+
+    use futures::FutureExt;
     use txn_types::LockType;
 
     use super::*;
 
+    #[derive(Clone)]
+    struct StubTimeProvider {
+        current_time: Arc<Mutex<Instant>>,
+    }
+
+    impl StubTimeProvider {
+        /// Creates a new MockTimeProvider initialized with the given instant.
+        fn new(start_time: Instant) -> Self {
+            StubTimeProvider {
+                current_time: Arc::new(Mutex::new(start_time)),
+            }
+        }
+
+        /// Advances the current time by the specified duration.
+        fn advance(&self, duration: Duration) {
+            let mut time = self.current_time.lock().unwrap();
+            // Note: Instant doesn't support addition, so we mock behavior.
+            // This simplistic approach assumes no overflow.
+            *time += duration;
+        }
+    }
+
+    impl TimeProvider for StubTimeProvider {
+        fn now(&self) -> Instant {
+            let time = self.current_time.lock().unwrap();
+            *time
+        }
+    }
+
+    #[derive(Clone)]
+    struct StubTsoProvider {
+        current_tso: Arc<Mutex<TimeStamp>>,
+        pending: bool,
+    }
+
+    impl StubTsoProvider {
+        fn new(start_tso: TimeStamp) -> Self {
+            StubTsoProvider {
+                current_tso: Arc::new(Mutex::new(start_tso)),
+                pending: false,
+            }
+        }
+
+        fn set_pending(&mut self) {
+            self.pending = true
+        }
+    }
+
+    impl TsoProvider for StubTsoProvider {
+        fn get_tso(&self) -> PdFuture<TimeStamp> {
+            if self.pending {
+                std::future::pending().boxed()
+            } else {
+                let current_tso = self.current_tso.lock().unwrap();
+                ready(Ok(*current_tso)).boxed()
+            }
+        }
+    }
+
     #[tokio::test]
     async fn test_lock_keys_order() {
         let concurrency_manager = ConcurrencyManager::new(1.into());
@@ -166,13 +690,13 @@ mod tests {
     #[tokio::test]
     async fn test_update_max_ts() {
         let concurrency_manager = ConcurrencyManager::new(10.into());
-        concurrency_manager.update_max_ts(20.into());
+        let _ = concurrency_manager.update_max_ts(20.into(), "");
         assert_eq!(concurrency_manager.max_ts(), 20.into());
 
-        concurrency_manager.update_max_ts(5.into());
+        let _ = concurrency_manager.update_max_ts(5.into(), "");
         assert_eq!(concurrency_manager.max_ts(), 20.into());
 
-        concurrency_manager.update_max_ts(TimeStamp::max());
+        let _ = concurrency_manager.update_max_ts(TimeStamp::max(), "");
         assert_eq!(concurrency_manager.max_ts(), 20.into());
     }
 
@@ -224,4 +748,200 @@ mod tests {
             assert_eq!(concurrency_manager.global_min_lock_ts(), Some(20.into()));
         }
     }
+
+    #[test]
+    fn test_max_ts_limit() {
+        let stub_pd = StubTsoProvider::new(160.into());
+        let stub_pd = Arc::new(stub_pd);
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Error,
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+
+        // Initially limit should be 0
+        cm.update_max_ts(TimeStamp::new(150), "").unwrap();
+
+        // Set initial limit to 200
+        cm.set_max_ts_limit(TimeStamp::new(200));
+
+        // Try to lower limit to 150 - should be ignored
+        cm.set_max_ts_limit(TimeStamp::new(150));
+        cm.update_max_ts(TimeStamp::new(180), "").unwrap(); // Should still work up to 200
+
+        // Increase limit to 300 - should work
+        cm.set_max_ts_limit(TimeStamp::new(300));
+        cm.update_max_ts(TimeStamp::new(250), "").unwrap();
+    }
+
+    #[test]
+    fn test_max_ts_limit_edge_cases() {
+        let cm = ConcurrencyManager::new(TimeStamp::new(100));
+
+        // Test transition from zero limit
+        assert_eq!(cm.max_ts_limit.load().limit, 0.into());
+        cm.set_max_ts_limit(TimeStamp::new(1000));
+        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+
+        // Try to lower from 1000 to 500 - should be ignored
+        cm.set_max_ts_limit(TimeStamp::new(500));
+        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+
+        // Test setting limit to max, should have no effect
+        cm.set_max_ts_limit(TimeStamp::max());
+        assert_eq!(cm.max_ts_limit.load().limit, 1000.into());
+    }
+
+    #[test]
+    fn test_max_ts_updates_with_monotonic_limit() {
+        // Assertion: should fail to update max_ts to 250 and query PD
+        let stub_pd = StubTsoProvider::new(201.into());
+        let stub_pd = Arc::new(stub_pd);
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Error,
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+
+        // Set limit to 200
+        cm.set_max_ts_limit(TimeStamp::new(200));
+
+        // Update max_ts to 150
+        cm.update_max_ts(TimeStamp::new(150), "").unwrap();
+        assert_eq!(cm.max_ts(), TimeStamp::new(150));
+
+        // Try to lower limit to 180 - should be ignored
+        cm.set_max_ts_limit(TimeStamp::new(180));
+
+        // Should still fail for values above 200
+        let result = cm.update_max_ts(TimeStamp::new(250), "");
+        assert!(result.is_err());
+        if let Err(e) = result {
+            assert_eq!(e.attempted_ts, TimeStamp::new(250));
+            assert_eq!(e.limit, TimeStamp::new(201));
+        }
+    }
+
+    #[test]
+    fn test_limit_valid_duration_boundary() {
+        let start_time = Instant::now();
+        let stub_time = StubTimeProvider::new(start_time);
+        let time_provider = Arc::new(stub_time);
+        let stub_pd = StubTsoProvider::new(200.into());
+        let stub_pd = Arc::new(stub_pd);
+
+        let cm = ConcurrencyManager::new_with_time_provider(
+            TimeStamp::new(100),
+            Duration::from_secs(60),
+            ActionOnInvalidMaxTs::Error,
+            time_provider.clone(),
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+
+        cm.set_max_ts_limit(TimeStamp::new(200));
+
+        time_provider.advance(Duration::from_secs(59));
+
+        time_provider.advance(Duration::from_secs(1));
+        cm.update_max_ts(TimeStamp::new(250), "").unwrap();
+        assert_eq!(cm.max_ts().into_inner(), 250);
+    }
+
+    #[test]
+    fn test_max_ts_limit_expired_allows_update() {
+        let start_time = Instant::now();
+        let stub_time = StubTimeProvider::new(start_time);
+        let time_provider = Arc::new(stub_time.clone());
+
+        let cm = ConcurrencyManager::new_with_time_provider(
+            TimeStamp::new(100),
+            Duration::from_secs(60),
+            ActionOnInvalidMaxTs::Error,
+            time_provider,
+            None,
+            Duration::ZERO,
+        );
+
+        cm.set_max_ts_limit(TimeStamp::new(200));
+
+        stub_time.advance(Duration::from_secs(61));
+
+        // Updating to 250 should be allowed, since the limit should be invalidated
+        cm.update_max_ts(TimeStamp::new(250), "test_source".to_string())
+            .unwrap();
+        assert_eq!(cm.max_ts().into_inner(), 250);
+    }
+
+    #[test]
+    #[should_panic(expected = "invalid max_ts update")]
+    fn test_panic_on_invalid_max_ts_enabled() {
+        let stub_pd = StubTsoProvider::new(201.into());
+        let stub_pd = Arc::new(stub_pd);
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Panic,
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+
+        cm.set_max_ts_limit(TimeStamp::new(200));
+
+        // should not panic
+        cm.update_max_ts(TimeStamp::new(250), "test_source".to_string())
+            .unwrap();
+    }
+
+    #[test]
+    fn test_update_max_ts_without_limit() {
+        let cm = ConcurrencyManager::new(TimeStamp::new(100));
+
+        cm.update_max_ts(TimeStamp::new(500), "test_source".to_string())
+            .unwrap();
+        assert_eq!(cm.max_ts().into_inner(), 500);
+    }
+
+    #[test]
+    fn test_pd_tso_jump_not_panic() {
+        // The double check procedure gets latest_ts=300 from TSO
+        let stub_pd = StubTsoProvider::new(300.into());
+        let stub_pd = Arc::new(stub_pd);
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Panic,
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+
+        cm.set_max_ts_limit(TimeStamp::new(200));
+        // PD TSO jumps from 100 to 300
+        cm.update_max_ts(TimeStamp::new(300), "test_source".to_string())
+            .unwrap();
+    }
+
+    #[test]
+    fn test_do_not_panic_under_pd_tso_jump_and_network_partition() {
+        let mut stub_pd = StubTsoProvider::new(300.into());
+        stub_pd.set_pending();
+        // Network partition between PD and TiKV
+        let stub_pd = Arc::new(stub_pd);
+        let cm = ConcurrencyManager::new_with_config(
+            TimeStamp::new(100),
+            DEFAULT_LIMIT_VALID_DURATION,
+            ActionOnInvalidMaxTs::Panic,
+            Some(stub_pd),
+            Duration::ZERO,
+        );
+        cm.set_max_ts_limit(200.into());
+
+        // PD TSO jumps from 100 to 300
+        cm.update_max_ts(TimeStamp::new(300), "test_source".to_string())
+            .unwrap();
+    }
 }
diff --git a/components/error_code/src/coprocessor.rs b/components/error_code/src/coprocessor.rs
index 31f0ed4224a..dfa4749795f 100644
--- a/components/error_code/src/coprocessor.rs
+++ b/components/error_code/src/coprocessor.rs
@@ -16,5 +16,7 @@ define_error_codes!(
     CORRUPTED_DATA => ("CorruptedData", "", ""),
 
     STORAGE_ERROR => ("StorageError", "", ""),
-    INVALID_CHARACTER_STRING => ("InvalidCharacterString", "", "")
+    INVALID_CHARACTER_STRING => ("InvalidCharacterString", "", ""),
+
+    INVALID_MAX_TS_UPDATE => ("InvalidMaxTsUpdate", "", "")
 );
diff --git a/components/error_code/src/storage.rs b/components/error_code/src/storage.rs
index b8eb3072391..cd243766b36 100644
--- a/components/error_code/src/storage.rs
+++ b/components/error_code/src/storage.rs
@@ -25,6 +25,7 @@ define_error_codes!(
     DEADLINE_EXCEEDED => ("DeadlineExceeded", "", ""),
     API_VERSION_NOT_MATCHED => ("ApiVersionNotMatched", "", ""),
     INVALID_KEY_MODE => ("InvalidKeyMode", "", ""),
+    INVALID_MAX_TS_UPDATE => ("InvalidMaxTsUpdate", "", ""),
 
     COMMITTED => ("Committed", "", ""),
     PESSIMISTIC_LOCK_ROLLED_BACK => ("PessimisticLockRolledBack", "", ""),
diff --git a/components/raftstore-v2/src/worker/pd/misc.rs b/components/raftstore-v2/src/worker/pd/misc.rs
index 6ade8d87de5..fd2848b95a4 100644
--- a/components/raftstore-v2/src/worker/pd/misc.rs
+++ b/components/raftstore-v2/src/worker/pd/misc.rs
@@ -12,7 +12,6 @@ use pd_client::PdClient;
 use raftstore::{store::TxnExt, Result};
 use slog::{info, warn};
 use tikv_util::{box_err, timer::GLOBAL_TIMER_HANDLE};
-use txn_types::TimeStamp;
 
 use super::Runner;
 
@@ -51,18 +50,22 @@ where
                 // And it won't break correctness of transaction commands, as
                 // causal_ts_provider.flush() is implemented as
                 // pd_client.get_tso() + renew TSO cached.
-                let res: Result<TimeStamp> = if let Some(causal_ts_provider) = &causal_ts_provider {
+                let res: Result<()> = if let Some(causal_ts_provider) = &causal_ts_provider {
                     causal_ts_provider
                         .async_flush()
                         .await
                         .map_err(|e| box_err!(e))
                 } else {
                     pd_client.get_tso().await.map_err(Into::into)
-                };
+                }
+                .and_then(|ts| {
+                    concurrency_manager
+                        .update_max_ts(ts, "raftstore-v2")
+                        .map_err(|e| crate::Error::Other(box_err!(e)))
+                });
 
                 match res {
-                    Ok(ts) => {
-                        concurrency_manager.update_max_ts(ts);
+                    Ok(()) => {
                         success = txn_ext
                             .max_ts_sync_status
                             .compare_exchange(
diff --git a/components/raftstore/src/store/peer.rs b/components/raftstore/src/store/peer.rs
index 0838927302f..b53ad039ecd 100644
--- a/components/raftstore/src/store/peer.rs
+++ b/components/raftstore/src/store/peer.rs
@@ -5642,7 +5642,7 @@ where
             txn_ext: self.txn_ext.clone(),
         }) {
             error!(
-                "failed to update max ts";
+                "failed to update max_ts";
                 "err" => ?e,
             );
         }
diff --git a/components/raftstore/src/store/worker/pd.rs b/components/raftstore/src/store/worker/pd.rs
index 964687bb35f..47c2548b616 100644
--- a/components/raftstore/src/store/worker/pd.rs
+++ b/components/raftstore/src/store/worker/pd.rs
@@ -51,7 +51,6 @@ use tikv_util::{
     worker::{Runnable, ScheduleError, Scheduler},
     InspectFactor,
 };
-use txn_types::TimeStamp;
 use yatp::Remote;
 
 use super::split_controller::AutoSplitControllerContext;
@@ -1816,7 +1815,7 @@ where
                 // And it won't break correctness of transaction commands, as
                 // causal_ts_provider.flush() is implemented as pd_client.get_tso() + renew TSO
                 // cached.
-                let res: crate::Result<TimeStamp> =
+                let res: crate::Result<()> =
                     if let Some(causal_ts_provider) = &causal_ts_provider {
                         causal_ts_provider
                             .async_flush()
@@ -1824,11 +1823,15 @@ where
                             .map_err(|e| box_err!(e))
                     } else {
                         pd_client.get_tso().await.map_err(Into::into)
-                    };
+                    }
+                    .and_then(|ts| {
+                        concurrency_manager
+                            .update_max_ts(ts, "raftstore")
+                            .map_err(|e| crate::Error::Other(box_err!(e)))
+                    });
 
                 match res {
-                    Ok(ts) => {
-                        concurrency_manager.update_max_ts(ts);
+                    Ok(()) => {
                         // Set the least significant bit to 1 to mark it as synced.
                         success = txn_ext
                             .max_ts_sync_status
diff --git a/components/resolved_ts/src/advance.rs b/components/resolved_ts/src/advance.rs
index 2667a43a304..9b7def6d8d8 100644
--- a/components/resolved_ts/src/advance.rs
+++ b/components/resolved_ts/src/advance.rs
@@ -117,7 +117,10 @@ impl AdvanceTsWorker {
             // optimizations like async commit is enabled.
             // Note: This step must be done before scheduling `Task::MinTs` task, and the
             // resolver must be checked in or after `Task::MinTs`' execution.
-            cm.update_max_ts(min_ts);
+            if let Err(e) = cm.update_max_ts(min_ts, "resolved-ts") {
+                error!("failed to advance resolved_ts: failed to update max_ts in concurrency manager"; "err" => ?e);
+                return;
+            }
             if let Some((min_mem_lock_ts, lock)) = cm.global_min_lock() {
                 if min_mem_lock_ts < min_ts {
                     min_ts = min_mem_lock_ts;
diff --git a/components/server/src/server.rs b/components/server/src/server.rs
index 54f159ed383..197ed91b83f 100644
--- a/components/server/src/server.rs
+++ b/components/server/src/server.rs
@@ -28,7 +28,10 @@ use backup_stream::{
 };
 use causal_ts::CausalTsProviderImpl;
 use cdc::CdcConfigManager;
-use concurrency_manager::ConcurrencyManager;
+use concurrency_manager::{
+    ActionOnInvalidMaxTs, ConcurrencyManager, DEFAULT_MAX_TS_DRIFT_ALLOWANCE,
+    DEFAULT_MAX_TS_SYNC_INTERVAL, LIMIT_VALID_TIME_MULTIPLIER,
+};
 use engine_rocks::{from_rocks_compression_type, RocksEngine, RocksStatistics};
 use engine_rocks_helper::sst_recovery::{RecoveryRunner, DEFAULT_CHECK_INTERVAL};
 use engine_traits::{
@@ -158,6 +161,7 @@ fn run_impl<CER: ConfiguredRaftEngine, F: KvFormat>(
     tikv.register_services();
     tikv.init_metrics_flusher(fetcher, engines_info);
     tikv.init_storage_stats_task(engines);
+    tikv.init_max_ts_updater();
     tikv.run_server(server_config);
     tikv.run_status_server();
     tikv.core.init_quota_tuning_task(tikv.quota_limiter.clone());
@@ -380,7 +384,13 @@ where
 
         // Initialize concurrency manager
         let latest_ts = block_on(pd_client.get_tso()).expect("failed to get timestamp from PD");
-        let concurrency_manager = ConcurrencyManager::new(latest_ts);
+        let concurrency_manager = ConcurrencyManager::new_with_config(
+            latest_ts,
+            DEFAULT_MAX_TS_SYNC_INTERVAL * LIMIT_VALID_TIME_MULTIPLIER,
+            ActionOnInvalidMaxTs::Log,
+            Some(pd_client.clone()),
+            DEFAULT_MAX_TS_DRIFT_ALLOWANCE,
+        );
 
         // use different quota for front-end and back-end requests
         let quota_limiter = Arc::new(QuotaLimiter::new(
@@ -1102,6 +1112,28 @@ where
         server_config
     }
 
+    fn init_max_ts_updater(&self) {
+        let cm = self.concurrency_manager.clone();
+        let pd_client = self.pd_client.clone();
+
+        let max_ts_sync_interval = DEFAULT_MAX_TS_SYNC_INTERVAL;
+        self.core
+            .background_worker
+            .spawn_interval_async_task(max_ts_sync_interval, move || {
+                let cm = cm.clone();
+                let pd_client = pd_client.clone();
+
+                async move {
+                    let pd_tso = pd_client.get_tso().await;
+                    if let Ok(ts) = pd_tso {
+                        cm.set_max_ts_limit(ts);
+                    } else {
+                        warn!("failed to get tso from pd in background, the max_ts validity check could be skipped");
+                    }
+                }
+            });
+    }
+
     fn register_services(&mut self) {
         let servers = self.servers.as_mut().unwrap();
         let engines = self.engines.as_ref().unwrap();
diff --git a/components/server/src/server2.rs b/components/server/src/server2.rs
index c6becff29c9..fc8c125d3fa 100644
--- a/components/server/src/server2.rs
+++ b/components/server/src/server2.rs
@@ -33,7 +33,10 @@ use backup_stream::{
 };
 use causal_ts::CausalTsProviderImpl;
 use cdc::CdcConfigManager;
-use concurrency_manager::ConcurrencyManager;
+use concurrency_manager::{
+    ActionOnInvalidMaxTs, ConcurrencyManager, DEFAULT_MAX_TS_DRIFT_ALLOWANCE,
+    DEFAULT_MAX_TS_SYNC_INTERVAL, LIMIT_VALID_TIME_MULTIPLIER,
+};
 use engine_rocks::{from_rocks_compression_type, RocksEngine, RocksStatistics};
 use engine_traits::{Engines, KvEngine, MiscExt, RaftEngine, TabletRegistry, CF_DEFAULT, CF_WRITE};
 use file_system::{get_io_rate_limiter, BytesFetcher, MetricsManager as IoMetricsManager};
@@ -154,6 +157,7 @@ fn run_impl<CER: ConfiguredRaftEngine, F: KvFormat>(
     tikv.register_services();
     tikv.init_metrics_flusher(fetcher, engines_info);
     tikv.init_storage_stats_task();
+    tikv.init_max_ts_updater();
     tikv.run_server(server_config);
     tikv.run_status_server();
     tikv.core.init_quota_tuning_task(tikv.quota_limiter.clone());
@@ -324,7 +328,13 @@ where
 
         // Initialize concurrency manager
         let latest_ts = block_on(pd_client.get_tso()).expect("failed to get timestamp from PD");
-        let concurrency_manager = ConcurrencyManager::new(latest_ts);
+        let concurrency_manager = ConcurrencyManager::new_with_config(
+            latest_ts,
+            DEFAULT_MAX_TS_SYNC_INTERVAL * LIMIT_VALID_TIME_MULTIPLIER,
+            ActionOnInvalidMaxTs::Log,
+            Some(pd_client.clone()),
+            DEFAULT_MAX_TS_DRIFT_ALLOWANCE,
+        );
 
         // use different quota for front-end and back-end requests
         let quota_limiter = Arc::new(QuotaLimiter::new(
@@ -929,6 +939,28 @@ where
         server_config
     }
 
+    fn init_max_ts_updater(&self) {
+        let cm = self.concurrency_manager.clone();
+        let pd_client = self.pd_client.clone();
+
+        let max_ts_sync_interval = DEFAULT_MAX_TS_SYNC_INTERVAL;
+        self.core
+            .background_worker
+            .spawn_interval_async_task(max_ts_sync_interval, move || {
+                let cm = cm.clone();
+                let pd_client = pd_client.clone();
+
+                async move {
+                    let pd_tso = pd_client.get_tso().await;
+                    if let Ok(ts) = pd_tso {
+                        cm.set_max_ts_limit(ts);
+                    } else {
+                        warn!("failed to get tso from pd in background");
+                    }
+                }
+            });
+    }
+
     fn register_services(&mut self) {
         let servers = self.servers.as_mut().unwrap();
         let engines = self.engines.as_ref().unwrap();
diff --git a/components/tikv_util/src/time.rs b/components/tikv_util/src/time.rs
index f329247c563..db56dfffcca 100644
--- a/components/tikv_util/src/time.rs
+++ b/components/tikv_util/src/time.rs
@@ -276,7 +276,7 @@ mod inner {
 /// A measurement of a monotonically increasing clock.
 /// It's similar and meant to replace `std::time::Instant`,
 /// for providing extra features.
-#[derive(Copy, Clone, Debug)]
+#[derive(Copy, Clone, Debug, Eq)]
 pub enum Instant {
     Monotonic(Timespec),
     MonotonicCoarse(Timespec),
diff --git a/src/coprocessor/endpoint.rs b/src/coprocessor/endpoint.rs
index f9a1e28dd19..48b3d236eb5 100644
--- a/src/coprocessor/endpoint.rs
+++ b/src/coprocessor/endpoint.rs
@@ -134,7 +134,8 @@ impl<E: Engine> Endpoint<E> {
     fn check_memory_locks(&self, req_ctx: &ReqContext) -> Result<()> {
         let start_ts = req_ctx.txn_start_ts;
         if !req_ctx.context.get_stale_read() {
-            self.concurrency_manager.update_max_ts(start_ts);
+            self.concurrency_manager
+                .update_max_ts(start_ts, || format!("coprocessor-{}", start_ts))?;
         }
         if need_check_locks(req_ctx.context.get_isolation_level()) {
             let begin_instant = Instant::now();
@@ -912,6 +913,12 @@ macro_rules! make_error_response_common {
                 errorpb.set_server_is_busy(server_is_busy_err);
                 $resp.set_region_error(errorpb);
             }
+            Error::InvalidMaxTsUpdate(e) => {
+                $tag = "invalid_max_ts_update";
+                let mut err = errorpb::Error::default();
+                err.set_message(e.to_string());
+                $resp.set_region_error(err);
+            }
             Error::Other(_) => {
                 $tag = "other";
                 warn!("unexpected other error encountered processing coprocessor task";
diff --git a/src/coprocessor/error.rs b/src/coprocessor/error.rs
index 5c3ce554cc2..1ecb4479da3 100644
--- a/src/coprocessor/error.rs
+++ b/src/coprocessor/error.rs
@@ -30,6 +30,9 @@ pub enum Error {
     #[error("Coprocessor task canceled due to exceeding memory quota")]
     MemoryQuotaExceeded,
 
+    #[error("{0}")]
+    InvalidMaxTsUpdate(#[from] concurrency_manager::InvalidMaxTsUpdate),
+
     #[error("{0}")]
     Other(String),
 }
@@ -137,6 +140,7 @@ impl ErrorCodeExt for Error {
             Error::DeadlineExceeded => error_code::coprocessor::DEADLINE_EXCEEDED,
             Error::MaxPendingTasksExceeded => error_code::coprocessor::MAX_PENDING_TASKS_EXCEEDED,
             Error::MemoryQuotaExceeded => error_code::coprocessor::MEMORY_QUOTA_EXCEEDED,
+            Error::InvalidMaxTsUpdate(_) => error_code::coprocessor::INVALID_MAX_TS_UPDATE,
             Error::Other(_) => error_code::UNKNOWN,
         }
     }
diff --git a/src/server/raftkv/mod.rs b/src/server/raftkv/mod.rs
index 9e6cc711558..0fafa22d7fb 100644
--- a/src/server/raftkv/mod.rs
+++ b/src/server/raftkv/mod.rs
@@ -797,7 +797,12 @@ impl ReadIndexObserver for ReplicaReadLockChecker {
             let begin_instant = Instant::now();
 
             let start_ts = request.get_start_ts().into();
-            self.concurrency_manager.update_max_ts(start_ts);
+            if let Err(e) = self
+                .concurrency_manager
+                .update_max_ts(start_ts, || format!("read_index-{}", start_ts))
+            {
+                error!("failed to update max_ts in concurrency manager"; "err" => ?e);
+            }
             for range in request.mut_key_ranges().iter_mut() {
                 let key_bound = |key: Vec<u8>| {
                     if key.is_empty() {
diff --git a/src/storage/errors.rs b/src/storage/errors.rs
index 0cd14d7ff8a..9dcf565e28d 100644
--- a/src/storage/errors.rs
+++ b/src/storage/errors.rs
@@ -303,6 +303,20 @@ pub fn extract_region_error_from_error(e: &Error) -> Option<errorpb::Error> {
             err.set_flashback_not_prepared(flashback_not_prepared_err);
             Some(err)
         }
+        Error(box ErrorInner::Txn(TxnError(box TxnErrorInner::InvalidMaxTsUpdate(
+            invalid_max_ts_update,
+        )))) => {
+            let mut err = errorpb::Error::default();
+            err.set_message(invalid_max_ts_update.to_string());
+            Some(err)
+        }
+        Error(box ErrorInner::Txn(TxnError(box TxnErrorInner::Mvcc(MvccError(
+            box MvccErrorInner::InvalidMaxTsUpdate(invalid_max_ts_update),
+        ))))) => {
+            let mut err = errorpb::Error::default();
+            err.set_message(invalid_max_ts_update.to_string());
+            Some(err)
+        }
         Error(box ErrorInner::SchedTooBusy) => {
             let mut err = errorpb::Error::default();
             let mut server_is_busy_err = errorpb::ServerIsBusy::default();
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index 60202d5ff84..1df9b11db9f 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -1237,7 +1237,9 @@ impl<E: Engine, L: LockManager, F: KvFormat> Storage<E, L, F> {
 
                 // Update max_ts and check the in-memory lock table before getting the snapshot
                 if !ctx.get_stale_read() {
-                    concurrency_manager.update_max_ts(start_ts);
+                    concurrency_manager
+                        .update_max_ts(start_ts, "scan")
+                        .map_err(txn::Error::from)?;
                 }
                 if need_check_locks(ctx.get_isolation_level()) {
                     let begin_instant = Instant::now();
@@ -1402,7 +1404,9 @@ impl<E: Engine, L: LockManager, F: KvFormat> Storage<E, L, F> {
 
                 let command_duration = Instant::now();
 
-                concurrency_manager.update_max_ts(max_ts);
+                concurrency_manager
+                    .update_max_ts(max_ts, "scan_lock")
+                    .map_err(txn::Error::from)?;
                 let begin_instant = Instant::now();
                 // TODO: Though it's very unlikely to find a conflicting memory lock here, it's
                 // not a good idea to return an error to the client, making the GC fail. A
@@ -3075,7 +3079,11 @@ fn prepare_snap_ctx<'a>(
 ) -> Result<SnapContext<'a>> {
     // Update max_ts and check the in-memory lock table before getting the snapshot
     if !pb_ctx.get_stale_read() {
-        concurrency_manager.update_max_ts(start_ts);
+        concurrency_manager
+            .update_max_ts(start_ts, || {
+                format!("prepare_snap_ctx-{}-{}", cmd, start_ts)
+            })
+            .map_err(txn::Error::from)?;
     }
     fail_point!("before-storage-check-memory-locks");
     let isolation_level = pb_ctx.get_isolation_level();
@@ -9747,7 +9755,7 @@ mod tests {
             .build()
             .unwrap();
         let cm = storage.concurrency_manager.clone();
-        cm.update_max_ts(10.into());
+        cm.update_max_ts(10.into(), "").unwrap();
 
         // Optimistic prewrite
         let (tx, rx) = channel();
@@ -9795,7 +9803,7 @@ mod tests {
             .unwrap();
         rx.recv().unwrap();
 
-        cm.update_max_ts(1000.into());
+        cm.update_max_ts(1000.into(), "").unwrap();
 
         let (tx, rx) = channel();
         storage
@@ -10901,7 +10909,7 @@ mod tests {
         // commit enabled, and max_ts changes when the second request arrives.
 
         // A retrying prewrite request arrives.
-        cm.update_max_ts(20.into());
+        cm.update_max_ts(20.into(), "").unwrap();
         let mut ctx = Context::default();
         ctx.set_is_retry_request(true);
         let (tx, rx) = channel();
@@ -11085,7 +11093,7 @@ mod tests {
 
         // 1PC update
         let (tx, rx) = channel();
-        cm.update_max_ts(59.into());
+        cm.update_max_ts(59.into(), "").unwrap();
         storage
             .sched_txn_command(
                 Prewrite::new(
diff --git a/src/storage/mvcc/mod.rs b/src/storage/mvcc/mod.rs
index 6e4848c8579..f71730f2148 100644
--- a/src/storage/mvcc/mod.rs
+++ b/src/storage/mvcc/mod.rs
@@ -174,6 +174,9 @@ pub enum ErrorInner {
     #[error("check_txn_status sent to secondary lock, current lock: {0:?}")]
     PrimaryMismatch(kvproto::kvrpcpb::LockInfo),
 
+    #[error("{0}")]
+    InvalidMaxTsUpdate(#[from] concurrency_manager::InvalidMaxTsUpdate),
+
     #[error("{0:?}")]
     Other(#[from] Box<dyn error::Error + Sync + Send>),
 }
@@ -304,6 +307,7 @@ impl ErrorInner {
                 })
             }
             ErrorInner::PrimaryMismatch(l) => Some(ErrorInner::PrimaryMismatch(l.clone())),
+            ErrorInner::InvalidMaxTsUpdate(e) => Some(ErrorInner::InvalidMaxTsUpdate(e.clone())),
             ErrorInner::Io(_) | ErrorInner::Other(_) => None,
         }
     }
@@ -407,6 +411,7 @@ impl ErrorCodeExt for Error {
             ErrorInner::AssertionFailed { .. } => error_code::storage::ASSERTION_FAILED,
             ErrorInner::LockIfExistsFailed { .. } => error_code::storage::LOCK_IF_EXISTS_FAILED,
             ErrorInner::PrimaryMismatch(_) => error_code::storage::PRIMARY_MISMATCH,
+            ErrorInner::InvalidMaxTsUpdate(_) => error_code::storage::INVALID_MAX_TS_UPDATE,
             ErrorInner::Other(_) => error_code::storage::UNKNOWN,
         }
     }
diff --git a/src/storage/txn/actions/acquire_pessimistic_lock.rs b/src/storage/txn/actions/acquire_pessimistic_lock.rs
index 2a9e49b45ff..6c71bb6843a 100644
--- a/src/storage/txn/actions/acquire_pessimistic_lock.rs
+++ b/src/storage/txn/actions/acquire_pessimistic_lock.rs
@@ -71,7 +71,9 @@ pub fn acquire_pessimistic_lock<S: Snapshot>(
     // it infers a read to the value, in which case max_ts need to be updated to
     // guarantee the linearizability and snapshot isolation.
     if should_not_exist || need_value || need_check_existence {
-        txn.concurrency_manager.update_max_ts(for_update_ts);
+        txn.concurrency_manager.update_max_ts(for_update_ts, || {
+            format!("pessimistic_lock-{}-{}", reader.start_ts, for_update_ts)
+        })?;
     }
 
     // When `need_value` is set, the value need to be loaded of course. If
diff --git a/src/storage/txn/actions/prewrite.rs b/src/storage/txn/actions/prewrite.rs
index 64e22a13585..22977e621a8 100644
--- a/src/storage/txn/actions/prewrite.rs
+++ b/src/storage/txn/actions/prewrite.rs
@@ -48,7 +48,10 @@ pub fn prewrite<S: Snapshot>(
     // Update max_ts for Insert operation to guarantee linearizability and snapshot
     // isolation
     if mutation.should_not_exist {
-        txn.concurrency_manager.update_max_ts(txn_props.start_ts);
+        txn.concurrency_manager
+            .update_max_ts(txn_props.start_ts, || {
+                format!("prewrite-{}", txn_props.start_ts)
+            })?;
     }
 
     fail_point!(
@@ -111,7 +114,10 @@ pub fn prewrite<S: Snapshot>(
     if mutation.should_not_write {
         // `checkNotExists` is equivalent to a get operation, so it should update the
         // max_ts.
-        txn.concurrency_manager.update_max_ts(txn_props.start_ts);
+        txn.concurrency_manager
+            .update_max_ts(txn_props.start_ts, || {
+                format!("prewrite-{}", txn_props.start_ts)
+            })?;
         let min_commit_ts = if mutation.need_min_commit_ts() {
             // Don't calculate the min_commit_ts according to the concurrency manager's
             // max_ts for a should_not_write mutation because it's not persisted and doesn't
@@ -998,7 +1004,7 @@ pub mod tests {
         .unwrap();
         assert_eq!(old_value, OldValue::None);
 
-        cm.update_max_ts(60.into());
+        cm.update_max_ts(60.into(), "").unwrap();
         // calculated commit_ts = 61 > 50, err
         let err = prewrite(
             &mut txn,
@@ -1183,7 +1189,7 @@ pub mod tests {
         .unwrap();
         assert_eq!(old_value, OldValue::None);
 
-        cm.update_max_ts(60.into());
+        cm.update_max_ts(60.into(), "").unwrap();
         // calculated commit_ts = 61 > 50, err
         let err = prewrite(
             &mut txn,
@@ -1285,7 +1291,7 @@ pub mod tests {
         // Pessimistic txn skips constraint check, does not read previous write.
         assert_eq!(old_value, OldValue::Unspecified);
 
-        cm.update_max_ts(60.into());
+        cm.update_max_ts(60.into(), "").unwrap();
         // calculated commit_ts = 61 > 50, ok
         prewrite(
             &mut txn,
@@ -1338,7 +1344,7 @@ pub mod tests {
         // Pessimistic txn skips constraint check, does not read previous write.
         assert_eq!(old_value, OldValue::Unspecified);
 
-        cm.update_max_ts(60.into());
+        cm.update_max_ts(60.into(), "").unwrap();
         // calculated commit_ts = 61 > 50, ok
         prewrite(
             &mut txn,
diff --git a/src/storage/txn/commands/check_secondary_locks.rs b/src/storage/txn/commands/check_secondary_locks.rs
index ceb169f79b2..4facd84f8a1 100644
--- a/src/storage/txn/commands/check_secondary_locks.rs
+++ b/src/storage/txn/commands/check_secondary_locks.rs
@@ -142,7 +142,11 @@ impl<S: Snapshot, L: LockManager> WriteCommand<S, L> for CheckSecondaryLocks {
         // It is not allowed for commit to overwrite a protected rollback. So we update
         // max_ts to prevent this case from happening.
         let region_id = self.ctx.get_region_id();
-        context.concurrency_manager.update_max_ts(self.start_ts);
+        context
+            .concurrency_manager
+            .update_max_ts(self.start_ts, || {
+                format!("check_secondary_locks-{}", self.start_ts)
+            })?;
 
         let mut txn = MvccTxn::new(self.start_ts, context.concurrency_manager);
         let mut reader = ReaderWithStats::new(
diff --git a/src/storage/txn/commands/check_txn_status.rs b/src/storage/txn/commands/check_txn_status.rs
index ce5774c0db6..6cbc30c94e3 100644
--- a/src/storage/txn/commands/check_txn_status.rs
+++ b/src/storage/txn/commands/check_txn_status.rs
@@ -85,7 +85,12 @@ impl<S: Snapshot, L: LockManager> WriteCommand<S, L> for CheckTxnStatus {
         if !self.caller_start_ts.is_max() && self.caller_start_ts > new_max_ts {
             new_max_ts = self.caller_start_ts;
         }
-        context.concurrency_manager.update_max_ts(new_max_ts);
+        context.concurrency_manager.update_max_ts(new_max_ts, || {
+            format!(
+                "check_txn_status-{}-{}-{}",
+                self.lock_ts, self.current_ts, self.caller_start_ts
+            )
+        })?;
 
         let mut txn = MvccTxn::new(self.lock_ts, context.concurrency_manager);
         let mut reader = ReaderWithStats::new(
diff --git a/src/storage/txn/commands/cleanup.rs b/src/storage/txn/commands/cleanup.rs
index 886094a7f34..e8ec3a3f2af 100644
--- a/src/storage/txn/commands/cleanup.rs
+++ b/src/storage/txn/commands/cleanup.rs
@@ -48,7 +48,9 @@ impl<S: Snapshot, L: LockManager> WriteCommand<S, L> for Cleanup {
     fn process_write(self, snapshot: S, context: WriteContext<'_, L>) -> Result<WriteResult> {
         // It is not allowed for commit to overwrite a protected rollback. So we update
         // max_ts to prevent this case from happening.
-        context.concurrency_manager.update_max_ts(self.start_ts);
+        context
+            .concurrency_manager
+            .update_max_ts(self.start_ts, || format!("cleanup-{}", self.start_ts))?;
 
         let mut txn = MvccTxn::new(self.start_ts, context.concurrency_manager);
         let mut reader = ReaderWithStats::new(
diff --git a/src/storage/txn/commands/prewrite.rs b/src/storage/txn/commands/prewrite.rs
index 34c98dab156..ac224281ece 100644
--- a/src/storage/txn/commands/prewrite.rs
+++ b/src/storage/txn/commands/prewrite.rs
@@ -1276,7 +1276,7 @@ mod tests {
         must_get(&mut engine, key, 12, value);
         must_get_commit_ts(&mut engine, key, 10, 11);
 
-        cm.update_max_ts(50.into());
+        cm.update_max_ts(50.into(), "").unwrap();
 
         let mutations = vec![Mutation::make_put(Key::from_raw(key), value.to_vec())];
 
@@ -1418,7 +1418,7 @@ mod tests {
         must_get_commit_ts(&mut engine, k1, 8, 13);
         must_get_commit_ts(&mut engine, k2, 8, 13);
 
-        cm.update_max_ts(50.into());
+        cm.update_max_ts(50.into(), "").unwrap();
         must_acquire_pessimistic_lock(&mut engine, k1, k1, 20, 20);
 
         let mutations = vec![(
@@ -1518,7 +1518,7 @@ mod tests {
         assert_eq!(res.one_pc_commit_ts, TimeStamp::zero());
         must_locked(&mut engine, key, 10);
 
-        cm.update_max_ts(50.into());
+        cm.update_max_ts(50.into(), "").unwrap();
 
         let (k1, v1) = (b"k1", b"v1");
         let (k2, v2) = (b"k2", b"v2");
@@ -1590,7 +1590,7 @@ mod tests {
         assert_eq!(res.one_pc_commit_ts, TimeStamp::zero());
         must_locked(&mut engine, key, 10);
 
-        cm.update_max_ts(50.into());
+        cm.update_max_ts(50.into(), "").unwrap();
 
         let (k1, v1) = (b"k1", b"v1");
         let (k2, v2) = (b"k2", b"v2");
@@ -2163,7 +2163,7 @@ mod tests {
         )
         .unwrap();
         let commit_ts = res.one_pc_commit_ts;
-        cm.update_max_ts(commit_ts.next());
+        cm.update_max_ts(commit_ts.next(), "").unwrap();
         // repeate the prewrite
         let res = pessimistic_prewrite_with_cm(
             &mut engine,
@@ -2673,7 +2673,7 @@ mod tests {
         must_commit(&mut engine, b"k2", 5, 18);
 
         // Update max_ts to be larger than the max_commit_ts.
-        cm.update_max_ts(50.into());
+        cm.update_max_ts(50.into(), "").unwrap();
 
         // Retry the prewrite on non-pessimistic key.
         // (is_retry_request flag is not set, here we don't rely on it.)
diff --git a/src/storage/txn/mod.rs b/src/storage/txn/mod.rs
index ca11fd6db30..192916f9e16 100644
--- a/src/storage/txn/mod.rs
+++ b/src/storage/txn/mod.rs
@@ -148,6 +148,9 @@ pub enum ErrorInner {
 
     #[error("region {0} not prepared the flashback")]
     FlashbackNotPrepared(u64),
+
+    #[error("{0}")]
+    InvalidMaxTsUpdate(#[from] concurrency_manager::InvalidMaxTsUpdate),
 }
 
 impl ErrorInner {
@@ -187,6 +190,9 @@ impl ErrorInner {
             ErrorInner::FlashbackNotPrepared(region_id) => {
                 Some(ErrorInner::FlashbackNotPrepared(region_id))
             }
+            ErrorInner::InvalidMaxTsUpdate(ref e) => {
+                Some(ErrorInner::InvalidMaxTsUpdate(e.clone()))
+            }
             ErrorInner::Other(_) | ErrorInner::ProtoBuf(_) | ErrorInner::Io(_) => None,
         }
     }
@@ -241,6 +247,7 @@ impl ErrorCodeExt for Error {
                 error_code::storage::MAX_TIMESTAMP_NOT_SYNCED
             }
             ErrorInner::FlashbackNotPrepared(_) => error_code::storage::FLASHBACK_NOT_PREPARED,
+            ErrorInner::InvalidMaxTsUpdate { .. } => error_code::storage::INVALID_MAX_TS_UPDATE,
         }
     }
 }


Test patch:
diff --git a/components/cdc/tests/integrations/test_cdc.rs b/components/cdc/tests/integrations/test_cdc.rs
index c1ac1706d52..aac2bd45814 100644
--- a/components/cdc/tests/integrations/test_cdc.rs
+++ b/components/cdc/tests/integrations/test_cdc.rs
@@ -1235,7 +1235,7 @@ fn test_cdc_resolve_ts_checking_concurrency_manager_impl<F: KvFormat>() {
         guard
     };
 
-    cm.update_max_ts(20.into());
+    cm.update_max_ts(20.into(), "").unwrap();
 
     let guard = lock_key(b"a", 80);
     suite.set_tso(99);
diff --git a/tests/failpoints/cases/test_transaction.rs b/tests/failpoints/cases/test_transaction.rs
index 0b6e6269e95..412c40e5238 100644
--- a/tests/failpoints/cases/test_transaction.rs
+++ b/tests/failpoints/cases/test_transaction.rs
@@ -344,7 +344,7 @@ fn test_max_commit_ts_error() {
     thread::sleep(Duration::from_millis(200));
     cm.read_key_check(&Key::from_raw(b"k1"), |_| Err(()))
         .unwrap_err();
-    cm.update_max_ts(200.into());
+    cm.update_max_ts(200.into(), "").unwrap();
 
     let res = prewrite_rx.recv().unwrap().unwrap();
     assert!(res.min_commit_ts.is_zero());
@@ -372,7 +372,7 @@ fn test_exceed_max_commit_ts_in_the_middle_of_prewrite() {
     // Pause between getting max ts and store the lock in memory
     fail::cfg("before-set-lock-in-memory", "pause").unwrap();
 
-    cm.update_max_ts(40.into());
+    cm.update_max_ts(40.into(), "").unwrap();
     let mutations = vec![
         Mutation::make_put(Key::from_raw(b"k1"), b"v".to_vec()),
         Mutation::make_put(Key::from_raw(b"k2"), b"v".to_vec()),
@@ -401,7 +401,7 @@ fn test_exceed_max_commit_ts_in_the_middle_of_prewrite() {
     // sleep a while so the first key gets max ts.
     thread::sleep(Duration::from_millis(200));
 
-    cm.update_max_ts(51.into());
+    cm.update_max_ts(51.into(), "").unwrap();
     fail::remove("before-set-lock-in-memory");
     let res = prewrite_rx.recv().unwrap().unwrap();
     assert!(res.min_commit_ts.is_zero());
diff --git a/tests/integrations/server/kv_service.rs b/tests/integrations/server/kv_service.rs
index 164fafdf964..a62fc368103 100644
--- a/tests/integrations/server/kv_service.rs
+++ b/tests/integrations/server/kv_service.rs
@@ -1829,7 +1829,7 @@ fn test_prewrite_check_max_commit_ts() {
     let (cluster, client, ctx) = new_cluster();
 
     let cm = cluster.sim.read().unwrap().get_concurrency_manager(1);
-    cm.update_max_ts(100.into());
+    cm.update_max_ts(100.into(), "").unwrap();
 
     let mut req = PrewriteRequest::default();
     req.set_context(ctx.clone());