*: protect transaction RPC compatibility

proto/VERSION

@@ -1,3 +1,3 @@
-pingcap/kvproto b41e86365ce0e0fba482005c0dfa5e2ca967d74c (2026-06-22)
-Matches client-go v2.0.8-0.20260708122311-01bd8f99f4da (go.mod: kvproto v0.0.0-20260622063236-b41e86365ce0).
-Re-vendor: copy proto/*.proto (matching names only) and include/ from that revision, then `make generate`.
+pingcap/kvproto d80b6c5e7821b55f78092caead501ab0f00102da (2026-09-10)
+Includes transaction RPC compatibility declarations and structured errors from kvproto PR #1534.
+Re-vendor: copy the vendored proto files and any newly imported dependencies from that revision, then `make generate`.

proto/errorpb.proto

@@ -88,8 +88,30 @@ message EpochNotMatch {
   repeated metapb.Region current_regions = 1;
 }
 
+// Reasons for rejecting an incompatible transaction protocol request.
+enum IncompatibleRequestReason {
+  IncompatibleRequestReasonUnknown = 0;
+  // The declared version is outside the compatible range returned in the error.
+  IncompatibleRequestReasonTxnProtocolVersionOutOfRange = 1;
+}
+
+// A transaction protocol compatibility rejection.
+// The compatible range is inclusive: admission reports the store's global range;
+// feature checks report [required feature version, store's maximum version].
+message IncompatibleRequest {
+  IncompatibleRequestReason reason = 1;
+  // Diagnostic text; clients must use structured fields for control flow.
+  string message = 2;
+  // The request's declaration; missing and explicit zero both mean legacy.
+  uint32 provided_txn_protocol_version = 3;
+  uint32 min_compatible_txn_protocol_version = 4;
+  uint32 max_compatible_txn_protocol_version = 5;
+}
+
 // ServerIsBusy is the error variant that tells the server is too busy to response.
 message ServerIsBusy {
+  // "txn_protocol_incompatible" is a legacy fallback for IncompatibleRequest,
+  // not an overload signal.
   string reason = 1;
   // The suggested backoff time
   uint64 backoff_ms = 2;
@@ -214,4 +236,14 @@ message Error {
 
   // UndeterminedResult is the error variant that tells the result is not determined yet.
   UndeterminedResult undetermined_result = 22;
+
+  // A compatibility rejection with no usable business result.
+  // For a zero declaration, also set server_is_busy.reason to
+  // "txn_protocol_incompatible"; omit that fallback for nonzero declarations.
+  // Clients must handle this error before fallback or ordinary retry logic.
+  // A client may reselect and resend only when the reason is version out of range,
+  // the returned range is valid, and provided > max. The selected version must
+  // satisfy the request's requirements. Such admission rejections must have no
+  // transaction side effects. All other incompatibilities are terminal.
+  IncompatibleRequest incompatible_request = 23;
 }

proto/kvrpcpb.proto

@@ -853,9 +853,16 @@ message MvccGetByStartTsResponse {
 
 // Helper messages.
 
+// Client-provided process/product identity for auditing, not authentication or
+// authorization. TiDB includes embedded BR; standalone BR uses RequestOriginBR.
+// Audit unknown values without mapping them to a known origin or rejecting
+// requests solely because of their origin in the initial rollout.
 enum RequestOrigin {
   RequestOriginUnknown = 0;
   RequestOriginTiDB = 1;
+  RequestOriginTiCDC = 2;
+  RequestOriginBR = 3;
+  RequestOriginTiFlash = 4;
 }
 
 // Miscellaneous metadata attached to most requests.
@@ -928,11 +935,7 @@ message Context {
   // Some information used for resource control.
   ResourceControlContext resource_control_context = 28;
 
-  // The declared origin of the request. TiDB server requests set this to TIDB.
-  // This field is client-provided metadata; consumers should rely on it only when
-  // the caller is authenticated/authorized by the transport or deployment boundary.
-  // UNKNOWN means the origin is unset or not recognized by this protocol version.
-  // Other components should add dedicated enum variants when they need origin-specific behavior.
+  // Declared process/product origin; see RequestOrigin.
   RequestOrigin request_origin = 29;
 
   // The keyspace that the request is sent to.
@@ -966,9 +969,12 @@ message Context {
   // This field is set by client-go based on an extractor function provided by TiDB.
   uint64 trace_control_flags = 37;
 
-  // TODO(slock): tidb should send requests with a feature flag to indicate whether it
-  // supports shared locks, so that tikv can fail the requests from old tidb versions
-  // when needed.
+  // Transaction protocol version declared for this request; see TxnProtocolVersion.
+  // Missing or zero means legacy. Any nonzero value promises structured
+  // IncompatibleRequest handling. BatchCommands declares this per inner request.
+  // This is a compatibility declaration, not a binary version or authorization.
+  // Keep uint32 so future version values remain representable.
+  uint32 txn_protocol_version = 39;
 }
 
 message ResourceControlContext {
@@ -1039,6 +1045,20 @@ enum APIVersion {
   V2 = 2;
 }
 
+// Transaction protocol version registry. Each version includes all earlier
+// semantics. Keep values stable; new versions require joint review by TiKV,
+// TiDB and relevant client owners. This registry sets neither client defaults
+// nor the store's admission range.
+enum TxnProtocolVersion {
+  // No promise to handle IncompatibleRequest.
+  TXN_VER_LEGACY = 0;
+  // Handles IncompatibleRequest according to its error contract.
+  TXN_VER_SUPPORT_INCOMPATIBLE_ERROR_HANDLING = 1;
+  // Also supports shared wrapper parsing and expansion, nested lock information
+  // preservation, and shared-lock pagination and resolve semantics.
+  TXN_VER_SUPPORT_SHARED_LOCK = 2;
+}
+
 message SourceStmt {
   uint64 start_ts = 1;
   uint64 connection_id = 2;
@@ -1067,9 +1087,8 @@ message LockInfo {
   // When lock_type is SharedLock, this describes transactions holding the shared lock.
   // Important: when lock_type is SharedLock, all shared locks must use shared_lock_infos;
   // DO NOT read from the wrapper LockInfo.
-  // TODO(slock): tidb should send requests with a feature flag to indicate whether it
-  // supports shared locks, so that tikv can fail the requests from old tidb versions
-  // when needed.
+  // Before returning a shared lock wrapper, the server must check the caller's
+  // txn_protocol_version and return IncompatibleRequest if support is insufficient.
   repeated LockInfo shared_lock_infos = 12;
 
   // Reserved for file based transaction.

proto/metapb.proto

@@ -67,6 +67,18 @@ message Store {
   bool physically_destroyed = 12;
   // NodeState is used to replace StoreState which will be deprecated in the future.
   NodeState node_state = 13;
+  // The store's current transaction RPC admission range, not a feature switch or
+  // a value derived from Store.version. Absence means legacy/unknown; clients
+  // conservatively use [0, 0] while preserving absence versus an explicit [0, 0].
+  // Consumers must fail closed if min > max. TiKV reports this on every startup,
+  // including upgrades and rollbacks; PD updates and persists it.
+  TxnProtocolVersionRange txn_protocol_version_range = 14;
+}
+
+// Inclusive transaction protocol version bounds; see kvrpcpb.TxnProtocolVersion.
+message TxnProtocolVersionRange {
+  uint32 min = 1;
+  uint32 max = 2;
 }
 
 message RegionEpoch {

src/common/errors.rs

@@ -14,6 +14,8 @@ use crate::BoundRange;
 /// future release even if the wire format is compatible.
 #[doc(inline)]
 pub use crate::proto::errorpb::Error as ProtoRegionError;
+#[doc(inline)]
+pub use crate::proto::errorpb::IncompatibleRequest as ProtoIncompatibleRequest;
 
 /// Protobuf-generated per-key error returned by TiKV.
 ///
@@ -78,6 +80,9 @@ pub enum Error {
     /// Errors caused by changes of region information
     #[error("Region error: {0:?}")]
     RegionError(Box<ProtoRegionError>),
+    /// The target Store cannot safely interpret this transaction RPC.
+    #[error("Incompatible transaction request: {0:?}")]
+    IncompatibleRequest(Box<ProtoIncompatibleRequest>),
     /// Whether the transaction is committed or not is undetermined
     #[error("Whether the transaction is committed or not is undetermined")]
     UndeterminedError(Box<Error>),
@@ -136,9 +141,54 @@ Use the async TransactionClient instead, or create and use SyncTransactionClient
     NestedRuntimeError(String),
 }
 
+/// Priority of direct RPC errors when several responses must be combined.
+/// Lower values win; callers retain their own wrapping and cleanup contracts.
+#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
+pub(crate) enum ErrorPriority {
+    Undetermined,
+    Incompatible,
+    Ordinary,
+}
+
+impl ProtoRegionError {
+    pub(crate) fn priority(&self) -> ErrorPriority {
+        if self.undetermined_result.is_some() {
+            ErrorPriority::Undetermined
+        } else if self.incompatible_request.is_some() {
+            ErrorPriority::Incompatible
+        } else {
+            ErrorPriority::Ordinary
+        }
+    }
+}
+
+impl Error {
+    pub(crate) fn priority(&self) -> ErrorPriority {
+        match self {
+            Self::RegionError(region) => region.priority(),
+            Self::IncompatibleRequest(_) => ErrorPriority::Incompatible,
+            _ => ErrorPriority::Ordinary,
+        }
+    }
+}
+
+pub(crate) fn is_grpc_error(error: &Error) -> bool {
+    matches!(error, Error::GrpcAPI(_) | Error::Grpc(_))
+}
+
+pub(crate) fn is_undetermined_region_error(error: &Error) -> bool {
+    matches!(error, Error::RegionError(_)) && error.priority() == ErrorPriority::Undetermined
+}
+
 impl From<ProtoRegionError> for Error {
     fn from(e: ProtoRegionError) -> Error {
-        Error::RegionError(Box::new(e))
+        if e.priority() == ErrorPriority::Undetermined {
+            Error::RegionError(Box::new(e))
+        } else if let Some(incompatible) = e.incompatible_request {
+            Error::IncompatibleRequest(Box::new(incompatible))
+        } else {
+            Error::RegionError(Box::new(e))
+        }
     }
 }
 

src/common/mod.rs

@@ -3,7 +3,11 @@
 mod errors;
 pub mod security;
 
+pub(crate) use self::errors::is_grpc_error;
+pub(crate) use self::errors::is_undetermined_region_error;
 pub use self::errors::Error;
+pub(crate) use self::errors::ErrorPriority;
+pub use self::errors::ProtoIncompatibleRequest;
 pub use self::errors::ProtoKeyError;
 pub use self::errors::ProtoRegionError;
 pub use self::errors::Result;

src/config.rs

@@ -3,9 +3,34 @@
 use std::path::PathBuf;
 use std::time::Duration;
 
+use crate::{Error, Result};
 use serde_derive::Deserialize;
 use serde_derive::Serialize;
 
+/// Identifies the process issuing transaction RPCs for compatibility auditing.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub enum RequestOrigin {
+    #[default]
+    Unknown,
+    TiDb,
+    TiCdc,
+    Br,
+    TiFlash,
+}
+
+impl RequestOrigin {
+    pub(crate) fn as_proto(self) -> i32 {
+        use crate::proto::kvrpcpb::RequestOrigin as ProtoOrigin;
+        match self {
+            Self::Unknown => ProtoOrigin::Unknown as i32,
+            Self::TiDb => ProtoOrigin::TiDb as i32,
+            Self::TiCdc => ProtoOrigin::TiCdc as i32,
+            Self::Br => ProtoOrigin::Br as i32,
+            Self::TiFlash => ProtoOrigin::TiFlash as i32,
+        }
+    }
+}
+
 /// The configuration for either a [`RawClient`](crate::RawClient) or a
 /// [`TransactionClient`](crate::TransactionClient).
 ///
@@ -26,10 +51,17 @@ pub struct Config {
     pub timeout: Duration,
     pub grpc_max_decoding_message_size: usize,
     pub keyspace: Option<String>,
+    #[serde(skip)]
+    pub(crate) default_txn_protocol_version: u32,
+    #[serde(skip)]
+    pub(crate) request_origin: RequestOrigin,
 }
 
 const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(2);
 const DEFAULT_GRPC_MAX_DECODING_MESSAGE_SIZE: usize = 4 * 1024 * 1024; // 4MB
+pub(crate) const MAX_SUPPORTED_TXN_PROTOCOL_VERSION: u32 =
+    crate::proto::kvrpcpb::TxnProtocolVersion::TxnVerSupportIncompatibleErrorHandling as u32;
+pub(crate) const DEFAULT_TXN_PROTOCOL_VERSION: u32 = MAX_SUPPORTED_TXN_PROTOCOL_VERSION;
 
 impl Default for Config {
     fn default() -> Self {
@@ -40,11 +72,32 @@ impl Default for Config {
             timeout: DEFAULT_REQUEST_TIMEOUT,
             grpc_max_decoding_message_size: DEFAULT_GRPC_MAX_DECODING_MESSAGE_SIZE,
             keyspace: None,
+            default_txn_protocol_version: DEFAULT_TXN_PROTOCOL_VERSION,
+            request_origin: RequestOrigin::Unknown,
         }
     }
 }
 
 impl Config {
+    /// Set the highest transaction protocol version this client can understand.
+    pub fn with_default_txn_protocol_version(mut self, version: u32) -> Result<Self> {
+        let max = MAX_SUPPORTED_TXN_PROTOCOL_VERSION;
+        if version > max {
+            return Err(Error::StringError(format!(
+                "unsupported transaction protocol version {version}; maximum is {max}"
+            )));
+        }
+        self.default_txn_protocol_version = version;
+        Ok(self)
+    }
+
+    /// Set the process origin attached to transaction RPCs.
+    #[must_use]
+    pub fn with_request_origin(mut self, origin: RequestOrigin) -> Self {
+        self.request_origin = origin;
+        self
+    }
+
     /// Set the certificate authority, certificate, and key locations for clients.
     ///
     /// By default, this client will use an insecure connection over instead of one protected by
@@ -118,3 +171,33 @@ impl Config {
         self
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn transaction_protocol_version_is_bounded() {
+        assert_eq!(Config::default().default_txn_protocol_version, 1);
+        assert_eq!(
+            Config::default()
+                .with_default_txn_protocol_version(0)
+                .unwrap()
+                .default_txn_protocol_version,
+            0
+        );
+        assert!(Config::default()
+            .with_default_txn_protocol_version(2)
+            .is_err());
+    }
+
+    #[test]
+    fn compatibility_fields_are_not_deserialized() {
+        let config: Config = serde_json::from_str(
+            r#"{"default-txn-protocol-version":0,"request-origin":"TiFlash"}"#,
+        )
+        .unwrap();
+        assert_eq!(config.default_txn_protocol_version, 1);
+        assert_eq!(config.request_origin, RequestOrigin::Unknown);
+    }
+}