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..924e8ab6 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,11 @@ class ResponseIter std::shared_ptr<::coprocessor::Response> resp; bool same_zone{true}; Exception error; + // The by-value error above is retained for existing callers, but slices derived + // exceptions and loses structured details such as ErrIncompatibleRequest's protobuf. + // Keep a polymorphic clone in shared ownership so Result remains copyable and callers + // can inspect the exception directly, without rethrowing an std::exception_ptr. + std::shared_ptr detailed_error; bool finished{false}; Result() = default; @@ -121,6 +127,7 @@ class ResponseIter {} explicit Result(const Exception & err) : error(err) + , detailed_error(err.clone()) {} explicit Result(bool finished_) : finished(finished_) @@ -131,6 +138,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..3b29fe6c 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,17 @@ struct Cluster void splitRegion(const std::string & split_key); void startBackgroundTasks(); + +private: + static const ClusterConfig & validateCompatibilityConfig(const ClusterConfig & config) + { + // This client currently implements protocol semantics only through incompatible-error-handling. + // The configured ceiling must not advertise support for higher versions (e.g. shared locks) + // merely because their enum values are available in kvproto; their semantics must be implemented first. + 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..faf6cf61 100644 --- a/src/test/CMakeLists.txt +++ b/src/test/CMakeLists.txt @@ -15,6 +15,9 @@ add_executable(kv_client_ut region_split_test.cc region_cache_test.cc lock_resolve_test.cc + txn_protocol_test.cc + txn_protocol_rpc_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_rpc_test.cc b/src/test/txn_protocol_rpc_test.cc new file mode 100644 index 00000000..b1381651 --- /dev/null +++ b/src/test/txn_protocol_rpc_test.cc @@ -0,0 +1,710 @@ +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +namespace pingcap::tests +{ +namespace +{ +using namespace kv; + +class CompatibilityPD : public pd::MockPDClient +{ +public: + explicit CompatibilityPD(const std::string & address) + { + region.set_id(9); + region.mutable_region_epoch()->set_conf_ver(2); + region.mutable_region_epoch()->set_version(3); + auto * peer = region.add_peers(); + peer->set_id(4); + peer->set_store_id(5); + store.set_id(5); + store.set_address(address); + setRange(0, 1); + } + + void setRange(uint32_t min, uint32_t max, bool present = true) + { + std::lock_guard lock(mutex); + store.clear_txn_protocol_version_range(); + if (present) + { + store.mutable_txn_protocol_version_range()->set_min(min); + store.mutable_txn_protocol_version_range()->set_max(max); + } + } + + void useTiFlashPeers() + { + std::lock_guard lock(mutex); + auto * label = store.add_labels(); + label->set_key("engine"); + label->set_value("tiflash"); + auto * peer = region.add_peers(); + peer->set_id(6); + peer->set_store_id(7); + } + + void useLegacySecondPeer() + { + std::lock_guard lock(mutex); + auto * peer = region.add_peers(); + peer->set_id(6); + peer->set_store_id(7); + legacy_second_peer = true; + } + + metapb::Store getStore(uint64_t id) override + { + std::lock_guard lock(mutex); + ++store_reads; + auto result = store; + result.set_id(id); + if (legacy_second_peer && id == 7) + { + result.mutable_txn_protocol_version_range()->set_min(0); + result.mutable_txn_protocol_version_range()->set_max(0); + } + return result; + } + + std::vector getAllStores(bool) override + { + std::lock_guard lock(mutex); + ++all_store_reads; + std::vector result; + for (const auto & peer : region.peers()) + { + auto current = store; + current.set_id(peer.store_id()); + if (legacy_second_peer && peer.store_id() == 7) + { + current.mutable_txn_protocol_version_range()->set_min(0); + current.mutable_txn_protocol_version_range()->set_max(0); + } + result.push_back(std::move(current)); + } + return result; + } + + pdpb::GetRegionResponse getRegionByKey(const std::string &) override { return regionResponse(); } + pdpb::GetRegionResponse getRegionByID(uint64_t) override { return regionResponse(); } + + std::atomic store_reads{0}; + std::atomic all_store_reads{0}; + std::atomic region_reads{0}; + +private: + pdpb::GetRegionResponse regionResponse() + { + std::lock_guard lock(mutex); + ++region_reads; + pdpb::GetRegionResponse response; + *response.mutable_region() = region; + *response.mutable_leader() = region.peers(0); + return response; + } + + std::mutex mutex; + metapb::Region region; + metapb::Store store; + bool legacy_second_peer = false; +}; + +struct Reply +{ + errorpb::Error error; + grpc::Status status; +}; + +Reply incompatible(uint32_t provided = 1, uint32_t min = 0, uint32_t max = 0) +{ + Reply reply; + auto * error = reply.error.mutable_incompatible_request(); + error->set_reason(errorpb::IncompatibleRequestReasonTxnProtocolVersionOutOfRange); + error->set_message("scripted incompatibility"); + error->set_provided_txn_protocol_version(provided); + error->set_min_compatible_txn_protocol_version(min); + error->set_max_compatible_txn_protocol_version(max); + return reply; +} + +class CompatibilityService : public tikvpb::Tikv::Service +{ +public: + void script(std::vector replies_) + { + std::lock_guard lock(mutex); + replies = std::move(replies_); + requests.clear(); + successful_prewrite = 0; + } + + std::vector contexts() + { + std::lock_guard lock(mutex); + return requests; + } + + grpc::Status KvGet(grpc::ServerContext *, const kvrpcpb::GetRequest * request, kvrpcpb::GetResponse * response) override + { + auto reply = next(request->context()); + if (reply.error.ByteSizeLong()) + *response->mutable_region_error() = reply.error; + else if (reply.status.ok()) + response->set_value("ok"); + return reply.status; + } + + grpc::Status KvPrewrite(grpc::ServerContext *, const kvrpcpb::PrewriteRequest * request, kvrpcpb::PrewriteResponse * response) override + { + auto reply = next(request->context()); + if (reply.error.ByteSizeLong()) + *response->mutable_region_error() = reply.error; + else if (reply.status.ok()) + ++successful_prewrite; + return reply.status; + } + + grpc::Status KvCheckTxnStatus(grpc::ServerContext *, const kvrpcpb::CheckTxnStatusRequest * request, kvrpcpb::CheckTxnStatusResponse * response) override + { + auto reply = next(request->context()); + if (reply.error.ByteSizeLong()) + *response->mutable_region_error() = reply.error; + return reply.status; + } + + grpc::Status Coprocessor(grpc::ServerContext *, const ::coprocessor::Request * request, ::coprocessor::Response * response) override + { + auto reply = next(request->context()); + if (reply.error.ByteSizeLong()) + *response->mutable_region_error() = reply.error; + else if (reply.status.ok()) + response->set_data("ok"); + return reply.status; + } + + grpc::Status KvPessimisticLock(grpc::ServerContext *, const kvrpcpb::PessimisticLockRequest *, kvrpcpb::PessimisticLockResponse *) override + { + ADD_FAILURE() << "shared mutation must be rejected before transport"; + return grpc::Status(grpc::StatusCode::INTERNAL, "unexpected shared mutation"); + } + + grpc::Status CoprocessorStream(grpc::ServerContext *, const ::coprocessor::Request * request, grpc::ServerWriter<::coprocessor::Response> * writer) override + { + auto reply = next(request->context()); + if (reply.status.ok()) + { + ::coprocessor::Response response; + if (reply.error.ByteSizeLong()) + *response.mutable_region_error() = reply.error; + else + response.set_data("ok"); + writer->Write(response); + } + return reply.status; + } + + std::atomic successful_prewrite{0}; + +private: + Reply next(const kvrpcpb::Context & context) + { + std::lock_guard lock(mutex); + const auto index = requests.size(); + requests.push_back(context); + if (index < replies.size()) + return replies[index]; + ADD_FAILURE() << "unexpected RPC attempt " << index; + return {{}, grpc::Status(grpc::StatusCode::UNIMPLEMENTED, "script exhausted")}; + } + + std::mutex mutex; + std::vector replies; + std::vector requests; +}; + +class TxnProtocolFixture : public testing::Test +{ +protected: + void SetUp() override + { + grpc::ServerBuilder builder; + int port = 0; + builder.AddListeningPort("127.0.0.1:0", grpc::InsecureServerCredentials(), &port); + builder.RegisterService(&service); + server = builder.BuildAndStart(); + ASSERT_NE(server, nullptr); + ASSERT_GT(port, 0); + pd = std::make_shared("127.0.0.1:" + std::to_string(port)); + // The mock Cluster starts no RegionCache updater. Attach a test-only PD/cache + // afterwards so metadata changes and refreshes remain under test control. + cluster = std::make_unique(); + cluster->pd_client = pd; + ClusterConfig config; + config.tiflash_engine_key = "engine"; + config.tiflash_engine_value = "tiflash"; + cluster->region_cache = std::make_unique(pd, config); + Backoffer bo(1000); + region = cluster->region_cache->locateKey(bo, "key").region; + } + + void TearDown() override + { + cluster.reset(); + if (server) + { + server->Shutdown(); + server->Wait(); + } + } + + RPCContextPtr context(Backoffer & bo) + { + return cluster->region_cache->getRPCContext(bo, region, store_type, false, [](const auto &) { return true; }); + } + + virtual bool streaming() const { return false; } + + void send(Backoffer & bo) + { + RegionClient client(cluster.get(), region); + const LabelFilter labels = [](const auto &) { return true; }; + if (streaming()) + { + ::coprocessor::Request request; + auto reader = client.sendStreamReqToRegion(bo, request, labels, 5, store_type); + ::coprocessor::Response response; + ASSERT_TRUE(reader->read(&response)); + EXPECT_FALSE(response.has_region_error()); + EXPECT_EQ(response.data(), "ok"); + EXPECT_FALSE(reader->read(&response)); + EXPECT_TRUE(reader->finish().ok()); + } + else + { + kvrpcpb::GetRequest request; + request.set_key("key"); + request.mutable_context()->set_txn_protocol_version(kvrpcpb::TXN_VER_SUPPORT_SHARED_LOCK); + kvrpcpb::GetResponse response; + client.sendReqToRegion(bo, request, &response, labels, 5, store_type); + EXPECT_FALSE(response.has_region_error()); + EXPECT_EQ(response.value(), "ok"); + } + } + + void expectIncompatible(Backoffer & bo, const errorpb::IncompatibleRequest & expected) + { + try + { + send(bo); + FAIL() << "expected incompatible request"; + } + catch (const ErrIncompatibleRequest & error) + { + EXPECT_EQ(error.code(), IncompatibleRequest); + EXPECT_EQ(error.error().SerializeAsString(), expected.SerializeAsString()); + } + } + + void expectVersions(std::initializer_list expected) + { + const auto requests = service.contexts(); + ASSERT_EQ(requests.size(), expected.size()); + size_t i = 0; + for (const auto version : expected) + { + EXPECT_EQ(requests[i].txn_protocol_version(), version); + ++i; + } + } + + std::future refillDroppedRegion(bool refresh_range) + { + return std::async(std::launch::async, [this, refresh_range] { + Backoffer refill_bo(1000); + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + while (std::chrono::steady_clock::now() < deadline) + { + if (!context(refill_bo)) + { + if (refresh_range) + pd->setRange(0, 0); + cluster->region_cache->getRegionByID(refill_bo, region); + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + return false; + }); + } + + CompatibilityService service; + std::unique_ptr server; + std::shared_ptr pd; + ClusterPtr cluster; + RegionVerID region; + StoreType store_type = StoreType::TiKV; +}; + +class TxnProtocolRPC : public TxnProtocolFixture, public testing::WithParamInterface +{ +protected: + bool streaming() const override { return GetParam(); } +}; + +TEST_P(TxnProtocolRPC, UpperBoundResendPinsExecutionPeerAndDoesNotUpdateCache) +{ + pd->useTiFlashPeers(); + store_type = StoreType::TiFlash; + Backoffer bo(1000); + // Reload the region to expose both replicas and exercise load balancing. + cluster->region_cache->dropRegion(region); + cluster->region_cache->getRegionByID(bo, region); + service.script({incompatible(), {}}); + send(bo); + expectVersions({1, 0}); + const auto requests = service.contexts(); + ASSERT_EQ(requests.size(), 2); + EXPECT_EQ(requests[0].peer().SerializeAsString(), requests[1].peer().SerializeAsString()); + EXPECT_EQ(requests[0].region_id(), requests[1].region_id()); + EXPECT_EQ(requests[0].region_epoch().SerializeAsString(), requests[1].region_epoch().SerializeAsString()); + EXPECT_EQ(pd->region_reads.load(), 2); + EXPECT_EQ(pd->store_reads.load(), 2); + EXPECT_EQ(pd->all_store_reads.load(), 0); + EXPECT_TRUE(bo.backoff_map.empty()); + EXPECT_EQ(cluster->region_cache->getStore(bo, requests[0].peer().store_id()).txn_protocol_version_max, 1); +} + +TEST_P(TxnProtocolRPC, SecondIncompatibilityIsTerminalAndPreservesOriginalError) +{ + auto second = incompatible(0, 1, 1); + service.script({incompatible(), second}); + Backoffer bo(1000); + expectIncompatible(bo, second.error.incompatible_request()); + expectVersions({1, 0}); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_P(TxnProtocolRPC, OrdinaryRegionRetryDoesNotResetCompatibilityBudget) +{ + Reply not_leader; + not_leader.error.mutable_not_leader()->mutable_leader()->set_id(4); + not_leader.error.mutable_not_leader()->mutable_leader()->set_store_id(5); + service.script({incompatible(), not_leader, incompatible()}); + Backoffer bo(1000); + expectIncompatible(bo, incompatible().error.incompatible_request()); + expectVersions({1, 0, 1}); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_P(TxnProtocolRPC, OrdinaryRetryReselectsVersionForNewExecutionStore) +{ + pd->useLegacySecondPeer(); + Backoffer bo(1000); + cluster->region_cache->dropRegion(region); + cluster->region_cache->getRegionByID(bo, region); + Reply not_leader; + auto * leader = not_leader.error.mutable_not_leader()->mutable_leader(); + leader->set_id(6); + leader->set_store_id(7); + service.script({not_leader, {}}); + send(bo); + expectVersions({1, 0}); + const auto requests = service.contexts(); + ASSERT_EQ(requests.size(), 2); + EXPECT_EQ(requests[0].peer().store_id(), 5); + EXPECT_EQ(requests[1].peer().store_id(), 7); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_P(TxnProtocolRPC, InvalidRecoveryResponsesNeverResend) +{ + std::vector invalid; + auto unknown = incompatible(); + unknown.error.mutable_incompatible_request()->set_reason(errorpb::IncompatibleRequestReasonUnknown); + invalid.push_back(unknown); + invalid.push_back(incompatible(1, 1, 0)); // Invalid range. + invalid.push_back(incompatible(0)); // Echo does not match the actual send. + invalid.push_back(incompatible(1, 0, 1)); // Not a strict upper-bound rejection. + invalid.push_back(incompatible(1, 2, 2)); // An upward recovery is forbidden. + for (const auto & reply : invalid) + { + SCOPED_TRACE(reply.error.DebugString()); + service.script({reply}); + Backoffer bo(1000); + expectIncompatible(bo, reply.error.incompatible_request()); + expectVersions({1}); + EXPECT_TRUE(bo.backoff_map.empty()); + } +} + +TEST_P(TxnProtocolRPC, MissingAndExplicitLegacyRangesRejectUpwardRecoveryBeforeBusyFallback) +{ + for (const bool present : {false, true}) + { + SCOPED_TRACE(present); + pd->setRange(0, 0, present); + cluster->region_cache->forceReloadAllStores(); + auto reply = incompatible(0, 1, 1); + reply.error.mutable_server_is_busy()->set_reason("txn_protocol_incompatible"); + service.script({reply}); + Backoffer bo(1000); + const auto cached = cluster->region_cache->getStore(bo, 5); + EXPECT_EQ(cached.has_txn_protocol_version_range, present); + EXPECT_EQ(cached.txn_protocol_version_min, 0); + EXPECT_EQ(cached.txn_protocol_version_max, 0); + expectIncompatible(bo, reply.error.incompatible_request()); + expectVersions({0}); + EXPECT_TRUE(bo.backoff_map.empty()); + } +} + +TEST_P(TxnProtocolRPC, UndeterminedWinsOverCompatibilityAndBusyWithoutRetry) +{ + auto reply = incompatible(); + reply.error.mutable_undetermined_result()->set_message("outcome unknown"); + reply.error.mutable_server_is_busy()->set_reason("txn_protocol_incompatible"); + service.script({reply}); + Backoffer bo(1000); + try + { + send(bo); + FAIL() << "expected undetermined result"; + } + catch (const Exception & error) + { + EXPECT_EQ(error.code(), UndeterminedResult); + EXPECT_EQ(error.message(), "outcome unknown"); + } + expectVersions({1}); + EXPECT_TRUE(bo.backoff_map.empty()); + ASSERT_NE(context(bo), nullptr); + EXPECT_EQ(pd->store_reads.load(), 1); +} + +TEST_P(TxnProtocolRPC, StoreRefreshChangesNewContextsButNotExistingSnapshots) +{ + Backoffer bo(1000); + const auto old = context(bo); + ASSERT_NE(old, nullptr); + pd->setRange(0, 0); + cluster->region_cache->forceReloadAllStores(); + const auto updated = context(bo); + ASSERT_NE(updated, nullptr); + EXPECT_EQ(old->store.txn_protocol_version_max, 1); + EXPECT_EQ(updated->store.txn_protocol_version_max, 0); + service.script({{}}); + send(bo); + expectVersions({0}); + pd->setRange(0, 1); + cluster->region_cache->forceReloadAllStores(); + service.script({{}}); + send(bo); + expectVersions({1}); +} + +TEST_P(TxnProtocolRPC, InvalidStoreRangeFailsLocallyWithoutTransport) +{ + pd->setRange(2, 1); + service.script({}); + Backoffer bo(1000); + try + { + send(bo); + FAIL() << "expected local incompatibility"; + } + catch (const ErrIncompatibleRequest & error) + { + EXPECT_EQ(error.error().reason(), errorpb::IncompatibleRequestReasonUnknown); + EXPECT_EQ(error.error().min_compatible_txn_protocol_version(), 2); + EXPECT_EQ(error.error().max_compatible_txn_protocol_version(), 1); + EXPECT_EQ(error.error().provided_txn_protocol_version(), 1); + } + EXPECT_TRUE(service.contexts().empty()); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_P(TxnProtocolRPC, ResendTransportFailureDropsRegionAndNextLogicalSendGetsNewBudget) +{ + Reply failure; + failure.status = grpc::Status(grpc::StatusCode::UNAVAILABLE, "scripted transport failure"); + service.script({incompatible(), failure}); + Backoffer bo(1000); + try + { + send(bo); + FAIL() << "expected missing region"; + } + catch (const Exception & error) + { + EXPECT_EQ(error.code(), RegionEpochNotMatch); + } + expectVersions({1, 0}); + EXPECT_EQ(context(bo), nullptr); + ASSERT_NE(bo.backoff_map.find(boTiKVRPC), bo.backoff_map.end()); + // PD still reports the old range; only a new logical send gets a new budget. + cluster->region_cache->getRegionByID(bo, region); + service.script({incompatible(), {}}); + send(bo); + expectVersions({1, 0}); +} + +TEST_P(TxnProtocolRPC, ConcurrentRegionRefillAfterTransportFailureKeepsBudgetConsumed) +{ + Reply failure; + failure.status = grpc::Status(grpc::StatusCode::UNAVAILABLE, "scripted transport failure"); + service.script({incompatible(), failure, incompatible()}); + Backoffer bo(3000); + // Give a concurrent cache user a fixed backoff window in which to reload the + // dropped region. The worker polls the public cache API, not private state. + bo.backoff_map[boTiKVRPC] = std::make_shared(1000, 1000, NoJitter); + auto refill = refillDroppedRegion(false); + expectIncompatible(bo, incompatible().error.incompatible_request()); + EXPECT_TRUE(refill.get()); + expectVersions({1, 0, 1}); + EXPECT_EQ(bo.backoff_map.at(boTiKVRPC)->attempts, 1); +} + +TEST_P(TxnProtocolRPC, RefreshedRangeAfterTransportFailureAvoidsAnotherRejection) +{ + Reply failure; + failure.status = grpc::Status(grpc::StatusCode::UNAVAILABLE, "scripted transport failure"); + service.script({incompatible(), failure, {}}); + Backoffer bo(3000); + bo.backoff_map[boTiKVRPC] = std::make_shared(1000, 1000, NoJitter); + auto refill = refillDroppedRegion(true); + send(bo); + EXPECT_TRUE(refill.get()); + expectVersions({1, 0, 0}); + EXPECT_EQ(bo.backoff_map.at(boTiKVRPC)->attempts, 1); +} + +TEST_F(TxnProtocolFixture, SharedMutationsNeverReachTransport) +{ + service.script({}); + Backoffer bo(1000); + RegionClient client(cluster.get(), region); + kvrpcpb::PrewriteRequest prewrite; + prewrite.add_mutations()->set_op(kvrpcpb::SharedLock); + kvrpcpb::PrewriteResponse prewrite_response; + EXPECT_THROW(client.sendReqToRegion(bo, prewrite, &prewrite_response), ErrIncompatibleRequest); + kvrpcpb::PessimisticLockRequest lock; + lock.add_mutations()->set_op(kvrpcpb::SharedPessimisticLock); + kvrpcpb::PessimisticLockResponse lock_response; + EXPECT_THROW(client.sendReqToRegion(bo, lock, &lock_response), ErrIncompatibleRequest); + EXPECT_TRUE(service.contexts().empty()); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_F(TxnProtocolFixture, MutatingResendRunsBusinessHandlerOnceAndClearsRejectedResponse) +{ + service.script({incompatible(), {}}); + Backoffer bo(1000); + kvrpcpb::PrewriteRequest request; + request.add_mutations()->set_op(kvrpcpb::Put); + kvrpcpb::PrewriteResponse response; + RegionClient client(cluster.get(), region); + client.sendReqToRegion(bo, request, &response); + EXPECT_FALSE(response.has_region_error()); + EXPECT_EQ(service.successful_prewrite.load(), 1); + expectVersions({1, 0}); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_F(TxnProtocolFixture, SnapshotPreservesTerminalErrorInsteadOfRetryingRegionMiss) +{ + auto reply = incompatible(); + reply.error.mutable_incompatible_request()->set_reason(errorpb::IncompatibleRequestReasonUnknown); + service.script({reply}); + Backoffer bo(1000); + Snapshot snapshot(cluster.get(), 10); + try + { + snapshot.Get(bo, "key"); + FAIL() << "expected incompatible request"; + } + catch (const ErrIncompatibleRequest & error) + { + EXPECT_EQ(error.error().SerializeAsString(), reply.error.incompatible_request().SerializeAsString()); + } + expectVersions({1}); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_F(TxnProtocolFixture, LockResolverPreservesTerminalErrorInsteadOfFallingBack) +{ + auto reply = incompatible(); + reply.error.mutable_incompatible_request()->set_reason(errorpb::IncompatibleRequestReasonUnknown); + service.script({reply}); + kvrpcpb::LockInfo info; + info.set_key("key"); + info.set_primary_lock("key"); + info.set_lock_version(10); + std::vector locks{std::make_shared(info)}; + std::vector pushed; + Backoffer bo(1000); + try + { + cluster->lock_resolver->resolveLocks(bo, 20, locks, pushed); + FAIL() << "expected incompatible request"; + } + catch (const ErrIncompatibleRequest & error) + { + EXPECT_EQ(error.error().SerializeAsString(), reply.error.incompatible_request().SerializeAsString()); + } + expectVersions({1}); + EXPECT_TRUE(pushed.empty()); + EXPECT_TRUE(bo.backoff_map.empty()); +} + +TEST_P(TxnProtocolRPC, CoprocessorWorkerDeliversTypedTerminalErrorThroughQueue) +{ + auto reply = incompatible(); + reply.error.mutable_incompatible_request()->set_reason(errorpb::IncompatibleRequestReasonUnknown); + service.script({reply}); + pingcap::coprocessor::CopTask task{}; + task.region_id = region; + task.ranges.emplace_back("key", "z"); + task.req = std::make_shared(); + task.req->tp = pingcap::coprocessor::DAG; + task.req->start_ts = 10; + task.req->schema_version = 0; + task.store_type = StoreType::TiKV; + task.keyspace_id = pd::NullspaceID; + pingcap::coprocessor::ResponseIter iter( + std::make_unique>(), + {task}, + cluster.get(), + 1, + &Logger::get("txn-protocol-test")); + if (streaming()) + iter.open(); + else + iter.open(); + const auto [result, available] = iter.next(); + ASSERT_TRUE(available); + EXPECT_EQ(result.error.code(), IncompatibleRequest); + const auto * detailed = dynamic_cast(result.exception()); + ASSERT_NE(detailed, nullptr); + EXPECT_EQ(detailed->error().SerializeAsString(), reply.error.incompatible_request().SerializeAsString()); + expectVersions({1}); +} + +INSTANTIATE_TEST_SUITE_P(UnaryAndStream, + TxnProtocolRPC, + testing::Bool(), + [](const testing::TestParamInfo & info) { return info.param ? "Stream" : "Unary"; }); +} // namespace +} // 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