From d342ebe8489808afb3fb774535b4ee3b0739ecc2 Mon Sep 17 00:00:00 2001 From: zyguan Date: Wed, 23 Sep 2026 04:38:12 +0000 Subject: [PATCH] feat: protect transaction RPC compatibility Signed-off-by: zyguan --- include/pingcap/Config.h | 3 + include/pingcap/Exception.h | 40 ++++- include/pingcap/coprocessor/Client.h | 7 + include/pingcap/kv/Cluster.h | 18 +- include/pingcap/kv/RegionCache.h | 9 +- include/pingcap/kv/RegionClient.h | 104 ++++++++++- include/pingcap/kv/Rpc.h | 12 +- include/pingcap/kv/internal/terminal_error.h | 50 ++++++ include/pingcap/kv/internal/txn_protocol.h | 79 +++++++++ src/coprocessor/Client.cc | 8 +- src/kv/Backoff.cc | 2 +- src/kv/LockResolver.cc | 18 ++ src/kv/RegionCache.cc | 13 +- src/kv/RegionClient.cc | 2 + src/test/CMakeLists.txt | 2 + src/test/terminal_error_test.cc | 177 +++++++++++++++++++ src/test/txn_protocol_test.cc | 89 ++++++++++ third_party/kvproto | 2 +- 18 files changed, 621 insertions(+), 14 deletions(-) create mode 100644 include/pingcap/kv/internal/terminal_error.h create mode 100644 include/pingcap/kv/internal/txn_protocol.h create mode 100644 src/test/terminal_error_test.cc create mode 100644 src/test/txn_protocol_test.cc diff --git a/include/pingcap/Config.h b/include/pingcap/Config.h index cae92cc1..a97784bd 100644 --- a/include/pingcap/Config.h +++ b/include/pingcap/Config.h @@ -9,6 +9,7 @@ #include +#include #include #include #include @@ -23,6 +24,8 @@ struct ClusterConfig std::string cert_path; std::string key_path; ::kvrpcpb::APIVersion api_version = ::kvrpcpb::APIVersion::V1; + ::kvrpcpb::RequestOrigin request_origin = ::kvrpcpb::RequestOriginUnknown; + uint32_t default_txn_protocol_version = ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING; ClusterConfig() = default; diff --git a/include/pingcap/Exception.h b/include/pingcap/Exception.h index 0fa278e3..549655fd 100644 --- a/include/pingcap/Exception.h +++ b/include/pingcap/Exception.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -29,7 +30,9 @@ enum ErrorCodes : int KeyspaceNotEnabled = 18, InternalError = 19, GRPCNotImplemented = 20, - UnknownError = 21 + UnknownError = 21, + IncompatibleRequest = 22, + UndeterminedResult = 23, }; class Exception : public Poco::Exception @@ -55,6 +58,41 @@ class Exception : public Poco::Exception bool empty() const { return code() == 0 && message().empty(); } }; +class ErrIncompatibleRequest : public Exception +{ +public: + explicit ErrIncompatibleRequest(const ::errorpb::IncompatibleRequest & error) + : Exception(error.message(), ErrorCodes::IncompatibleRequest) + , error_(error) + {} + + ErrIncompatibleRequest(const std::string & message, const ::errorpb::IncompatibleRequest & error) + : Exception(message, ErrorCodes::IncompatibleRequest) + , error_(error) + {} + + const ::errorpb::IncompatibleRequest & error() const { return error_; } + + ErrIncompatibleRequest * clone() const override { return new ErrIncompatibleRequest(*this); } + void rethrow() const override { throw *this; } + +private: + ::errorpb::IncompatibleRequest error_; +}; + +inline bool isTerminalTransactionError(const Exception & exception) +{ + return exception.code() == ErrorCodes::IncompatibleRequest || exception.code() == ErrorCodes::UndeterminedResult; +} + +inline void rethrowTerminalRegionError(const ::errorpb::Error & error) +{ + if (error.has_undetermined_result()) + throw Exception(error.undetermined_result().message(), UndeterminedResult); + if (error.has_incompatible_request()) + throw ErrIncompatibleRequest(error.incompatible_request()); +} + inline std::string getCurrentExceptionMsg(const std::string & prefix_msg) { std::string msg = prefix_msg; diff --git a/include/pingcap/coprocessor/Client.h b/include/pingcap/coprocessor/Client.h index 945d6810..7b53a4de 100644 --- a/include/pingcap/coprocessor/Client.h +++ b/include/pingcap/coprocessor/Client.h @@ -7,6 +7,7 @@ #include #include #include +#include #include #include @@ -113,6 +114,9 @@ class ResponseIter std::shared_ptr<::coprocessor::Response> resp; bool same_zone{true}; Exception error; + // Preserve the dynamic exception and, for ErrIncompatibleRequest, its + // complete structured protobuf across the asynchronous queue boundary. + std::shared_ptr detailed_error; bool finished{false}; Result() = default; @@ -121,6 +125,7 @@ class ResponseIter {} explicit Result(const Exception & err) : error(err) + , detailed_error(err.clone()) {} explicit Result(bool finished_) : finished(finished_) @@ -131,6 +136,8 @@ class ResponseIter {} const std::string & data() const { return resp->data(); } + + const Exception * exception() const { return detailed_error.get(); } }; ResponseIter(std::unique_ptr> && queue_, diff --git a/include/pingcap/kv/Cluster.h b/include/pingcap/kv/Cluster.h index fa6d012f..abcc0ce0 100644 --- a/include/pingcap/kv/Cluster.h +++ b/include/pingcap/kv/Cluster.h @@ -30,6 +30,8 @@ struct Cluster LockResolverPtr lock_resolver; ::kvrpcpb::APIVersion api_version = ::kvrpcpb::APIVersion::V1; + const ::kvrpcpb::RequestOrigin request_origin; + const uint32_t default_txn_protocol_version; std::unique_ptr thread_pool; std::unique_ptr mpp_prober; @@ -39,6 +41,8 @@ struct Cluster , rpc_client(std::make_unique(pd_client, ClusterConfig{})) , oracle(std::make_unique(pd_client, std::chrono::milliseconds(oracle_update_interval))) , lock_resolver(std::make_unique(this)) + , request_origin(ClusterConfig{}.request_origin) + , default_txn_protocol_version(ClusterConfig{}.default_txn_protocol_version) , thread_pool(std::make_unique(mock_cluster_background_workers)) , mpp_prober(std::make_unique(this)) { @@ -46,12 +50,14 @@ struct Cluster } Cluster(const std::vector & pd_addrs, const ClusterConfig & config) - : pd_client(std::make_shared(pd_addrs, config)) + : pd_client(std::make_shared(pd_addrs, validateCompatibilityConfig(config))) , region_cache(std::make_unique(pd_client, config)) , rpc_client(std::make_unique(pd_client, config)) , oracle(std::make_unique(pd_client, std::chrono::milliseconds(oracle_update_interval))) , lock_resolver(std::make_unique(this)) , api_version(config.api_version) + , request_origin(config.request_origin) + , default_txn_protocol_version(config.default_txn_protocol_version) , thread_pool(std::make_unique(cluster_background_workers)) , mpp_prober(std::make_unique(this)) { @@ -60,6 +66,8 @@ struct Cluster void update(const std::vector & pd_addrs, const ClusterConfig & config) const { + if (config.request_origin != request_origin || config.default_txn_protocol_version != default_txn_protocol_version) + throw Exception("request origin and transaction protocol version are immutable after Cluster creation", LogicalError); pd_client->update(pd_addrs, config); rpc_client->update(config); } @@ -81,6 +89,14 @@ struct Cluster void splitRegion(const std::string & split_key); void startBackgroundTasks(); + +private: + static const ClusterConfig & validateCompatibilityConfig(const ClusterConfig & config) + { + if (config.default_txn_protocol_version > ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING) + throw Exception("default transaction protocol version must be legacy or incompatible-error-handling", LogicalError); + return config; + } }; struct MinCommitTSPushed diff --git a/include/pingcap/kv/RegionCache.h b/include/pingcap/kv/RegionCache.h index 10d33e57..7414535a 100644 --- a/include/pingcap/kv/RegionCache.h +++ b/include/pingcap/kv/RegionCache.h @@ -9,6 +9,7 @@ #include #include +#include #include #include @@ -30,14 +31,20 @@ struct Store std::map labels; StoreType store_type; ::metapb::StoreState state; + bool has_txn_protocol_version_range; + uint32_t txn_protocol_version_min; + uint32_t txn_protocol_version_max; - Store(uint64_t id_, const std::string & addr_, const std::string & peer_addr_, const std::map & labels_, StoreType store_type_, const ::metapb::StoreState state_) + Store(uint64_t id_, const std::string & addr_, const std::string & peer_addr_, const std::map & labels_, StoreType store_type_, const ::metapb::StoreState state_, bool has_txn_protocol_version_range_ = false, uint32_t txn_protocol_version_min_ = 0, uint32_t txn_protocol_version_max_ = 0) : id(id_) , addr(addr_) , peer_addr(peer_addr_) , labels(labels_) , store_type(store_type_) , state(state_) + , has_txn_protocol_version_range(has_txn_protocol_version_range_) + , txn_protocol_version_min(txn_protocol_version_min_) + , txn_protocol_version_max(txn_protocol_version_max_) {} }; diff --git a/include/pingcap/kv/RegionClient.h b/include/pingcap/kv/RegionClient.h index 4ba701de..bb3d2c9e 100644 --- a/include/pingcap/kv/RegionClient.h +++ b/include/pingcap/kv/RegionClient.h @@ -4,6 +4,7 @@ #include #include #include +#include namespace pingcap { @@ -53,9 +54,15 @@ struct RegionClient { throw Exception("should setup proper label_filter for tiflash"); } + bool compatibility_resend_used = false; + RPCContextPtr compatibility_resend_ctx; + uint32_t compatibility_resend_version = 0; for (;;) { - RPCContextPtr ctx = cluster->region_cache->getRPCContext(bo, region_id, store_type, /*load_balance=*/true, tiflash_label_filter, store_id_blocklist, prefer_store_id); + const bool is_compatibility_resend = compatibility_resend_ctx != nullptr; + RPCContextPtr ctx = is_compatibility_resend + ? compatibility_resend_ctx + : cluster->region_cache->getRPCContext(bo, region_id, store_type, /*load_balance=*/true, tiflash_label_filter, store_id_blocklist, prefer_store_id); if (ctx == nullptr) { // If the region is not found in cache, it must be out @@ -64,8 +71,19 @@ struct RegionClient auto s = store_id_blocklist != nullptr ? ", store_filter_size=" + std::to_string(store_id_blocklist->size()) + "." : std::string("."); throw Exception("Region epoch not match after retries: Region " + region_id.toString() + " not in region cache" + s, RegionEpochNotMatch); } + auto selection = internal::selectTxnProtocolVersion( + req, cluster->default_txn_protocol_version, ctx->store.txn_protocol_version_min, ctx->store.txn_protocol_version_max); + if (is_compatibility_resend) + { + selection.selected = compatibility_resend_version; + selection.allowed = !selection.protected_request || selection.required <= selection.selected; + compatibility_resend_ctx.reset(); + } + if (!selection.allowed) + throw localIncompatibleRequest(ctx, selection); + RpcCall rpc(cluster->rpc_client, ctx->addr); - rpc.setRequestCtx(req, ctx, cluster->api_version); + rpc.setRequestCtx(req, ctx, cluster->api_version, cluster->request_origin, selection.selected); grpc::ClientContext context; rpc.setClientContext(context, timeout, meta_data); @@ -87,6 +105,29 @@ struct RegionClient if (resp->has_region_error()) { log->warning("region_id " + region_id.toString() + " find error: " + resp->region_error().DebugString()); + const auto & error = resp->region_error(); + if (error.has_undetermined_result()) + throw Exception(error.undetermined_result().message(), UndeterminedResult); + if (error.has_incompatible_request()) + { + const auto & incompatible = error.incompatible_request(); + if (!compatibility_resend_used && !is_compatibility_resend + && internal::isValidUpperBoundRejection(incompatible, selection.selected)) + { + auto updated = internal::selectTxnProtocolVersion( + req, + cluster->default_txn_protocol_version, + incompatible.min_compatible_txn_protocol_version(), + incompatible.max_compatible_txn_protocol_version()); + if (updated.allowed && updated.selected != selection.selected) + { + compatibility_resend_used = true; + compatibility_resend_ctx = ctx; + compatibility_resend_version = updated.selected; + continue; + } + } + } onRegionError(bo, ctx, resp->region_error()); continue; } @@ -153,9 +194,15 @@ struct RegionClient { throw Exception("should setup proper label_filter for tiflash"); } + bool compatibility_resend_used = false; + RPCContextPtr compatibility_resend_ctx; + uint32_t compatibility_resend_version = 0; for (;;) { - RPCContextPtr ctx = cluster->region_cache->getRPCContext(bo, region_id, store_type, /*load_balance=*/true, tiflash_label_filter, store_id_blocklist, prefer_store_id); + const bool is_compatibility_resend = compatibility_resend_ctx != nullptr; + RPCContextPtr ctx = is_compatibility_resend + ? compatibility_resend_ctx + : cluster->region_cache->getRPCContext(bo, region_id, store_type, /*load_balance=*/true, tiflash_label_filter, store_id_blocklist, prefer_store_id); if (ctx == nullptr) { // If the region is not found in cache, it must be out @@ -164,9 +211,20 @@ struct RegionClient throw Exception("Region epoch not match after retries: Region " + region_id.toString() + " not in region cache.", RegionEpochNotMatch); } + auto selection = internal::selectTxnProtocolVersion( + req, cluster->default_txn_protocol_version, ctx->store.txn_protocol_version_min, ctx->store.txn_protocol_version_max); + if (is_compatibility_resend) + { + selection.selected = compatibility_resend_version; + selection.allowed = !selection.protected_request || selection.required <= selection.selected; + compatibility_resend_ctx.reset(); + } + if (!selection.allowed) + throw localIncompatibleRequest(ctx, selection); + auto stream_reader = std::make_unique>(); RpcCall rpc(cluster->rpc_client, ctx->addr); - rpc.setRequestCtx(req, ctx, cluster->api_version); + rpc.setRequestCtx(req, ctx, cluster->api_version, cluster->request_origin, selection.selected); rpc.setClientContext(stream_reader->context, timeout, meta_data); stream_reader->reader = rpc.call(&stream_reader->context, req); @@ -175,6 +233,29 @@ struct RegionClient if (stream_reader->first_resp.has_region_error()) { log->warning("region_id " + region_id.toString() + " find error: " + stream_reader->first_resp.region_error().message()); + const auto & error = stream_reader->first_resp.region_error(); + if (error.has_undetermined_result()) + throw Exception(error.undetermined_result().message(), UndeterminedResult); + if (error.has_incompatible_request()) + { + const auto & incompatible = error.incompatible_request(); + if (!compatibility_resend_used && !is_compatibility_resend + && internal::isValidUpperBoundRejection(incompatible, selection.selected)) + { + auto updated = internal::selectTxnProtocolVersion( + req, + cluster->default_txn_protocol_version, + incompatible.min_compatible_txn_protocol_version(), + incompatible.max_compatible_txn_protocol_version()); + if (updated.allowed && updated.selected != selection.selected) + { + compatibility_resend_used = true; + compatibility_resend_ctx = ctx; + compatibility_resend_version = updated.selected; + continue; + } + } + } onRegionError(bo, ctx, stream_reader->first_resp.region_error()); continue; } @@ -207,6 +288,21 @@ struct RegionClient } protected: + static ErrIncompatibleRequest localIncompatibleRequest(const RPCContextPtr & ctx, const internal::TxnProtocolSelection & selection) + { + ::errorpb::IncompatibleRequest error; + error.set_reason(::errorpb::IncompatibleRequestReasonUnknown); + error.set_min_compatible_txn_protocol_version(ctx->store.txn_protocol_version_min); + error.set_max_compatible_txn_protocol_version(ctx->store.txn_protocol_version_max); + error.set_provided_txn_protocol_version(selection.selected); + const auto message = "transaction protocol is incompatible with store " + std::to_string(ctx->store.id) + " range [" + + std::to_string(ctx->store.txn_protocol_version_min) + "," + std::to_string(ctx->store.txn_protocol_version_max) + + "], process ceiling " + std::to_string(selection.process_ceiling) + ", candidate " + std::to_string(selection.selected) + + ", required " + std::to_string(selection.required); + error.set_message(message); + return ErrIncompatibleRequest(message, error); + } + void onRegionError(Backoffer & bo, RPCContextPtr rpc_ctx, const errorpb::Error & err) const; // Normally, it happens when machine down or network partition between tidb and kv or process crash. diff --git a/include/pingcap/kv/Rpc.h b/include/pingcap/kv/Rpc.h index 6dd30fe5..9b6ef93a 100644 --- a/include/pingcap/kv/Rpc.h +++ b/include/pingcap/kv/Rpc.h @@ -7,6 +7,7 @@ #include #include +#include #include #include #include @@ -103,7 +104,11 @@ class RpcCall {} template - void setRequestCtx(REQ & req, RPCContextPtr rpc_ctx, kvrpcpb::APIVersion api_version) + void setRequestCtx(REQ & req, + RPCContextPtr rpc_ctx, + kvrpcpb::APIVersion api_version, + ::kvrpcpb::RequestOrigin default_origin, + uint32_t txn_protocol_version) { ::kvrpcpb::Context * context = req.mutable_context(); // Set api_version to this context, it's caller's duty to ensure the api_version. @@ -113,6 +118,11 @@ class RpcCall context->set_allocated_region_epoch(new metapb::RegionEpoch(rpc_ctx->meta.region_epoch())); context->set_allocated_peer(new metapb::Peer(rpc_ctx->peer)); context->set_cluster_id(rpc_ctx->cluster_id); + if (context->request_origin() == ::kvrpcpb::RequestOriginUnknown) + context->set_request_origin(default_origin); + // This is a capability declaration selected for this physical send. Never + // retain a version supplied by a logical request creator. + context->set_txn_protocol_version(txn_protocol_version); } void setClientContext(::grpc::ClientContext & context, int timeout, const GRPCMetaData & meta_data = {}) diff --git a/include/pingcap/kv/internal/terminal_error.h b/include/pingcap/kv/internal/terminal_error.h new file mode 100644 index 00000000..896d3597 --- /dev/null +++ b/include/pingcap/kv/internal/terminal_error.h @@ -0,0 +1,50 @@ +#pragma once + +#include + +#include +#include + +namespace pingcap::kv::internal +{ +// Collect terminal errors from concurrent workers without allowing completion +// order to alter the externally visible transaction outcome. +class TerminalErrorCollector +{ +public: + void capture(const Exception & exception) + { + if (!isTerminalTransactionError(exception)) + return; + + const auto captured = std::current_exception(); + if (!captured) + return; + + std::lock_guard lock(mutex); + auto & destination = exception.code() == ErrorCodes::UndeterminedResult ? undetermined : incompatible; + if (!destination) + destination = captured; + } + + void rethrowIfPresent() const + { + std::exception_ptr undetermined_snapshot; + std::exception_ptr incompatible_snapshot; + { + std::lock_guard lock(mutex); + undetermined_snapshot = undetermined; + incompatible_snapshot = incompatible; + } + if (undetermined_snapshot) + std::rethrow_exception(undetermined_snapshot); + if (incompatible_snapshot) + std::rethrow_exception(incompatible_snapshot); + } + +private: + mutable std::mutex mutex; + std::exception_ptr undetermined; + std::exception_ptr incompatible; +}; +} // namespace pingcap::kv::internal diff --git a/include/pingcap/kv/internal/txn_protocol.h b/include/pingcap/kv/internal/txn_protocol.h new file mode 100644 index 00000000..098e0ae3 --- /dev/null +++ b/include/pingcap/kv/internal/txn_protocol.h @@ -0,0 +1,79 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include + +namespace pingcap::kv::internal +{ +struct TxnProtocolSelection +{ + bool protected_request = false; + uint32_t required = ::kvrpcpb::TXN_VER_LEGACY; + uint32_t process_ceiling = ::kvrpcpb::TXN_VER_LEGACY; + uint32_t selected = ::kvrpcpb::TXN_VER_LEGACY; + bool allowed = true; +}; + +template +constexpr bool isProtectedTxnRequest() +{ + return std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v + || std::is_same_v || std::is_same_v; +} + +template +uint32_t requiredTxnProtocolVersion(const REQ & req) +{ + if constexpr (std::is_same_v) + { + for (const auto & mutation : req.mutations()) + { + if (mutation.op() == ::kvrpcpb::SharedLock) + return ::kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK; + } + } + else if constexpr (std::is_same_v) + { + for (const auto & mutation : req.mutations()) + { + if (mutation.op() == ::kvrpcpb::SharedPessimisticLock) + return ::kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK; + } + } + return ::kvrpcpb::TXN_VER_LEGACY; +} + +template +TxnProtocolSelection selectTxnProtocolVersion(const REQ & req, uint32_t process_ceiling, uint32_t store_min, uint32_t store_max) +{ + TxnProtocolSelection result; + result.protected_request = isProtectedTxnRequest(); + result.process_ceiling = process_ceiling; + if (!result.protected_request) + return result; + + result.required = requiredTxnProtocolVersion(req); + result.selected = std::min(result.process_ceiling, store_max); + result.allowed = store_min <= store_max && result.selected >= store_min && result.selected >= result.required; + return result; +} + +inline bool isValidUpperBoundRejection(const ::errorpb::IncompatibleRequest & err, uint32_t selected) +{ + return err.reason() == ::errorpb::IncompatibleRequestReasonTxnProtocolVersionOutOfRange + && err.min_compatible_txn_protocol_version() <= err.max_compatible_txn_protocol_version() + && err.provided_txn_protocol_version() == selected && selected > err.max_compatible_txn_protocol_version(); +} +} // namespace pingcap::kv::internal diff --git a/src/coprocessor/Client.cc b/src/coprocessor/Client.cc index 92419005..63110b99 100644 --- a/src/coprocessor/Client.cc +++ b/src/coprocessor/Client.cc @@ -791,6 +791,11 @@ std::vector ResponseIter::handleTaskImpl(kv::Backoffer & bo, const CopT resp = std::make_shared<::coprocessor::Response>(); if (!reader->read(resp.get())) break; + if (resp->has_region_error()) + { + rethrowTerminalRegionError(resp->region_error()); + throw Exception("Coprocessor stream subsequent response has a region error: " + resp->region_error().message(), ErrorCodes::CoprocessorError); + } if (is_first_resp) { is_first_resp = false; @@ -806,9 +811,6 @@ std::vector ResponseIter::handleTaskImpl(kv::Backoffer & bo, const CopT throw Exception("Coprocessor stream subsequent response has a lock error", ErrorCodes::CoprocessorError); } - if (resp->has_region_error()) - throw Exception("Coprocessor stream subsequent response has a region error: " + resp->region_error().message(), ErrorCodes::CoprocessorError); - const std::string & err_msg = resp->other_error(); if (!err_msg.empty()) throw Exception("Coprocessor other error: " + err_msg, ErrorCodes::CoprocessorError); diff --git a/src/kv/Backoff.cc b/src/kv/Backoff.cc index f27e7c67..fcd7f150 100644 --- a/src/kv/Backoff.cc +++ b/src/kv/Backoff.cc @@ -46,7 +46,7 @@ void Backoffer::backoff(pingcap::kv::BackoffType tp, const pingcap::Exception & void Backoffer::backoffWithMaxSleep(pingcap::kv::BackoffType tp, int max_sleep_time, const pingcap::Exception & exc) { - if (exc.code() == MismatchClusterIDCode) + if (exc.code() == MismatchClusterIDCode || exc.code() == IncompatibleRequest || exc.code() == UndeterminedResult) { exc.rethrow(); } diff --git a/src/kv/LockResolver.cc b/src/kv/LockResolver.cc index a2329637..874d8a7d 100644 --- a/src/kv/LockResolver.cc +++ b/src/kv/LockResolver.cc @@ -1,6 +1,7 @@ #include #include #include +#include #include namespace pingcap @@ -60,6 +61,8 @@ TryGetBypassLockResult LockResolver::tryGetBypassLock( } catch (Exception & e) { + if (isTerminalTransactionError(e)) + e.rethrow(); log->warning("get txn status failed: " + e.displayText()); continue; } @@ -90,6 +93,8 @@ TryGetBypassLockResult LockResolver::tryGetBypassLock( } catch (Exception & e) { + if (isTerminalTransactionError(e)) + e.rethrow(); log->warning("tryGetBypassLock failed: " + e.displayText()); } catch (...) @@ -138,6 +143,8 @@ int64_t LockResolver::resolveLocksImpl( } catch (Exception & e) { + if (isTerminalTransactionError(e)) + e.rethrow(); log->warning("get txn status failed: " + e.displayText()); before_txn_expired.update(0); return before_txn_expired.value(); @@ -195,6 +202,8 @@ int64_t LockResolver::resolveLocksImpl( } catch (Exception & e) { + if (isTerminalTransactionError(e)) + e.rethrow(); log->warning("resolve txn failed: " + e.displayText()); before_txn_expired.update(0); return before_txn_expired.value(); @@ -414,6 +423,7 @@ void LockResolver::resolveLockAsync(Backoffer & bo, LockPtr lock, TxnStatus & st std::vector threads; std::atomic errors{}; + internal::TerminalErrorCollector terminal_errors; threads.reserve(keys_by_region.size()); for (auto & pair : keys_by_region) { @@ -425,6 +435,8 @@ void LockResolver::resolveLockAsync(Backoffer & bo, LockPtr lock, TxnStatus & st } catch (Exception & e) { + if (isTerminalTransactionError(e)) + terminal_errors.capture(e); errors.fetch_add(1); log->warning("ResolveRegionLocks error: " + e.displayText()); } @@ -438,6 +450,8 @@ void LockResolver::resolveLockAsync(Backoffer & bo, LockPtr lock, TxnStatus & st log->debug("resolve lock async done"); + terminal_errors.rethrowIfPresent(); + if (errors.load() > 0) { throw Exception("AsyncCommit recovery finished with errors", ErrorCodes::UnknownError); @@ -501,6 +515,7 @@ AsyncResolveDataPtr LockResolver::checkAllSecondaries(Backoffer & bo, LockPtr lo auto shared_data = std::make_shared(status.primary_lock->min_commit_ts(), false); std::vector threads; std::atomic_int8_t errors{0}; + internal::TerminalErrorCollector terminal_errors; threads.reserve(regions.size()); for (auto & pair : regions) { @@ -512,6 +527,8 @@ AsyncResolveDataPtr LockResolver::checkAllSecondaries(Backoffer & bo, LockPtr lo } catch (Exception & e) { + if (isTerminalTransactionError(e)) + terminal_errors.capture(e); if (e.code() == ErrorCodes::NonAsyncCommit) { need_fallback.store(true); @@ -527,6 +544,7 @@ AsyncResolveDataPtr LockResolver::checkAllSecondaries(Backoffer & bo, LockPtr lo t.join(); } + terminal_errors.rethrowIfPresent(); if (need_fallback.load()) { throw Exception("CheckSecondaryLocks receives a non-async-commit lock", ErrorCodes::NonAsyncCommit); diff --git a/src/kv/RegionCache.cc b/src/kv/RegionCache.cc index c31f209e..fb80e6d4 100644 --- a/src/kv/RegionCache.cc +++ b/src/kv/RegionCache.cc @@ -335,7 +335,18 @@ Store RegionCache::reloadStoreWithoutLock(const metapb::Store & store) store_type = StoreType::TiFlash; } } - auto res = stores.insert_or_assign(id, Store(id, store.address(), store.peer_address(), labels, store_type, store.state())); + const bool has_range = store.has_txn_protocol_version_range(); + const auto min_version = has_range ? store.txn_protocol_version_range().min() : 0; + const auto max_version = has_range ? store.txn_protocol_version_range().max() : 0; + const bool invalid_range = min_version > max_version; + auto old = stores.find(id); + if (invalid_range + && (old == stores.end() || old->second.txn_protocol_version_min != min_version || old->second.txn_protocol_version_max != max_version + || old->second.txn_protocol_version_min <= old->second.txn_protocol_version_max)) + { + log->warning("store_id " + std::to_string(id) + " has invalid transaction protocol range [" + std::to_string(min_version) + "," + std::to_string(max_version) + "]"); + } + auto res = stores.insert_or_assign(id, Store(id, store.address(), store.peer_address(), labels, store_type, store.state(), has_range, min_version, max_version)); return res.first->second; } diff --git a/src/kv/RegionClient.cc b/src/kv/RegionClient.cc index 3b53fc20..9a7632fd 100644 --- a/src/kv/RegionClient.cc +++ b/src/kv/RegionClient.cc @@ -6,6 +6,8 @@ namespace kv { void RegionClient::onRegionError(Backoffer & bo, RPCContextPtr rpc_ctx, const errorpb::Error & err) const { + rethrowTerminalRegionError(err); + if (err.has_not_leader()) { const auto & not_leader = err.not_leader(); diff --git a/src/test/CMakeLists.txt b/src/test/CMakeLists.txt index afad481c..6fde2fec 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -15,6 +15,8 @@ add_executable(kv_client_ut region_split_test.cc region_cache_test.cc lock_resolve_test.cc + txn_protocol_test.cc + terminal_error_test.cc coprocessor_test.cc batch_coprocessor_test.cc ) diff --git a/src/test/terminal_error_test.cc b/src/test/terminal_error_test.cc new file mode 100644 index 00000000..dbb9f50a --- /dev/null +++ b/src/test/terminal_error_test.cc @@ -0,0 +1,177 @@ +#include +#include +#include +#include +#include + +#include +#include +#include +#include + +namespace pingcap::tests +{ +namespace +{ +::errorpb::IncompatibleRequest incompatibleError() +{ + ::errorpb::IncompatibleRequest error; + error.set_message("incompatible"); + error.set_provided_txn_protocol_version(1); + error.set_min_compatible_txn_protocol_version(0); + error.set_max_compatible_txn_protocol_version(0); + return error; +} + +template +void captureTerminal(kv::internal::TerminalErrorCollector & collector, F && throw_error) +{ + try + { + throw_error(); + } + catch (const Exception & error) + { + collector.capture(error); + } +} + +void expectUndetermined(kv::internal::TerminalErrorCollector & collector) +{ + try + { + collector.rethrowIfPresent(); + FAIL() << "expected an undetermined-result error"; + } + catch (const Exception & error) + { + EXPECT_EQ(error.code(), UndeterminedResult); + } +} +} // namespace + +TEST(TerminalErrorCollector, UndeterminedWinsRegardlessOfWorkerCompletionOrder) +{ + for (const bool incompatible_first : {false, true}) + { + kv::internal::TerminalErrorCollector collector; + auto capture_incompatible = [&] { captureTerminal(collector, [] { throw ErrIncompatibleRequest(incompatibleError()); }); }; + auto capture_undetermined = [&] { captureTerminal(collector, [] { throw Exception("undetermined", UndeterminedResult); }); }; + if (incompatible_first) + { + capture_incompatible(); + capture_undetermined(); + } + else + { + capture_undetermined(); + capture_incompatible(); + } + expectUndetermined(collector); + } +} + +TEST(TerminalErrorCollector, ConcurrentWorkersKeepUndeterminedPriority) +{ + kv::internal::TerminalErrorCollector collector; + std::atomic_bool start{false}; + std::thread incompatible_worker([&] { + while (!start.load()) + std::this_thread::yield(); + captureTerminal(collector, [] { throw ErrIncompatibleRequest(incompatibleError()); }); + }); + std::thread undetermined_worker([&] { + while (!start.load()) + std::this_thread::yield(); + captureTerminal(collector, [] { throw Exception("undetermined", UndeterminedResult); }); + }); + + start.store(true); + incompatible_worker.join(); + undetermined_worker.join(); + expectUndetermined(collector); +} + +TEST(TerminalErrorCollector, IncompatibleSurvivesFallbackAndKeepsProto) +{ + kv::internal::TerminalErrorCollector collector; + captureTerminal(collector, [] { throw Exception("fallback", NonAsyncCommit); }); + captureTerminal(collector, [] { throw ErrIncompatibleRequest(incompatibleError()); }); + + try + { + collector.rethrowIfPresent(); + FAIL() << "expected an incompatible-request error"; + } + catch (const ErrIncompatibleRequest & error) + { + EXPECT_EQ(error.error().message(), "incompatible"); + EXPECT_EQ(error.error().provided_txn_protocol_version(), 1); + } +} + +TEST(TerminalErrorCollector, RegionErrorPriorityAndCoprocessorQueuePreserveDetails) +{ + ::errorpb::Error region_error; + *region_error.mutable_incompatible_request() = incompatibleError(); + region_error.mutable_undetermined_result()->set_message("undetermined"); + bool terminal_error_thrown = false; + try + { + rethrowTerminalRegionError(region_error); + } + catch (const Exception & error) + { + terminal_error_thrown = true; + EXPECT_EQ(error.code(), UndeterminedResult); + } + EXPECT_TRUE(terminal_error_thrown); + + ErrIncompatibleRequest incompatible(incompatibleError()); + coprocessor::ResponseIter::Result result(incompatible); + ASSERT_NE(result.exception(), nullptr); + const auto * detailed = dynamic_cast(result.exception()); + ASSERT_NE(detailed, nullptr); + EXPECT_EQ(detailed->error().message(), "incompatible"); + EXPECT_EQ(result.error.code(), IncompatibleRequest); + + common::MPMCQueue queue; + ASSERT_EQ(queue.push(std::move(result)), common::MPMCQueueResult::OK); + coprocessor::ResponseIter::Result popped; + ASSERT_EQ(queue.pop(popped), common::MPMCQueueResult::OK); + ASSERT_NE(dynamic_cast(popped.exception()), nullptr); +} + +TEST(TerminalErrorCollector, RpcCallInjectsOriginAndAlwaysOverwritesVersion) +{ + kv::RpcClientPtr client; + kv::RpcCall call(client, "unused"); + ::metapb::Region region; + region.set_id(9); + region.mutable_region_epoch()->set_conf_ver(2); + region.mutable_region_epoch()->set_version(3); + ::metapb::Peer peer; + peer.set_id(4); + peer.set_store_id(5); + kv::Store store(5, "unused", "", {}, kv::StoreType::TiKV, ::metapb::StoreState::Up); + auto context = std::make_shared(7, kv::RegionVerID(9, 2, 3), region, peer, store, "unused"); + + ::kvrpcpb::GetRequest request; + request.mutable_context()->set_request_origin(::kvrpcpb::RequestOriginTiCDC); + request.mutable_context()->set_txn_protocol_version(::kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK); + call.setRequestCtx( + request, + context, + ::kvrpcpb::APIVersion::V1, + ::kvrpcpb::RequestOriginTiFlash, + ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING); + EXPECT_EQ(request.context().request_origin(), ::kvrpcpb::RequestOriginTiCDC); + EXPECT_EQ(request.context().txn_protocol_version(), ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING); + EXPECT_EQ(request.context().region_id(), 9); + + request.mutable_context()->set_request_origin(::kvrpcpb::RequestOriginUnknown); + call.setRequestCtx(request, context, ::kvrpcpb::APIVersion::V1, ::kvrpcpb::RequestOriginTiFlash, ::kvrpcpb::TXN_VER_LEGACY); + EXPECT_EQ(request.context().request_origin(), ::kvrpcpb::RequestOriginTiFlash); + EXPECT_EQ(request.context().txn_protocol_version(), ::kvrpcpb::TXN_VER_LEGACY); +} +} // namespace pingcap::tests diff --git a/src/test/txn_protocol_test.cc b/src/test/txn_protocol_test.cc new file mode 100644 index 00000000..889749e7 --- /dev/null +++ b/src/test/txn_protocol_test.cc @@ -0,0 +1,89 @@ +#include +#include +#include +#include +#include + +#include + +namespace pingcap::kv::internal +{ +TEST(TxnProtocolPolicy, DefaultsDeclareStructuredErrorHandling) +{ + ClusterConfig config; + EXPECT_EQ(config.request_origin, ::kvrpcpb::RequestOriginUnknown); + EXPECT_EQ(config.default_txn_protocol_version, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING); +} + +TEST(TxnProtocolPolicy, SelectsFromExecutionStoreRange) +{ + ::kvrpcpb::GetRequest get; + auto selected = selectTxnProtocolVersion(get, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 0, 0); + EXPECT_TRUE(selected.protected_request); + EXPECT_TRUE(selected.allowed); + EXPECT_EQ(selected.selected, ::kvrpcpb::TXN_VER_LEGACY); + + selected = selectTxnProtocolVersion(get, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 1, 1); + EXPECT_TRUE(selected.allowed); + EXPECT_EQ(selected.selected, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING); + + selected = selectTxnProtocolVersion(get, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 2, 1); + EXPECT_FALSE(selected.allowed); +} + +TEST(TxnProtocolPolicy, RejectsSharedLockPayloadAtVersionOneCeiling) +{ + ::kvrpcpb::PrewriteRequest prewrite; + prewrite.add_mutations()->set_op(::kvrpcpb::SharedLock); + auto selected = selectTxnProtocolVersion(prewrite, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 0, 2); + EXPECT_TRUE(selected.protected_request); + EXPECT_EQ(selected.required, ::kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK); + EXPECT_FALSE(selected.allowed); + + ::kvrpcpb::PessimisticLockRequest pessimistic_lock; + pessimistic_lock.add_mutations()->set_op(::kvrpcpb::SharedPessimisticLock); + selected = selectTxnProtocolVersion(pessimistic_lock, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 0, 2); + EXPECT_EQ(selected.required, ::kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK); + EXPECT_FALSE(selected.allowed); +} + +TEST(TxnProtocolPolicy, LeavesUnprotectedRequestsAtLegacy) +{ + ::kvrpcpb::RawGetRequest raw_get; + auto selected = selectTxnProtocolVersion(raw_get, ::kvrpcpb::TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING, 2, 2); + EXPECT_FALSE(selected.protected_request); + EXPECT_TRUE(selected.allowed); + EXPECT_EQ(selected.selected, ::kvrpcpb::TXN_VER_LEGACY); +} + +TEST(TxnProtocolPolicy, AcceptsOnlyStrictUpperBoundRejection) +{ + ::errorpb::IncompatibleRequest error; + error.set_reason(::errorpb::IncompatibleRequestReasonTxnProtocolVersionOutOfRange); + error.set_provided_txn_protocol_version(1); + error.set_min_compatible_txn_protocol_version(0); + error.set_max_compatible_txn_protocol_version(0); + EXPECT_TRUE(isValidUpperBoundRejection(error, 1)); + + error.set_provided_txn_protocol_version(0); + EXPECT_FALSE(isValidUpperBoundRejection(error, 1)); + error.set_provided_txn_protocol_version(1); + error.set_max_compatible_txn_protocol_version(1); + EXPECT_FALSE(isValidUpperBoundRejection(error, 1)); +} + +TEST(TxnProtocolPolicy, TerminalErrorsPreserveTypeAndSkipBackoff) +{ + ::errorpb::IncompatibleRequest error; + error.set_message("incompatible"); + ErrIncompatibleRequest incompatible(error); + Exception & base = incompatible; + std::unique_ptr copy(base.clone()); + EXPECT_NE(dynamic_cast(copy.get()), nullptr); + EXPECT_EQ(dynamic_cast(copy.get())->error().message(), "incompatible"); + + Backoffer backoffer(1000); + EXPECT_THROW(backoffer.backoff(boRegionMiss, incompatible), ErrIncompatibleRequest); + EXPECT_THROW(backoffer.backoff(boRegionMiss, Exception("undetermined", UndeterminedResult)), Exception); +} +} // namespace pingcap::kv::internal diff --git a/third_party/kvproto b/third_party/kvproto index 03ba9a75..d80b6c5e 160000 --- a/third_party/kvproto +++ b/third_party/kvproto @@ -1 +1 @@ -Subproject commit 03ba9a7536b70dadcc04239347b98fa1d160e639 +Subproject commit d80b6c5e7821b55f78092caead501ab0f00102da