Repository: tikv/client-go
Pull request #1565 (2025-01-26 10:15:53)

locate: refactor RegionRequestSender.SendReqCtx
This PR extracts the iteration of retry loop in `SendReqCtx` into `sendReqState.next`. We will reuse `sendReqState` in async client API.

ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [zyguan](https://github.com/zyguan), **ensuring that each of them provides their approval before proceeding**. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=tikv%2Fclient-go).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/tikv/client-go/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["zyguan"]} -->

Patch:
diff --git a/OWNERS b/OWNERS
index f70f2cb..af3aca1 100644
--- a/OWNERS
+++ b/OWNERS
@@ -41,6 +41,7 @@ approvers:
   - tabokie
   - TennyZhuang
   - tonyxuqqi
+  - v01dstar
   - yiwu-arbug
   - you06
   - youjiali1995
@@ -59,6 +60,7 @@ reviewers:
   - gozssky
   - haojinming
   - hbisheng
+  - hhwyt
   - HuSharp
   - jayzhan211
   - Jibbow
@@ -73,7 +75,6 @@ reviewers:
   - rleungx
   - Rustin170506
   - tier-cap
-  - v01dstar
   - wjhuang2016
   - wshwsh12
   - Xuanwo
diff --git a/examples/gcworker/go.mod b/examples/gcworker/go.mod
index 7af1d6b..5e675bd 100644
--- a/examples/gcworker/go.mod
+++ b/examples/gcworker/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/rawkv/go.mod b/examples/rawkv/go.mod
index b779427..83fe302 100644
--- a/examples/rawkv/go.mod
+++ b/examples/rawkv/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/1pc_txn/go.mod b/examples/txnkv/1pc_txn/go.mod
index 4926479..17e0129 100644
--- a/examples/txnkv/1pc_txn/go.mod
+++ b/examples/txnkv/1pc_txn/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/async_commit/go.mod b/examples/txnkv/async_commit/go.mod
index 066120e..2b53c80 100644
--- a/examples/txnkv/async_commit/go.mod
+++ b/examples/txnkv/async_commit/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/delete_range/go.mod b/examples/txnkv/delete_range/go.mod
index 2f9d244..246fd19 100644
--- a/examples/txnkv/delete_range/go.mod
+++ b/examples/txnkv/delete_range/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/go.mod b/examples/txnkv/go.mod
index 5a23a19..3db30c1 100644
--- a/examples/txnkv/go.mod
+++ b/examples/txnkv/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/pessimistic_txn/go.mod b/examples/txnkv/pessimistic_txn/go.mod
index 016964a..d97309c 100644
--- a/examples/txnkv/pessimistic_txn/go.mod
+++ b/examples/txnkv/pessimistic_txn/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/examples/txnkv/unsafedestoryrange/go.mod b/examples/txnkv/unsafedestoryrange/go.mod
index 7f5c8d1..4ba2252 100644
--- a/examples/txnkv/unsafedestoryrange/go.mod
+++ b/examples/txnkv/unsafedestoryrange/go.mod
@@ -5,6 +5,7 @@ go 1.23
 require github.com/tikv/client-go/v2 v2.0.0
 
 require (
+	github.com/VividCortex/ewma v1.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudfoundry/gosigar v1.3.6 // indirect
@@ -18,10 +19,11 @@ require (
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.1.0 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c // indirect
-	github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c // indirect
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 // indirect
+	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 // indirect
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 // indirect
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_golang v1.20.5 // indirect
@@ -30,7 +32,7 @@ require (
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a // indirect
-	github.com/tikv/pd/client v0.0.0-20241220053006-461b86adc78d // indirect
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 // indirect
 	github.com/twmb/murmur3 v1.1.3 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.10 // indirect
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
@@ -39,10 +41,10 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sync v0.10.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
 	google.golang.org/grpc v1.64.0 // indirect
diff --git a/go.mod b/go.mod
index 374cba6..d13a2a8 100644
--- a/go.mod
+++ b/go.mod
@@ -16,21 +16,21 @@ require (
 	github.com/pingcap/errors v0.11.5-0.20211224045212-9687c2b0f87c
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86
 	github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989
-	github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1
 	github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.20.5
 	github.com/prometheus/client_model v0.6.1
 	github.com/stretchr/testify v1.9.0
 	github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a
-	github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42
 	github.com/twmb/murmur3 v1.1.3
 	go.etcd.io/etcd/api/v3 v3.5.10
 	go.etcd.io/etcd/client/v3 v3.5.10
 	go.uber.org/atomic v1.11.0
 	go.uber.org/goleak v1.2.0
 	go.uber.org/zap v1.26.0
-	golang.org/x/sync v0.7.0
+	golang.org/x/sync v0.10.0
 	google.golang.org/grpc v1.63.2
 )
 
@@ -51,9 +51,9 @@ require (
 	go.etcd.io/etcd/client/pkg/v3 v3.5.10 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/exp v0.0.0-20230711005742-c3f37128e5a4 // indirect
-	golang.org/x/net v0.26.0 // indirect
-	golang.org/x/sys v0.22.0 // indirect
-	golang.org/x/text v0.16.0 // indirect
+	golang.org/x/net v0.33.0 // indirect
+	golang.org/x/sys v0.28.0 // indirect
+	golang.org/x/text v0.21.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240304212257-790db918fca8 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240401170217-c3f982113cda // indirect
 	google.golang.org/protobuf v1.34.2 // indirect
diff --git a/go.sum b/go.sum
index 15476ef..e7d5f9a 100644
--- a/go.sum
+++ b/go.sum
@@ -79,8 +79,8 @@ github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86 h1:tdMsjOqUR7YXH
 github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86/go.mod h1:exzhVYca3WRtd6gclGNErRWb1qEgff3LYta0LvRmON4=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
-github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037 h1:xYNSJjYNur4Dr5bV+9BXK9n5E0T1zlcAN25XX68+mOg=
-github.com/pingcap/kvproto v0.0.0-20241120071417-b5b7843d9037/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 h1:rTAyiswGyWSGHJVa4Mkhdi8YfGqfA4LrUVKsH9nrJ8E=
+github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3 h1:HR/ylkkLmGdSSDaD8IDP+SZrdhV1Kibl9KrHxJ9eciw=
 github.com/pingcap/log v1.1.1-0.20221110025148-ca232912c9f3/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
@@ -113,8 +113,8 @@ github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsT
 github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a h1:J/YdBZ46WKpXsxsW93SG+q0F8KI+yFrcIDT4c/RNoc4=
 github.com/tiancaiamao/gp v0.0.0-20221230034425-4025bc8a4d4a/go.mod h1:h4xBhSNtOeEosLJ4P7JyKXX7Cabg7AVkWCK5gV2vOrM=
-github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de h1:fefo6U56UMca1CfwY/FVhfVcinyDgXgffQQp9lfMeLg=
-github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de/go.mod h1:6xcCIcECSowarZ9+AqdURngHzS+t5w2x3qhwg1moj4o=
+github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 h1:hvB0tf4u5ZU1vAew3MmbyxdYuogDLjIZYzSFt1ACFqo=
+github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42/go.mod h1:veL5JbFL99G5cvS/Q2cLiFQklIiSrVgsBCwFnXc+Zgo=
 github.com/twmb/murmur3 v1.1.3 h1:D83U0XYKcHRYwYIpBKf3Pks91Z0Byda/9SJ8B6EMRcA=
 github.com/twmb/murmur3 v1.1.3/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
@@ -163,27 +163,27 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.26.0 h1:soB7SVo0PWrY4vPW/+ay0jKDNScG2X9wFeYlXIvJsOQ=
-golang.org/x/net v0.26.0/go.mod h1:5YKkiSynbBIh3p6iOc/vibscux0x38BZDkn8sCUPxHE=
+golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I=
+golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M=
-golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
+golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.22.0 h1:RI27ohtqKCnwULzJLqkv897zojh5/DwS/ENaMzUOaWI=
-golang.org/x/sys v0.22.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.28.0 h1:Fksou7UEQUWlKvIdsqzJmUmCX3cZuD2+P3XyyzwMhlA=
+golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.16.0 h1:a94ExnEXNtEwYLGJSIUxnWoxoRz/ZcCsV63ROupILh4=
-golang.org/x/text v0.16.0/go.mod h1:GhwF1Be+LQoKShO3cGOHzqOgRrGaYc9AvblQOmPVHnI=
+golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
+golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/internal/locate/region_request.go b/internal/locate/region_request.go
index 16e5688..b6b215a 100644
--- a/internal/locate/region_request.go
+++ b/internal/locate/region_request.go
@@ -45,7 +45,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -62,6 +61,7 @@ import (
 	"github.com/tikv/client-go/v2/internal/logutil"
 	"github.com/tikv/client-go/v2/kv"
 	"github.com/tikv/client-go/v2/metrics"
+	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikvrpc"
 	"github.com/tikv/client-go/v2/util"
 	"github.com/tikv/pd/client/errs"
@@ -732,328 +732,189 @@ func IsFakeRegionError(err *errorpb.Error) bool {
 
 const slowLogSendReqTime = 100 * time.Millisecond
 
-// SendReqCtx sends a request to tikv server and return response and RPCCtx of this RPC.
-func (s *RegionRequestSender) SendReqCtx(
+// sendReqState represents the state of sending request with retry, which allows us to construct a state and start to
+// retry from that state.
+type sendReqState struct {
+	*RegionRequestSender
+
+	vars struct {
+		rpcCtx    *RPCContext
+		resp      *tikvrpc.Response
+		regionErr *errorpb.Error
+		err       error
+		msg       string
+		sendTimes int
+	}
+}
+
+// next encapsulates one iteration of the retry loop. calling `next` will handle send error (s.vars.err) or region error
+// (s.vars.regionErr) if one of them exists. When the error is retriable, `next` then constructs a new RPCContext and
+// sends the request again. `next` returns true if the retry loop should stop, either because the request is done or
+// exhausted (cannot complete by retrying).
+func (s *sendReqState) next(
 	bo *retry.Backoffer,
 	req *tikvrpc.Request,
 	regionID RegionVerID,
 	timeout time.Duration,
 	et tikvrpc.EndpointType,
-	opts ...StoreSelectorOption,
-) (
-	resp *tikvrpc.Response,
-	rpcCtx *RPCContext,
-	retryTimes int,
-	err error,
-) {
-	if span := opentracing.SpanFromContext(bo.GetCtx()); span != nil && span.Tracer() != nil {
-		span1 := span.Tracer().StartSpan("regionRequest.SendReqCtx", opentracing.ChildOf(span.Context()))
-		defer span1.Finish()
-		bo.SetCtx(opentracing.ContextWithSpan(bo.GetCtx(), span1))
+	opts []StoreSelectorOption,
+) (done bool) {
+	// check whether the session/query is killed during the Next()
+	if err := bo.CheckKilled(); err != nil {
+		s.vars.resp, s.vars.err = nil, err
+		return true
 	}
 
-	if resp, err = failpointSendReqResult(req, et); err != nil || resp != nil {
-		return
+	// handle send error
+	if s.vars.err != nil {
+		if e := s.onSendFail(bo, s.vars.rpcCtx, req, s.vars.err); e != nil {
+			s.vars.rpcCtx, s.vars.resp = nil, nil
+			s.vars.msg = fmt.Sprintf("failed to handle send error: %v", s.vars.err)
+			return true
+		}
+		s.vars.err = nil
 	}
 
-	if err = s.validateReadTS(bo.GetCtx(), req); err != nil {
-		logutil.Logger(bo.GetCtx()).Error("validate read ts failed for request", zap.Stringer("reqType", req.Type), zap.Stringer("req", req.Req.(fmt.Stringer)), zap.Stringer("context", &req.Context), zap.Stack("stack"), zap.Error(err))
-		return nil, nil, 0, err
+	// handle region error
+	if s.vars.regionErr != nil {
+		retry, err := s.onRegionError(bo, s.vars.rpcCtx, req, s.vars.regionErr)
+		if err != nil {
+			s.vars.rpcCtx, s.vars.resp = nil, nil
+			s.vars.err = err
+			s.vars.msg = fmt.Sprintf("failed to handle region error: %v", err)
+			return true
+		}
+		if !retry {
+			s.vars.msg = fmt.Sprintf("met unretriable region error: %T", s.vars.regionErr)
+			return true
+		}
+		s.vars.regionErr = nil
 	}
 
-	// If the MaxExecutionDurationMs is not set yet, we set it to be the RPC timeout duration
-	// so TiKV can give up the requests whose response TiDB cannot receive due to timeout.
-	if req.Context.MaxExecutionDurationMs == 0 {
-		req.Context.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
+	s.vars.rpcCtx, s.vars.resp = nil, nil
+	if !req.IsRetryRequest && s.vars.sendTimes > 0 {
+		req.IsRetryRequest = true
 	}
 
-	s.reset()
-	startTime := time.Now()
-	startBackOff := bo.GetTotalSleep()
-	retryTimes = 0
-	defer func() {
-		if retryTimes > 0 {
-			metrics.TiKVRequestRetryTimesHistogram.Observe(float64(retryTimes))
-		}
-	}()
-
-	if req.StaleRead {
-		defer func() {
-			if retryTimes == 0 {
-				metrics.StaleReadHitCounter.Add(1)
-			} else {
-				metrics.StaleReadMissCounter.Add(1)
-			}
-		}()
+	s.vars.rpcCtx, s.vars.err = s.getRPCContext(bo, req, regionID, et, opts...)
+	if s.vars.err != nil {
+		return true
 	}
 
-	for {
-		if retryTimes > 0 {
-			if retryTimes%100 == 0 {
-				logutil.Logger(bo.GetCtx()).Warn(
-					"retry",
-					zap.Uint64("region", regionID.GetID()),
-					zap.Int("times", retryTimes),
-				)
-			}
+	if _, err := util.EvalFailpoint("invalidCacheAndRetry"); err == nil {
+		// cooperate with tikvclient/setGcResolveMaxBackoff
+		if c := bo.GetCtx().Value("injectedBackoff"); c != nil {
+			s.vars.regionErr = &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}
+			s.vars.resp, s.vars.err = tikvrpc.GenRegionErrorResp(req, s.vars.regionErr)
+			return true
 		}
+	}
 
-		rpcCtx, err = s.getRPCContext(bo, req, regionID, et, opts...)
-		if err != nil {
-			return nil, nil, retryTimes, err
-		}
+	if s.vars.rpcCtx == nil {
+		// TODO(youjiali1995): remove it when using the replica selector for all requests.
+		// If the region is not found in cache, it must be out
+		// of date and already be cleaned up. We can skip the
+		// RPC by returning RegionError directly.
 
-		if _, err := util.EvalFailpoint("invalidCacheAndRetry"); err == nil {
-			// cooperate with tikvclient/setGcResolveMaxBackoff
-			if c := bo.GetCtx().Value("injectedBackoff"); c != nil {
-				resp, err = tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}})
-				return resp, nil, retryTimes, err
-			}
-		}
-		if rpcCtx == nil {
-			// TODO(youjiali1995): remove it when using the replica selector for all requests.
-			// If the region is not found in cache, it must be out
-			// of date and already be cleaned up. We can skip the
-			// RPC by returning RegionError directly.
-
-			// TODO: Change the returned error to something like "region missing in cache",
-			// and handle this error like EpochNotMatch, which means to re-split the request and retry.
-			if s.replicaSelector != nil {
-				if err := s.replicaSelector.backoffOnNoCandidate(bo); err != nil {
-					return nil, nil, retryTimes, err
-				}
-				if cost := time.Since(startTime); cost > slowLogSendReqTime || cost > timeout || bo.GetTotalSleep() > 1000 {
-					s.logSendReqError(bo, "throwing pseudo region error due to no replica available", regionID, retryTimes, req, cost, bo.GetTotalSleep()-startBackOff, timeout)
-				}
-			}
-			resp, err = tikvrpc.GenRegionErrorResp(req, &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}})
-			return resp, nil, retryTimes, err
-		}
-		// patch the access location if it is not set under region request sender. which includes the coprocessor,
-		// txn relative tikv request.
-		// note: MPP not use this path. need specified in the MPP layer.
-		patchAccessLocation := func() {
-			if s.replicaSelector.target.store.IsLabelsMatch(s.replicaSelector.option.labels) {
-				req.AccessLocation = kv.AccessLocalZone
-			} else {
-				req.AccessLocation = kv.AccessCrossZone
+		// TODO: Change the returned error to something like "region missing in cache",
+		// and handle this error like EpochNotMatch, which means to re-split the request and retry.
+		if s.replicaSelector != nil {
+			if s.vars.err = s.replicaSelector.backoffOnNoCandidate(bo); s.vars.err != nil {
+				return true
 			}
 		}
-		if s.replicaSelector != nil &&
-			s.replicaSelector.target != nil &&
-			req.AccessLocation == kv.AccessUnknown &&
-			len(s.replicaSelector.option.labels) != 0 {
-			// patch the access location if it is not set under region request sender.
-			patchAccessLocation()
-		}
-		logutil.Eventf(bo.GetCtx(), "send %s request to region %d at %s", req.Type, regionID.id, rpcCtx.Addr)
-		s.storeAddr = rpcCtx.Addr
+		s.vars.regionErr = &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}
+		s.vars.resp, s.vars.err = tikvrpc.GenRegionErrorResp(req, s.vars.regionErr)
+		s.vars.msg = "throwing pseudo region error due to no replica available"
+		return true
+	}
 
-		if _, err := util.EvalFailpoint("beforeSendReqToRegion"); err == nil {
-			if hook := bo.GetCtx().Value("sendReqToRegionHook"); hook != nil {
-				h := hook.(func(*tikvrpc.Request))
-				h(req)
-			}
+	if s.replicaSelector != nil &&
+		s.replicaSelector.target != nil &&
+		req.AccessLocation == kv.AccessUnknown &&
+		len(s.replicaSelector.option.labels) != 0 {
+		// patch the access location if it is not set under region request sender.
+		if s.replicaSelector.target.store.IsLabelsMatch(s.replicaSelector.option.labels) {
+			req.AccessLocation = kv.AccessLocalZone
+		} else {
+			req.AccessLocation = kv.AccessCrossZone
 		}
+	}
 
-		req.Context.ClusterId = rpcCtx.ClusterID
-		if req.InputRequestSource != "" && s.replicaSelector != nil {
-			patchRequestSource(req, s.replicaSelector.replicaType())
-		}
-		// RPCClient.SendRequest will attach `req.Context` thus skip attaching here to reduce overhead.
-		if err := tikvrpc.SetContextNoAttach(req, rpcCtx.Meta, rpcCtx.Peer); err != nil {
-			return nil, nil, retryTimes, err
-		}
-		if s.replicaSelector != nil {
-			if err := s.replicaSelector.backoffOnRetry(rpcCtx.Store, bo); err != nil {
-				return nil, nil, retryTimes, err
-			}
-		}
+	logutil.Eventf(bo.GetCtx(), "send %s request to region %d at %s", req.Type, regionID.id, s.vars.rpcCtx.Addr)
+	s.storeAddr = s.vars.rpcCtx.Addr
 
-		var retry bool
-		resp, retry, err = s.sendReqToRegion(bo, rpcCtx, req, timeout)
-		req.IsRetryRequest = true
-		if err != nil {
-			if cost := time.Since(startTime); cost > slowLogSendReqTime || cost > timeout || bo.GetTotalSleep() > 1000 {
-				msg := fmt.Sprintf("send request failed, err: %v", err.Error())
-				s.logSendReqError(bo, msg, regionID, retryTimes, req, cost, bo.GetTotalSleep()-startBackOff, timeout)
-			}
-			return nil, nil, retryTimes, err
+	req.Context.ClusterId = s.vars.rpcCtx.ClusterID
+	if req.InputRequestSource != "" && s.replicaSelector != nil {
+		patchRequestSource(req, s.replicaSelector.replicaType())
+	}
+	// RPCClient.SendRequest will attach `req.Context` thus skip attaching here to reduce overhead.
+	if s.vars.err = tikvrpc.SetContextNoAttach(req, s.vars.rpcCtx.Meta, s.vars.rpcCtx.Peer); s.vars.err != nil {
+		return true
+	}
+	if s.replicaSelector != nil {
+		if s.vars.err = s.replicaSelector.backoffOnRetry(s.vars.rpcCtx.Store, bo); s.vars.err != nil {
+			return true
 		}
+	}
 
-		if _, err1 := util.EvalFailpoint("afterSendReqToRegion"); err1 == nil {
-			if hook := bo.GetCtx().Value("sendReqToRegionFinishHook"); hook != nil {
-				h := hook.(func(*tikvrpc.Request, *tikvrpc.Response, error))
-				h(req, resp, err)
-			}
+	if _, err := util.EvalFailpoint("beforeSendReqToRegion"); err == nil {
+		if hook := bo.GetCtx().Value("sendReqToRegionHook"); hook != nil {
+			h := hook.(func(*tikvrpc.Request))
+			h(req)
 		}
+	}
 
-		// recheck whether the session/query is killed during the Next()
-		if err2 := bo.CheckKilled(); err2 != nil {
-			return nil, nil, retryTimes, err2
-		}
-		if val, err := util.EvalFailpoint("mockRetrySendReqToRegion"); err == nil {
-			if val.(bool) {
-				retry = true
-			}
-		}
-		if retry {
-			retryTimes++
-			continue
+	// judge the store limit switch.
+	if limit := kv.StoreLimit.Load(); limit > 0 {
+		if s.vars.err = s.getStoreToken(s.vars.rpcCtx.Store, limit); s.vars.err != nil {
+			return true
 		}
+		defer s.releaseStoreToken(s.vars.rpcCtx.Store)
+	}
 
-		var regionErr *errorpb.Error
-		regionErr, err = resp.GetRegionError()
-		if err != nil {
-			return nil, nil, retryTimes, err
+	sendCtx := s.send(bo, req, timeout)
+	s.vars.sendTimes++
+
+	if s.vars.err != nil {
+		// Because in rpc logic, context.Cancel() will be transferred to rpcContext.Cancel error. For rpcContext cancel,
+		// we need to retry the request. But for context cancel active, for example, limitExec gets the required rows,
+		// we shouldn't retry the request, it will go to backoff and hang in retry logic.
+		if sendCtx.Err() != nil && errors.Cause(sendCtx.Err()) == context.Canceled {
+			metrics.TiKVRPCErrorCounter.WithLabelValues("context-canceled", storeIDLabel(s.vars.rpcCtx)).Inc()
+			return true
 		}
-		if regionErr != nil {
-			retry, err = s.onRegionError(bo, rpcCtx, req, regionErr)
-			if err != nil {
-				if cost := time.Since(startTime); cost > slowLogSendReqTime || cost > timeout || bo.GetTotalSleep() > 1000 {
-					msg := fmt.Sprintf("send request on region error failed, err: %v", err.Error())
-					s.logSendReqError(bo, msg, regionID, retryTimes, req, cost, bo.GetTotalSleep()-startBackOff, timeout)
-				}
-				return nil, nil, retryTimes, err
-			}
-			if retry {
-				retryTimes++
-				continue
-			}
-			if cost := time.Since(startTime); cost > slowLogSendReqTime || cost > timeout || bo.GetTotalSleep() > 1000 {
-				s.logSendReqError(bo, "send request meet region error without retry", regionID, retryTimes, req, cost, bo.GetTotalSleep()-startBackOff, timeout)
-			}
-		} else {
-			if s.replicaSelector != nil {
-				s.replicaSelector.onSendSuccess(req)
-			}
+		if val, e := util.EvalFailpoint("noRetryOnRpcError"); e == nil && val.(bool) {
+			return true
 		}
-
-		return resp, rpcCtx, retryTimes, nil
+		// need to handle send error
+		return false
 	}
-}
 
-func (s *RegionRequestSender) logSendReqError(bo *retry.Backoffer, msg string, regionID RegionVerID, retryTimes int, req *tikvrpc.Request, cost time.Duration, currentBackoffMs int, timeout time.Duration) {
-	var builder strings.Builder
-	// build the total round stats string.
-	builder.WriteString("{total-backoff: ")
-	builder.WriteString(util.FormatDuration(time.Duration(bo.GetTotalSleep() * int(time.Millisecond))))
-	builder.WriteString(", total-backoff-times: ")
-	builder.WriteString(strconv.Itoa(bo.GetTotalBackoffTimes()))
-	if s.Stats != nil {
-		builder.WriteString(", total-rpc: {")
-		builder.WriteString(s.Stats.String())
-		builder.WriteString("}")
+	if val, err := util.EvalFailpoint("mockRetrySendReqToRegion"); err == nil && val.(bool) {
+		// force retry
+		return false
 	}
-	builder.WriteString("}")
-	totalRoundStats := builder.String()
 
-	// build the current round stats string.
-	builder.Reset()
-	builder.WriteString("{time: ")
-	builder.WriteString(util.FormatDuration(cost))
-	builder.WriteString(", backoff: ")
-	builder.WriteString(util.FormatDuration(time.Duration(currentBackoffMs * int(time.Millisecond))))
-	builder.WriteString(", timeout: ")
-	builder.WriteString(util.FormatDuration(timeout))
-	builder.WriteString(", req-max-exec-timeout: ")
-	builder.WriteString(util.FormatDuration(time.Duration(int64(req.Context.MaxExecutionDurationMs) * int64(time.Millisecond))))
-	builder.WriteString(", retry-times: ")
-	builder.WriteString(strconv.Itoa(retryTimes))
-	if s.AccessStats != nil {
-		builder.WriteString(", replica-access: {")
-		builder.WriteString(s.AccessStats.String())
-		builder.WriteString("}")
+	s.vars.regionErr, s.vars.err = s.vars.resp.GetRegionError()
+	if s.vars.err != nil {
+		s.vars.rpcCtx, s.vars.resp = nil, nil
+		return true
+	} else if s.vars.regionErr != nil {
+		// need to handle region error
+		return false
 	}
-	builder.WriteString("}")
-	currentRoundStats := builder.String()
-	logutil.Logger(bo.GetCtx()).Info(msg,
-		zap.Uint64("req-ts", req.GetStartTS()),
-		zap.String("req-type", req.Type.String()),
-		zap.String("region", regionID.String()),
-		zap.String("replica-read-type", req.ReplicaReadType.String()),
-		zap.Bool("stale-read", req.StaleRead),
-		zap.Stringer("request-sender", s),
-		zap.String("total-round-stats", totalRoundStats),
-		zap.String("current-round-stats", currentRoundStats))
-}
-
-// RPCCancellerCtxKey is context key attach rpc send cancelFunc collector to ctx.
-type RPCCancellerCtxKey struct{}
 
-// RPCCanceller is rpc send cancelFunc collector.
-type RPCCanceller struct {
-	sync.Mutex
-	allocID   int
-	cancels   map[int]func()
-	cancelled bool
-}
-
-// NewRPCanceller creates RPCCanceller with init state.
-func NewRPCanceller() *RPCCanceller {
-	return &RPCCanceller{cancels: make(map[int]func())}
-}
-
-// WithCancel generates new context with cancel func.
-func (h *RPCCanceller) WithCancel(ctx context.Context) (context.Context, func()) {
-	nctx, cancel := context.WithCancel(ctx)
-	h.Lock()
-	if h.cancelled {
-		h.Unlock()
-		cancel()
-		return nctx, func() {}
-	}
-	id := h.allocID
-	h.allocID++
-	h.cancels[id] = cancel
-	h.Unlock()
-	return nctx, func() {
-		cancel()
-		h.Lock()
-		delete(h.cancels, id)
-		h.Unlock()
+	if s.replicaSelector != nil {
+		s.replicaSelector.onSendSuccess(req)
 	}
-}
 
-// CancelAll cancels all inflight rpc context.
-func (h *RPCCanceller) CancelAll() {
-	h.Lock()
-	for _, c := range h.cancels {
-		c()
-	}
-	h.cancelled = true
-	h.Unlock()
+	return true
 }
 
-func fetchRespInfo(resp *tikvrpc.Response) string {
-	var extraInfo string
-	if resp == nil || resp.Resp == nil {
-		extraInfo = "nil response"
-	} else {
-		regionErr, e := resp.GetRegionError()
-		if e != nil {
-			extraInfo = e.Error()
-		} else if regionErr != nil {
-			extraInfo = regionErr.String()
-		} else if prewriteResp, ok := resp.Resp.(*kvrpcpb.PrewriteResponse); ok {
-			extraInfo = prewriteResp.String()
-		}
-	}
-	return extraInfo
-}
-
-func (s *RegionRequestSender) sendReqToRegion(
-	bo *retry.Backoffer, rpcCtx *RPCContext, req *tikvrpc.Request, timeout time.Duration,
-) (resp *tikvrpc.Response, retry bool, err error) {
-	// judge the store limit switch.
-	if limit := kv.StoreLimit.Load(); limit > 0 {
-		if err := s.getStoreToken(rpcCtx.Store, limit); err != nil {
-			return nil, false, err
-		}
-		defer s.releaseStoreToken(rpcCtx.Store)
-	}
-
-	ctx := bo.GetCtx()
+func (s *sendReqState) send(bo *retry.Backoffer, req *tikvrpc.Request, timeout time.Duration) (ctx context.Context) {
+	rpcCtx := s.vars.rpcCtx
+	ctx = bo.GetCtx()
 	if rawHook := ctx.Value(RPCCancellerCtxKey{}); rawHook != nil {
 		var cancel context.CancelFunc
 		ctx, cancel = rawHook.(*RPCCanceller).WithCancel(ctx)
@@ -1107,13 +968,13 @@ func (s *RegionRequestSender) sendReqToRegion(
 				zap.Stringer("req", req.Req.(fmt.Stringer)), zap.Stringer("ctx", &req.Context),
 			)
 			injectFailOnSend = true
-			err = errors.New("injected RPC error on send")
+			s.vars.err = errors.New("injected RPC error on send")
 		}
 	}
 
 	if !injectFailOnSend {
 		start := time.Now()
-		resp, err = s.client.SendRequest(ctx, sendToAddr, req, timeout)
+		s.vars.resp, s.vars.err = s.client.SendRequest(ctx, sendToAddr, req, timeout)
 		rpcDuration := time.Since(start)
 		if s.replicaSelector != nil {
 			recordAttemptedTime(s.replicaSelector, rpcDuration)
@@ -1127,9 +988,10 @@ func (s *RegionRequestSender) sendReqToRegion(
 			if val, fpErr := util.EvalFailpoint("tikvStoreRespResult"); fpErr == nil {
 				if val.(bool) {
 					if req.Type == tikvrpc.CmdCop && bo.GetTotalSleep() == 0 {
-						return &tikvrpc.Response{
+						s.vars.resp, s.vars.err = &tikvrpc.Response{
 							Resp: &coprocessor.Response{RegionError: &errorpb.Error{EpochNotMatch: &errorpb.EpochNotMatch{}}},
-						}, false, nil
+						}, nil
+						return
 					}
 				}
 			}
@@ -1152,10 +1014,9 @@ func (s *RegionRequestSender) sendReqToRegion(
 				logutil.Logger(ctx).Info(
 					"[failpoint] injected RPC error on recv", zap.Stringer("type", req.Type),
 					zap.Stringer("req", req.Req.(fmt.Stringer)), zap.Stringer("ctx", &req.Context),
-					zap.Error(err), zap.String("extra response info", fetchRespInfo(resp)),
+					zap.Error(s.vars.err), zap.String("extra response info", fetchRespInfo(s.vars.resp)),
 				)
-				err = errors.New("injected RPC error on recv")
-				resp = nil
+				s.vars.resp, s.vars.err = nil, errors.New("injected RPC error on recv")
 			}
 		}
 
@@ -1165,15 +1026,14 @@ func (s *RegionRequestSender) sendReqToRegion(
 				cancel()
 				<-ctx1.Done()
 				ctx = ctx1
-				err = ctx.Err()
-				resp = nil
+				s.vars.resp, s.vars.err = nil, ctx.Err()
 			}
 		}
 
 		if _, e := util.EvalFailpoint("onRPCFinishedHook"); e == nil {
 			if hook := bo.GetCtx().Value("onRPCFinishedHook"); hook != nil {
 				h := hook.(func(*tikvrpc.Request, *tikvrpc.Response, error) (*tikvrpc.Response, error))
-				resp, err = h(req, resp, err)
+				s.vars.resp, s.vars.err = h(req, s.vars.resp, s.vars.err)
 			}
 		}
 	}
@@ -1182,42 +1042,215 @@ func (s *RegionRequestSender) sendReqToRegion(
 		fromStore := strconv.FormatUint(rpcCtx.ProxyStore.storeID, 10)
 		toStore := strconv.FormatUint(rpcCtx.Store.storeID, 10)
 		result := "ok"
-		if err != nil {
+		if s.vars.err != nil {
 			result = "fail"
 		}
 		metrics.TiKVForwardRequestCounter.WithLabelValues(fromStore, toStore, req.Type.String(), result).Inc()
 	}
 
-	if err != nil {
+	if err := s.vars.err; err != nil {
 		if isRPCError(err) {
 			s.rpcError = err
 		}
 		if s.Stats != nil {
 			errStr := getErrMsg(err)
 			s.Stats.RecordRPCErrorStats(errStr)
-			s.recordRPCAccessInfo(req, rpcCtx, errStr)
-		}
-		// Because in rpc logic, context.Cancel() will be transferred to rpcContext.Cancel error. For rpcContext cancel,
-		// we need to retry the request. But for context cancel active, for example, limitExec gets the required rows,
-		// we shouldn't retry the request, it will go to backoff and hang in retry logic.
-		if ctx.Err() != nil && errors.Cause(ctx.Err()) == context.Canceled {
-			metrics.TiKVRPCErrorCounter.WithLabelValues("context-canceled", storeIDLabel(rpcCtx)).Inc()
-			return nil, false, errors.WithStack(ctx.Err())
+			s.recordRPCAccessInfo(req, s.vars.rpcCtx, errStr)
 		}
+	}
+	return
+}
 
-		if val, e := util.EvalFailpoint("noRetryOnRpcError"); e == nil {
-			if val.(bool) {
-				return nil, false, err
+// SendReqCtx sends a request to tikv server and return response and RPCCtx of this RPC.
+func (s *RegionRequestSender) SendReqCtx(
+	bo *retry.Backoffer,
+	req *tikvrpc.Request,
+	regionID RegionVerID,
+	timeout time.Duration,
+	et tikvrpc.EndpointType,
+	opts ...StoreSelectorOption,
+) (
+	resp *tikvrpc.Response,
+	rpcCtx *RPCContext,
+	retryTimes int,
+	err error,
+) {
+	if span := opentracing.SpanFromContext(bo.GetCtx()); span != nil && span.Tracer() != nil {
+		span1 := span.Tracer().StartSpan("regionRequest.SendReqCtx", opentracing.ChildOf(span.Context()))
+		defer span1.Finish()
+		bo.SetCtx(opentracing.ContextWithSpan(bo.GetCtx(), span1))
+	}
+
+	if resp, err = failpointSendReqResult(req, et); err != nil || resp != nil {
+		return
+	}
+
+	if err = s.validateReadTS(bo.GetCtx(), req); err != nil {
+		logutil.Logger(bo.GetCtx()).Error("validate read ts failed for request", zap.Stringer("reqType", req.Type), zap.Stringer("req", req.Req.(fmt.Stringer)), zap.Stringer("context", &req.Context), zap.Stack("stack"), zap.Error(err))
+		return nil, nil, 0, err
+	}
+
+	// If the MaxExecutionDurationMs is not set yet, we set it to be the RPC timeout duration
+	// so TiKV can give up the requests whose response TiDB cannot receive due to timeout.
+	if req.Context.MaxExecutionDurationMs == 0 {
+		req.Context.MaxExecutionDurationMs = uint64(timeout.Milliseconds())
+	}
+
+	state := &sendReqState{RegionRequestSender: s}
+	defer func() {
+		if retryTimes := state.vars.sendTimes - 1; retryTimes > 0 {
+			metrics.TiKVRequestRetryTimesHistogram.Observe(float64(retryTimes))
+		}
+		if req.StaleRead {
+			if state.vars.sendTimes == 1 {
+				metrics.StaleReadHitCounter.Add(1)
+			} else {
+				metrics.StaleReadMissCounter.Add(1)
 			}
 		}
-		if e := s.onSendFail(bo, rpcCtx, req, err); e != nil {
-			return nil, false, err
+	}()
+
+	s.reset()
+	startTime := time.Now()
+	startBackOff := bo.GetTotalSleep()
+
+	for !state.next(bo, req, regionID, timeout, et, opts) {
+		if retryTimes := state.vars.sendTimes - 1; retryTimes > 0 && retryTimes%100 == 0 {
+			logutil.Logger(bo.GetCtx()).Warn("retry", zap.Uint64("region", regionID.GetID()), zap.Int("times", retryTimes))
 		}
-		return nil, true, nil
 	}
+
+	if state.vars.err == nil {
+		resp, rpcCtx = state.vars.resp, state.vars.rpcCtx
+	} else {
+		err = state.vars.err
+	}
+	if state.vars.sendTimes > 1 {
+		retryTimes = state.vars.sendTimes - 1
+	}
+
+	if len(state.vars.msg) > 0 || err != nil {
+		if cost := time.Since(startTime); cost > slowLogSendReqTime || cost > timeout || bo.GetTotalSleep() > 1000 {
+			msg := state.vars.msg
+			if len(msg) == 0 {
+				msg = fmt.Sprintf("send request failed: %v", err)
+			}
+			s.logSendReqError(bo, msg, regionID, retryTimes, req, cost, bo.GetTotalSleep()-startBackOff, timeout)
+		}
+	}
+
 	return
 }
 
+func (s *RegionRequestSender) logSendReqError(bo *retry.Backoffer, msg string, regionID RegionVerID, retryTimes int, req *tikvrpc.Request, cost time.Duration, currentBackoffMs int, timeout time.Duration) {
+	var builder strings.Builder
+	// build the total round stats string.
+	builder.WriteString("{total-backoff: ")
+	builder.WriteString(util.FormatDuration(time.Duration(bo.GetTotalSleep() * int(time.Millisecond))))
+	builder.WriteString(", total-backoff-times: ")
+	builder.WriteString(strconv.Itoa(bo.GetTotalBackoffTimes()))
+	if s.Stats != nil {
+		builder.WriteString(", total-rpc: {")
+		builder.WriteString(s.Stats.String())
+		builder.WriteString("}")
+	}
+	builder.WriteString("}")
+	totalRoundStats := builder.String()
+
+	// build the current round stats string.
+	builder.Reset()
+	builder.WriteString("{time: ")
+	builder.WriteString(util.FormatDuration(cost))
+	builder.WriteString(", backoff: ")
+	builder.WriteString(util.FormatDuration(time.Duration(currentBackoffMs * int(time.Millisecond))))
+	builder.WriteString(", timeout: ")
+	builder.WriteString(util.FormatDuration(timeout))
+	builder.WriteString(", req-max-exec-timeout: ")
+	builder.WriteString(util.FormatDuration(time.Duration(int64(req.Context.MaxExecutionDurationMs) * int64(time.Millisecond))))
+	builder.WriteString(", retry-times: ")
+	builder.WriteString(strconv.Itoa(retryTimes))
+	if s.AccessStats != nil {
+		builder.WriteString(", replica-access: {")
+		builder.WriteString(s.AccessStats.String())
+		builder.WriteString("}")
+	}
+	builder.WriteString("}")
+	currentRoundStats := builder.String()
+	logutil.Logger(bo.GetCtx()).Info(msg,
+		zap.Uint64("req-ts", req.GetStartTS()),
+		zap.String("req-type", req.Type.String()),
+		zap.String("region", regionID.String()),
+		zap.String("replica-read-type", req.ReplicaReadType.String()),
+		zap.Bool("stale-read", req.StaleRead),
+		zap.Stringer("request-sender", s),
+		zap.String("total-round-stats", totalRoundStats),
+		zap.String("current-round-stats", currentRoundStats))
+}
+
+// RPCCancellerCtxKey is context key attach rpc send cancelFunc collector to ctx.
+type RPCCancellerCtxKey struct{}
+
+// RPCCanceller is rpc send cancelFunc collector.
+type RPCCanceller struct {
+	sync.Mutex
+	allocID   int
+	cancels   map[int]func()
+	cancelled bool
+}
+
+// NewRPCanceller creates RPCCanceller with init state.
+func NewRPCanceller() *RPCCanceller {
+	return &RPCCanceller{cancels: make(map[int]func())}
+}
+
+// WithCancel generates new context with cancel func.
+func (h *RPCCanceller) WithCancel(ctx context.Context) (context.Context, func()) {
+	nctx, cancel := context.WithCancel(ctx)
+	h.Lock()
+	if h.cancelled {
+		h.Unlock()
+		cancel()
+		return nctx, func() {}
+	}
+	id := h.allocID
+	h.allocID++
+	h.cancels[id] = cancel
+	h.Unlock()
+	return nctx, func() {
+		cancel()
+		h.Lock()
+		delete(h.cancels, id)
+		h.Unlock()
+	}
+}
+
+// CancelAll cancels all inflight rpc context.
+func (h *RPCCanceller) CancelAll() {
+	h.Lock()
+	for _, c := range h.cancels {
+		c()
+	}
+	h.cancelled = true
+	h.Unlock()
+}
+
+func fetchRespInfo(resp *tikvrpc.Response) string {
+	var extraInfo string
+	if resp == nil || resp.Resp == nil {
+		extraInfo = "nil response"
+	} else {
+		regionErr, e := resp.GetRegionError()
+		if e != nil {
+			extraInfo = e.Error()
+		} else if regionErr != nil {
+			extraInfo = regionErr.String()
+		} else if prewriteResp, ok := resp.Resp.(*kvrpcpb.PrewriteResponse); ok {
+			extraInfo = prewriteResp.String()
+		}
+	}
+	return extraInfo
+}
+
 func isRPCError(err error) bool {
 	// exclude ErrClientResourceGroupThrottled
 	return err != nil && errs.ErrClientResourceGroupThrottled.NotEqual(err)
diff --git a/internal/locate/replica_selector.go b/internal/locate/replica_selector.go
index 901592c..c873d32 100644
--- a/internal/locate/replica_selector.go
+++ b/internal/locate/replica_selector.go
@@ -307,8 +307,8 @@ func (s *ReplicaSelectMixedStrategy) next(selector *replicaSelector) *replica {
 func (s *ReplicaSelectMixedStrategy) canSendReplicaRead(selector *replicaSelector) bool {
 	replicas := selector.replicas
 	replica := replicas[s.leaderIdx]
-	if replica.hasFlag(deadlineErrUsingConfTimeoutFlag) || replica.hasFlag(serverIsBusyFlag) {
-		// don't overwhelm the leader if it is busy
+	if replica.attempts == 0 || replica.hasFlag(deadlineErrUsingConfTimeoutFlag) || replica.hasFlag(serverIsBusyFlag) {
+		// don't do the replica read if leader is not exhausted or leader has timeout or server busy error.
 		return false
 	}
 	return true
diff --git a/internal/unionstore/arena/arena.go b/internal/unionstore/arena/arena.go
index ef5c081..6ae3e89 100644
--- a/internal/unionstore/arena/arena.go
+++ b/internal/unionstore/arena/arena.go
@@ -38,8 +38,10 @@ import (
 	"encoding/binary"
 	"math"
 
+	"github.com/tikv/client-go/v2/internal/logutil"
 	"github.com/tikv/client-go/v2/kv"
 	"go.uber.org/atomic"
+	"go.uber.org/zap"
 )
 
 const (
@@ -223,6 +225,20 @@ func (cp *MemDBCheckpoint) IsSamePosition(other *MemDBCheckpoint) bool {
 	return cp.blocks == other.blocks && cp.offsetInBlock == other.offsetInBlock
 }
 
+// LessThan compares two checkpoints.
+func (cp *MemDBCheckpoint) LessThan(cp2 *MemDBCheckpoint) bool {
+	if cp == nil || cp2 == nil {
+		logutil.BgLogger().Panic("unexpected nil checkpoint", zap.Any("cp", cp), zap.Any("cp2", cp2))
+	}
+	if cp.blocks < cp2.blocks {
+		return true
+	}
+	if cp.blocks == cp2.blocks && cp.offsetInBlock < cp2.offsetInBlock {
+		return true
+	}
+	return false
+}
+
 func (a *MemdbArena) Checkpoint() MemDBCheckpoint {
 	snap := MemDBCheckpoint{
 		blockSize: a.blockSize,
diff --git a/internal/unionstore/art/art.go b/internal/unionstore/art/art.go
index 36acb90..4df6a86 100644
--- a/internal/unionstore/art/art.go
+++ b/internal/unionstore/art/art.go
@@ -47,11 +47,23 @@ type ART struct {
 	len             int
 	size            int
 
+	// These variables serve the caching mechanism, meaning they can be concurrently updated by read operations, thus
+	// they are protected by atomic operations.
 	// The lastTraversedNode stores addr in uint64 of the last traversed node, includes search and recursiveInsert.
-	// Compare to atomic.Pointer, atomic.Uint64 can avoid heap allocation, so it's more efficient.
+	// Compared with atomic.Pointer, atomic.Uint64 can avoid heap allocation, so it's more efficient.
 	lastTraversedNode atomic.Uint64
 	hitCount          atomic.Uint64
 	missCount         atomic.Uint64
+
+	// The counter of every write operation, used to invalidate iterators that were created before the write operation.
+	// The purpose of the counter is to check interleaving of write and read operations (via iterator).
+	// It does not protect against data race. If it happens, there must be a bug in the caller code.
+	// invariant: no concurrent write/write or read/write access to it
+	WriteSeqNo int
+	// Increased by 1 when an operation that may affect the content returned by "snapshot" (i.e. stage[0]) happens.
+	// It's used to invalidate snapshot iterators.
+	// invariant: no concurrent write/write or read/write access to it
+	SnapshotSeqNo int
 }
 
 func New() *ART {
@@ -115,8 +127,12 @@ func (t *ART) Set(key artKey, value []byte, ops ...kv.FlagsOp) error {
 		}
 	}
 
+	t.WriteSeqNo++
 	if len(t.stages) == 0 {
 		t.dirty = true
+
+		// note: there is no such usage in TiDB
+		t.SnapshotSeqNo++
 	}
 	// 1. create or search the existing leaf in the tree.
 	addr, leaf := t.traverse(key, true)
@@ -479,6 +495,10 @@ func (t *ART) RevertToCheckpoint(cp *arena.MemDBCheckpoint) {
 	t.allocator.vlogAllocator.RevertToCheckpoint(t, cp)
 	t.allocator.vlogAllocator.Truncate(cp)
 	t.allocator.vlogAllocator.OnMemChange()
+	t.WriteSeqNo++
+	if len(t.stages) == 0 || t.stages[0].LessThan(cp) {
+		t.SnapshotSeqNo++
+	}
 }
 
 func (t *ART) Stages() []arena.MemDBCheckpoint {
@@ -498,7 +518,9 @@ func (t *ART) Release(h int) {
 	if h != len(t.stages) {
 		panic("cannot release staging buffer")
 	}
+	t.WriteSeqNo++
 	if h == 1 {
+		t.SnapshotSeqNo++
 		tail := t.checkpoint()
 		if !t.stages[0].IsSamePosition(&tail) {
 			t.dirty = true
@@ -519,6 +541,11 @@ func (t *ART) Cleanup(h int) {
 		panic(fmt.Sprintf("cannot cleanup staging buffer, h=%v, len(tree.stages)=%v", h, len(t.stages)))
 	}
 
+	t.WriteSeqNo++
+	if h == 1 {
+		t.SnapshotSeqNo++
+	}
+
 	cp := &t.stages[h-1]
 	if !t.vlogInvalid {
 		curr := t.checkpoint()
@@ -542,6 +569,8 @@ func (t *ART) Reset() {
 	t.allocator.nodeAllocator.Reset()
 	t.allocator.vlogAllocator.Reset()
 	t.lastTraversedNode.Store(arena.NullU64Addr)
+	t.SnapshotSeqNo++
+	t.WriteSeqNo++
 }
 
 // DiscardValues releases the memory used by all values.
diff --git a/internal/unionstore/art/art_iterator.go b/internal/unionstore/art/art_iterator.go
index 2bf4fdb..fd5f328 100644
--- a/internal/unionstore/art/art_iterator.go
+++ b/internal/unionstore/art/art_iterator.go
@@ -20,8 +20,10 @@ import (
 	"sort"
 
 	"github.com/pkg/errors"
+	"github.com/tikv/client-go/v2/internal/logutil"
 	"github.com/tikv/client-go/v2/internal/unionstore/arena"
 	"github.com/tikv/client-go/v2/kv"
+	"go.uber.org/zap"
 )
 
 func (t *ART) Iter(lowerBound, upperBound []byte) (*Iterator, error) {
@@ -56,6 +58,7 @@ func (t *ART) iter(lowerBound, upperBound []byte, reverse, includeFlags bool) (*
 		// this avoids the initial value of currAddr equals to endAddr.
 		currAddr: arena.BadAddr,
 		endAddr:  arena.NullAddr,
+		seqNo:    t.WriteSeqNo,
 	}
 	it.init(lowerBound, upperBound)
 	if !it.valid {
@@ -76,12 +79,41 @@ type Iterator struct {
 	currLeaf     *artLeaf
 	currAddr     arena.MemdbArenaAddr
 	endAddr      arena.MemdbArenaAddr
+
+	// only when seqNo == art.WriteSeqNo, the iterator is valid.
+	seqNo int
+	// ignoreSeqNo is used to ignore the seqNo check, used for snapshot iter before its full deprecation.
+	ignoreSeqNo bool
+}
+
+func (it *Iterator) checkSeqNo() {
+	if it.seqNo != it.tree.WriteSeqNo && !it.ignoreSeqNo {
+		logutil.BgLogger().Panic(
+			"seqNo mismatch",
+			zap.Int("it seqNo", it.seqNo),
+			zap.Int("art seqNo", it.tree.WriteSeqNo),
+			zap.Stack("stack"),
+		)
+	}
+}
+
+func (it *Iterator) Valid() bool {
+	it.checkSeqNo()
+	return it.valid
+}
+
+func (it *Iterator) Key() []byte {
+	it.checkSeqNo()
+	return it.currLeaf.GetKey()
+}
+
+func (it *Iterator) Flags() kv.KeyFlags {
+	it.checkSeqNo()
+	return it.currLeaf.GetKeyFlags()
 }
 
-func (it *Iterator) Valid() bool        { return it.valid }
-func (it *Iterator) Key() []byte        { return it.currLeaf.GetKey() }
-func (it *Iterator) Flags() kv.KeyFlags { return it.currLeaf.GetKeyFlags() }
 func (it *Iterator) Value() []byte {
+	it.checkSeqNo()
 	if it.currLeaf.vLogAddr.IsNull() {
 		return nil
 	}
@@ -102,6 +134,7 @@ func (it *Iterator) Next() error {
 		// iterate is finished
 		return errors.New("Art: iterator is finished")
 	}
+	it.checkSeqNo()
 	if it.currAddr == it.endAddr {
 		it.valid = false
 		return nil
diff --git a/internal/unionstore/art/art_snapshot.go b/internal/unionstore/art/art_snapshot.go
index 454634b..b899d7c 100644
--- a/internal/unionstore/art/art_snapshot.go
+++ b/internal/unionstore/art/art_snapshot.go
@@ -21,6 +21,8 @@ import (
 	"github.com/tikv/client-go/v2/internal/unionstore/arena"
 )
 
+// getSnapshot returns the "snapshot" for snapshotGetter or snapshotIterator, which is usually the snapshot
+// of stage[0]
 func (t *ART) getSnapshot() arena.MemDBCheckpoint {
 	if len(t.stages) > 0 {
 		return t.stages[0]
@@ -49,6 +51,7 @@ func (t *ART) newSnapshotIterator(start, end []byte, desc bool) *SnapIter {
 	if err != nil {
 		panic(err)
 	}
+	inner.ignoreSeqNo = true
 	it := &SnapIter{
 		Iterator: inner,
 		cp:       t.getSnapshot(),
diff --git a/internal/unionstore/memdb_art.go b/internal/unionstore/memdb_art.go
index c7c1b21..16438d5 100644
--- a/internal/unionstore/memdb_art.go
+++ b/internal/unionstore/memdb_art.go
@@ -16,9 +16,12 @@ package unionstore
 
 import (
 	"context"
+	"fmt"
 	"sync"
 
+	"github.com/pingcap/errors"
 	tikverr "github.com/tikv/client-go/v2/error"
+	"github.com/tikv/client-go/v2/internal/logutil"
 	"github.com/tikv/client-go/v2/internal/unionstore/arena"
 	"github.com/tikv/client-go/v2/internal/unionstore/art"
 	"github.com/tikv/client-go/v2/kv"
@@ -151,6 +154,32 @@ func (db *artDBWithContext) IterReverse(upper, lower []byte) (Iterator, error) {
 	return db.ART.IterReverse(upper, lower)
 }
 
+func (db *artDBWithContext) ForEachInSnapshotRange(lower []byte, upper []byte, f func(k, v []byte) (stop bool, err error), reverse bool) error {
+	db.RLock()
+	defer db.RUnlock()
+	var iter Iterator
+	if reverse {
+		iter = db.SnapshotIterReverse(upper, lower)
+	} else {
+		iter = db.SnapshotIter(lower, upper)
+	}
+	defer iter.Close()
+	for iter.Valid() {
+		stop, err := f(iter.Key(), iter.Value())
+		if err != nil {
+			return err
+		}
+		err = iter.Next()
+		if err != nil {
+			return err
+		}
+		if stop {
+			break
+		}
+	}
+	return nil
+}
+
 // SnapshotIter returns an Iterator for a snapshot of MemBuffer.
 func (db *artDBWithContext) SnapshotIter(lower, upper []byte) Iterator {
 	return db.ART.SnapshotIter(lower, upper)
@@ -165,3 +194,163 @@ func (db *artDBWithContext) SnapshotIterReverse(upper, lower []byte) Iterator {
 func (db *artDBWithContext) SnapshotGetter() Getter {
 	return db.ART.SnapshotGetter()
 }
+
+type snapshotBatchedIter struct {
+	db            *artDBWithContext
+	snapshotSeqNo int
+	lower         []byte
+	upper         []byte
+	reverse       bool
+	err           error
+
+	// current batch
+	keys      [][]byte
+	values    [][]byte
+	pos       int
+	batchSize int
+	nextKey   []byte
+}
+
+func (db *artDBWithContext) BatchedSnapshotIter(lower, upper []byte, reverse bool) Iterator {
+	if len(db.Stages()) == 0 {
+		logutil.BgLogger().Error("should not use BatchedSnapshotIter for a memdb without any staging buffer")
+	}
+	iter := &snapshotBatchedIter{
+		db:            db,
+		snapshotSeqNo: db.SnapshotSeqNo,
+		lower:         lower,
+		upper:         upper,
+		reverse:       reverse,
+		batchSize:     32,
+	}
+
+	iter.err = iter.fillBatch()
+	return iter
+}
+
+func (it *snapshotBatchedIter) fillBatch() error {
+	// The check of sequence numbers don't have to be protected by the rwlock, as the invariant is that
+	// there cannot be concurrent writes to the seqNo variables.
+	if it.snapshotSeqNo != it.db.SnapshotSeqNo {
+		return errors.Errorf(
+			"invalid iter: snapshotSeqNo changed, iter's=%d, db's=%d",
+			it.snapshotSeqNo,
+			it.db.SnapshotSeqNo,
+		)
+	}
+
+	it.db.RLock()
+	defer it.db.RUnlock()
+
+	if it.keys == nil || it.values == nil || cap(it.keys) < it.batchSize || cap(it.values) < it.batchSize {
+		it.keys = make([][]byte, 0, it.batchSize)
+		it.values = make([][]byte, 0, it.batchSize)
+	} else {
+		it.keys = it.keys[:0]
+		it.values = it.values[:0]
+	}
+
+	var snapshotIter Iterator
+	if it.reverse {
+		searchUpper := it.upper
+		if it.nextKey != nil {
+			searchUpper = it.nextKey
+		}
+		snapshotIter = it.db.SnapshotIterReverse(searchUpper, it.lower)
+	} else {
+		searchLower := it.lower
+		if it.nextKey != nil {
+			searchLower = it.nextKey
+		}
+		snapshotIter = it.db.SnapshotIter(searchLower, it.upper)
+	}
+	defer snapshotIter.Close()
+
+	// fill current batch
+	// Further optimization: let the underlying memdb support batch iter.
+	for i := 0; i < it.batchSize && snapshotIter.Valid(); i++ {
+		it.keys = it.keys[:i+1]
+		it.values = it.values[:i+1]
+		it.keys[i] = snapshotIter.Key()
+		it.values[i] = snapshotIter.Value()
+		if err := snapshotIter.Next(); err != nil {
+			return err
+		}
+	}
+
+	// update state
+	it.pos = 0
+	if len(it.keys) > 0 {
+		lastKey := it.keys[len(it.keys)-1]
+		keyLen := len(lastKey)
+
+		if it.reverse {
+			if cap(it.nextKey) >= keyLen {
+				it.nextKey = it.nextKey[:keyLen]
+			} else {
+				it.nextKey = make([]byte, keyLen)
+			}
+			copy(it.nextKey, lastKey)
+		} else {
+			if cap(it.nextKey) >= keyLen+1 {
+				it.nextKey = it.nextKey[:keyLen+1]
+			} else {
+				it.nextKey = make([]byte, keyLen+1)
+			}
+			copy(it.nextKey, lastKey)
+			it.nextKey[keyLen] = 0
+		}
+	} else {
+		it.nextKey = nil
+	}
+
+	it.batchSize = min(it.batchSize*2, 4096)
+	return nil
+}
+
+func (it *snapshotBatchedIter) Valid() bool {
+	return it.snapshotSeqNo == it.db.SnapshotSeqNo &&
+		it.pos < len(it.keys) &&
+		it.err == nil
+}
+
+func (it *snapshotBatchedIter) Next() error {
+	if it.err != nil {
+		return it.err
+	}
+	if it.snapshotSeqNo != it.db.SnapshotSeqNo {
+		return errors.New(
+			fmt.Sprintf(
+				"invalid snapshotBatchedIter: snapshotSeqNo changed, iter's=%d, db's=%d",
+				it.snapshotSeqNo,
+				it.db.SnapshotSeqNo,
+			),
+		)
+	}
+
+	it.pos++
+	if it.pos >= len(it.keys) {
+		return it.fillBatch()
+	}
+	return nil
+}
+
+func (it *snapshotBatchedIter) Key() []byte {
+	if !it.Valid() {
+		return nil
+	}
+	return it.keys[it.pos]
+}
+
+func (it *snapshotBatchedIter) Value() []byte {
+	if !it.Valid() {
+		return nil
+	}
+	return it.values[it.pos]
+}
+
+func (it *snapshotBatchedIter) Close() {
+	it.keys = nil
+	it.values = nil
+	it.nextKey = nil
+}
diff --git a/internal/unionstore/memdb_rbt.go b/internal/unionstore/memdb_rbt.go
index c805f49..3524150 100644
--- a/internal/unionstore/memdb_rbt.go
+++ b/internal/unionstore/memdb_rbt.go
@@ -161,6 +161,32 @@ func (db *rbtDBWithContext) IterReverse(upper, lower []byte) (Iterator, error) {
 	return db.RBT.IterReverse(upper, lower)
 }
 
+func (db *rbtDBWithContext) ForEachInSnapshotRange(lower []byte, upper []byte, f func(k, v []byte) (stop bool, err error), reverse bool) error {
+	db.RLock()
+	defer db.RUnlock()
+	var iter Iterator
+	if reverse {
+		iter = db.SnapshotIterReverse(upper, lower)
+	} else {
+		iter = db.SnapshotIter(lower, upper)
+	}
+	defer iter.Close()
+	for iter.Valid() {
+		stop, err := f(iter.Key(), iter.Value())
+		if err != nil {
+			return err
+		}
+		err = iter.Next()
+		if err != nil {
+			return err
+		}
+		if stop {
+			break
+		}
+	}
+	return nil
+}
+
 // SnapshotIter returns an Iterator for a snapshot of MemBuffer.
 func (db *rbtDBWithContext) SnapshotIter(lower, upper []byte) Iterator {
 	return db.RBT.SnapshotIter(lower, upper)
@@ -175,3 +201,12 @@ func (db *rbtDBWithContext) SnapshotIterReverse(upper, lower []byte) Iterator {
 func (db *rbtDBWithContext) SnapshotGetter() Getter {
 	return db.RBT.SnapshotGetter()
 }
+
+func (db *rbtDBWithContext) BatchedSnapshotIter(lower, upper []byte, reverse bool) Iterator {
+	// TODO: implement *batched* iter
+	if reverse {
+		return db.SnapshotIterReverse(upper, lower)
+	} else {
+		return db.SnapshotIter(lower, upper)
+	}
+}
diff --git a/internal/unionstore/pipelined_memdb.go b/internal/unionstore/pipelined_memdb.go
index 163a289..888f2ec 100644
--- a/internal/unionstore/pipelined_memdb.go
+++ b/internal/unionstore/pipelined_memdb.go
@@ -412,6 +412,10 @@ func (p *PipelinedMemDB) IterReverse([]byte, []byte) (Iterator, error) {
 	return nil, errors.New("pipelined memdb does not support IterReverse")
 }
 
+func (db *PipelinedMemDB) ForEachInSnapshotRange(lower []byte, upper []byte, f func(k, v []byte) (bool, error), reverse bool) error {
+	return errors.New("pipelined memdb does not support ForEachInSnapshotRange")
+}
+
 // SetEntrySizeLimit sets the size limit for each entry and total buffer.
 func (p *PipelinedMemDB) SetEntrySizeLimit(entryLimit, _ uint64) {
 	p.entryLimit = entryLimit
@@ -550,3 +554,7 @@ func (p *PipelinedMemDB) GetMetrics() Metrics {
 func (p *PipelinedMemDB) MemHookSet() bool {
 	return p.memChangeHook != nil
 }
+
+func (p *PipelinedMemDB) BatchedSnapshotIter(lower, upper []byte, reverse bool) Iterator {
+	panic("BatchedSnapshotIter is not supported for PipelinedMemDB")
+}
diff --git a/internal/unionstore/union_store.go b/internal/unionstore/union_store.go
index 1a5f1a3..c6d091f 100644
--- a/internal/unionstore/union_store.go
+++ b/internal/unionstore/union_store.go
@@ -193,15 +193,45 @@ type MemBuffer interface {
 	Delete([]byte) error
 	// DeleteWithFlags deletes the key k in the MemBuffer with flags.
 	DeleteWithFlags([]byte, ...kv.FlagsOp) error
+
 	// Iter implements the Retriever interface.
+	// Any write operation to the memdb invalidates this iterator immediately after its creation.
+	// Attempting to use such an invalidated iterator will result in a panic.
 	Iter([]byte, []byte) (Iterator, error)
 	// IterReverse implements the Retriever interface.
+	// Any write operation to the memdb invalidates this iterator immediately after its creation.
+	// Attempting to use such an invalidated iterator will result in a panic.
 	IterReverse([]byte, []byte) (Iterator, error)
 	// SnapshotIter returns an Iterator for a snapshot of MemBuffer.
+	// Deprecated: use ForEachInSnapshotRange or BatchedSnapshotIter instead.
 	SnapshotIter([]byte, []byte) Iterator
 	// SnapshotIterReverse returns a reversed Iterator for a snapshot of MemBuffer.
+	// Deprecated: use ForEachInSnapshotRange or BatchedSnapshotIter instead.
 	SnapshotIterReverse([]byte, []byte) Iterator
-	// SnapshotGetter returns a Getter for a snapshot of MemBuffer.
+
+	// ForEachInSnapshotRange scans the key-value pairs in the state[0] snapshot if it exists,
+	// otherwise it uses the current checkpoint as snapshot.
+	//
+	// NOTE: returned kv-pairs are only valid during the iteration. If you want to use them after the iteration,
+	// you need to make a copy.
+	//
+	// The method is protected by a RWLock to prevent potential iterator invalidation, i.e.
+	// You cannot modify the MemBuffer during the iteration.
+	//
+	// Use it when you need to scan the whole range, otherwise consider using BatchedSnapshotIter.
+	ForEachInSnapshotRange(lower []byte, upper []byte, f func(k, v []byte) (stop bool, err error), reverse bool) error
+
+	// BatchedSnapshotIter returns an iterator of the "snapshot", namely stage[0].
+	// It iterates in batches and prevents iterator invalidation.
+	//
+	// Use it when you need on-demand "next", otherwise consider using ForEachInSnapshotRange.
+	// NOTE: you should never use it when there are no stages.
+	//
+	// The iterator becomes invalid when any operation that may modify the "snapshot",
+	// e.g. RevertToCheckpoint or releasing stage[0].
+	BatchedSnapshotIter(lower, upper []byte, reverse bool) Iterator
+
+	//SnapshotGetter returns a Getter for a snapshot of MemBuffer.
 	SnapshotGetter() Getter
 	// InspectStage iterates all buffered keys and values in MemBuffer.
 	InspectStage(handle int, f func([]byte, kv.KeyFlags, []byte))
diff --git a/tikv/kv.go b/tikv/kv.go
index 70b7d84..cd5f17d 100644
--- a/tikv/kv.go
+++ b/tikv/kv.go
@@ -103,6 +103,12 @@ func createEtcdKV(addrs []string, tlsConfig *tls.Config) (*clientv3.Client, erro
 	if err != nil {
 		return nil, errors.WithStack(err)
 	}
+	ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
+	defer cancel()
+	if err := cli.Sync(ctx); err != nil {
+		_ = cli.Close()
+		return nil, errors.WithStack(err)
+	}
 	return cli, nil
 }
 
diff --git a/txnkv/transaction/2pc.go b/txnkv/transaction/2pc.go
index 6a739b3..14307bf 100644
--- a/txnkv/transaction/2pc.go
+++ b/txnkv/transaction/2pc.go
@@ -559,6 +559,7 @@ func (c *twoPhaseCommitter) initKeysAndMutations(ctx context.Context) error {
 
 	var err error
 	var assertionError error
+	toUpdatePrewriteOnly := make([][]byte, 0)
 	for it := memBuf.IterWithFlags(nil, nil); it.Valid(); err = it.Next() {
 		_ = err
 		key := it.Key()
@@ -607,7 +608,7 @@ func (c *twoPhaseCommitter) initKeysAndMutations(ctx context.Context) error {
 					// due to `Op_CheckNotExists` doesn't prewrite lock, so mark those keys should not be used in commit-phase.
 					op = kvrpcpb.Op_CheckNotExists
 					checkCnt++
-					memBuf.UpdateFlags(key, kv.SetPrewriteOnly)
+					toUpdatePrewriteOnly = append(toUpdatePrewriteOnly, key)
 				} else {
 					if flags.HasNewlyInserted() {
 						// The delete-your-write keys in pessimistic transactions, only lock needed keys and skip
@@ -682,6 +683,10 @@ func (c *twoPhaseCommitter) initKeysAndMutations(ctx context.Context) error {
 		}
 	}
 
+	for _, key := range toUpdatePrewriteOnly {
+		memBuf.UpdateFlags(key, kv.SetPrewriteOnly)
+	}
+
 	if c.mutations.Len() == 0 {
 		return nil
 	}
diff --git a/txnkv/transaction/pipelined_flush.go b/txnkv/transaction/pipelined_flush.go
index d2fe35c..6e19c94 100644
--- a/txnkv/transaction/pipelined_flush.go
+++ b/txnkv/transaction/pipelined_flush.go
@@ -484,6 +484,7 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end
 		handler,
 	)
 	runner.SetStatLogInterval(30 * time.Second)
+	runner.SetRegionsPerTask(1)
 
 	c.txn.spawnWithStorePool(func() {
 		if err = runner.RunOnRange(bo.GetCtx(), start, end); err != nil {


Test patch:
diff --git a/integration_tests/go.mod b/integration_tests/go.mod
index 472980b..2558f02 100644
--- a/integration_tests/go.mod
+++ b/integration_tests/go.mod
@@ -1,18 +1,18 @@
 module integration_tests
 
-go 1.23.5
+go 1.23.6
 
 require (
 	github.com/ninedraft/israce v0.0.3
 	github.com/pingcap/errors v0.11.5-0.20240318064555-6bd07397691f
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86
-	github.com/pingcap/kvproto v0.0.0-20250108041715-3b77f2c65c63
-	github.com/pingcap/tidb v1.1.0-beta.0.20250123092444-cf4d252af8aa
+	github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1
+	github.com/pingcap/tidb v1.1.0-beta.0.20250208075453-ad2c9f464fde
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.10.0
 	github.com/tidwall/gjson v1.14.1
-	github.com/tikv/client-go/v2 v2.0.8-0.20250117034919-61e09c6539bd
-	github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de
+	github.com/tikv/client-go/v2 v2.0.8-0.20250207065327-ec9ad0fd70cf
+	github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42
 	go.uber.org/goleak v1.3.0
 )
 
@@ -98,9 +98,9 @@ require (
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
 	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect
-	golang.org/x/sys v0.29.0 // indirect
+	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/text v0.21.0 // indirect
-	golang.org/x/time v0.7.0 // indirect
+	golang.org/x/time v0.9.0 // indirect
 	golang.org/x/tools v0.29.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240515191416-fc5f0ca64291 // indirect
diff --git a/integration_tests/go.sum b/integration_tests/go.sum
index d69849a..479f656 100644
--- a/integration_tests/go.sum
+++ b/integration_tests/go.sum
@@ -1057,8 +1057,8 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfU
 github.com/golang/glog v1.0.0/go.mod h1:EWib/APOK0SL3dFbYqvxE3UYd8E6s1ouQ7iEp/0LWV4=
 github.com/golang/glog v1.1.0/go.mod h1:pfYeQZ3JWZoXTV5sFc986z3HTpwQs9At6P4ImfuP3NQ=
 github.com/golang/glog v1.1.2/go.mod h1:zR+okUeTbrL6EL3xHUDxZuEtGv04p5shwip1+mL/rLQ=
-github.com/golang/glog v1.2.0 h1:uCdmnmatrKCgMBlM4rMuJZWOkPDqdbZPnrMXDY4gI68=
-github.com/golang/glog v1.2.0/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w=
+github.com/golang/glog v1.2.4 h1:CNNw5U8lSiiBk7druxtSHHTsRWcxKoac6kZKm2peBBc=
+github.com/golang/glog v1.2.4/go.mod h1:6AhwSGph0fcJtXVM/PEHPqZlFeoLxhs7/t5UDAwmO+w=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -1351,16 +1351,16 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
 github.com/pingcap/kvproto v0.0.0-20241113043844-e1fa7ea8c302/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
-github.com/pingcap/kvproto v0.0.0-20250108041715-3b77f2c65c63 h1:ThJ7ddLJVk96Iai2HDeyJGuuhrcBtc3HwYKJfuKPLsI=
-github.com/pingcap/kvproto v0.0.0-20250108041715-3b77f2c65c63/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
+github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1 h1:rTAyiswGyWSGHJVa4Mkhdi8YfGqfA4LrUVKsH9nrJ8E=
+github.com/pingcap/kvproto v0.0.0-20250117122752-2b87602a94a1/go.mod h1:rXxWk2UnwfUhLXha1jxRWPADw9eMZGWEWCg92Tgmb/8=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
 github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a h1:WIhmJBlNGmnCWH6TLMdZfNEDaiU8cFpZe3iaqDbQ0M8=
 github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA=
 github.com/pingcap/sysutil v1.0.1-0.20241113070546-23b50de46fd3 h1:Q9CMGKUztbM0RWHdQu0pD9b9OC47sbISRiMvf9vJ2RY=
 github.com/pingcap/sysutil v1.0.1-0.20241113070546-23b50de46fd3/go.mod h1:tyo4AX5P7udiSKN0Mv3nD9DcUnuLLmFfE22+dEs4vbU=
-github.com/pingcap/tidb v1.1.0-beta.0.20250123092444-cf4d252af8aa h1:ksDHCLdajx/IX2tGLYeZv9537/txJjdH9n1ipeeT3Js=
-github.com/pingcap/tidb v1.1.0-beta.0.20250123092444-cf4d252af8aa/go.mod h1:E8XJR02i6CQAQCFkhMIW2KuVtZ7PqsoSmTagsEt6NjQ=
+github.com/pingcap/tidb v1.1.0-beta.0.20250208075453-ad2c9f464fde h1:H0TA7G0+QDgvylJN+XpL1I+VY+rhk0fbWWschPOS1ec=
+github.com/pingcap/tidb v1.1.0-beta.0.20250208075453-ad2c9f464fde/go.mod h1:c7tIXSA8zZEgobZi3+j6FMGvts7rd6tkjf/m2o9GZCs=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20250123092444-cf4d252af8aa h1:YTfITllgt53Htbu0gDYOYpeBn6k9ZM8kt1ZmtZHbOlA=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20250123092444-cf4d252af8aa/go.mod h1:Hju1TEWZvrctQKbztTRwXH7rd41Yq0Pgmq4PrEKcq7o=
 github.com/pingcap/tipb v0.0.0-20241212101007-246f91188357 h1:s58UXyaWMNeaoeuVPZdrkm5Uk7NcODHqICGCUQ3A9s4=
@@ -1485,8 +1485,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
-github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de h1:fefo6U56UMca1CfwY/FVhfVcinyDgXgffQQp9lfMeLg=
-github.com/tikv/pd/client v0.0.0-20250107032658-5c4ab57d68de/go.mod h1:6xcCIcECSowarZ9+AqdURngHzS+t5w2x3qhwg1moj4o=
+github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42 h1:hvB0tf4u5ZU1vAew3MmbyxdYuogDLjIZYzSFt1ACFqo=
+github.com/tikv/pd/client v0.0.0-20250213082949-e8930327be42/go.mod h1:veL5JbFL99G5cvS/Q2cLiFQklIiSrVgsBCwFnXc+Zgo=
 github.com/tklauser/go-sysconf v0.3.9/go.mod h1:11DU/5sG7UexIrp/O6g35hrWzu0JxlwQ3LSFUzyeuhs=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
@@ -1684,8 +1684,8 @@ golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.9.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.10.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/mod v0.11.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/mod v0.22.0 h1:D4nJWe9zXqHOmWqj4VMOJhvzj7bEZg4wEYa759z1pH4=
-golang.org/x/mod v0.22.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
+golang.org/x/mod v0.23.0 h1:Zb7khfcRGKk+kqfxFaP5tZqCnDZMjC5VtUBs87Hr6QM=
+golang.org/x/mod v0.23.0/go.mod h1:6SkKJ3Xj0I0BrPOZoBy3bdMptDDU9oJrpohJ3eWZ1fY=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -1785,8 +1785,8 @@ golang.org/x/oauth2 v0.7.0/go.mod h1:hPLQkd9LyjfXTiRohC/41GhcFqxisoUQ99sCUOHO9x4
 golang.org/x/oauth2 v0.8.0/go.mod h1:yr7u4HXZRm1R1kBWqr/xKNqewf0plRYoB7sla+BCIXE=
 golang.org/x/oauth2 v0.10.0/go.mod h1:kTpgurOux7LqtuxjuyZa4Gj2gdezIt/jQtGnNFfypQI=
 golang.org/x/oauth2 v0.11.0/go.mod h1:LdF7O/8bLR/qWK9DrpXmbHLTouvRHK0SgJl0GmDBchk=
-golang.org/x/oauth2 v0.24.0 h1:KTBBxWqUa0ykRPLtV69rRto9TLXcqYkeswu48x/gvNE=
-golang.org/x/oauth2 v0.24.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
+golang.org/x/oauth2 v0.25.0 h1:CY4y7XT9v0cRI9oupztF8AgiIu99L/ksR/Xp/6jrZ70=
+golang.org/x/oauth2 v0.25.0/go.mod h1:XYTD2NtWslqkgxebSiOHnXEap4TF09sJSc7H1sXbhtI=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -1897,8 +1897,8 @@ golang.org/x/sys v0.9.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.10.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.13.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
-golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
+golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
@@ -1943,8 +1943,8 @@ golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20220922220347-f3bd1da661af/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.1.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.7.0 h1:ntUhktv3OPE6TgYxXWv9vKvUSJyIFJlyohwbkEwPrKQ=
-golang.org/x/time v0.7.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
+golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY=
+golang.org/x/time v0.9.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180525024113-a5b4c53f6e8b/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go
index 1459aa8..e7a5070 100644
--- a/internal/locate/region_request3_test.go
+++ b/internal/locate/region_request3_test.go
@@ -1412,9 +1412,6 @@ func (s *testRegionRequestToThreeStoresSuite) TestDoNotTryUnreachableLeader() {
 	follower, _, _, _ := region.FollowerStorePeer(regionStore, 0, &storeSelectorOp{})
 
 	s.regionRequestSender.client = &fnClient{fn: func(ctx context.Context, addr string, req *tikvrpc.Request, timeout time.Duration) (response *tikvrpc.Response, err error) {
-		if req.StaleRead && addr == follower.addr {
-			return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{RegionError: &errorpb.Error{DataIsNotReady: &errorpb.DataIsNotReady{}}}}, nil
-		}
 		return &tikvrpc.Response{Resp: &kvrpcpb.GetResponse{
 			Value: []byte(addr),
 		}}, nil
diff --git a/internal/locate/region_request_state_test.go b/internal/locate/region_request_state_test.go
index a5b7280..96b5932 100644
--- a/internal/locate/region_request_state_test.go
+++ b/internal/locate/region_request_state_test.go
@@ -333,7 +333,7 @@ func TestRegionCacheStaleRead(t *testing.T) {
 			leaderRegionValid:       true,
 			leaderAsyncReload:       util.Some(true),
 			leaderSuccessReplica:    []string{"z2", "z3"},
-			leaderSuccessReadType:   SuccessFollowerRead,
+			leaderSuccessReadType:   SuccessStaleRead,
 			followerRegionValid:     true,
 			followerAsyncReload:     util.None[bool](),
 			followerSuccessReplica:  []string{"z2"},
diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go
index 3721729..980a4cf 100644
--- a/internal/locate/replica_selector_test.go
+++ b/internal/locate/replica_selector_test.go
@@ -2144,7 +2144,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) {
 			beforeRun: func() { /* don't resetStoreState */ },
 			expect: &accessPathResult{
 				accessPath: []string{
-					"{addr: store3, replica-read: true, stale-read: false}",
+					"{addr: store3, replica-read: false, stale-read: true}",
 				},
 				respErr:         "",
 				respRegionError: fakeEpochNotMatch,
diff --git a/internal/unionstore/memdb_bench_test.go b/internal/unionstore/memdb_bench_test.go
index 8a2c3e5..07fa6d3 100644
--- a/internal/unionstore/memdb_bench_test.go
+++ b/internal/unionstore/memdb_bench_test.go
@@ -172,13 +172,36 @@ func BenchmarkMemDbBufferRandom(b *testing.B) {
 }
 
 func BenchmarkMemDbIter(b *testing.B) {
-	fn := func(b *testing.B, buffer MemBuffer) {
+	fnIter := func(b *testing.B, buffer MemBuffer) {
 		benchIterator(b, buffer)
 		b.ReportAllocs()
 	}
 
-	b.Run("RBT", func(b *testing.B) { fn(b, newRbtDBWithContext()) })
-	b.Run("ART", func(b *testing.B) { fn(b, newArtDBWithContext()) })
+	b.Run("RBT", func(b *testing.B) { fnIter(b, newRbtDBWithContext()) })
+	b.Run("ART", func(b *testing.B) { fnIter(b, newArtDBWithContext()) })
+}
+
+func BenchmarkSnapshotIter(b *testing.B) {
+	f := func(b *testing.B, buffer MemBuffer) {
+		benchSnapshotIter(b, buffer)
+		b.ReportAllocs()
+	}
+
+	fBatched := func(b *testing.B, buffer MemBuffer) {
+		benchBatchedSnapshotIter(b, buffer)
+		b.ReportAllocs()
+	}
+
+	fForEach := func(b *testing.B, buffer MemBuffer) {
+		benchForEachInSnapshot(b, buffer)
+		b.ReportAllocs()
+	}
+
+	b.Run("RBT-SnapshotIter", func(b *testing.B) { f(b, newRbtDBWithContext()) })
+	// unimplemented for RBT
+	b.Run("ART-SnapshotIter", func(b *testing.B) { f(b, newArtDBWithContext()) })
+	b.Run("ART-BatchedSnapshotIter", func(b *testing.B) { fBatched(b, newArtDBWithContext()) })
+	b.Run("ART-ForEachInSnapshot", func(b *testing.B) { fForEach(b, newArtDBWithContext()) })
 }
 
 func BenchmarkMemDbCreation(b *testing.B) {
@@ -224,6 +247,40 @@ func benchIterator(b *testing.B, buffer MemBuffer) {
 		if err != nil {
 			b.Error(err)
 		}
+		for iter.Valid() {
+			_ = iter.Key()
+			_ = iter.Value()
+			iter.Next()
+		}
+		iter.Close()
+	}
+}
+
+func benchSnapshotIter(b *testing.B, buffer MemBuffer) {
+	for k := 0; k < opCnt; k++ {
+		buffer.Set(encodeInt(k), encodeInt(k))
+	}
+	buffer.Staging()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		iter := buffer.SnapshotIter(nil, nil)
+		for iter.Valid() {
+			_ = iter.Value()
+			_ = iter.Key()
+			iter.Next()
+		}
+		iter.Close()
+	}
+}
+
+func benchBatchedSnapshotIter(b *testing.B, buffer MemBuffer) {
+	for k := 0; k < opCnt; k++ {
+		buffer.Set(encodeInt(k), encodeInt(k))
+	}
+	buffer.Staging()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		iter := buffer.BatchedSnapshotIter(nil, nil, false)
 		for iter.Valid() {
 			iter.Next()
 		}
@@ -231,6 +288,23 @@ func benchIterator(b *testing.B, buffer MemBuffer) {
 	}
 }
 
+func benchForEachInSnapshot(b *testing.B, buffer MemBuffer) {
+	for k := 0; k < opCnt; k++ {
+		buffer.Set(encodeInt(k), encodeInt(k))
+	}
+	buffer.Staging()
+	b.ResetTimer()
+	f := func(key, value []byte) (bool, error) {
+		return false, nil
+	}
+	for i := 0; i < b.N; i++ {
+		err := buffer.ForEachInSnapshotRange(nil, nil, f, false)
+		if err != nil {
+			b.Error(err)
+		}
+	}
+}
+
 func BenchmarkMemBufferCache(b *testing.B) {
 	fn := func(b *testing.B, buffer MemBuffer) {
 		buf := make([][keySize]byte, b.N)
diff --git a/internal/unionstore/memdb_test.go b/internal/unionstore/memdb_test.go
index 4721f83..ccd0699 100644
--- a/internal/unionstore/memdb_test.go
+++ b/internal/unionstore/memdb_test.go
@@ -1364,3 +1364,253 @@ func TestSnapshotReaderWithWrite(t *testing.T) {
 	check(newRbtDBWithContext(), 48)
 	check(newArtDBWithContext(), 48)
 }
+
+func TestBatchedSnapshotIter(t *testing.T) {
+	check := func(db *artDBWithContext, num int) {
+		// Insert test data
+		for i := 0; i < num; i++ {
+			db.Set([]byte{0, byte(i)}, []byte{0, byte(i)})
+		}
+		h := db.Staging()
+		defer db.Release(h)
+
+		// Create iterator - should be positioned at first key
+		iter := db.BatchedSnapshotIter([]byte{0, 0}, []byte{0, 255}, false)
+		defer iter.Close()
+
+		// Should be able to read first key immediately
+		require.True(t, iter.Valid())
+		require.Equal(t, []byte{0, 0}, iter.Key())
+
+		// Write additional data
+		db.Set([]byte{0, byte(num)}, []byte{0, byte(num)})
+		for i := 0; i < num; i++ {
+			db.Set([]byte{1, byte(i)}, []byte{1, byte(i)})
+		}
+
+		// Verify iteration
+		i := 0
+		for ; i < num; i++ {
+			require.True(t, iter.Valid())
+			require.Equal(t, []byte{0, byte(i)}, iter.Key())
+			require.Equal(t, []byte{0, byte(i)}, iter.Value())
+			require.NoError(t, iter.Next())
+		}
+		require.False(t, iter.Valid())
+	}
+
+	checkReverse := func(db *artDBWithContext, num int) {
+		for i := 0; i < num; i++ {
+			db.Set([]byte{0, byte(i)}, []byte{0, byte(i)})
+		}
+		h := db.Staging()
+		defer db.Release(h)
+
+		iter := db.BatchedSnapshotIter([]byte{0, 0}, []byte{0, 255}, true)
+		defer iter.Close()
+
+		// Should be positioned at last key
+		require.True(t, iter.Valid())
+		require.Equal(t, []byte{0, byte(num - 1)}, iter.Key())
+
+		db.Set([]byte{0, byte(num)}, []byte{0, byte(num)})
+		for i := 0; i < num; i++ {
+			db.Set([]byte{1, byte(i)}, []byte{1, byte(i)})
+		}
+
+		i := num - 1
+		for ; i >= 0; i-- {
+			require.True(t, iter.Valid())
+			require.Equal(t, []byte{0, byte(i)}, iter.Key())
+			require.Equal(t, []byte{0, byte(i)}, iter.Value())
+			require.NoError(t, iter.Next())
+		}
+		require.False(t, iter.Valid())
+	}
+
+	// Run size test cases
+	check(newArtDBWithContext(), 3)
+	check(newArtDBWithContext(), 17)
+	check(newArtDBWithContext(), 64)
+
+	checkReverse(newArtDBWithContext(), 3)
+	checkReverse(newArtDBWithContext(), 17)
+	checkReverse(newArtDBWithContext(), 64)
+}
+
+func TestBatchedSnapshotIterEdgeCase(t *testing.T) {
+	t.Run("EdgeCases", func(t *testing.T) {
+		db := newArtDBWithContext()
+		h := db.Staging()
+		// invalid range - should be invalid immediately
+		iter := db.BatchedSnapshotIter([]byte{1}, []byte{1}, false)
+		require.False(t, iter.Valid())
+		iter.Close()
+
+		// empty range - should be invalid immediately
+		iter = db.BatchedSnapshotIter([]byte{0}, []byte{1}, false)
+		require.False(t, iter.Valid())
+		iter.Close()
+
+		// Single element range
+		_ = db.Set([]byte{1}, []byte{1})
+		db.Release(h)
+		h = db.Staging()
+		iter = db.BatchedSnapshotIter([]byte{1}, []byte{2}, false)
+		require.True(t, iter.Valid())
+		require.Equal(t, []byte{1}, iter.Key())
+		require.NoError(t, iter.Next())
+		require.False(t, iter.Valid())
+		iter.Close()
+
+		// Multiple elements
+		_ = db.Set([]byte{2}, []byte{2})
+		_ = db.Set([]byte{3}, []byte{3})
+		_ = db.Set([]byte{4}, []byte{4})
+		db.Release(h)
+		_ = db.Staging()
+
+		// Forward iteration [2,4)
+		iter = db.BatchedSnapshotIter([]byte{2}, []byte{4}, false)
+		vals := []byte{}
+		for iter.Valid() {
+			vals = append(vals, iter.Key()[0])
+			require.NoError(t, iter.Next())
+		}
+		require.Equal(t, []byte{2, 3}, vals)
+		iter.Close()
+
+		// Reverse iteration [2,4)
+		iter = db.BatchedSnapshotIter([]byte{2}, []byte{4}, true)
+		vals = []byte{}
+		for iter.Valid() {
+			vals = append(vals, iter.Key()[0])
+			require.NoError(t, iter.Next())
+		}
+		require.Equal(t, []byte{3, 2}, vals)
+		iter.Close()
+	})
+
+	t.Run("BoundaryTests", func(t *testing.T) {
+		db := newArtDBWithContext()
+		keys := [][]byte{
+			{1, 0}, {1, 2}, {1, 4}, {1, 6}, {1, 8},
+		}
+		for _, k := range keys {
+			_ = db.Set(k, k)
+		}
+
+		// lower bound included
+		h := db.Staging()
+		defer db.Release(h)
+		iter := db.BatchedSnapshotIter([]byte{1, 2}, []byte{1, 9}, false)
+		vals := []byte{}
+		for iter.Valid() {
+			vals = append(vals, iter.Key()[1])
+			require.NoError(t, iter.Next())
+		}
+		require.Equal(t, []byte{2, 4, 6, 8}, vals)
+		iter.Close()
+
+		// upper bound excluded
+		iter = db.BatchedSnapshotIter([]byte{1, 0}, []byte{1, 6}, false)
+		vals = []byte{}
+		for iter.Valid() {
+			vals = append(vals, iter.Key()[1])
+			require.NoError(t, iter.Next())
+		}
+		require.Equal(t, []byte{0, 2, 4}, vals)
+		iter.Close()
+
+		// reverse
+		iter = db.BatchedSnapshotIter([]byte{1, 0}, []byte{1, 6}, true)
+		vals = []byte{}
+		for iter.Valid() {
+			vals = append(vals, iter.Key()[1])
+			require.NoError(t, iter.Next())
+		}
+		require.Equal(t, []byte{4, 2, 0}, vals)
+		iter.Close()
+	})
+
+	t.Run("AlphabeticalOrder", func(t *testing.T) {
+		db := newArtDBWithContext()
+		keys := [][]byte{
+			{2},
+			{2, 1},
+			{2, 1, 1},
+			{2, 1, 1, 1},
+		}
+		for _, k := range keys {
+			_ = db.Set(k, k)
+		}
+		h := db.Staging()
+		defer db.Release(h)
+		// forward
+		iter := db.BatchedSnapshotIter([]byte{2}, []byte{3}, false)
+		count := 0
+		for iter.Valid() {
+			require.Equal(t, keys[count], iter.Key())
+			require.NoError(t, iter.Next())
+			count++
+		}
+		require.Equal(t, len(keys), count)
+		iter.Close()
+
+		// reverse
+		iter = db.BatchedSnapshotIter([]byte{2}, []byte{3}, true)
+		count = len(keys) - 1
+		for iter.Valid() {
+			require.Equal(t, keys[count], iter.Key())
+			require.NoError(t, iter.Next())
+			count--
+		}
+		require.Equal(t, -1, count)
+		iter.Close()
+	})
+
+	t.Run("BatchSizeGrowth", func(t *testing.T) {
+		db := newArtDBWithContext()
+
+		for i := 0; i < 100; i++ {
+			_ = db.Set([]byte{3, byte(i)}, []byte{3, byte(i)})
+		}
+		h := db.Staging()
+		defer db.Release(h)
+		// forward
+		iter := db.BatchedSnapshotIter([]byte{3, 0}, []byte{3, 255}, false)
+		count := 0
+		for iter.Valid() {
+			require.Equal(t, []byte{3, byte(count)}, iter.Key())
+			require.NoError(t, iter.Next())
+			count++
+		}
+		require.Equal(t, 100, count)
+		iter.Close()
+
+		// reverse
+		iter = db.BatchedSnapshotIter([]byte{3, 0}, []byte{3, 255}, true)
+		count = 99
+		for iter.Valid() {
+			require.Equal(t, []byte{3, byte(count)}, iter.Key())
+			require.NoError(t, iter.Next())
+			count--
+		}
+		require.Equal(t, -1, count)
+		iter.Close()
+	})
+
+	t.Run("SnapshotChange", func(t *testing.T) {
+		db := newArtDBWithContext()
+		_ = db.Set([]byte{0}, []byte{0})
+		h := db.Staging()
+		_ = db.Set([]byte{byte(1)}, []byte{byte(1)})
+		iter := db.BatchedSnapshotIter([]byte{0}, []byte{255}, false)
+		require.True(t, iter.Valid())
+		require.NoError(t, iter.Next())
+		db.Release(h)
+		db.Staging()
+		require.False(t, iter.Valid())
+		require.Error(t, iter.Next())
+	})
+}