[fix][store] Mark rollback when lock miss.

visualYJD · ketor · commit f6499b170a67 · 2026-01-28T08:03:50.000Z
diff --git a/src/engine/mono_store_engine.cc b/src/engine/mono_store_engine.cc
@@ -677,11 +677,12 @@ butil::Status MonoStoreEngine::TxnWriter::TxnCheckTxnStatus(std::shared_ptr<Cont
 butil::Status MonoStoreEngine::TxnWriter::TxnCheckSecondaryLocks(std::shared_ptr<Context> ctx, store::RegionPtr region,
                                                                  int64_t start_ts,
                                                                  const std::vector<std::string>& keys) {
-  return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, ctx, region, start_ts, keys);
+  return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, mono_engine_, ctx, region, start_ts, keys);
 }
 
 butil::Status MonoStoreEngine::TxnWriter::TxnResolveLock(std::shared_ptr<Context> ctx, int64_t start_ts,
-                                                         int64_t commit_ts, const std::vector<std::string>& keys, const std::map<int64_t, int64_t>& txn_infos) {
+                                                         int64_t commit_ts, const std::vector<std::string>& keys,
+                                                         const std::map<int64_t, int64_t>& txn_infos) {
   return TxnEngineHelper::ResolveLock(txn_writer_raw_engine_, mono_engine_, ctx, start_ts, commit_ts, keys, txn_infos);
 }
 
diff --git a/src/engine/raft_store_engine.cc b/src/engine/raft_store_engine.cc
@@ -658,11 +658,12 @@ butil::Status RaftStoreEngine::TxnWriter::TxnCheckTxnStatus(std::shared_ptr<Cont
 butil::Status RaftStoreEngine::TxnWriter::TxnCheckSecondaryLocks(std::shared_ptr<Context> ctx, store::RegionPtr region,
                                                                  int64_t start_ts,
                                                                  const std::vector<std::string>& keys) {
-  return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, ctx, region, start_ts, keys);
+  return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, raft_engine_, ctx, region, start_ts, keys);
 }
 
 butil::Status RaftStoreEngine::TxnWriter::TxnResolveLock(std::shared_ptr<Context> ctx, int64_t start_ts,
-                                                         int64_t commit_ts, const std::vector<std::string>& keys, const std::map<int64_t, int64_t>& txn_infos) {
+                                                         int64_t commit_ts, const std::vector<std::string>& keys,
+                                                         const std::map<int64_t, int64_t>& txn_infos) {
   return TxnEngineHelper::ResolveLock(txn_writer_raw_engine_, raft_engine_, ctx, start_ts, commit_ts, keys, txn_infos);
 }
 
diff --git a/src/engine/txn_engine_helper.cc b/src/engine/txn_engine_helper.cc
@@ -3394,7 +3394,7 @@ butil::Status TxnEngineHelper::CheckTxnStatus(RawEnginePtr raw_engine, std::shar
                      Helper::StringToHex(primary_key))
       << ", region_epoch: " << ctx->RegionEpoch().ShortDebugString() << ", lock_ts: " << lock_ts
       << ", caller_start_ts: " << caller_start_ts << ", current_ts: " << current_ts
-      << ", force_sync_commit: " << force_sync_commit;
+      << ", force_sync_commit: " << force_sync_commit << ", rollback_if_not_exist: " << rollback_if_not_exist;
 
   // we need to do if primay_key is in this region'range in service before apply to raft state machine
   // use reader to get if the lock is exists, if lock is exists, check if the lock is expired its ttl, if expired do
@@ -3460,9 +3460,9 @@ butil::Status TxnEngineHelper::CheckTxnStatus(RawEnginePtr raw_engine, std::shar
       if (force_sync_commit) {
         DINGO_LOG_IF(INFO, FLAGS_dingo_log_switch_txn_detail)
             << fmt::format("[txn][region({})] CheckTxnStatus,", region->Id())
-            << "fallback is set, check_txn_status treats it as a non-async-commit txn" << "primary_key: " << primary_key
-            << ", lock_info: " << lock_info.ShortDebugString() << ", lock_ts: " << lock_ts
-            << ", caller_start_ts: " << caller_start_ts << ", current_ts: " << current_ts;
+            << "fallback is set, check_txn_status treats it as a non-async-commit txn"
+            << "primary_key: " << primary_key << ", lock_info: " << lock_info.ShortDebugString()
+            << ", lock_ts: " << lock_ts << ", caller_start_ts: " << caller_start_ts << ", current_ts: " << current_ts;
       } else {
         *txn_result->mutable_locked() = lock_info;
         // async-commit locks can't be resolved until they expire.
@@ -3552,6 +3552,7 @@ butil::Status TxnEngineHelper::CheckTxnStatus(RawEnginePtr raw_engine, std::shar
     // lock is expired, do rollback
     std::vector<std::string> keys_to_rollback_with_data;
     std::vector<std::string> keys_to_rollback_without_data;
+    std::vector<std::string> keys_miss_lock_to_rollback;
     if (lock_info.short_value().empty()) {
       keys_to_rollback_with_data.push_back(primary_key);
     } else {
@@ -3560,8 +3561,8 @@ butil::Status TxnEngineHelper::CheckTxnStatus(RawEnginePtr raw_engine, std::shar
     DINGO_LOG(INFO) << fmt::format("[txn][region({})] CheckTxnStatus,", region->Id())
                     << ", do rollback, primary_key: " << Helper::StringToHex(primary_key) << ", lock_ts: " << lock_ts
                     << ", lock_info: " << lock_info.ShortDebugString();
-    auto ret =
-        DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data, lock_ts);
+    auto ret = DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data,
+                          keys_miss_lock_to_rollback, lock_ts);
     if (!ret.ok()) {
       DINGO_LOG(ERROR) << fmt::format("[txn][region({})] CheckTxnStatus,", region->Id())
                        << ", rollback failed, primary_key: " << Helper::StringToHex(primary_key)
@@ -3693,6 +3694,7 @@ butil::Status TxnEngineHelper::BatchRollback(RawEnginePtr raw_engine, std::share
 
   std::vector<std::string> keys_to_rollback_with_data;
   std::vector<std::string> keys_to_rollback_without_data;
+  std::vector<std::string> keys_miss_lock_to_rollback;
   for (const auto &key : keys) {
     pb::store::LockInfo lock_info;
     auto ret = txn_reader.GetLockInfo(key, lock_info);
@@ -3702,14 +3704,44 @@ butil::Status TxnEngineHelper::BatchRollback(RawEnginePtr raw_engine, std::share
                        << ", status: " << ret.error_str();
     }
 
-    // if lock is not exist, nothing to do
+    // when concurrency prewrite, prewrite secondary key success,but prewrite primary meet write conflict,rollback
+    // transaction,primary key lock not exist. we need record rollback for primary key for speeding up transaction
+    // conflict handling.
     if (lock_info.primary_lock().empty()) {
-      DINGO_LOG(WARNING) << fmt::format("[txn][region({})] BatchRollback", region->Id())
-                         << ", txn_not_found with lock_info empty, key: " << Helper::StringToHex(key)
-                         << ", start_ts: " << start_ts;
+      DINGO_LOG(INFO) << fmt::format("[txn][region({})] BatchRollback", region->Id())
+                      << ", txn_not_found with lock_info empty, key: " << Helper::StringToHex(key)
+                      << ", start_ts: " << start_ts;
 
-      // auto *txn_not_found = txn_result->mutable_txn_not_found();
-      // txn_not_found->set_start_ts(start_ts);
+      // the lock is not exists, check if it is rollbacked or committed
+      // try to get if there is a rollback to lock_ts
+      pb::store::WriteInfo write_info;
+      auto ret1 = txn_reader.GetRollbackInfo(start_ts, key, write_info);
+      if (!ret1.ok()) {
+        DINGO_LOG(FATAL) << fmt::format("[txn][region({})] BatchRollback, ", region->Id())
+                         << ", get rollback info failed, key: " << Helper::StringToHex(key)
+                         << ", start_ts: " << start_ts << ", status: " << ret1.error_str();
+      }
+
+      if (write_info.start_ts() == start_ts) {
+        // has been rollbacked.
+        continue;
+      }
+
+      // if there is not a rollback to lock_ts, try to get the commit_ts
+      int64_t commit_ts = 0;
+      auto ret2 =
+          txn_reader.GetWriteInfo(start_ts, Constant::kMaxVer, start_ts, key, false, true, true, write_info, commit_ts);
+      if (!ret2.ok()) {
+        DINGO_LOG(FATAL) << fmt::format("[txn][region({})] BatchRollback,", region->Id())
+                         << ", get write info failed, key: " << Helper::StringToHex(key) << ", lock_ts: " << start_ts
+                         << ", status: " << ret2.error_str();
+      }
+
+      if (commit_ts == 0) {
+        keys_miss_lock_to_rollback.push_back(key);
+        DINGO_LOG(INFO) << fmt::format("[txn][region({})] BatchRollback,start_ts:{} key:{} MarkRollBackOnMissingLock.",
+                                       region->Id(), start_ts, Helper::StringToHex(key));
+      }
       continue;
     }
 
@@ -3752,8 +3784,8 @@ butil::Status TxnEngineHelper::BatchRollback(RawEnginePtr raw_engine, std::share
   }
 
   // do rollback
-  auto ret =
-      DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data, start_ts);
+  auto ret = DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data,
+                        keys_miss_lock_to_rollback, start_ts);
   if (!ret.ok()) {
     DINGO_LOG(ERROR) << fmt::format("[txn][region({})] BatchRollback, ", region->Id())
                      << ", rollback failed, status: " << ret.error_str();
@@ -3769,14 +3801,17 @@ bvar::LatencyRecorder g_txn_do_rollback_latency("dingo_txn_do_rollback");
 butil::Status TxnEngineHelper::DoRollback(RawEnginePtr /*raw_engine*/, std::shared_ptr<Engine> raft_engine,
                                           std::shared_ptr<Context> ctx,
                                           std::vector<std::string> &keys_to_rollback_with_data,
-                                          std::vector<std::string> &keys_to_rollback_without_data, int64_t start_ts) {
+                                          std::vector<std::string> &keys_to_rollback_without_data,
+                                          std::vector<std::string> &keys_miss_lock_to_rollback, int64_t start_ts) {
   BvarLatencyGuard bvar_guard(&g_txn_do_rollback_latency);
 
   DINGO_LOG_IF(INFO, FLAGS_dingo_log_switch_txn_detail)
       << "[txn]Rollback start_ts: " << start_ts << ", keys_count_with_data: " << keys_to_rollback_with_data.size()
-      << ", keys_count_without_data: " << keys_to_rollback_without_data.size();
+      << ", keys_count_without_data: " << keys_to_rollback_without_data.size()
+      << ", keys_miss_lock_to_rollback:" << keys_miss_lock_to_rollback.size();
 
-  if (keys_to_rollback_without_data.empty() && keys_to_rollback_with_data.empty()) {
+  if (keys_to_rollback_without_data.empty() && keys_to_rollback_with_data.empty() &&
+      keys_miss_lock_to_rollback.empty()) {
     DINGO_LOG_IF(INFO, FLAGS_dingo_log_switch_txn_detail) << "[txn]Rollback nothing to do, start_ts: " << start_ts;
     return butil::Status::OK();
   }
@@ -3818,6 +3853,18 @@ butil::Status TxnEngineHelper::DoRollback(RawEnginePtr /*raw_engine*/, std::shar
     kv_puts_write.emplace_back(kv);
   }
 
+  for (const auto &key : keys_miss_lock_to_rollback) {
+    // add write
+    pb::store::WriteInfo write_info;
+    write_info.set_start_ts(start_ts);
+    write_info.set_op(::dingodb::pb::store::Op::Rollback);
+
+    pb::common::KeyValue kv;
+    kv.set_key(mvcc::Codec::EncodeKey(key, start_ts));
+    kv.set_value(write_info.SerializeAsString());
+    kv_puts_write.emplace_back(kv);
+  }
+
   if (kv_puts_write.empty() && kv_deletes_lock.empty() && kv_deletes_data.empty()) {
     DINGO_LOG_IF(INFO, FLAGS_dingo_log_switch_txn_detail) << "[txn]Rollback nothing to do, start_ts: " << start_ts;
     return butil::Status::OK();
@@ -3908,9 +3955,9 @@ bvar::LatencyRecorder g_txn_check_secondary_locks_latency("dingo_txn_check_secon
 ///
 /// If all prewritten locks exist, the lock information is returned.
 /// Otherwise, it returns the commit timestamp of the transaction.
-butil::Status TxnEngineHelper::TxnCheckSecondaryLocks(RawEnginePtr raw_engine, std::shared_ptr<Context> ctx,
-                                                      store::RegionPtr region, int64_t start_ts,
-                                                      const std::vector<std::string> &keys) {
+butil::Status TxnEngineHelper::TxnCheckSecondaryLocks(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
+                                                      std::shared_ptr<Context> ctx, store::RegionPtr region,
+                                                      int64_t start_ts, const std::vector<std::string> &keys) {
   BvarLatencyGuard bvar_guard(&g_txn_check_secondary_locks_latency);
 
   DINGO_LOG_IF(INFO, FLAGS_dingo_log_switch_txn_detail)
@@ -3992,14 +4039,19 @@ butil::Status TxnEngineHelper::TxnCheckSecondaryLocks(RawEnginePtr raw_engine, s
       }
 
       if (commit_ts == 0) {
-        // it seems there is a lock previously exists, but it is not committed, and there is no rollback, there must
-        // be some error, return TxnNotFound
+        auto ret3 = MarkRollBackOnMissingLock(raw_engine, raft_engine, ctx, key, start_ts);
+        if (!ret3.ok()) {
+          DINGO_LOG(ERROR) << fmt::format("[txn][region({})] CheckTxnStatus", region->Id())
+                           << ", MarkRollBackOnMissingLock failed";
+          return butil::Status(pb::error::Errno::EINTERNAL, "MarkRollBackOnMissingLock failed");
+        }
+
         auto *txn_not_found = txn_result->mutable_txn_not_found();
         txn_not_found->set_primary_key(key);
         txn_not_found->set_start_ts(start_ts);
-        DINGO_LOG(ERROR)
+        DINGO_LOG(WARNING)
             << fmt::format("[txn][region({})] CheckSecondaryLocks,", region->Id())
-            << ", cannot found the transaction, maybe some error ocurred, return txn_not_found, secondary_key: "
+            << ", cannot found the transaction, mark rollback on missing lock, return txn_not_found, secondary_key: "
             << Helper::StringToHex(key) << ", start_ts: " << start_ts
             << ", lock_info: " << lock_info.ShortDebugString();
         return butil::Status::OK();
@@ -4026,6 +4078,7 @@ butil::Status TxnEngineHelper::BatchResolveLock(RawEnginePtr raw_engine, std::sh
     std::vector<pb::store::LockInfo> lock_infos_to_commit;
     std::vector<std::string> keys_to_rollback_with_data;
     std::vector<std::string> keys_to_rollback_without_data;
+    std::vector<std::string> keys_miss_lock_to_rollback;
 
     auto stream = Stream::New(FLAGS_stream_message_max_limit_size);
     std::vector<pb::store::LockInfo> tmp_lock_infos;
@@ -4087,8 +4140,8 @@ butil::Status TxnEngineHelper::BatchResolveLock(RawEnginePtr raw_engine, std::sh
         DINGO_LOG(INFO) << fmt::format("[txn][region({})] BatchResolveLock, ", region->Id()) << "primary key:" << key
                         << ", do rollback without data, start_ts: " << start_ts;
       }
-      auto ret =
-          DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data, start_ts);
+      auto ret = DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data,
+                            keys_miss_lock_to_rollback, start_ts);
       if (!ret.ok()) {
         DINGO_LOG(ERROR) << fmt::format("[txn][region({})] BatchResolveLock, ", region->Id())
                          << ", rollback failed, start_ts: " << start_ts << ", status: " << ret.error_str();
@@ -4153,6 +4206,7 @@ butil::Status TxnEngineHelper::ResolveLock(RawEnginePtr raw_engine, std::shared_
   std::vector<pb::store::LockInfo> lock_infos_to_commit;
   std::vector<std::string> keys_to_rollback_with_data;
   std::vector<std::string> keys_to_rollback_without_data;
+  std::vector<std::string> keys_miss_lock_to_rollback;
 
   // if keys is not empty, we only do resolve lock for these keys
   if (!keys.empty()) {
@@ -4266,8 +4320,8 @@ butil::Status TxnEngineHelper::ResolveLock(RawEnginePtr raw_engine, std::shared_
       DINGO_LOG(INFO) << fmt::format("[txn][region({})] ResolveLock, ", region->Id()) << "primary key:" << key
                       << ", do rollback without data, start_ts: " << start_ts;
     }
-    auto ret =
-        DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data, start_ts);
+    auto ret = DoRollback(raw_engine, raft_engine, ctx, keys_to_rollback_with_data, keys_to_rollback_without_data,
+                          keys_miss_lock_to_rollback, start_ts);
     if (!ret.ok()) {
       DINGO_LOG(ERROR) << fmt::format("[txn][region({})] ResolveLock, ", region->Id())
                        << ", rollback failed, start_ts: " << start_ts << ", status: " << ret.error_str();
diff --git a/src/engine/txn_engine_helper.h b/src/engine/txn_engine_helper.h
@@ -158,7 +158,8 @@ class TxnEngineHelper {
 
   static butil::Status DoRollback(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
                                   std::shared_ptr<Context> ctx, std::vector<std::string> &keys_to_rollback_with_data,
-                                  std::vector<std::string> &keys_to_rollback_without_data, int64_t start_ts);
+                                  std::vector<std::string> &keys_to_rollback_without_data,
+                                  std::vector<std::string> &keys_miss_lock_to_rollback, int64_t start_ts);
 
   static butil::Status MarkRollBackOnMissingLock(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
                                                  std::shared_ptr<Context> ctx, std::string primary_key,
@@ -200,15 +201,18 @@ class TxnEngineHelper {
                                       int64_t caller_start_ts, int64_t current_ts, bool force_sync_commit,
                                       bool rollback_if_not_exist);
 
-  static butil::Status TxnCheckSecondaryLocks(RawEnginePtr raw_engine, std::shared_ptr<Context> ctx,
-                                              store::RegionPtr region, int64_t start_ts,
+  static butil::Status TxnCheckSecondaryLocks(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
+                                              std::shared_ptr<Context> ctx, store::RegionPtr region, int64_t start_ts,
                                               const std::vector<std::string> &keys);
 
   static butil::Status ResolveLock(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
                                    std::shared_ptr<Context> ctx, int64_t start_ts, int64_t commit_ts,
                                    const std::vector<std::string> &keys, const std::map<int64_t, int64_t> &txn_infos);
 
-  static butil::Status BatchResolveLock(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine, std::shared_ptr<Context> ctx, store::RegionPtr region, const std::map<int64_t, int64_t> &txn_infos, pb::store::TxnResultInfo* txn_result);
+  static butil::Status BatchResolveLock(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
+                                        std::shared_ptr<Context> ctx, store::RegionPtr region,
+                                        const std::map<int64_t, int64_t> &txn_infos,
+                                        pb::store::TxnResultInfo *txn_result);
 
   static butil::Status HeartBeat(RawEnginePtr raw_engine, std::shared_ptr<Engine> raft_engine,
                                  std::shared_ptr<Context> ctx, const std::string &primary_lock, int64_t start_ts,

Original file line number	Diff line number	Diff line change
`@@ -677,11 +677,12 @@ butil::Status MonoStoreEngine::TxnWriter::TxnCheckTxnStatus(std::shared_ptr<Cont`
`677`	`677`	`butil::Status MonoStoreEngine::TxnWriter::TxnCheckSecondaryLocks(std::shared_ptr<Context> ctx, store::RegionPtr region,`
`678`	`678`	`int64_t start_ts,`
`679`	`679`	`const std::vector<std::string>& keys) {`
`680`		`- return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, ctx, region, start_ts, keys);`
	`680`	`+ return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, mono_engine_, ctx, region, start_ts, keys);`
`681`	`681`	`}`
`682`	`682`
`683`	`683`	`butil::Status MonoStoreEngine::TxnWriter::TxnResolveLock(std::shared_ptr<Context> ctx, int64_t start_ts,`
`684`		`- int64_t commit_ts, const std::vector<std::string>& keys, const std::map<int64_t, int64_t>& txn_infos) {`
	`684`	`+ int64_t commit_ts, const std::vector<std::string>& keys,`
	`685`	`+ const std::map<int64_t, int64_t>& txn_infos) {`
`685`	`686`	`return TxnEngineHelper::ResolveLock(txn_writer_raw_engine_, mono_engine_, ctx, start_ts, commit_ts, keys, txn_infos);`
`686`	`687`	`}`
`687`	`688`
Original file line number	Diff line number	Diff line change
`@@ -658,11 +658,12 @@ butil::Status RaftStoreEngine::TxnWriter::TxnCheckTxnStatus(std::shared_ptr<Cont`
`658`	`658`	`butil::Status RaftStoreEngine::TxnWriter::TxnCheckSecondaryLocks(std::shared_ptr<Context> ctx, store::RegionPtr region,`
`659`	`659`	`int64_t start_ts,`
`660`	`660`	`const std::vector<std::string>& keys) {`
`661`		`- return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, ctx, region, start_ts, keys);`
	`661`	`+ return TxnEngineHelper::TxnCheckSecondaryLocks(txn_writer_raw_engine_, raft_engine_, ctx, region, start_ts, keys);`
`662`	`662`	`}`
`663`	`663`
`664`	`664`	`butil::Status RaftStoreEngine::TxnWriter::TxnResolveLock(std::shared_ptr<Context> ctx, int64_t start_ts,`
`665`		`- int64_t commit_ts, const std::vector<std::string>& keys, const std::map<int64_t, int64_t>& txn_infos) {`
	`665`	`+ int64_t commit_ts, const std::vector<std::string>& keys,`
	`666`	`+ const std::map<int64_t, int64_t>& txn_infos) {`
`666`	`667`	`return TxnEngineHelper::ResolveLock(txn_writer_raw_engine_, raft_engine_, ctx, start_ts, commit_ts, keys, txn_infos);`
`667`	`668`	`}`
`668`	`669`