From 4cb0639ebcd70343b2c494f1d268c84883939021 Mon Sep 17 00:00:00 2001 From: Yixuan Wang Date: Thu, 3 Sep 2026 16:02:50 +0800 Subject: [PATCH] [fix](cloud) Avoid conflicts when recycling transactions with the same label (#65746) ### What problem does this PR solve? Concurrent recycling of transactions with the same label updates the same label key and may cause transaction conflicts. Group transactions by label and recycle each group sequentially while processing different label groups concurrently. --- cloud/src/recycler/recycler.cpp | 104 ++++++++------ cloud/test/recycler_test.cpp | 231 +++++++++++++++++++++++++++++--- 2 files changed, 280 insertions(+), 55 deletions(-) diff --git a/cloud/src/recycler/recycler.cpp b/cloud/src/recycler/recycler.cpp index 4023cd1489fab0..a7c551a74d6208 100644 --- a/cloud/src/recycler/recycler.cpp +++ b/cloud/src/recycler/recycler.cpp @@ -6706,7 +6706,7 @@ int InstanceRecycler::recycle_expired_txn_label() { std::string end_recycle_txn_key; recycle_txn_key(recycle_txn_key_info0, &begin_recycle_txn_key); recycle_txn_key(recycle_txn_key_info1, &end_recycle_txn_key); - std::vector recycle_txn_info_keys; + std::unordered_map> recycle_txn_keys_by_label; LOG_WARNING("begin to recycle expired txn").tag("instance_id", instance_id_); @@ -6747,7 +6747,17 @@ int InstanceRecycler::recycle_expired_txn_label() { current_time_ms)) { VLOG_DEBUG << "found recycle txn, key=" << hex(k); num_expired++; - recycle_txn_info_keys.emplace_back(k); + + std::string_view k1 = k; + k1.remove_prefix(1); // Remove key space + std::vector, int, int>> out; + if (decode_key(&k1, &out) != 0) { + LOG_ERROR("failed to decode key").tag("key", hex(k)); + return -1; + } + int64_t db_id = std::get(std::get<0>(out[3])); + auto label_key = txn_label_key({instance_id_, db_id, recycle_txn_pb.label()}); + recycle_txn_keys_by_label[label_key].emplace_back(k); } return 0; }; @@ -6856,35 +6866,48 @@ int InstanceRecycler::recycle_expired_txn_label() { auto loop_done = [&]() -> int { DORIS_CLOUD_DEFER { - recycle_txn_info_keys.clear(); + recycle_txn_keys_by_label.clear(); }; TEST_SYNC_POINT_CALLBACK( - "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys", - &recycle_txn_info_keys); - for (const auto& k : recycle_txn_info_keys) { - concurrent_delete_executor.add([&]() { - int ret = delete_recycle_txn_kv(k); - if (ret == 1) { - const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times); - for (int i = 1; i <= max_retry; ++i) { - LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k); - ret = delete_recycle_txn_kv(k); - // clang-format off - TEST_SYNC_POINT_CALLBACK( - "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", &ret); - // clang-format off - if (ret != 1) { - break; - } - // random sleep 0-100 ms to retry - std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100)); + "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label", + &recycle_txn_keys_by_label); + auto delete_recycle_txn_kv_with_retry = [&](const std::string& k) -> int { + int ret = delete_recycle_txn_kv(k); + TEST_SYNC_POINT_CALLBACK( + "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_error", + &ret); + if (ret == 1) { + const int max_retry = std::max(1, config::recycle_txn_delete_max_retry_times); + for (int i = 1; i <= max_retry; ++i) { + LOG(WARNING) << "txn conflict, retry times=" << i << " key=" << hex(k); + ret = delete_recycle_txn_kv(k); + TEST_SYNC_POINT_CALLBACK( + "InstanceRecycler::recycle_expired_txn_label.delete_recycle_txn_kv_" + "error", + &ret); + if (ret != 1) { + break; } + // random sleep 0-100 ms to retry + std::this_thread::sleep_for(std::chrono::milliseconds(rand() % 100)); } - if (ret != 0) { - LOG_WARNING("failed to delete recycle txn kv") - .tag("instance id", instance_id_) - .tag("key", hex(k)); - return -1; + } + return ret; + }; + + for (auto& [label_key, txn_keys] : recycle_txn_keys_by_label) { + concurrent_delete_executor.add([&, txn_keys = std::move(txn_keys), label_key]() { + VLOG_DEBUG << "recycle txn label group, key=" << hex(label_key) + << " txn_count=" << txn_keys.size(); + for (const auto& k : txn_keys) { + int ret = delete_recycle_txn_kv_with_retry(k); + if (ret != 0) { + LOG_WARNING("failed to delete recycle txn kv") + .tag("instance id", instance_id_) + .tag("key", hex(k)) + .tag("label_key", hex(label_key)); + return -1; + } } return 0; }); @@ -7646,8 +7669,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() { std::string recyc_rs_key0; std::string recyc_rs_key1; recycle_rowset_key(recyc_rs_key_info0, &recyc_rs_key0); - recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1); - int64_t earlest_ts = std::numeric_limits::max(); + recycle_rowset_key(recyc_rs_key_info1, &recyc_rs_key1); + int64_t earlest_ts = std::numeric_limits::max(); auto handle_rowset_kv = [&, this](std::string_view k, std::string_view v) -> int { RecycleRowsetPB rowset; @@ -7671,7 +7694,8 @@ int InstanceRecycler::scan_and_statistics_rowsets() { metrics_context.total_need_recycle_num++; metrics_context.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size(); segment_metrics_context_.total_need_recycle_num += rowset.rowset_meta().num_segments(); - segment_metrics_context_.total_need_recycle_data_size += rowset.rowset_meta().total_disk_size(); + segment_metrics_context_.total_need_recycle_data_size += + rowset.rowset_meta().total_disk_size(); return 0; } @@ -7724,7 +7748,7 @@ int InstanceRecycler::scan_and_statistics_tmp_rowsets() { DCHECK_GT(rowset.txn_id(), 0) << "txn_id=" << rowset.txn_id() << " rowset=" << rowset.ShortDebugString(); - if(!rowset.has_is_recycled() || !rowset.is_recycled()) { + if (!rowset.has_is_recycled() || !rowset.is_recycled()) { return 0; } @@ -7801,7 +7825,8 @@ int InstanceRecycler::scan_and_statistics_abort_timeout_txn() { return 0; }; - int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, std::move(handle_abort_timeout_txn_kv)); + int ret = scan_and_recycle(begin_txn_running_key, end_txn_running_key, + std::move(handle_abort_timeout_txn_kv)); metrics_context.report(true); return ret; } @@ -7835,7 +7860,8 @@ int InstanceRecycler::scan_and_statistics_expired_txn_label() { return 0; }; - int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, std::move(handle_expired_txn_label_kv)); + int ret = scan_and_recycle(begin_recycle_txn_key, end_recycle_txn_key, + std::move(handle_expired_txn_label_kv)); metrics_context.report(true); return ret; } @@ -8066,7 +8092,7 @@ int InstanceRecycler::scan_and_statistics_restore_jobs() { return 0; } metrics_context.total_need_recycle_num++; - if(restore_job_pb.need_recycle_data()) { + if (restore_job_pb.need_recycle_data()) { scan_tablet_and_statistics(restore_job_pb.tablet_id(), metrics_context); } return 0; @@ -8108,7 +8134,7 @@ void InstanceRecycler::scan_and_statistics_operation_logs() { OperationLogReferenceInfo ref_info; if (recycle_checker.can_recycle(log_versionstamp, operation_log.min_timestamp(), - &ref_info)) { + &ref_info)) { metrics_context.total_need_recycle_num++; metrics_context.total_need_recycle_data_size += operation_log.ByteSizeLong(); } @@ -8264,8 +8290,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector::max(), std::numeric_limits::max()}); + {reference_instance_id, tablet_id, rowset_id, std::numeric_limits::max(), + std::numeric_limits::max()}); txn->remove(dbm_start_key, dbm_end_key); LOG_INFO("remove delete bitmap kv in cleanup phase") .tag("instance_id", instance_id_) @@ -8288,8 +8314,8 @@ int InstanceRecycler::cleanup_rowset_metadata(const std::vector( - txn.get(), task.versioned_rowset_key, task.versionstamp); + versioned::document_remove(txn.get(), task.versioned_rowset_key, + task.versionstamp); LOG_INFO("remove versioned meta rowset key in cleanup phase") .tag("instance_id", instance_id_) .tag("tablet_id", tablet_id) diff --git a/cloud/test/recycler_test.cpp b/cloud/test/recycler_test.cpp index d23264c0fbea13..56a736f048dc4c 100644 --- a/cloud/test/recycler_test.cpp +++ b/cloud/test/recycler_test.cpp @@ -9233,7 +9233,9 @@ void make_single_txn_related_kvs(std::shared_ptr txn_kv, int64_t i } else { recycle_txn_pb.set_creation_time(current_time); } - recycle_txn_pb.set_label("recycle_txn_key_info_label_" + std::to_string(i)); + // Production writes RecycleTxnPB.label and TxnInfoPB.label from the same transaction label. + const std::string label = "txn_label_" + std::to_string(i); + recycle_txn_pb.set_label(label); if (!recycle_txn_pb.SerializeToString(&recycle_txn_info_val)) { LOG_WARNING("failed to serialize recycle txn info") .tag("key", hex(recycle_txn_info_key)) @@ -9263,7 +9265,7 @@ void make_single_txn_related_kvs(std::shared_ptr txn_kv, int64_t i std::string info_val; TxnInfoPB txn_info_pb; txn_info_pb.add_sub_txn_ids(sub_txn_id); - txn_info_pb.set_label("txn_info_label_" + std::to_string(i)); + txn_info_pb.set_label(label); if (!txn_info_pb.SerializeToString(&info_val)) { LOG_WARNING("failed to serialize txn info") .tag("key", hex(info_key)) @@ -9489,20 +9491,24 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) { auto txn_kv = mem_txn_kv; ASSERT_TRUE(txn_kv.get()) << "exit get MemTxnKv error" << std::endl; - make_multiple_txn_info_kvs(txn_kv, 20000, 15000); - check_multiple_txn_info_kvs(txn_kv, 20000); + make_multiple_txn_info_kvs(txn_kv, 40000, 30000); + check_multiple_txn_info_kvs(txn_kv, 40000); auto* sp = SyncPoint::get_instance(); DORIS_CLOUD_DEFER { SyncPoint::get_instance()->clear_all_call_backs(); }; - size_t recycle_txn_info_keys_cnt = 0; - sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_info_keys", - [&](auto&& args) { - auto* recycle_txn_info_keys = - try_any_cast*>(args[0]); - recycle_txn_info_keys_cnt += recycle_txn_info_keys->size(); - }); + size_t recycle_txn_keys_cnt = 0; + sp->set_call_back( + "InstanceRecycler::recycle_expired_txn_label.check_recycle_txn_keys_by_label", + [&](auto&& args) { + auto* recycle_txn_keys_by_label = + try_any_cast>*>( + args[0]); + for (const auto& entry : *recycle_txn_keys_by_label) { + recycle_txn_keys_cnt += entry.second.size(); + } + }); sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.failure", [](auto&& args) { auto* ret = try_any_cast(args[0]); *ret = -1; @@ -9520,7 +9526,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_failure_test) { std::cout << "recycle expired txn label cost=" << std::chrono::duration_cast(finish - start).count() << "ms" << std::endl; - check_multiple_txn_info_kvs(txn_kv, (20000 - recycle_txn_info_keys_cnt)); + check_multiple_txn_info_kvs(txn_kv, (40000 - recycle_txn_keys_cnt)); } TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { config::label_keep_max_second = 0; @@ -9671,7 +9677,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { std::cout << "Update label after count: " << update_label_after_count << std::endl; std::cout << "Transaction conflict count: " << txn_conflict_count << std::endl; - EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered"; + EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group"; std::unique_ptr verify_txn; ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); @@ -9699,7 +9705,7 @@ TEST(RecyclerTest, concurrent_recycle_txn_label_conflict_test) { } } -TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) { +TEST(RecyclerTest, recycle_txn_label_propagate_delete_error_test) { config::label_keep_max_second = 0; config::recycle_pool_parallelism = 20; @@ -9845,10 +9851,203 @@ TEST(RecyclerTest, recycle_txn_label_deal_with_conflict_error_test) { std::make_shared(mem_txn_kv)); ASSERT_EQ(recycler.init(), 0); - // deal with conflict but error during recycle + // Propagate a recycle error without relying on an internal label conflict. + ASSERT_EQ(recycler.recycle_expired_txn_label(), -1); + + EXPECT_EQ(txn_conflict_count, 0) << "txn conflicts should not occur within one label group"; +} + +TEST(RecyclerTest, recycle_txn_label_retry_after_conflict_test) { + config::label_keep_max_second = 0; + + auto txn_kv = std::dynamic_pointer_cast(std::make_shared()); + ASSERT_NE(txn_kv.get(), nullptr); + ASSERT_EQ(txn_kv->init(), 0); + auto resource_mgr = std::make_shared(txn_kv); + auto rate_limiter = std::make_shared(); + auto snapshot = std::make_shared(txn_kv); + auto meta_service = + std::make_unique(txn_kv, resource_mgr, rate_limiter, snapshot); + + constexpr int64_t db_id = 10001; + constexpr int64_t table_id = 20001; + const std::string cloud_unique_id = "recycle_txn_label_retry_after_conflict_test"; + const std::string label = "recycle_txn_label_retry_after_conflict_test"; + + int64_t recycled_txn_id = -1; + { + brpc::Controller cntl; + BeginTxnRequest req; + BeginTxnResponse res; + req.set_cloud_unique_id(cloud_unique_id); + auto* txn_info = req.mutable_txn_info(); + txn_info->set_db_id(db_id); + txn_info->set_label(label); + txn_info->add_table_ids(table_id); + txn_info->set_timeout_ms(36000); + meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, + &res, nullptr); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString(); + ASSERT_TRUE(res.has_txn_id()); + recycled_txn_id = res.txn_id(); + } + { + brpc::Controller cntl; + AbortTxnRequest req; + AbortTxnResponse res; + req.set_cloud_unique_id(cloud_unique_id); + req.set_db_id(db_id); + req.set_txn_id(recycled_txn_id); + req.set_reason("test"); + meta_service->abort_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, + &res, nullptr); + ASSERT_EQ(res.status().code(), MetaServiceCode::OK) << res.ShortDebugString(); + } + + auto* sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + SyncPoint::get_instance()->disable_processing(); + }; + + std::atomic before_commit_count {0}; + std::atomic txn_conflict_count {0}; + std::atomic external_begin_code {-1}; + std::atomic new_txn_id {-1}; + sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.before_commit", [&](auto&&) { + if (before_commit_count.fetch_add(1) != 0) { + return; + } + + brpc::Controller cntl; + BeginTxnRequest req; + BeginTxnResponse res; + req.set_cloud_unique_id(cloud_unique_id); + auto* txn_info = req.mutable_txn_info(); + txn_info->set_db_id(db_id); + txn_info->set_label(label); + txn_info->add_table_ids(table_id); + txn_info->set_timeout_ms(36000); + meta_service->begin_txn(reinterpret_cast<::google::protobuf::RpcController*>(&cntl), &req, + &res, nullptr); + external_begin_code.store(static_cast(res.status().code())); + if (res.has_txn_id()) { + new_txn_id.store(res.txn_id()); + } + }); + sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.txn_conflict", + [&](auto&&) { txn_conflict_count.fetch_add(1); }); + sp->enable_processing(); + + InstanceInfoPB instance; + instance.set_instance_id(mock_instance); + InstanceRecycler recycler(txn_kv, instance, thread_group, + std::make_shared(txn_kv)); + ASSERT_EQ(recycler.init(), 0); + + ASSERT_EQ(recycler.recycle_expired_txn_label(), 0); + EXPECT_EQ(external_begin_code.load(), static_cast(MetaServiceCode::OK)); + EXPECT_GT(new_txn_id.load(), 0); + EXPECT_EQ(txn_conflict_count.load(), 1); + EXPECT_EQ(before_commit_count.load(), 2); + + std::unique_ptr verify_txn; + ASSERT_EQ(txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); + const std::string recycle_key = recycle_txn_key({mock_instance, db_id, recycled_txn_id}); + std::string recycle_value; + EXPECT_EQ(verify_txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_KEY_NOT_FOUND); + + const std::string label_key = txn_label_key({mock_instance, db_id, label}); + std::string label_value; + ASSERT_EQ(verify_txn->get(label_key, &label_value), TxnErrorCode::TXN_OK); + TxnLabelPB txn_label; + ASSERT_TRUE( + txn_label.ParseFromArray(label_value.data(), label_value.size() - VERSION_STAMP_LEN)); + ASSERT_EQ(txn_label.txn_ids_size(), 1); + EXPECT_EQ(txn_label.txn_ids(0), new_txn_id.load()); + + std::string new_info_value; + ASSERT_EQ(verify_txn->get(txn_info_key({mock_instance, db_id, new_txn_id.load()}), + &new_info_value), + TxnErrorCode::TXN_OK); + TxnInfoPB new_txn_info; + ASSERT_TRUE(new_txn_info.ParseFromString(new_info_value)); + EXPECT_EQ(new_txn_info.label(), label); +} + +TEST(RecyclerTest, recycle_txn_label_retry_exhausted_then_recover_test) { + const int old_max_retry_times = config::recycle_txn_delete_max_retry_times; + DORIS_CLOUD_DEFER { + config::recycle_txn_delete_max_retry_times = old_max_retry_times; + }; + config::label_keep_max_second = 0; + config::recycle_txn_delete_max_retry_times = 2; + + auto mem_txn_kv = std::make_shared(); + ASSERT_EQ(mem_txn_kv->init(), 0); + make_single_txn_related_kvs(mem_txn_kv, 0, 1); + + const std::string recycle_key = recycle_txn_key({instance_id, 0, 1000000}); + const std::string label_key = txn_label_key({instance_id, 0, "txn_label_0"}); + + auto* sp = SyncPoint::get_instance(); + DORIS_CLOUD_DEFER { + SyncPoint::get_instance()->clear_all_call_backs(); + SyncPoint::get_instance()->disable_processing(); + }; + + std::atomic external_write_count {0}; + std::atomic external_write_error_count {0}; + std::atomic txn_conflict_count {0}; + sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.before_commit", [&](auto&&) { + std::unique_ptr txn; + if (mem_txn_kv->create_txn(&txn) != TxnErrorCode::TXN_OK) { + external_write_error_count.fetch_add(1); + return; + } + std::string label_value; + if (txn->get(label_key, &label_value) != TxnErrorCode::TXN_OK) { + external_write_error_count.fetch_add(1); + return; + } + txn->put(label_key, label_value); + if (txn->commit() != TxnErrorCode::TXN_OK) { + external_write_error_count.fetch_add(1); + return; + } + external_write_count.fetch_add(1); + }); + sp->set_call_back("InstanceRecycler::recycle_expired_txn_label.txn_conflict", + [&](auto&&) { txn_conflict_count.fetch_add(1); }); + sp->enable_processing(); + + InstanceInfoPB instance; + instance.set_instance_id(instance_id); + InstanceRecycler recycler(mem_txn_kv, instance, thread_group, + std::make_shared(mem_txn_kv)); + ASSERT_EQ(recycler.init(), 0); + ASSERT_EQ(recycler.recycle_expired_txn_label(), -1); + EXPECT_EQ(external_write_error_count.load(), 0); + EXPECT_EQ(external_write_count.load(), 3); + EXPECT_EQ(txn_conflict_count.load(), 3); - EXPECT_GT(txn_conflict_count, 0) << "txn_conflict sync point should be triggered"; + { + std::unique_ptr verify_txn; + ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); + std::string recycle_value; + EXPECT_EQ(verify_txn->get(recycle_key, &recycle_value), TxnErrorCode::TXN_OK); + } + + sp->clear_all_call_backs(); + sp->disable_processing(); + ASSERT_EQ(recycler.recycle_expired_txn_label(), 0); + + std::unique_ptr verify_txn; + ASSERT_EQ(mem_txn_kv->create_txn(&verify_txn), TxnErrorCode::TXN_OK); + std::string value; + EXPECT_EQ(verify_txn->get(recycle_key, &value), TxnErrorCode::TXN_KEY_NOT_FOUND); + EXPECT_EQ(verify_txn->get(label_key, &value), TxnErrorCode::TXN_KEY_NOT_FOUND); } TEST(RecyclerTest, recycle_restore_job_complete_state) {