From 97ae75cb12f532df0ee42da4bc288559bd19b8a3 Mon Sep 17 00:00:00 2001 From: bobhan1 Date: Tue, 27 Oct 2026 10:10:24 +0800 Subject: [PATCH] [fix](cloud) Resume committed transactions through lazy publication (#69606) ### What problem does this PR solve? Related PR: #67989 A transaction commit retry can scan the temporary rowsets left by concurrent lazy publication, then start its write transaction after those rowsets and their pending markers have been removed. The transaction is still COMMITTED, so processing the old scan can publish the same rowsets again or return only part of the transaction's partition versions. Handle COMMITTED explicitly in the immediate, eventual and sub-transaction commit paths through a shared wait_for_committed_txn helper. After the non-snapshot txn_info read in the actual write transaction, release that transaction and submit or join the lazy task. After publication succeeds, restart and return the durable VISIBLE transaction through the existing idempotent branch, with empty version results so FE invalidates its version caches. A concurrent transition from PREPARED to COMMITTED conflicts with the same txn_info read at KV commit. If the lazy task fails, the transaction has already committed and the RPC still returns OK. Match the existing lazy response contract: return txn_info as VISIBLE for FE callbacks, with is_lazy_commit=true and is_lazy_commit_incomplete=true, without changing the durable COMMITTED state or publishing the old scan. The two flags also identify incomplete publication when the retry originally selected immediate, so FE does not treat it as completed visibility. Also clear the response before falling back from a failed immediate attempt to lazy commit. Immediate may populate version results before its KV commit fails with TXN_BYTES_TOO_LARGE; those results must not be combined with the successful lazy response or retained in an already-visible response. #### Commit structure 1. Add three deterministic real-FDB retry tests and strengthen the existing fallback response test. 2. Resume COMMITTED transactions through lazy publication in all three commit paths, preserving commit success if publication fails. 3. Clear failed immediate response data before lazy fallback. #### Validation - Real-FDB retry tests cover an immediate retry with a stale scan, a retry that still selects lazy, and publication failure after the transaction has committed. The failure case checks both immediate and lazy retry responses: OK, incomplete-lazy flags, no version/statistics hints, unchanged business metadata and durable COMMITTED status. - The immediate regression reproduced real version-3 duplicates on the base implementation. The lazy regression fails before its existing DCHECK aborts the whole test binary. The existing MemTxnKv fallback test rejects duplicate version results from a failed immediate attempt. - Final focused ASAN run: **all four related cases passed, zero failures/errors**. A complete-file rerun exited 137 during ConcurrentCommitTxnEventuallyCase3Test before producing an XML report; no assertion failure was logged before interruption, and this run is not counted as passing. FDB 7.1.57 was deployed as a private instance with the repository's tools/fdb/fdb_ctl.sh; its processes were stopped and temporary configuration restored afterward. - Changed-line clang-format 16 checks and git diff --check passed. The build-hygiene check reports an existing BE util/threadpool.h reach overrun (1088 TUs versus a limit of 1087), also reproduced from archived HEAD sources before this refactor; no BE files were changed. Clang-tidy was not run. The successful retry cases use bounded gates to synchronize actual partition cleanup and final visibility, compare business metadata including statistics, check rowset identities and versions, and verify the empty VISIBLE response used by FE cache invalidation. These are MS tests; the failure case does not run an FE process or independently reproduce the sub-transaction COMMITTED retry timeline. ### Release note Prevent duplicate publication and incomplete version responses when transaction commit retries race lazy publication, while preserving commit success when lazy publication is incomplete. ### Check List (For Author) - Test - [x] Unit Test - Behavior changed: - [x] Yes. Resume lazy publication for COMMITTED retries, preserve commit success on publication failure, and discard failed immediate response data before fallback. - Does this need documentation? - [x] No. No new user-facing setting or syntax. ### Check List (For Reviewer who merge this PR) - [ ] Confirm the release note - [ ] Confirm test cases - [ ] Confirm document - [ ] Add branch pick label --- cloud/src/meta-service/meta_service.h | 4 + cloud/src/meta-service/meta_service_txn.cpp | 55 +++ cloud/src/meta-service/txn_lazy_committer.cpp | 3 +- cloud/test/txn_lazy_commit_test.cpp | 359 ++++++++++++++++++ 4 files changed, 420 insertions(+), 1 deletion(-) diff --git a/cloud/src/meta-service/meta_service.h b/cloud/src/meta-service/meta_service.h index 7023ccf56f6..368676c9cbd 100644 --- a/cloud/src/meta-service/meta_service.h +++ b/cloud/src/meta-service/meta_service.h @@ -490,6 +490,10 @@ private: bool is_version_read_enabled(std::string_view instance_id) const; bool is_version_write_enabled(std::string_view instance_id) const; + bool wait_for_committed_txn(const std::string& instance_id, const TxnInfoPB& txn_info, + std::unique_ptr& txn, CommitTxnResponse* response, + MetaServiceCode& code, std::string& msg, KVStats& stats); + void commit_txn_immediately( const CommitTxnRequest* request, CommitTxnResponse* response, MetaServiceCode& code, std::string& msg, const std::string& instance_id, int64_t db_id, diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index caa967f2ee7..2b71443e008 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/cloud/src/meta-service/meta_service_txn.cpp @@ -1840,6 +1840,36 @@ std::pair get_partition_versions( return {MetaServiceCode::OK, ""}; } +// Returns true when the caller should reread VISIBLE. +bool MetaServiceImpl::wait_for_committed_txn(const std::string& instance_id, + const TxnInfoPB& txn_info, + std::unique_ptr& txn, + CommitTxnResponse* response, MetaServiceCode& code, + std::string& msg, KVStats& stats) { + // Lazy publication owns this transaction; do not reuse its stale rowset scan. + stats.get_bytes += txn->get_bytes(); + stats.get_counter += txn->num_get_keys(); + txn.reset(); + auto txn_id = txn_info.txn_id(); + auto task = txn_lazy_committer_->submit(instance_id, txn_id); + TEST_SYNC_POINT_CALLBACK("commit_txn::wait_committed", &txn_id); + std::tie(code, msg) = task->wait(); + if (code == MetaServiceCode::OK) { + return true; + } + + LOG(WARNING) << "txn lazy commit failed txn_id=" << txn_id << " code=" << code + << " msg=" << msg; + // The transaction is committed; match the incomplete lazy response contract. + code = MetaServiceCode::OK; + msg.clear(); + response->mutable_txn_info()->CopyFrom(txn_info); + response->mutable_txn_info()->set_status(TxnStatusPB::TXN_STATUS_VISIBLE); + response->set_is_lazy_commit(true); + response->set_is_lazy_commit_incomplete(true); + return false; +} + /** * 0. Extract txn_id from request * 1. Get db id from TxnKv with txn_id @@ -1957,6 +1987,13 @@ void MetaServiceImpl::commit_txn_immediately( return; } + if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) { + if (!wait_for_committed_txn(instance_id, txn_info, txn, response, code, msg, stats)) { + return; + } + continue; + } + if (request->enable_check_commit_tso_fence() && config::enable_check_commit_tso_fence && txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED && !check_txn_commit_tso_fence(txn.get(), instance_id, commit_tso, response, code, msg)) { @@ -2823,6 +2860,13 @@ void MetaServiceImpl::commit_txn_eventually( return; } + if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) { + if (!wait_for_committed_txn(instance_id, txn_info, txn, response, code, msg, stats)) { + return; + } + continue; + } + if (request->enable_check_commit_tso_fence() && config::enable_check_commit_tso_fence && txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED && !check_txn_commit_tso_fence(txn.get(), instance_id, commit_tso, response, code, msg)) { @@ -2844,6 +2888,7 @@ void MetaServiceImpl::commit_txn_eventually( if (request->has_commit_attachment()) { txn_info.mutable_commit_attachment()->CopyFrom(request->commit_attachment()); } + TEST_SYNC_POINT_RETURN_WITH_VOID("commit_txn_eventually::before_mark_committed", &code); DCHECK(txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED); // set status TXN_STATUS_COMMITTED not TXN_STATUS_VISIBLE !!! // lazy commit task will advance txn to make txn visible @@ -3229,6 +3274,13 @@ void MetaServiceImpl::commit_txn_with_sub_txn(const CommitTxnRequest* request, return; } + if (txn_info.status() == TxnStatusPB::TXN_STATUS_COMMITTED) { + if (!wait_for_committed_txn(instance_id, txn_info, txn, response, code, msg, stats)) { + return; + } + continue; + } + if (request->enable_check_commit_tso_fence() && config::enable_check_commit_tso_fence && txn_info.status() != TxnStatusPB::TXN_STATUS_COMMITTED && !check_txn_commit_tso_fence(txn.get(), instance_id, commit_tso, response, code, msg)) { @@ -3825,6 +3877,9 @@ void MetaServiceImpl::commit_txn(::google::protobuf::RpcController* controller, << " tmp_rowsets_meta.size=" << tmp_rowsets_meta.size(); code = MetaServiceCode::OK; msg.clear(); + // Drop failed immediate results so FE's CloudGlobalTransactionMgr.updateVersion() + // can invalidate caches when lazy returns only VISIBLE txn_info. + response->Clear(); commit_txn_eventually(request, response, code, msg, instance_id, db_id, tmp_rowsets_meta, stats); } diff --git a/cloud/src/meta-service/txn_lazy_committer.cpp b/cloud/src/meta-service/txn_lazy_committer.cpp index bec4b13f068..382cbb06963 100644 --- a/cloud/src/meta-service/txn_lazy_committer.cpp +++ b/cloud/src/meta-service/txn_lazy_committer.cpp @@ -918,7 +918,7 @@ std::pair TxnLazyCommitTask::commit_partition( LOG(INFO) << "remove tmp_rowset_key=" << hex(tmp_rowset_key) << " txn_id=" << txn_id_; } - TEST_SYNC_POINT_CALLBACK("TxnLazyCommitter::commit"); + TEST_SYNC_POINT_CALLBACK("TxnLazyCommitter::commit", &partition_id); err = txn->commit(); if (err != TxnErrorCode::TXN_OK) { code = cast_as(err); @@ -927,6 +927,7 @@ std::pair TxnLazyCommitTask::commit_partition( return {code, msg}; } } + TEST_SYNC_POINT_CALLBACK("TxnLazyCommitTask::commit_partition::finish", &partition_id); return {MetaServiceCode::OK, ""}; } diff --git a/cloud/test/txn_lazy_commit_test.cpp b/cloud/test/txn_lazy_commit_test.cpp index e7875171892..e17056b61bc 100644 --- a/cloud/test/txn_lazy_commit_test.cpp +++ b/cloud/test/txn_lazy_commit_test.cpp @@ -24,11 +24,14 @@ #include #include +#include #include #include #include #include +#include #include +#include #include #include #include @@ -1450,6 +1453,354 @@ TEST(TxnLazyCommitTest, CommitTxnWithSubTxnRepairTabletIndexTest) { } } +// Retry tests use a real lazy publisher and real FDB transactions. Two partitions +// have two tablets each, so a retry can scan a strict subset of the original load. +class CommittedTxnRetryTest : public testing::Test { +protected: + using Metadata = std::map; + static constexpr int64_t DB_ID = 93001; + static constexpr int64_t TABLE_ID = 93002; + static constexpr int64_t INDEX_ID = 93003; + static constexpr int64_t PARTITION_ID = 93004; + static constexpr int64_t TABLET_ID = 93100; + static constexpr auto INSTANCE_ID = "test_instance"; + + void SetUp() override { + config::txn_lazy_commit_rowsets_thresold = 2; + config::enable_cloud_parallel_txn_lazy_commit = true; + // Both partition workers must be able to reach their synchronization point. + config::parallel_txn_lazy_commit_num_threads = 2; + config::cloud_txn_lazy_commit_fuzzy_possibility = 0; + kv_ = get_fdb_txn_kv(); + service_ = get_meta_service(kv_, true); + sp_->enable_processing(); + + BeginTxnRequest begin; + begin.set_cloud_unique_id("test_cloud_unique_id"); + auto* info = begin.mutable_txn_info(); + info->set_db_id(DB_ID); + info->set_label("committed_txn_retry"); + info->add_table_ids(TABLE_ID); + info->set_timeout_ms(60000); + BeginTxnResponse response; + brpc::Controller controller; + service_->begin_txn(&controller, &begin, &response, nullptr); + ASSERT_EQ(response.status().code(), MetaServiceCode::OK); + request_.set_cloud_unique_id("test_cloud_unique_id"); + request_.set_db_id(DB_ID); + request_.set_txn_id(response.txn_id()); + request_.set_is_2pc(false); + request_.set_enable_txn_lazy_commit(true); + + for (int i = 0; i < 4; ++i) { + create_tablet_with_db_id(service_.get(), DB_ID, TABLE_ID, INDEX_ID, + PARTITION_ID + i / 2, TABLET_ID + i); + auto rowset = + create_rowset(request_.txn_id(), TABLET_ID + i, INDEX_ID, PARTITION_ID + i / 2); + rowset.set_num_rows(7 + i); + rowset.set_num_segments(1); + rowset.set_data_disk_size(100); + rowset.set_index_disk_size(20); + rowset.set_total_disk_size(120); + CreateRowsetResponse result; + prepare_rowset(service_.get(), rowset, result); + ASSERT_EQ(result.status().code(), MetaServiceCode::OK); + commit_rowset(service_.get(), rowset, result); + ASSERT_EQ(result.status().code(), MetaServiceCode::OK); + rowsets_.push_back(std::move(rowset)); + } + } + + void TearDown() override { + // Release workers before joining, including when a test exits via ASSERT. + // Keep callbacks and their captured fixture state alive until all work ends. + { + std::lock_guard lock(mutex_); + stop_ = true; + cv_.notify_all(); + } + join_requests(); + service_.reset(); + sp_->clear_all_call_backs(); + sp_->clear_trace(); + sp_->disable_processing(); + config::txn_lazy_commit_rowsets_thresold = old_threshold_; + config::enable_cloud_parallel_txn_lazy_commit = old_parallel_; + config::parallel_txn_lazy_commit_num_threads = old_parallel_threads_; + config::cloud_txn_lazy_commit_fuzzy_possibility = old_fuzzy_; + } + + CommitTxnResponse commit(const CommitTxnRequest& request) { + CommitTxnResponse response; + brpc::Controller controller; + service_->commit_txn(&controller, &request, &response, nullptr); + return response; + } + + template + bool await(Predicate predicate) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, std::chrono::seconds(10), [&] { return stop_ || predicate(); }) && + !stop_; + } + + void start_partial_lazy_commit() { + sp_->set_call_back("TxnLazyCommitter::commit", [&](auto&& args) { + auto partition_id = *try_any_cast(args[0]); + std::unique_lock lock(mutex_); + ++partitions_entered_; + cv_.notify_all(); + if (partition_id == PARTITION_ID + 1) { + EXPECT_TRUE(cv_.wait_for(lock, std::chrono::seconds(10), + [&] { return stop_ || release_last_; })); + } + }); + sp_->set_call_back("TxnLazyCommitTask::commit_partition::finish", [&](auto&& args) { + if (*try_any_cast(args[0]) == PARTITION_ID) { + std::lock_guard lock(mutex_); + first_finished_ = true; + cv_.notify_all(); + } + }); + sp_->set_call_back("TxnLazyCommitTask::make_committed_txn_visible::commit", [&](auto&&) { + std::unique_lock lock(mutex_); + before_visible_ = true; + cv_.notify_all(); + EXPECT_TRUE(cv_.wait_for(lock, std::chrono::seconds(10), + [&] { return stop_ || release_visible_; })); + }); + sp_->set_call_back("commit_txn::wait_committed", [&](auto&& args) { + EXPECT_EQ(*try_any_cast(args[0]), request_.txn_id()); + std::lock_guard lock(mutex_); + retry_waiting_ = true; + cv_.notify_all(); + }); + original_ = std::thread([&] { original_response_ = commit(request_); }); + } + + void finish_partition_before_retry() { + { + std::lock_guard lock(mutex_); + if (release_last_) return; // The restarted write transaction also enters this hook. + release_last_ = true; + cv_.notify_all(); + } + ASSERT_TRUE(await([&] { return before_visible_; })); + check_status(TxnStatusPB::TXN_STATUS_COMMITTED); + // Both partitions are published now, but the final visibility transaction + // is paused. Save versions, rowsets and all statistics before the retry. + before_retry_ = business_metadata(); + } + + void start_retry() { + retry_ = std::thread([&] { + retry_response_ = commit(request_); + std::lock_guard lock(mutex_); + retry_done_ = true; + cv_.notify_all(); + }); + } + + void release_visibility() { + std::lock_guard lock(mutex_); + release_visible_ = true; + cv_.notify_all(); + } + + void join_requests() { + if (original_.joinable()) original_.join(); + if (retry_.joinable()) retry_.join(); + } + + Metadata business_metadata() { + Metadata result; + std::unique_ptr txn; + EXPECT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK); + FullRangeGetOptions options; + options.snapshot = true; + auto iter = txn->full_range_get("\x00", "\xfe", std::move(options)); + for (auto item = iter->next(); item.has_value(); item = iter->next()) { + // The original request legitimately changes these three keys when it + // becomes VISIBLE. All other keys must remain identical across retry. + if (item->first == txn_info_key({INSTANCE_ID, DB_ID, request_.txn_id()}) || + item->first == txn_running_key({INSTANCE_ID, DB_ID, request_.txn_id()}) || + item->first == recycle_txn_key({INSTANCE_ID, DB_ID, request_.txn_id()})) { + continue; + } + result.emplace(item->first, item->second); + } + EXPECT_TRUE(iter->is_valid()) << iter->error_code(); + return result; + } + + void check_status(TxnStatusPB expected) { + std::unique_ptr txn; + ASSERT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK); + std::string value; + ASSERT_EQ(txn->get(txn_info_key({INSTANCE_ID, DB_ID, request_.txn_id()}), &value), + TxnErrorCode::TXN_OK); + TxnInfoPB info; + ASSERT_TRUE(info.ParseFromString(value)); + EXPECT_EQ(info.status(), expected); + } + + void check_visible_retry() { + EXPECT_EQ(original_response_.status().code(), MetaServiceCode::OK); + ASSERT_EQ(retry_response_.status().code(), MetaServiceCode::OK) + << retry_response_.ShortDebugString(); + EXPECT_EQ(retry_response_.txn_info().status(), TxnStatusPB::TXN_STATUS_VISIBLE); + ASSERT_EQ(retry_response_.txn_info().table_ids_size(), 1); + EXPECT_EQ(retry_response_.txn_info().table_ids(0), TABLE_ID); + // This is the existing FE updateVersion() cache-invalidation response: + // VISIBLE, no partition/table versions, and no incomplete-lazy flag. + EXPECT_TRUE(retry_response_.partition_ids().empty()); + EXPECT_TRUE(retry_response_.table_ids().empty()); + EXPECT_TRUE(retry_response_.versions().empty()); + EXPECT_TRUE(retry_response_.table_stats().empty()); + EXPECT_FALSE(retry_response_.is_lazy_commit_incomplete()); + EXPECT_EQ(business_metadata(), before_retry_); + check_status(TxnStatusPB::TXN_STATUS_VISIBLE); + std::unique_ptr txn; + ASSERT_EQ(kv_->create_txn(&txn), TxnErrorCode::TXN_OK); + for (int i = 0; i < 4; ++i) { + std::string value; + ASSERT_EQ(txn->get(meta_rowset_key({INSTANCE_ID, TABLET_ID + i, 2}), &value), + TxnErrorCode::TXN_OK); + RowsetMetaCloudPB rowset; + ASSERT_TRUE(rowset.ParseFromString(value)); + EXPECT_EQ(rowset.rowset_id_v2(), rowsets_[i].rowset_id_v2()); + EXPECT_EQ(rowset.start_version(), 2); + EXPECT_EQ(rowset.end_version(), 2); + EXPECT_EQ(rowset.num_rows(), rowsets_[i].num_rows()); + check_rowset_meta_not_exist(txn, TABLET_ID + i, 3); + check_tmp_rowset_not_exist(txn, TABLET_ID + i, request_.txn_id()); + } + } + + std::shared_ptr kv_; + std::unique_ptr service_; + SyncPoint* sp_ = SyncPoint::get_instance(); + CommitTxnRequest request_; + std::vector rowsets_; + Metadata before_retry_; + CommitTxnResponse original_response_; + CommitTxnResponse retry_response_; + std::thread original_; + std::thread retry_; + std::mutex mutex_; + std::condition_variable cv_; + int partitions_entered_ = 0; + bool first_finished_ = false; + bool release_last_ = false; + bool before_visible_ = false; + bool release_visible_ = false; + bool retry_waiting_ = false; + bool retry_done_ = false; + bool stop_ = false; + +private: + const int32_t old_threshold_ = config::txn_lazy_commit_rowsets_thresold; + const bool old_parallel_ = config::enable_cloud_parallel_txn_lazy_commit; + const int32_t old_parallel_threads_ = config::parallel_txn_lazy_commit_num_threads; + const int32_t old_fuzzy_ = config::cloud_txn_lazy_commit_fuzzy_possibility; +}; + +TEST_F(CommittedTxnRetryTest, ImmediateRetryWaitsForVisibility) { + // 1. Lazy publishes both partitions; pause the second partition's tmp cleanup. + // 2. Retry scans its two tmp rowsets and selects immediate (threshold = 2). + // 3. Finish cleanup before retry creates its write transaction, but hold the + // global status at COMMITTED. Retry must wait rather than publish version 3. + start_partial_lazy_commit(); + ASSERT_TRUE(await([&] { return first_finished_ && partitions_entered_ == 2; })); + sp_->set_call_back("commit_txn_immediately:begin", + [&](auto&&) { finish_partition_before_retry(); }); + start_retry(); + ASSERT_TRUE(await([&] { return retry_waiting_ || retry_done_; })); + { + std::lock_guard lock(mutex_); + EXPECT_TRUE(retry_waiting_); + EXPECT_FALSE(retry_done_); + } + release_visibility(); + join_requests(); + EXPECT_FALSE(retry_response_.is_lazy_commit()); + check_visible_retry(); +} + +TEST_F(CommittedTxnRetryTest, LazyRetryWaitsForVisibility) { + config::txn_lazy_commit_rowsets_thresold = 1; + start_partial_lazy_commit(); + ASSERT_TRUE(await([&] { return first_finished_ && partitions_entered_ == 2; })); + // The same remaining two rowsets now select lazy. Remove the last pending + // marker before it reads versions, so the existing pending-wait path cannot + // mask a missing COMMITTED-state check. + sp_->set_call_back("commit_txn_eventually:begin", + [&](auto&&) { finish_partition_before_retry(); }); + sp_->set_call_back("commit_txn_eventually::before_mark_committed", [&](auto&& args) { + // Stop the unfixed path before its DCHECK aborts the entire test binary. + // Reaching this point is itself the failure: a committed retry is about + // to change commit metadata again. The immediate case reproduces the + // actual duplicate FDB writes without injecting a conflict or error. + ADD_FAILURE() << "A committed retry must not mark the transaction committed again"; + *try_any_cast(args[0]) = MetaServiceCode::TXN_INVALID_STATUS; + *try_any_cast(args.back()) = true; + }); + start_retry(); + ASSERT_TRUE(await([&] { return retry_waiting_ || retry_done_; })); + { + std::lock_guard lock(mutex_); + EXPECT_TRUE(retry_waiting_); + EXPECT_FALSE(retry_done_); + } + release_visibility(); + join_requests(); + EXPECT_TRUE(retry_response_.is_lazy_commit()); + check_visible_retry(); +} + +TEST_F(CommittedTxnRetryTest, CommittedRetryReturnsSuccessWhenLazyFails) { + sp_->set_call_back("TxnLazyCommitTask::make_committed_txn_visible::commit", [&](auto&& args) { + // Fail publication after the transaction has durably committed. Commit + // retries must preserve success while reporting incomplete lazy work. + *try_any_cast(args[0]) = MetaServiceCode::KV_TXN_COMMIT_ERR; + *try_any_cast(args.back()) = true; + }); + auto original = commit(request_); + ASSERT_EQ(original.status().code(), MetaServiceCode::OK); + ASSERT_TRUE(original.is_lazy_commit_incomplete()); + check_status(TxnStatusPB::TXN_STATUS_COMMITTED); + auto before = business_metadata(); + auto check_incomplete_retry = [&](const CommitTxnResponse& retry) { + ASSERT_EQ(retry.status().code(), MetaServiceCode::OK) << retry.ShortDebugString(); + EXPECT_EQ(retry.txn_info().txn_id(), request_.txn_id()); + EXPECT_EQ(retry.txn_info().status(), TxnStatusPB::TXN_STATUS_VISIBLE); + ASSERT_EQ(retry.txn_info().table_ids_size(), 1); + EXPECT_EQ(retry.txn_info().table_ids(0), TABLE_ID); + // FE uses both flags to recognize incomplete publication, including + // when this request originally selected the immediate handler. + EXPECT_TRUE(retry.is_lazy_commit()); + EXPECT_TRUE(retry.is_lazy_commit_incomplete()); + EXPECT_TRUE(retry.partition_ids().empty()); + EXPECT_TRUE(retry.table_ids().empty()); + EXPECT_TRUE(retry.versions().empty()); + EXPECT_TRUE(retry.table_stats().empty()); + EXPECT_EQ(business_metadata(), before); + check_status(TxnStatusPB::TXN_STATUS_COMMITTED); + }; + + // All tmp rowsets have gone, so this retry selects immediate. + request_.set_enable_txn_lazy_commit(false); + auto immediate_retry = commit(request_); + check_incomplete_retry(immediate_retry); + + // Force the other entry point despite the empty tmp scan. It must return + // the same committed-but-incomplete response when its lazy task fails. + config::cloud_txn_lazy_commit_fuzzy_possibility = 100; + request_.set_enable_txn_lazy_commit(true); + auto lazy_retry = commit(request_); + check_incomplete_retry(lazy_retry); +} + TEST(TxnLazyCommitTest, CommitTxnEventuallyWithFailedLazyCommitTaskTest) { auto txn_kv = get_mem_txn_kv(); @@ -1696,6 +2047,14 @@ TEST(TxnLazyCommitTest, FallThroughCommitTxnEventuallyTest) { ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString(); ASSERT_TRUE(commit_txn_immediatelly_hit); ASSERT_TRUE(commit_txn_eventually_finish_hit); + // The failed immediate attempt already populated partition versions. + // Falling back must return only the successful lazy attempt's results. + ASSERT_EQ(res.partition_ids_size(), 1); + ASSERT_EQ(res.table_ids_size(), 1); + ASSERT_EQ(res.versions_size(), 1); + EXPECT_EQ(res.partition_ids(0), partition_id); + EXPECT_EQ(res.table_ids(0), table_id); + EXPECT_EQ(res.versions(0), 2); } {