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) {