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 153efaedb01..69c61aa83b6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1022,12 +1022,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.0",
+ "pd_client",
+ "prometheus",
  "rand 0.8.5",
+ "serde",
+ "slog",
+ "slog-global",
+ "thiserror",
  "tikv_alloc",
  "tikv_util",
  "tokio",
@@ -3841,9 +3850,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",
@@ -3991,7 +4000,7 @@ dependencies = [
  "log",
  "multimap",
  "petgraph",
- "prettyplease 0.1.21",
+ "prettyplease 0.1.25",
  "prost",
  "prost-types",
  "regex",
@@ -6735,7 +6744,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 b5ddb6bff85..5619069bebc 100644
--- a/components/backup-stream/src/endpoint.rs
+++ b/components/backup-stream/src/endpoint.rs
@@ -783,7 +783,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 95f840e9f91..bfd37ee2d9e 100644
--- a/components/backup/src/endpoint.rs
+++ b/components/backup/src/endpoint.rs
@@ -307,7 +307,9 @@ impl BackupRange {
         ctx.set_peer(self.leader.clone());
 
         // 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 c8413d968fd..4b4608ad54b 100644
--- a/components/cdc/src/endpoint.rs
+++ b/components/cdc/src/endpoint.rs
@@ -1084,7 +1084,7 @@ impl<T: 'static + RaftStoreRouter<E>, E: KvEngine> 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 b391c1d239a..d41449b460e 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", features = ["derive"] }
+slog = { version = "2.3", features = ["max_level_trace", "release_max_level_debug"] }
+slog-global = { version = "0.1", git = "https://github.com/breeswish/slog-global.git", rev = "d592f88e4dbba5eb439998463054f1a44fbf17b9" }
+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 dcd516716eb..3135284ecdf 100644
--- a/components/concurrency_manager/src/lib.rs
+++ b/components/concurrency_manager/src/lib.rs
@@ -16,13 +16,24 @@ mod key_handle;
 mod lock_table;
 
 use std::{
+    convert::TryFrom,
+    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 +41,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 +187,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 +450,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 +691,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());
     }
 
@@ -214,4 +739,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 d98c85162bf..5cad4756168 100644
--- a/components/error_code/src/coprocessor.rs
+++ b/components/error_code/src/coprocessor.rs
@@ -15,5 +15,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 8b41e7a797e..4474de846dc 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/update_max_timestamp.rs b/components/raftstore-v2/src/worker/pd/update_max_timestamp.rs
index ab03b86cf95..56993b3d7d8 100644
--- a/components/raftstore-v2/src/worker/pd/update_max_timestamp.rs
+++ b/components/raftstore-v2/src/worker/pd/update_max_timestamp.rs
@@ -13,7 +13,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;
 
@@ -50,18 +49,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 cd32343b31b..a411c0f5c8f 100644
--- a/components/raftstore/src/store/peer.rs
+++ b/components/raftstore/src/store/peer.rs
@@ -5654,7 +5654,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 c2186bcdc45..53852115b61 100644
--- a/components/raftstore/src/store/worker/pd.rs
+++ b/components/raftstore/src/store/worker/pd.rs
@@ -48,7 +48,6 @@ use tikv_util::{
     warn,
     worker::{Runnable, RunnableWithTimer, ScheduleError, Scheduler},
 };
-use txn_types::TimeStamp;
 use yatp::Remote;
 
 use crate::{
@@ -1678,7 +1677,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()
@@ -1686,11 +1685,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 ecf3222726c..dc77d101084 100644
--- a/components/resolved_ts/src/advance.rs
+++ b/components/resolved_ts/src/advance.rs
@@ -121,7 +121,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 2bb54fe7310..151f10662a0 100644
--- a/components/server/src/server.rs
+++ b/components/server/src/server.rs
@@ -35,7 +35,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 encryption_export::{data_key_manager_from_config, DataKeyManager};
 use engine_rocks::{
     from_rocks_compression_type,
@@ -174,6 +177,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.init_quota_tuning_task(tikv.quota_limiter.clone());
@@ -374,7 +378,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(
@@ -1248,6 +1258,27 @@ 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.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/tikv_util/src/time.rs b/components/tikv_util/src/time.rs
index 72276b54a03..a48fea3e856 100644
--- a/components/tikv_util/src/time.rs
+++ b/components/tikv_util/src/time.rs
@@ -277,7 +277,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 91bbd8b92e4..654f4600eb8 100644
--- a/src/coprocessor/endpoint.rs
+++ b/src/coprocessor/endpoint.rs
@@ -111,7 +111,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();
@@ -770,6 +771,12 @@ fn make_error_batch_response(batch_resp: &mut coppb::StoreBatchTaskResponse, e:
             errorpb.set_server_is_busy(server_is_busy_err);
             batch_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());
+            batch_resp.set_region_error(err);
+        }
         Error::Other(_) => {
             tag = "other";
             batch_resp.set_other_error(e.to_string());
@@ -807,6 +814,12 @@ fn make_error_response(e: Error) -> coppb::Response {
             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";
             resp.set_other_error(e.to_string());
diff --git a/src/coprocessor/error.rs b/src/coprocessor/error.rs
index 25751553c28..7d2061f8191 100644
--- a/src/coprocessor/error.rs
+++ b/src/coprocessor/error.rs
@@ -26,6 +26,9 @@ pub enum Error {
     #[error("Coprocessor task canceled due to exceeding max pending tasks")]
     MaxPendingTasksExceeded,
 
+    #[error("{0}")]
+    InvalidMaxTsUpdate(#[from] concurrency_manager::InvalidMaxTsUpdate),
+
     #[error("{0}")]
     Other(String),
 }
@@ -126,6 +129,7 @@ impl ErrorCodeExt for Error {
             Error::Locked(_) => error_code::coprocessor::LOCKED,
             Error::DeadlineExceeded => error_code::coprocessor::DEADLINE_EXCEEDED,
             Error::MaxPendingTasksExceeded => error_code::coprocessor::MAX_PENDING_TASKS_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 a3f9ca3bb68..56b0b1adfe7 100644
--- a/src/server/raftkv/mod.rs
+++ b/src/server/raftkv/mod.rs
@@ -723,7 +723,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 8fd4f3d8ccc..372d127f9c8 100644
--- a/src/storage/errors.rs
+++ b/src/storage/errors.rs
@@ -300,6 +300,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 5e9e808663b..7ce3f0ddfad 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -1163,7 +1163,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();
@@ -1320,7 +1322,9 @@ impl<E: Engine, L: LockManager, F: KvFormat> Storage<E, L, F> {
 
                 let command_duration = tikv_util::time::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
@@ -2860,7 +2864,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();
@@ -9476,7 +9484,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();
@@ -9524,7 +9532,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
diff --git a/src/storage/mvcc/mod.rs b/src/storage/mvcc/mod.rs
index cc010dec488..234d8edf151 100644
--- a/src/storage/mvcc/mod.rs
+++ b/src/storage/mvcc/mod.rs
@@ -168,6 +168,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>),
 }
@@ -295,6 +298,7 @@ impl ErrorInner {
                 })
             }
             ErrorInner::PrimaryMismatch(l) => Some(ErrorInner::PrimaryMismatch(l.clone())),
+            ErrorInner::InvalidMaxTsUpdate(e) => Some(ErrorInner::InvalidMaxTsUpdate(e.clone())),
             ErrorInner::Io(_) | ErrorInner::Other(_) => None,
         }
     }
@@ -398,6 +402,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 107e25c7d2c..b896fe36fc6 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 d4e16331ac9..de2cf481c94 100644
--- a/src/storage/txn/actions/prewrite.rs
+++ b/src/storage/txn/actions/prewrite.rs
@@ -45,7 +45,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!(
@@ -108,7 +111,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
@@ -921,7 +927,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,
@@ -1098,7 +1104,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,
@@ -1197,7 +1203,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,
@@ -1248,7 +1254,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 4802535c054..4690eb5167f 100644
--- a/src/storage/txn/commands/check_secondary_locks.rs
+++ b/src/storage/txn/commands/check_secondary_locks.rs
@@ -58,7 +58,11 @@ impl<S: Snapshot, L: LockManager> WriteCommand<S, L> for CheckSecondaryLocks {
     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!("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 0108c247897..99fd9496932 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 a6c529420d3..6fbff271eb2 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 b34c4eb752b..7f6dfd258c5 100644
--- a/src/storage/txn/commands/prewrite.rs
+++ b/src/storage/txn/commands/prewrite.rs
@@ -1142,7 +1142,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())];
 
@@ -1284,7 +1284,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![(
@@ -1384,7 +1384,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");
@@ -1455,7 +1455,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");
@@ -2019,7 +2019,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,
@@ -2525,7 +2525,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 706d3a82e99..26fb80bb8ea 100644
--- a/src/storage/txn/mod.rs
+++ b/src/storage/txn/mod.rs
@@ -147,6 +147,9 @@ pub enum ErrorInner {
 
     #[error("region {0} not prepared the flashback")]
     FlashbackNotPrepared(u64),
+
+    #[error("{0}")]
+    InvalidMaxTsUpdate(#[from] concurrency_manager::InvalidMaxTsUpdate),
 }
 
 impl ErrorInner {
@@ -186,6 +189,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,
         }
     }
@@ -240,6 +246,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 73f46fe6427..4255014e2e7 100644
--- a/components/cdc/tests/integrations/test_cdc.rs
+++ b/components/cdc/tests/integrations/test_cdc.rs
@@ -1234,7 +1234,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 31b79fa4276..e9042cceb9d 100644
--- a/tests/failpoints/cases/test_transaction.rs
+++ b/tests/failpoints/cases/test_transaction.rs
@@ -328,7 +328,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());
@@ -356,7 +356,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()),
@@ -385,7 +385,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 ab0e776fb22..11e1fa14518 100644
--- a/tests/integrations/server/kv_service.rs
+++ b/tests/integrations/server/kv_service.rs
@@ -1650,7 +1650,7 @@ fn test_prewrite_check_max_commit_ts() {
     cluster.run();
 
     let cm = cluster.sim.read().unwrap().get_concurrency_manager(1);
-    cm.update_max_ts(100.into());
+    cm.update_max_ts(100.into(), "").unwrap();
 
     let (client, ctx) = build_client(&cluster);