Skip to content

Commit ac95421

Browse files
pingyupingyu
authored
transaction: Handle "commit ts expired" error (#491)
Signed-off-by: Ping Yu <[email protected]>
1 parent fa78931 commit ac95421

File tree

6 files changed

+67
-15
lines changed

6 files changed

+67
-15
lines changed

‎src/common/errors.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ pub enum Error {
103103
#[error("{}", message)]
104104
InternalError { message: String },
105105
#[error("{0}")]
106-
StringError(String),
106+
OtherError(String),
107107
#[error("PessimisticLock error: {:?}", inner)]
108108
PessimisticLockError {
109109
inner: Box<Error>,

‎src/kv/mod.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ pub use key::Key;
1313
pub use kvpair::KvPair;
1414
pub use value::Value;
1515

16-
struct HexRepr<'a>(pub &'a [u8]);
16+
pub struct HexRepr<'a>(pub &'a [u8]);
1717

1818
impl fmt::Display for HexRepr<'_> {
1919
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {

‎src/region_cache.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ impl<C: RetryClientTrait> RegionCache<C> {
117117
return self.read_through_region_by_id(id).await;
118118
}
119119
}
120-
Err(Error::StringError(format!(
120+
Err(Error::OtherError(format!(
121121
"Concurrent PD requests failed for {MAX_RETRY_WAITING_CONCURRENT_REQUEST} times"
122122
)))
123123
}
@@ -315,7 +315,7 @@ mod test {
315315
.filter(|(_, r)| r.contains(&key.clone().into()))
316316
.map(|(_, r)| r.clone())
317317
.next()
318-
.ok_or_else(|| Error::StringError("MockRetryClient: region not found".to_owned()))
318+
.ok_or_else(|| Error::OtherError("MockRetryClient: region not found".to_owned()))
319319
}
320320

321321
async fn get_region_by_id(
@@ -330,7 +330,7 @@ mod test {
330330
.filter(|(id, _)| id == &&region_id)
331331
.map(|(_, r)| r.clone())
332332
.next()
333-
.ok_or_else(|| Error::StringError("MockRetryClient: region not found".to_owned()))
333+
.ok_or_else(|| Error::OtherError("MockRetryClient: region not found".to_owned()))
334334
}
335335

336336
async fn get_store(

‎src/store/errors.rs‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,5 @@
11
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
22

3-
use std::fmt::Display;
4-
53
use crate::proto::kvrpcpb;
64
use crate::Error;
75

@@ -162,11 +160,15 @@ impl HasKeyErrors for kvrpcpb::PessimisticRollbackResponse {
162160
}
163161
}
164162

165-
impl<T: HasKeyErrors, E: Display> HasKeyErrors for Result<T, E> {
163+
impl<T: HasKeyErrors> HasKeyErrors for Result<T, Error> {
166164
fn key_errors(&mut self) -> Option<Vec<Error>> {
167165
match self {
168166
Ok(x) => x.key_errors(),
169-
Err(e) => Some(vec![Error::StringError(e.to_string())]),
167+
Err(Error::MultipleKeyErrors(errs)) => Some(std::mem::take(errs)),
168+
Err(e) => Some(vec![std::mem::replace(
169+
e,
170+
Error::OtherError("".to_string()), // placeholder, no use.
171+
)]),
170172
}
171173
}
172174
}

‎src/transaction/transaction.rs‎

Lines changed: 53 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,15 @@ use derive_new::new;
1111
use fail::fail_point;
1212
use futures::prelude::*;
1313
use log::debug;
14+
use log::error;
15+
use log::info;
1416
use log::warn;
1517
use tokio::time::Duration;
1618

1719
use crate::backoff::Backoff;
1820
use crate::backoff::DEFAULT_REGION_BACKOFF;
1921
use crate::codec::ApiV1TxnCodec;
22+
use crate::kv::HexRepr;
2023
use crate::pd::PdClient;
2124
use crate::pd::PdRpcClient;
2225
use crate::proto::kvrpcpb;
@@ -1246,7 +1249,7 @@ impl<PdC: PdClient> Committer<PdC> {
12461249
let min_commit_ts = self.prewrite().await?;
12471250

12481251
fail_point!("after-prewrite", |_| {
1249-
Err(Error::StringError(
1252+
Err(Error::OtherError(
12501253
"failpoint: after-prewrite return error".to_owned(),
12511254
))
12521255
});
@@ -1260,7 +1263,7 @@ impl<PdC: PdClient> Committer<PdC> {
12601263
// FIXME: min_commit_ts == 0 => fallback to normal 2PC
12611264
min_commit_ts.unwrap()
12621265
} else {
1263-
match self.commit_primary().await {
1266+
match self.commit_primary_with_retry().await {
12641267
Ok(commit_ts) => commit_ts,
12651268
Err(e) => {
12661269
return if self.undetermined {
@@ -1365,6 +1368,11 @@ impl<PdC: PdClient> Committer<PdC> {
13651368
.plan();
13661369
plan.execute()
13671370
.inspect_err(|e| {
1371+
debug!(
1372+
"commit primary error: {:?}, start_ts: {}",
1373+
e,
1374+
self.start_version.version()
1375+
);
13681376
// We don't know whether the transaction is committed or not if we fail to receive
13691377
// the response. Then, we mark the transaction as undetermined and propagate the
13701378
// error to the user.
@@ -1377,6 +1385,48 @@ impl<PdC: PdClient> Committer<PdC> {
13771385
Ok(commit_version)
13781386
}
13791387

1388+
async fn commit_primary_with_retry(&mut self) -> Result<Timestamp> {
1389+
loop {
1390+
match self.commit_primary().await {
1391+
Ok(commit_version) => return Ok(commit_version),
1392+
Err(Error::ExtractedErrors(mut errors)) => match errors.pop() {
1393+
Some(Error::KeyError(key_err)) => {
1394+
if let Some(expired) = key_err.commit_ts_expired {
1395+
// Ref: https://github.com/tikv/client-go/blob/tidb-8.5/txnkv/transaction/commit.go
1396+
info!("2PC commit_ts rejected by TiKV, retry with a newer commit_ts, start_ts: {}",
1397+
self.start_version.version());
1398+
1399+
let primary_key = self.primary_key.as_ref().unwrap();
1400+
if primary_key != expired.key.as_ref() {
1401+
error!("2PC commit_ts rejected by TiKV, but the key is not the primary key, start_ts: {}, key: {}, primary: {:?}",
1402+
self.start_version.version(), HexRepr(&expired.key), primary_key);
1403+
return Err(Error::OtherError("2PC commitTS rejected by TiKV, but the key is not the primary key".to_string()));
1404+
}
1405+
1406+
// Do not retry for a txn which has a too large min_commit_ts.
1407+
// 3600000 << 18 = 943718400000
1408+
if expired
1409+
.min_commit_ts
1410+
.saturating_sub(expired.attempted_commit_ts)
1411+
> 943718400000
1412+
{
1413+
let msg = format!("2PC min_commit_ts is too large, we got min_commit_ts: {}, and attempted_commit_ts: {}",
1414+
expired.min_commit_ts, expired.attempted_commit_ts);
1415+
return Err(Error::OtherError(msg));
1416+
}
1417+
continue;
1418+
} else {
1419+
return Err(Error::KeyError(key_err));
1420+
}
1421+
}
1422+
Some(err) => return Err(err),
1423+
None => unreachable!(),
1424+
},
1425+
Err(err) => return Err(err),
1426+
}
1427+
}
1428+
}
1429+
13801430
async fn commit_secondary(self, commit_version: Timestamp) -> Result<()> {
13811431
debug!("committing secondary");
13821432
let mutations_len = self.mutations.len();
@@ -1394,7 +1444,7 @@ impl<PdC: PdClient> Committer<PdC> {
13941444
let percent = percent.unwrap().parse::<usize>().unwrap();
13951445
new_len = mutations_len * percent / 100;
13961446
if new_len == 0 {
1397-
Err(Error::StringError(
1447+
Err(Error::OtherError(
13981448
"failpoint: before-commit-secondary return error".to_owned(),
13991449
))
14001450
} else {

‎tests/common/ctl.rs‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -10,16 +10,16 @@ use crate::common::Result;
1010
pub async fn get_region_count() -> Result<u64> {
1111
let res = reqwest::get(format!("http://{}/pd/api/v1/regions", pd_addrs()[0]))
1212
.await
13-
.map_err(|e| Error::StringError(e.to_string()))?;
13+
.map_err(|e| Error::OtherError(e.to_string()))?;
1414

1515
let body = res
1616
.text()
1717
.await
18-
.map_err(|e| Error::StringError(e.to_string()))?;
18+
.map_err(|e| Error::OtherError(e.to_string()))?;
1919
let value: serde_json::Value = serde_json::from_str(body.as_ref()).unwrap_or_else(|err| {
2020
panic!("invalid body: {:?}, error: {:?}", body, err);
2121
});
2222
value["count"]
2323
.as_u64()
24-
.ok_or_else(|| Error::StringError("pd region count does not return an integer".to_owned()))
24+
.ok_or_else(|| Error::OtherError("pd region count does not return an integer".to_owned()))
2525
}

0 commit comments

Comments
 (0)