fixup

include/dingosdk/client.h

@@ -191,13 +191,21 @@ struct TraceMetrics {
   std::atomic<uint64_t> sleep_time_us{0};
   std::atomic<uint64_t> sleep_count{0};
 
+  // Wall time spent acquiring timestamps from TSO on the transaction critical
+  // path (Begin start_ts, prewrite physical_ts/commit_ts, 2PC commit_ts).
+  // NOTE: the prewrite/commit portion is also contained in the corresponding
+  // sdk_time_us, so tso_time_us overlaps with those metrics.
+  std::atomic<uint64_t> tso_time_us{0};
+  std::atomic<uint64_t> tso_count{0};
+
   std::string ToString() const {
     return fmt::format(
         "total_time_us({}) read({}) prewrite({}) commit({}) "
-        "resolve_lock({}) sleep({} {})",
+        "resolve_lock({}) sleep({} {}) tso({} {})",
         total_time_us.load(std::memory_order_relaxed), read_metric.ToString(), prewrite_metric.ToString(),
         commit_metric.ToString(), resolve_lock_time_us.load(std::memory_order_relaxed),
-        sleep_time_us.load(std::memory_order_relaxed), sleep_count.load(std::memory_order_relaxed));
+        sleep_time_us.load(std::memory_order_relaxed), sleep_count.load(std::memory_order_relaxed),
+        tso_time_us.load(std::memory_order_relaxed), tso_count.load(std::memory_order_relaxed));
   }
 };
 

src/sdk/common/param_config.cc

@@ -78,3 +78,5 @@ DEFINE_int64(txn_check_status_interval_ms, 5, "txn check status interval ms");
 
 DEFINE_uint32(stale_period_us, 1000, "stale period us default 1000 us, used for tso provider");
 DEFINE_uint32(tso_batch_size, 256, "tso batch size default 256, used for tso provider");
+DEFINE_uint32(tso_anchor_max_age_us, 60000000,
+              "max age of the physical-time anchor before GetPhysicalTs falls back to a real tso fetch");

src/sdk/common/param_config.h

@@ -90,5 +90,6 @@ DECLARE_int64(txn_check_status_interval_ms);
 
 DECLARE_uint32(stale_period_us);
 DECLARE_uint32(tso_batch_size);
+DECLARE_uint32(tso_anchor_max_age_us);
 
 #endif  // DINGODB_SDK_PARAM_CONFIG_H_

src/sdk/common/tracker.h

@@ -52,6 +52,9 @@ class Tracker {
 
     std::atomic<uint64_t> sleep_time_us{0};
     std::atomic<uint64_t> sleep_count{0};
+
+    std::atomic<uint64_t> tso_time_us{0};
+    std::atomic<uint64_t> tso_count{0};
   };
 
   void SetTotalTransactionTime() { metrics_.total_transaction_time_us.store(TimestampUs() - start_time_); }
@@ -104,6 +107,13 @@ class Tracker {
   void IncrementSleepCount(uint64_t count) { metrics_.sleep_count.fetch_add(count); }
   uint64_t SleepTimeCount() const { return metrics_.sleep_count.load(); }
 
+  void IncrementTsoTime(uint64_t elapsed_time) {
+    metrics_.tso_time_us.fetch_add(elapsed_time);
+    metrics_.tso_count.fetch_add(1);
+  }
+  uint64_t TsoTime() const { return metrics_.tso_time_us.load(); }
+  uint64_t TsoCount() const { return metrics_.tso_count.load(); }
+
  private:
   uint64_t start_time_;
 

src/sdk/transaction/tso.cc

@@ -104,6 +104,31 @@ Status TsoProvider::GenPhysicalTs(int32_t count, int64_t& physical_ts) {
   return status;
 }
 
+static int64_t SteadyUs() {
+  return std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now().time_since_epoch())
+      .count();
+}
+
+Status TsoProvider::GetPhysicalTs(int64_t& physical_ts) {
+  {
+    ReadLockGuard guard(rwlock_);
+    if (anchor_physical_ms_ > 0) {
+      int64_t age_us = SteadyUs() - anchor_steady_us_;
+      if (age_us >= 0 && age_us < FLAGS_tso_anchor_max_age_us) {
+        int64_t est = anchor_physical_ms_ + age_us / 1000;
+        int64_t prev = max_physical_ms_.load(std::memory_order_relaxed);
+        while (est > prev && !max_physical_ms_.compare_exchange_weak(prev, est)) {
+        }
+        physical_ts = est > prev ? est : prev;
+        return Status::OK();
+      }
+    }
+  }
+
+  // no anchor yet or anchor too old: re-anchor with a real fetch
+  return GenPhysicalTs(2, physical_ts);
+}
+
 void TsoProvider::Refresh() {
   last_time_us_ = TimestampUs();
   physical_ = 0;
@@ -138,6 +163,9 @@ Status TsoProvider::FetchTso(uint32_t count) {
   next_logical_ = tso.logical();
   max_logical_ = next_logical_ + ts_count - 1;
 
+  anchor_physical_ms_ = tso.physical();
+  anchor_steady_us_ = SteadyUs();
+
   DINGO_LOG(DEBUG) << fmt::format("[sdk.tso] fetch tso ts({}) count({}).", Tso2Timestamp(tso), ts_count);
 
   return Status::OK();

src/sdk/transaction/tso.h

@@ -15,6 +15,7 @@
 #ifndef DINGODB_SDK_TRANSACTION_TSO_H_
 #define DINGODB_SDK_TRANSACTION_TSO_H_
 
+#include <atomic>
 #include <cstdint>
 #include <memory>
 
@@ -38,6 +39,13 @@ class TsoProvider {
 
   Status GenPhysicalTs(int32_t count, int64_t& physical_ts);
 
+  // Current tso physical time in ms, extrapolated locally from the last
+  // FetchTso anchor. For lock-ttl style deadlines only: monotonic and
+  // bounded-staleness, but NOT a globally unique timestamp — use GenTs for
+  // start_ts/commit_ts. Falls back to a real fetch when the anchor is older
+  // than FLAGS_tso_anchor_max_age_us.
+  Status GetPhysicalTs(int64_t& physical_ts);
+
   void Refresh();
 
  private:
@@ -57,6 +65,13 @@ class TsoProvider {
   int64_t max_logical_{0};
 
   uint64_t last_time_us_{0};
+
+  // physical-time anchor, written on every successful FetchTso (under write
+  // lock), read by GetPhysicalTs (under read lock)
+  int64_t anchor_physical_ms_{0};
+  int64_t anchor_steady_us_{0};
+  // largest physical ever returned; deadlines must never regress
+  std::atomic<int64_t> max_physical_ms_{0};
 };
 
 using TsoProviderSPtr = std::shared_ptr<TsoProvider>;

src/sdk/transaction/txn_impl.cc

@@ -58,7 +58,9 @@ TxnImplSPtr TxnImpl::GetSelfPtr() { return std::dynamic_pointer_cast<TxnImpl>(sh
 
 Status TxnImpl::Begin() {
   int64_t start_ts;
+  uint64_t tso_start_us = TimestampUs();
   Status status = stub_.GetTsoProvider()->GenTs(2, start_ts);
+  tracker_->IncrementTsoTime(TimestampUs() - tso_start_us);
   if (status.ok()) {
     state_.store(kActive);
     start_ts_.store(start_ts);
@@ -701,7 +703,9 @@ Status TxnImpl::CommitPrimaryKey() {
     status = task.Run();
     if (status.IsTxnCommitTsExpired()) {
       int64_t commit_ts;
+      uint64_t tso_start_us = TimestampUs();
       Status s = stub_.GetTsoProvider()->GenTs(2, commit_ts);
+      tracker_->IncrementTsoTime(TimestampUs() - tso_start_us);
       if (!s.ok()) {
         DINGO_LOG(ERROR) << fmt::format("[sdk.txn.{}] commit primary key regen ts fail, status({}).", ID(),
                                         s.ToString());
@@ -777,7 +781,10 @@ Status TxnImpl::DoCommit() {
 
   if (commit_ts == 0) {
     // only init once, if commit_ts_ not set, get a new one
-    DINGO_RETURN_NOT_OK(stub_.GetTsoProvider()->GenTs(2, commit_ts));
+    uint64_t tso_start_us = TimestampUs();
+    Status tso_status = stub_.GetTsoProvider()->GenTs(2, commit_ts);
+    tracker_->IncrementTsoTime(TimestampUs() - tso_start_us);
+    DINGO_RETURN_NOT_OK(tso_status);
     commit_ts_.store(commit_ts);
   }
 
@@ -932,6 +939,8 @@ void TxnImpl::GetTraceMetrics(TraceMetrics& metrics) {
   metrics.resolve_lock_time_us = tracker_->ResolveLockSdkTime();
   metrics.sleep_time_us = tracker_->SleepTime();
   metrics.sleep_count = tracker_->SleepTimeCount();
+  metrics.tso_time_us = tracker_->TsoTime();
+  metrics.tso_count = tracker_->TsoCount();
 }
 }  // namespace sdk
 }  // namespace dingodb

src/sdk/transaction/txn_task/txn_heartbeat_task.cc

@@ -40,7 +40,7 @@ void TxnHeartbeatTask::DoAsync() {
     return;
   }
 
-  s = stub.GetTsoProvider()->GenPhysicalTs(2, physical_ts_);
+  s = stub.GetTsoProvider()->GetPhysicalTs(physical_ts_);
   if (!s.ok()) {
     DoAsyncDone(s);
     return;

src/sdk/transaction/txn_task/txn_prewrite_task.cc

@@ -131,7 +131,9 @@ void TxnPrewriteTask::DoAsync() {
   }
 
   int64_t physical_ts{0};
-  Status status = stub.GetTsoProvider()->GenPhysicalTs(2, physical_ts);
+  uint64_t tso_start_us = TimestampUs();
+  Status status = stub.GetTsoProvider()->GetPhysicalTs(physical_ts);
+  txn_impl_->GetTracer()->IncrementTsoTime(TimestampUs() - tso_start_us);
 
   if (!status.ok()) {
     {
@@ -146,7 +148,9 @@ void TxnPrewriteTask::DoAsync() {
 
   int64_t commit_ts{0};
   if (is_one_pc_ || use_async_commit_) {
+    tso_start_us = TimestampUs();
     status = stub.GetTsoProvider()->GenTs(2, commit_ts);
+    txn_impl_->GetTracer()->IncrementTsoTime(TimestampUs() - tso_start_us);
     if (!status.ok()) {
       {
         WriteLockGuard guard(rw_lock_);