diff --git a/src/common/errors.rs b/src/common/errors.rs index f9a99d2f..663b93a0 100644 --- a/src/common/errors.rs +++ b/src/common/errors.rs @@ -84,12 +84,24 @@ pub enum Error { /// Wraps a per-key error returned by TiKV. #[error("{0:?}")] KeyError(Box), - /// Multiple errors generated from the ExtractError plan. - #[error("Multiple errors: {0:?}")] - ExtractedErrors(Vec), - /// Multiple key errors + /// Errors the server reported about the request itself — one per affected key, + /// or one per failed raw operation. + /// + /// Elements are [`Error::KeyError`] for transactional endpoints (from a + /// `kvrpcpb::KeyError` field) or [`Error::KvError`] for raw endpoints (from a + /// string `error` field). Match both if you handle raw and transactional + /// requests through one path. + /// + /// Which of this and [`Error::MultipleRegionErrors`] you receive is decided by + /// *what the server reported*, never by how the request's plan happened to be + /// composed — matching on one of them is therefore sufficient to handle that + /// class of failure. #[error("Multiple key errors: {0:?}")] MultipleKeyErrors(Vec), + /// Multiple region errors returned by TiKV. Every element is a + /// [`Error::RegionError`]. See [`Error::MultipleKeyErrors`]. + #[error("Multiple region errors: {0:?}")] + MultipleRegionErrors(Vec), /// Invalid ColumnFamily #[error("Unsupported column family {}", _0)] ColumnFamilyError(String), diff --git a/src/request/plan.rs b/src/request/plan.rs index b7fac56f..4f2c17a8 100644 --- a/src/request/plan.rs +++ b/src/request/plan.rs @@ -132,6 +132,19 @@ where .collect()) } +/// Did this error come from the response body, rather than from a failure to obtain a +/// response at all? +/// +/// These are the two error types the `HasKeyErrors` response impls construct — see +/// `has_key_error!` / `extract_errors` (a `kvrpcpb::KeyError` field) and `has_str_error!` +/// (a raw endpoint's string `error` field). Everything else reaching `key_errors()` got +/// there through the catch-all arm of `HasKeyErrors for Result`, which reports +/// any per-shard `Err` — a transport failure, an exhausted retry — and must not be +/// relabelled as something the server said about a key. +fn is_response_error(e: &Error) -> bool { + matches!(e, Error::KeyError(_) | Error::KvError { .. }) +} + pub struct RetryableMultiRegion { pub(super) inner: P, pub pd_client: Arc, @@ -862,15 +875,23 @@ where Ok(()) => { result.resolved_locks += lock_size; } - Err(Error::ExtractedErrors(mut errors)) => { - // Propagate errors to `retry_multi_region` for retry. - if let Error::RegionError(e) = errors.pop().unwrap() { + // Propagate errors to `retry_multi_region` for retry. The variant now + // says which kind arrived, so neither arm has to pop an element to find + // out. That also fixes a real leak in the old single-variant code: it + // popped the last error to type-test it, and on the key-error path + // assigned the *remaining* vec to `key_error` — so a lone key error was + // dropped and `key_errors()` went on to report `Some(vec![])`, an error + // carrying no errors. + Err(Error::MultipleRegionErrors(mut errors)) => { + if let Some(Error::RegionError(e)) = errors.pop() { result.region_error = Some(*e); - } else { - result.key_error = Some(errors); } return Ok(result); } + Err(Error::MultipleKeyErrors(errors)) => { + result.key_error = Some(errors); + return Ok(result); + } Err(e) => { return Err(e); } @@ -915,10 +936,34 @@ where async fn execute(&self) -> Result { let mut result = self.inner.execute().await?; + // Each branch names what it found. Previously both produced one + // `ExtractedErrors`, so a caller had to pop an element and inspect its type to + // learn which kind it was holding — and whether a caller saw that variant at + // all depended on whether a collapsing merge sat between the retry layer and + // this plan (see `MultipleKeyErrors`' docs). if let Some(errors) = result.key_errors() { - Err(Error::ExtractedErrors(errors)) + // `HasKeyErrors for Result` yields *any* per-shard `Err`, not only + // errors the response carried — that is how a plan keeping its + // `Vec>` (see `retry_multi_region_preserve_results`) carries a hard + // failure through. Wrap only what the response itself reported; a transport + // or exhausted-retry error keeps its own class instead of being relabelled. + // + // "Reported by the response" is exactly the two error types the `HasKeyErrors` + // response impls construct: `KeyError` from a `kvrpcpb::KeyError` field + // (`has_key_error!`, `extract_errors`) and `KvError` from a raw endpoint's + // string `error` field (`has_str_error!`). Both must be treated alike, or + // plan shape would still decide the variant for raw operations — the very + // thing this split removes. + if errors.iter().all(is_response_error) { + Err(Error::MultipleKeyErrors(errors)) + } else { + Err(errors + .into_iter() + .find(|e| !is_response_error(e)) + .expect("a non-response error exists: the all() test above failed")) + } } else if let Some(errors) = result.region_errors() { - Err(Error::ExtractedErrors( + Err(Error::MultipleRegionErrors( errors .into_iter() .map(|e| Error::RegionError(Box::new(e))) @@ -1062,4 +1107,215 @@ mod test { assert_eq!(results, vec![0, 1, 2]); } + + /// A single-shard plan whose response carries a key error or a region error. + #[derive(Clone)] + struct FailingPlan { + key_error: bool, + } + + #[async_trait] + impl Plan for FailingPlan { + type Result = kvrpcpb::GetResponse; + + async fn execute(&self) -> Result { + let mut resp = kvrpcpb::GetResponse::default(); + if self.key_error { + resp.error = Some(kvrpcpb::KeyError { + abort: "boom".to_owned(), + ..Default::default() + }); + } else { + resp.region_error = Some(errorpb::Error { + server_is_busy: Some(errorpb::ServerIsBusy::default()), + ..Default::default() + }); + } + Ok(resp) + } + } + + impl Shardable for FailingPlan { + type Shard = (); + + fn shards( + &self, + _: &Arc, + ) -> BoxStream<'static, crate::Result<(Self::Shard, RegionWithLeader)>> { + Box::pin(stream::iter(vec![Ok(((), MockPdClient::region1()))])).boxed() + } + + fn apply_shard(&mut self, _: Self::Shard) {} + + fn apply_store(&mut self, _: &crate::store::RegionStore) -> Result<()> { + Ok(()) + } + } + + fn retryable(key_error: bool) -> RetryableMultiRegion { + RetryableMultiRegion { + inner: FailingPlan { key_error }, + pd_client: Arc::new(MockPdClient::default()), + backoff: Backoff::no_backoff(), + preserve_region_results: false, + } + } + + /// THE INVARIANT THIS REFACTOR BUYS. A response key error must surface as + /// `MultipleKeyErrors` whether or not a collapsing merge sits between the retry + /// layer and `ExtractError`. + /// + /// Before the split these two shapes produced *different* variants — without a + /// merge, `ExtractError` normalized the per-shard `Vec>` into + /// `ExtractedErrors`; with `CollectSingle`, the merge popped the `Result` out first + /// so `MultipleKeyErrors` escaped untouched. Callers matched one and silently missed + /// the other, and adding a merge to any plan flipped it with no compile error. + #[tokio::test] + async fn key_errors_surface_under_one_variant_whatever_the_plan_shape() { + // Shape A: no collapsing merge — ExtractError sees Vec>. + let unmerged = ExtractError { + inner: retryable(true), + }; + assert!( + matches!(unmerged.execute().await, Err(Error::MultipleKeyErrors(_))), + "unmerged plan must report key errors as MultipleKeyErrors" + ); + + // Shape B: CollectSingle pops the single Result out before ExtractError. + let merged = ExtractError { + inner: MergeResponse { + inner: retryable(true), + merge: CollectSingle, + phantom: PhantomData, + }, + }; + assert!( + matches!(merged.execute().await, Err(Error::MultipleKeyErrors(_))), + "merged plan must report key errors as MultipleKeyErrors too" + ); + } + + /// A bare plan whose response carries a region error, with no retry layer above it. + #[derive(Clone)] + struct RegionErrPlan; + + #[async_trait] + impl Plan for RegionErrPlan { + type Result = kvrpcpb::GetResponse; + + async fn execute(&self) -> Result { + Ok(kvrpcpb::GetResponse { + region_error: Some(errorpb::Error { + server_is_busy: Some(errorpb::ServerIsBusy::default()), + ..Default::default() + }), + ..Default::default() + }) + } + } + + /// Raw endpoints report failures as a string `error` field, which `has_str_error!` + /// turns into `Error::KvError` rather than `Error::KeyError`. That must be treated + /// as response-originated too — otherwise `batch_put_with_ttl` (no merge) would + /// return a bare `KvError` while `put_with_ttl` (CollectSingle) returned + /// `MultipleKeyErrors`, and plan shape would still decide the public variant. + #[tokio::test] + async fn raw_string_errors_are_response_errors_under_either_plan_shape() { + #[derive(Clone)] + struct RawErrPlan; + + #[async_trait] + impl Plan for RawErrPlan { + type Result = kvrpcpb::RawPutResponse; + + async fn execute(&self) -> Result { + Ok(kvrpcpb::RawPutResponse { + error: "boom".to_owned(), + ..Default::default() + }) + } + } + + // Unmerged: ExtractError classifies the response's string error. + let unmerged = ExtractError { inner: RawErrPlan }; + match unmerged.execute().await { + Err(Error::MultipleKeyErrors(errors)) => { + assert!(matches!(errors.as_slice(), [Error::KvError { .. }])); + } + other => panic!("want MultipleKeyErrors([KvError]), got {other:?}"), + } + + // And a shard error carried through as a Vec> classifies the same way. + #[derive(Clone)] + struct PreservedRawErr; + + #[async_trait] + impl Plan for PreservedRawErr { + type Result = Vec>; + + async fn execute(&self) -> Result { + Ok(vec![Ok(kvrpcpb::RawPutResponse { + error: "boom".to_owned(), + ..Default::default() + })]) + } + } + + let preserved = ExtractError { + inner: PreservedRawErr, + }; + assert!( + matches!(preserved.execute().await, Err(Error::MultipleKeyErrors(_))), + "the same response error must not change variant with plan shape" + ); + } + + /// A hard per-shard failure must keep its own class. `HasKeyErrors for + /// Result` reports *any* `Err` as a key error, so without a guard in + /// `ExtractError` a gRPC or exhausted-retry failure carried through a + /// preserve-results plan would be relabelled `MultipleKeyErrors` — the exact + /// mislabelling this refactor exists to remove. + #[tokio::test] + async fn a_non_key_error_is_not_relabelled_as_a_key_error() { + #[derive(Clone)] + struct PreservedShardErr; + + #[async_trait] + impl Plan for PreservedShardErr { + type Result = Vec>; + + async fn execute(&self) -> Result { + Ok(vec![Err(Error::Unimplemented)]) + } + } + + let plan = ExtractError { + inner: PreservedShardErr, + }; + match plan.execute().await { + Err(Error::Unimplemented) => {} + other => panic!("want the hard error propagated unchanged, got {other:?}"), + } + } + + /// The region-error counterpart: a distinct variant, so a caller no longer has to + /// pop an element and type-test it to learn which kind it is holding. + /// + /// Note the shape. `RetryableMultiRegion` consumes region errors itself — it retries + /// them and, once the backoff is spent, returns a bare `Error::RegionError`. So + /// `ExtractError` only ever sees a region error for a plan with no retry layer above + /// it, which is exactly `resolve_lock_with_retry`'s shape: it drives the backoff + /// itself and needs the error handed back rather than retried underneath it. + #[tokio::test] + async fn region_errors_surface_as_multiple_region_errors() { + let plan = ExtractError { + inner: RegionErrPlan, + }; + match plan.execute().await { + Err(Error::MultipleRegionErrors(errors)) => { + assert!(matches!(errors.as_slice(), [Error::RegionError(_)])); + } + other => panic!("want MultipleRegionErrors, got {other:?}"), + } + } } diff --git a/src/transaction/lock.rs b/src/transaction/lock.rs index 4a8adb63..43d0fcce 100644 --- a/src/transaction/lock.rs +++ b/src/transaction/lock.rs @@ -184,37 +184,40 @@ async fn resolve_lock_with_retry( Ok(_) => { return Ok(ver_id); } - // Retry on region error - Err(Error::ExtractedErrors(mut errors)) => { - // ResolveLockResponse can have at most 1 error - match errors.pop() { - Some(Error::RegionError(e)) => match backoff.next_delay_duration() { - Some(duration) => { - let region_error_resolved = - handle_region_error(pd_client.clone(), *e, store.clone()).await?; - if !region_error_resolved { - sleep(duration).await; - } - continue; + // Retry on region error. `ResolveLockResponse` can have at most 1 error, and + // the variant now says which kind it is rather than requiring a type test on + // a popped element. + Err(Error::MultipleRegionErrors(mut errors)) => match errors.pop() { + Some(Error::RegionError(e)) => match backoff.next_delay_duration() { + Some(duration) => { + let region_error_resolved = + handle_region_error(pd_client.clone(), *e, store.clone()).await?; + if !region_error_resolved { + sleep(duration).await; } - None => return Err(Error::RegionError(e)), - }, - Some(Error::KeyError(key_err)) => { - // Keyspace is not truncated here because we need full key info for logging. - error!( - "resolve_lock error, unexpected resolve err: {:?}, lock: {{key: {}, start_version: {}, commit_version: {}, is_txn_file: {}}}", - key_err, - format_key_for_log(key), - start_version, - commit_version, - is_txn_file, - ); - return Err(Error::KeyError(key_err)); + continue; } - Some(e) => return Err(e), - None => unreachable!(), + None => return Err(Error::RegionError(e)), + }, + Some(e) => return Err(e), + None => unreachable!(), + }, + Err(Error::MultipleKeyErrors(mut errors)) => match errors.pop() { + Some(Error::KeyError(key_err)) => { + // Keyspace is not truncated here because we need full key info for logging. + error!( + "resolve_lock error, unexpected resolve err: {:?}, lock: {{key: {}, start_version: {}, commit_version: {}, is_txn_file: {}}}", + key_err, + format_key_for_log(key), + start_version, + commit_version, + is_txn_file, + ); + return Err(Error::KeyError(key_err)); } - } + Some(e) => return Err(e), + None => unreachable!(), + }, Err(e) if is_grpc_error(&e) => match backoff.next_delay_duration() { Some(duration) => { pd_client.invalidate_region_cache(ver_id.clone()).await; @@ -467,17 +470,26 @@ impl LockResolver { .plan(); let mut status: TransactionStatus = match plan.execute().await { Ok(status) => status, - Err(Error::ExtractedErrors(mut errors)) => match errors.pop() { - Some(Error::KeyError(key_err)) => { - if let Some(txn_not_found) = key_err.txn_not_found { - return Err(Error::TxnNotFound(txn_not_found)); + // Response key errors now reach us under exactly one variant, whichever + // shape the plan has: `single_shard_handler` and `ExtractError` both produce + // `MultipleKeyErrors`. Before the variants were split this arm had to accept + // two, because a collapsing merge decided which one surfaced — and matching + // only one left the lock-resolution heal path (`rollback_if_not_exist` after + // TTL expiry in `get_txn_status_from_lock`) unreachable, so an orphaned + // secondary lock poisoned its key forever. + Err(Error::MultipleKeyErrors(mut errors)) => { + match errors.pop() { + Some(Error::KeyError(key_err)) => { + if let Some(txn_not_found) = key_err.txn_not_found { + return Err(Error::TxnNotFound(txn_not_found)); + } + // TODO: handle primary mismatch error. + return Err(Error::KeyError(key_err)); } - // TODO: handle primary mismatch error. - return Err(Error::KeyError(key_err)); + Some(err) => return Err(err), + None => unreachable!(), } - Some(err) => return Err(err), - None => unreachable!(), - }, + } Err(err) => return Err(err), }; @@ -610,6 +622,7 @@ mod tests { use std::any::Any; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; + use std::sync::Mutex; use fail::FailScenario; use serial_test::serial; @@ -712,6 +725,80 @@ mod tests { assert_eq!(resolve_lock_count.load(Ordering::SeqCst), 1); } + /// An orphaned lock whose primary was never written: `CheckTxnStatus` answers + /// `txn_not_found`, which reaches us as `MultipleKeyErrors`. It must convert to + /// `Error::TxnNotFound` so that `get_txn_status_from_lock` retries with + /// `rollback_if_not_exist` and heals the key; failing to match that variant leaves + /// the lock unresolved forever. + #[tokio::test] + #[serial] + async fn test_resolve_locks_rolls_back_expired_lock_whose_primary_is_missing() { + let check_txn_status_reqs = Arc::new(Mutex::new(Vec::new())); + let resolve_lock_count = Arc::new(AtomicUsize::new(0)); + + let check_txn_status_reqs_captured = check_txn_status_reqs.clone(); + let resolve_lock_count_captured = resolve_lock_count.clone(); + let client = Arc::new(MockPdClient::new(MockKvClient::with_dispatch_hook( + move |req: &dyn Any| { + if let Some(req) = req.downcast_ref::() { + check_txn_status_reqs_captured + .lock() + .unwrap() + .push(req.rollback_if_not_exist); + + // Until the client asks TiKV to roll back a missing primary, TiKV can + // only report that it never saw the transaction. + if !req.rollback_if_not_exist { + let resp = kvrpcpb::CheckTxnStatusResponse { + error: Some(kvrpcpb::KeyError { + txn_not_found: Some(kvrpcpb::TxnNotFound { + start_ts: 1, + primary_key: vec![2], + }), + ..Default::default() + }), + ..Default::default() + }; + return Ok(Box::new(resp) as Box); + } + + // With rollback_if_not_exist set, TiKV writes the rollback tombstone. + // commit_version 0 + lock_ttl 0 + no lock_info => RolledBack. + let resp = kvrpcpb::CheckTxnStatusResponse { + action: kvrpcpb::Action::LockNotExistRollback as i32, + ..Default::default() + }; + return Ok(Box::new(resp) as Box); + } + if req.is::() { + resolve_lock_count_captured.fetch_add(1, Ordering::SeqCst); + return Ok(Box::::default() as Box); + } + panic!("unexpected request type: {:?}", req.type_id()); + }, + ))); + + let mut lock = kvrpcpb::LockInfo::default(); + lock.key = vec![1]; + lock.primary_lock = vec![2]; // a primary that was never written + lock.lock_version = 1; + lock.lock_ttl = 0; // expired under MockPdClient's Timestamp::default() + + let live_locks = resolve_locks(vec![lock], Timestamp::default(), client, Keyspace::Disable) + .await + .unwrap(); + + // The lock is gone, not left live. + assert!(live_locks.is_empty()); + // Asked twice: once optimistically, then again escalating to a rollback. + assert_eq!( + *check_txn_status_reqs.lock().unwrap(), + vec![false, true], + "txn_not_found must escalate to rollback_if_not_exist" + ); + assert_eq!(resolve_lock_count.load(Ordering::SeqCst), 1); + } + #[test] fn format_key_for_log_hex_encodes_the_prefix() { assert_eq!(format_key_for_log(b"hello"), "len=5, prefix=68656C6C6F"); diff --git a/src/transaction/transaction.rs b/src/transaction/transaction.rs index 61b87423..fbc39e5e 100644 --- a/src/transaction/transaction.rs +++ b/src/transaction/transaction.rs @@ -1490,7 +1490,7 @@ impl Committer { loop { match self.commit_primary().await { Ok(commit_version) => return Ok(commit_version), - Err(Error::ExtractedErrors(mut errors)) => match errors.pop() { + Err(Error::MultipleKeyErrors(mut errors)) => match errors.pop() { Some(Error::KeyError(key_err)) => { if let Some(expired) = key_err.commit_ts_expired { // Ref: https://github.com/tikv/client-go/blob/tidb-8.5/txnkv/transaction/commit.go