plan: Handle no leader error

.github/workflows/ci.yml

@@ -3,6 +3,7 @@ on:
   push:
     branches:
       - master
+      - cse
 concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
@@ -52,8 +53,13 @@ jobs:
 
   integration-test:
     name: integration test
+    strategy:
+      fail-fast: false
+      matrix:
+        case: ["integration-test-txn", "integration-test-raw"]
     env:
       CARGO_INCREMENTAL: 0
+      TIKV_VERSION: v7.5.5
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
@@ -69,14 +75,22 @@ jobs:
       - name: start tiup playground
         run: |
           # use latest stable version
-          ~/.tiup/bin/tiup install tikv pd
-          ~/.tiup/bin/tiup playground --mode tikv-slim --kv 3 --without-monitor --kv.config config/tikv.toml --pd.config config/pd.toml &
+          ~/.tiup/bin/tiup install tikv:${{ env.TIKV_VERSION }} pd:${{ env.TIKV_VERSION }}
+          ~/.tiup/bin/tiup playground ${{ env.TIKV_VERSION }} --mode tikv-slim --kv 3 --tag cluster --without-monitor --kv.config config/tikv.toml --pd.config config/pd.toml &
           while :; do
             echo "waiting cluster to be ready"
             [[ "$(curl -I http://127.0.0.1:2379/pd/api/v1/regions 2>/dev/null | head -n 1 | cut -d$' ' -f2)" -ne "405" ]] || break
             sleep 1
           done
       - name: Install latest nextest release
         uses: taiki-e/install-action@nextest
-      - name: integration test
-        run: MULTI_REGION=1 make integration-test
+      - name: Integration test
+        run: MULTI_REGION=1 make ${{ matrix.case }}
+      - name: Upload logs
+        if: failure()
+        uses: actions/upload-artifact@v4
+        with:
+          name: cluster-logs
+          path: |
+            ~/.tiup/data/cluster/tikv*/*.log
+            ~/.tiup/data/cluster/pd*/*.log

Cargo.toml

@@ -50,7 +50,7 @@ clap = "2"
 env_logger = "0.10"
 fail = { version = "0.4", features = ["failpoints"] }
 proptest = "1"
-proptest-derive = "0.3"
+proptest-derive = "0.5.1"
 reqwest = { version = "0.11", default-features = false, features = [
     "native-tls-vendored",
 ] }

Makefile

@@ -1,13 +1,17 @@
 export RUSTFLAGS=-Dwarnings
 
-.PHONY: default check unit-test integration-tests test doc docker-pd docker-kv docker all
+.PHONY: default check unit-test generate integration-tests integration-tests-txn integration-tests-raw test doc docker-pd docker-kv docker all
 
 export PD_ADDRS     ?= 127.0.0.1:2379
 export MULTI_REGION ?= 1
 
 ALL_FEATURES := integration-tests
 
-INTEGRATION_TEST_ARGS := --features "integration-tests"
+NEXTEST_ARGS := --config-file $(shell pwd)/config/nextest.toml -P ci
+
+INTEGRATION_TEST_ARGS := --features "integration-tests" --test-threads 1
+
+RUN_INTEGRATION_TEST := cargo nextest run ${NEXTEST_ARGS} --all ${INTEGRATION_TEST_ARGS}
 
 default: check
 
@@ -20,12 +24,15 @@ check: generate
 	cargo clippy --all-targets --features "${ALL_FEATURES}" -- -D clippy::all
 
 unit-test: generate
-	cargo nextest run --all --no-default-features
+	cargo nextest run ${NEXTEST_ARGS} --all --no-default-features
+
+integration-test: integration-test-txn integration-test-raw
+
+integration-test-txn: generate
+	$(RUN_INTEGRATION_TEST) txn_
 
-integration-test: generate
-	cargo test txn_ --all ${INTEGRATION_TEST_ARGS} -- --nocapture
-	cargo test raw_ --all ${INTEGRATION_TEST_ARGS} -- --nocapture
-	cargo test misc_ --all ${INTEGRATION_TEST_ARGS} -- --nocapture
+integration-test-raw: generate
+	$(RUN_INTEGRATION_TEST) raw_
 
 test: unit-test integration-test
 

config/nextest.toml

@@ -0,0 +1,8 @@
+[profile.ci]
+retries = 0
+fail-fast = false
+slow-timeout = { period = "60s", terminate-after = 3 } # Timeout 3m.
+failure-output = "final"
+
+[profile.ci.junit]
+path = "junit.xml"

config/tikv.toml

@@ -1,17 +1,19 @@
 [coprocessor]
-region-max-keys = 10
-region-split-keys = 7
+region-split-keys = 150
 batch-split-limit = 100
 
 [raftstore]
 region-split-check-diff = "1B"
 pd-heartbeat-tick-interval = "2s"
 pd-store-heartbeat-tick-interval = "5s"
 split-region-check-tick-interval = "1s"
-raft-entry-max-size = "1MB"
+raft-entry-max-size = "256KiB"
 
 [rocksdb]
 max-open-files = 10000
 
 [raftdb]
 max-open-files = 10000
+
+[storage]
+reserve-space = "0MiB"

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
-channel = "stable"
-components = ["rustfmt", "clippy"]
+channel = "1.84.1"
+components = ["rustfmt", "clippy", "rust-analyzer"]

src/common/errors.rs

@@ -5,6 +5,7 @@ use std::result;
 use thiserror::Error;
 
 use crate::proto::kvrpcpb;
+use crate::region::RegionVerId;
 use crate::BoundRange;
 
 /// An error originating from the TiKV client or dependencies.
@@ -89,8 +90,8 @@ pub enum Error {
     #[error("Region {} is not found in the response", region_id)]
     RegionNotFoundInResponse { region_id: u64 },
     /// No leader is found for the given id.
-    #[error("Leader of region {} is not found", region_id)]
-    LeaderNotFound { region_id: u64 },
+    #[error("Leader of region {} is not found", region.id)]
+    LeaderNotFound { region: RegionVerId },
     /// Scan limit exceeds the maximum
     #[error("Limit {} exceeds max scan limit {}", limit, max_limit)]
     MaxScanLimitExceeded { limit: u32, max_limit: u32 },

src/generated/backup.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// The message save the metadata of a backup.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

src/generated/cdcpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Header {

src/generated/configpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Status {

src/generated/coprocessor.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// \[start, end)
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

src/generated/deadlock.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct WaitForEntriesRequest {}

src/generated/debugpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct GetRequest {

src/generated/diagnosticspb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SearchLogRequest {

src/generated/disk_usage.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
 #[repr(i32)]
 pub enum DiskUsage {

src/generated/encryptionpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// General encryption metadata for any data type.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

src/generated/enginepb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CommandRequestHeader {

src/generated/eraftpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// The entry is a type of change that needs to be applied. It contains two data fields.
 /// While the fields are built into the model; their usage is determined by the entry_type.
 ///

src/generated/errorpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// NotLeader is the error variant that tells a request be handle by raft leader
 /// is sent to raft follower or learner.
 #[allow(clippy::derive_partial_eq_without_eq)]

src/generated/google.api.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// Defines the HTTP configuration for an API service. It contains a list of
 /// \[HttpRule\]\[google.api.HttpRule\], each specifying the mapping of an RPC method
 /// to one or more HTTP REST API methods.

src/generated/import_kvpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SwitchModeRequest {

src/generated/import_sstpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SwitchModeRequest {

src/generated/kvrpcpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// A transactional get command. Lookup a value for `key` in the transaction with
 /// starting timestamp = `version`.
 #[allow(clippy::derive_partial_eq_without_eq)]

src/generated/metapb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct Cluster {

src/generated/mod.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 pub mod backup {
     include!("backup.rs");
 }

src/generated/mpp.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// TaskMeta contains meta of a mpp plan, including query's ts and task address.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

src/generated/pdpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct RequestHeader {

src/generated/raft_cmdpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct GetRequest {

src/generated/raft_serverpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct RaftMessage {

src/generated/replication_modepb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 /// The replication status sync from PD to TiKV.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

src/generated/resource_usage_agent.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct CpuTimeRecord {

src/generated/span.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct SpanSet {

src/generated/tikvpb.rs

@@ -1,3 +1,4 @@
+// This file is @generated by prost-build.
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]
 pub struct BatchCommandsRequest {

src/kv/key.rs

@@ -2,7 +2,6 @@
 
 use std::fmt;
 use std::ops::Bound;
-use std::u8;
 
 #[allow(unused_imports)]
 #[cfg(test)]

src/kv/mod.rs

@@ -1,6 +1,5 @@
 // Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
 use std::fmt;
-use std::u8;
 
 mod bound_range;
 pub mod codec;
@@ -16,7 +15,7 @@ pub use value::Value;
 
 struct HexRepr<'a>(pub &'a [u8]);
 
-impl<'a> fmt::Display for HexRepr<'a> {
+impl fmt::Display for HexRepr<'_> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         for byte in self.0 {
             write!(f, "{byte:02X}")?;

src/kv/value.rs

@@ -11,5 +11,4 @@ const _PROPTEST_VALUE_MAX: usize = 1024 * 16; // 16 KB
 /// Since `Value` is just an alias for `Vec<u8>`, conversions to and from it are easy.
 ///
 /// Many functions which accept a `Value` accept an `Into<Value>`.
-
 pub type Value = Vec<u8>;

src/mock.rs

@@ -228,6 +228,8 @@ impl PdClient for MockPdClient {
 
     async fn invalidate_region_cache(&self, _ver_id: crate::region::RegionVerId) {}
 
+    async fn invalidate_store_cache(&self, _store_id: crate::region::StoreId) {}
+
     fn get_codec(&self) -> &Self::Codec {
         &self.codec
     }

src/pd/client.rs

@@ -19,6 +19,7 @@ use crate::proto::metapb;
 use crate::region::RegionId;
 use crate::region::RegionVerId;
 use crate::region::RegionWithLeader;
+use crate::region::StoreId;
 use crate::region_cache::RegionCache;
 use crate::request::codec::{ApiV1TxnCodec, Codec};
 use crate::store::KvConnect;
@@ -208,6 +209,8 @@ pub trait PdClient: Send + Sync + 'static {
 
     async fn invalidate_region_cache(&self, ver_id: RegionVerId);
 
+    async fn invalidate_store_cache(&self, store_id: StoreId);
+
     /// Get the codec carried by `PdClient`.
     /// The purpose of carrying the codec is to avoid passing it on so many calling paths.
     fn get_codec(&self) -> &Self::Codec;
@@ -283,6 +286,10 @@ impl<Cod: Codec, KvC: KvConnect + Send + Sync + 'static> PdClient for PdRpcClien
         self.region_cache.invalidate_region_cache(ver_id).await
     }
 
+    async fn invalidate_store_cache(&self, store_id: StoreId) {
+        self.region_cache.invalidate_store_cache(store_id).await
+    }
+
     fn get_codec(&self) -> &Self::Codec {
         self.codec
             .as_ref()

src/proto.rs

@@ -2,6 +2,7 @@
 
 #![allow(clippy::large_enum_variant)]
 #![allow(clippy::enum_variant_names)]
+#![allow(clippy::doc_lazy_continuation)]
 
 pub use protos::*;
 

src/raw/client.rs

@@ -3,7 +3,6 @@
 use core::ops::Range;
 use std::str::FromStr;
 use std::sync::Arc;
-use std::u32;
 
 use futures::StreamExt;
 use log::debug;

src/region.rs

@@ -47,7 +47,7 @@ impl RegionWithLeader {
         self.leader
             .as_ref()
             .ok_or(Error::LeaderNotFound {
-                region_id: self.region.id,
+                region: self.ver_id(),
             })
             .map(|l| {
                 let mut ctx = kvrpcpb::Context::default();
@@ -89,7 +89,7 @@ impl RegionWithLeader {
             .as_ref()
             .cloned()
             .ok_or_else(|| Error::LeaderNotFound {
-                region_id: self.id(),
+                region: self.ver_id(),
             })
             .map(|s| s.store_id)
     }

src/region_cache.rs

@@ -233,6 +233,11 @@ impl<C: RetryClientTrait> RegionCache<C> {
         }
     }
 
+    pub async fn invalidate_store_cache(&self, store_id: StoreId) {
+        let mut cache = self.store_cache.write().await;
+        cache.remove(&store_id);
+    }
+
     pub async fn read_through_all_stores(&self) -> Result<Vec<Store>> {
         let stores = self
             .inner_client