From 0f6fa252631be6e152da5eef128b881e597b9f6f Mon Sep 17 00:00:00 2001 From: Vlad Dmitriev Date: Sun, 16 Feb 2025 20:12:30 -0800 Subject: [PATCH 1/6] sync etcd endpoints immediately after initializing the client (#1573) X-Original-Commit: ba68268e435ce3b2d392247a2a72caf9992b1f54 --- tikv/kv.go | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/tikv/kv.go b/tikv/kv.go index 49066c7921..a1f1e2d861 100644 --- a/tikv/kv.go +++ b/tikv/kv.go @@ -98,6 +98,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 } From e4b47c998ad7cef75a2c3618cfe1d1982a3cbcd0 Mon Sep 17 00:00:00 2001 From: mittalrishabh Date: Sun, 1 Dec 2024 23:37:11 -0800 Subject: [PATCH 2/6] retry as stale read on replica (#1509) X-Original-Commit: 7d0c64842f2926449f9668893426dcb2b4a43836 --- internal/locate/region_request_state_test.go | 14 ++--- internal/locate/replica_selector.go | 35 ++++++++--- internal/locate/replica_selector_test.go | 63 ++++++++++++-------- 3 files changed, 70 insertions(+), 42 deletions(-) diff --git a/internal/locate/region_request_state_test.go b/internal/locate/region_request_state_test.go index 18b297e915..80b5eb963e 100644 --- a/internal/locate/region_request_state_test.go +++ b/internal/locate/region_request_state_test.go @@ -383,7 +383,7 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z2", "z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z2"}, @@ -408,11 +408,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderServerIsBusy, @@ -421,11 +421,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z2", "z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z2", "z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderServerIsBusy, @@ -434,11 +434,11 @@ func testRegionCacheStaleRead(t *testing.T) { leaderRegionValid: true, leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z3"}, - leaderSuccessReadType: SuccessFollowerRead, + leaderSuccessReadType: SuccessStaleRead, followerRegionValid: true, followerAsyncReload: util.Some(false), followerSuccessReplica: []string{"z3"}, - followerSuccessReadType: SuccessFollowerRead, + followerSuccessReadType: SuccessStaleRead, }, { do: leaderDown, diff --git a/internal/locate/replica_selector.go b/internal/locate/replica_selector.go index 499b0fd10d..992ebd9049 100644 --- a/internal/locate/replica_selector.go +++ b/internal/locate/replica_selector.go @@ -182,17 +182,21 @@ func (s *replicaSelector) nextForReplicaReadMixed(req *tikvrpc.Request) { } s.target = strategy.next(s) if s.target != nil { - if s.isStaleRead && s.attempts == 1 { - // stale-read request first access. - if !s.target.store.IsLabelsMatch(s.option.labels) && s.target.peer.Id != s.region.GetLeaderPeerID() { - // If the target replica's labels is not match and not leader, use replica read. - // This is for compatible with old version. - req.StaleRead = false - req.ReplicaRead = true - } else { - // use stale read. + if s.isStaleRead { + isStaleRead := true + if s.attempts != 1 || (!s.target.store.IsLabelsMatch(s.option.labels) && s.target.peer.Id != s.region.GetLeaderPeerID()) { + // retry or target replica's labels does not match and not leader + if strategy.canSendReplicaRead(s) { + // use replica read. + isStaleRead = false + } + } + if isStaleRead { req.StaleRead = true req.ReplicaRead = false + } else { + req.StaleRead = false + req.ReplicaRead = true } } else { // use replica read only if the target is not leader. @@ -300,6 +304,16 @@ func (s *ReplicaSelectMixedStrategy) next(selector *replicaSelector) *replica { return nil } +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 + return false + } + return true +} + func hasDeadlineExceededError(replicas []*replica) bool { for _, replica := range replicas { if replica.hasFlag(deadlineErrUsingConfTimeoutFlag) { @@ -546,6 +560,9 @@ func (s *replicaSelector) onServerIsBusy( backoffErr := errors.Errorf("server is busy, ctx: %v", ctx) if s.canFastRetry() { s.addPendingBackoff(store, retry.BoTiKVServerBusy, backoffErr) + if s.target != nil { + s.target.addFlag(serverIsBusyFlag) + } return true, nil } err = bo.Backoff(retry.BoTiKVServerBusy, backoffErr) diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go index cb688831fa..a8842aebfc 100644 --- a/internal/locate/replica_selector_test.go +++ b/internal/locate/replica_selector_test.go @@ -376,7 +376,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store2, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -418,7 +418,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, // store2 has DeadLineExceededErr, so don't retry store2 even it is new leader. + "{addr: store3, replica-read: false, stale-read: true}"}, // store2 has DeadLineExceededErr, so don't retry store2 even it is new leader. respErr: "", respRegionError: nil, backoffCnt: 0, @@ -518,8 +518,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: fakeEpochNotMatch, backoffCnt: 1, @@ -562,7 +562,7 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store3, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store2, replica-read: true, stale-read: false}"}, + "{addr: store2, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -582,8 +582,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -604,8 +604,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -625,8 +625,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -668,8 +668,8 @@ func testReplicaReadAccessPathByCase(s *testReplicaSelectorSuite) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -948,7 +948,7 @@ func testReplicaReadAccessPathByCase2(s *testReplicaSelectorSuite) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}"}, + "{addr: store3, replica-read: false, stale-read: true}"}, respErr: "", respRegionError: nil, backoffCnt: 0, @@ -1094,9 +1094,16 @@ func testReplicaReadAccessPathByBasicCase(s *testReplicaSelectorSuite) { backoff = []string{} } if staleRead { - accessPath = []string{ - "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", + if tp == ServerIsBusyErr || tp == ServerIsBusyWithEstimatedWaitMsErr { + accessPath = []string{ + "{addr: store1, replica-read: false, stale-read: true}", + "{addr: store2, replica-read: false, stale-read: true}", + } + } else { + accessPath = []string{ + "{addr: store1, replica-read: false, stale-read: true}", + "{addr: store2, replica-read: true, stale-read: false}", + } } } default: @@ -2014,8 +2021,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -2036,8 +2043,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", "{addr: store2, replica-read: false, stale-read: false}", // try leader with leader read. - "{addr: store3, replica-read: true, stale-read: false}", - "{addr: store1, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", + "{addr: store1, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: fakeEpochNotMatch, @@ -2146,8 +2153,8 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { expect: &accessPathResult{ accessPath: []string{ "{addr: store1, replica-read: false, stale-read: true}", - "{addr: store2, replica-read: true, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store2, replica-read: false, stale-read: true}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2169,7 +2176,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2191,7 +2198,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) { accessPath: []string{ "{addr: store2, replica-read: false, stale-read: true}", "{addr: store1, replica-read: false, stale-read: false}", - "{addr: store3, replica-read: true, stale-read: false}", + "{addr: store3, replica-read: false, stale-read: true}", }, respErr: "", respRegionError: nil, @@ -2762,7 +2769,11 @@ func TestReplicaReadAvoidSlowStore(t *testing.T) { if expectedFirstStore == 3 { // Retry on store 2 which is a follower. // Stale-read mode falls back to replica-read mode. - expectedSecondPath = "{addr: store2, replica-read: true, stale-read: false}" + if staleRead { + expectedSecondPath = "{addr: store2, replica-read: false, stale-read: true}" + } else { + expectedSecondPath = "{addr: store2, replica-read: true, stale-read: false}" + } } else { // Retry in leader read mode will not use replica-read. expectedSecondPath = "{addr: store3, replica-read: false, stale-read: false}" From 39a0588b22798f33939ef77052573068b2b4b797 Mon Sep 17 00:00:00 2001 From: mittalrishabh Date: Tue, 4 Feb 2025 19:01:03 -0800 Subject: [PATCH 3/6] retry stale read as stale if leader is not accessible (#1566) X-Original-Commit: 004b333873cc38e4719b49440d752247c1e704cd --- internal/locate/region_request3_test.go | 3 --- internal/locate/region_request_state_test.go | 2 +- internal/locate/replica_selector.go | 4 ++-- internal/locate/replica_selector_test.go | 2 +- 4 files changed, 4 insertions(+), 7 deletions(-) diff --git a/internal/locate/region_request3_test.go b/internal/locate/region_request3_test.go index 9ded6f3169..31b5549723 100644 --- a/internal/locate/region_request3_test.go +++ b/internal/locate/region_request3_test.go @@ -1410,9 +1410,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 80b5eb963e..2e9dafcc52 100644 --- a/internal/locate/region_request_state_test.go +++ b/internal/locate/region_request_state_test.go @@ -348,7 +348,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.go b/internal/locate/replica_selector.go index 992ebd9049..0c05390385 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/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go index a8842aebfc..fa49bef1b6 100644 --- a/internal/locate/replica_selector_test.go +++ b/internal/locate/replica_selector_test.go @@ -2240,7 +2240,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, From 065ec5e1b85344eb5e4bfffbf1f68206d58beee0 Mon Sep 17 00:00:00 2001 From: mittalrishabh Date: Thu, 19 Mar 2026 22:02:14 -0700 Subject: [PATCH 4/6] hard-invalidate region on RegionNotFound instead of async reload (#1916) X-Original-Commit: 978018487d8aaed2aad5f1d6221539c3b682f6d0 --- internal/locate/region_cache.go | 12 ------- internal/locate/region_request_state_test.go | 6 ++-- internal/locate/replica_selector.go | 33 +++++++++++++------- internal/locate/replica_selector_test.go | 6 ++-- 4 files changed, 29 insertions(+), 28 deletions(-) diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go index 38b7727e3c..6deb0e9c36 100644 --- a/internal/locate/region_cache.go +++ b/internal/locate/region_cache.go @@ -1827,18 +1827,6 @@ func (c *RegionCache) BatchLoadRegionsFromKey(bo *retry.Backoffer, startKey []by return regions[len(regions)-1].EndKey(), nil } -// AsyncInvalidateCachedRegion marks a cached region for reload on next access -// without invalidating it, so in-flight retries can still proceed. -// It sets needDelayedReloadReady directly, skipping the GC promotion from -// needDelayedReloadPending for faster reload. -func (c *RegionCache) AsyncInvalidateCachedRegion(id RegionVerID) { - cachedRegion := c.GetCachedRegionWithRLock(id) - if cachedRegion == nil { - return - } - cachedRegion.setSyncFlags(needDelayedReloadReady) -} - // InvalidateCachedRegion removes a cached Region. func (c *RegionCache) InvalidateCachedRegion(id RegionVerID) { c.InvalidateCachedRegionWithReason(id, Other) diff --git a/internal/locate/region_request_state_test.go b/internal/locate/region_request_state_test.go index 2e9dafcc52..cc9fd75074 100644 --- a/internal/locate/region_request_state_test.go +++ b/internal/locate/region_request_state_test.go @@ -303,9 +303,9 @@ func testRegionCacheStaleRead(t *testing.T) { leaderAsyncReload: util.Some(false), leaderSuccessReplica: []string{"z1"}, leaderSuccessReadType: SuccessStaleRead, - followerRegionValid: true, - followerAsyncReload: util.Some(true), - // may async reload region and access it from leader. + followerRegionValid: false, + followerAsyncReload: util.Some(false), + // region is hard-invalidated but the current request succeeds via leader retry. followerSuccessReplica: []string{"z1"}, followerSuccessReadType: SuccessStaleRead, }, diff --git a/internal/locate/replica_selector.go b/internal/locate/replica_selector.go index 0c05390385..65f9947083 100644 --- a/internal/locate/replica_selector.go +++ b/internal/locate/replica_selector.go @@ -31,13 +31,14 @@ import ( type replicaSelector struct { baseReplicaSelector - replicaReadType kv.ReplicaReadType - isStaleRead bool - isReadOnlyReq bool - option storeSelectorOp - target *replica - proxy *replica - attempts int + replicaReadType kv.ReplicaReadType + isStaleRead bool + isReadOnlyReq bool + option storeSelectorOp + target *replica + proxy *replica + attempts int + regionInvalidatedForRetry bool // set when region is hard-invalidated but this selector should still retry on leader } func newReplicaSelector( @@ -98,7 +99,13 @@ func buildTiKVReplicas(region *Region) []*replica { } func (s *replicaSelector) next(bo *retry.Backoffer, req *tikvrpc.Request) (rpcCtx *RPCContext, err error) { - if !s.region.isValid() { + if s.regionInvalidatedForRetry { + // Allow one more attempt on this selector even though the region is + // invalidated. This is set by onRegionNotFound so that the current + // request can retry on the leader while concurrent requests are + // stopped by the hard invalidation. + s.regionInvalidatedForRetry = false + } else if !s.region.isValid() { metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("invalid").Inc() return nil, nil } @@ -523,11 +530,15 @@ func (s *replicaSelector) onRegionNotFound( leaderIdx := s.region.getStore().workTiKVIdx leader := s.replicas[leaderIdx] if !leader.isExhausted(1, 0) { - // if the request is not sent to leader, we can retry it with leader and invalidate the region cache asynchronously. It helps in the scenario - // where region is split by the leader but not yet created in replica due to replica down. + // if the request is not sent to leader, we can retry it with leader and invalidate the region cache + // immediately. It helps in the scenario where region is split by the leader but not yet created + // in replica due to replica down. + // Hard-invalidate the region so that concurrent requests stop using the stale region, + // but allow this selector to retry on the leader via regionInvalidatedForRetry. req.SetReplicaReadType(kv.ReplicaReadLeader) s.replicaReadType = kv.ReplicaReadLeader - s.regionCache.AsyncInvalidateCachedRegion(ctx.Region) + s.regionInvalidatedForRetry = true + s.regionCache.InvalidateCachedRegion(ctx.Region) return true, nil } s.regionCache.InvalidateCachedRegion(ctx.Region) diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go index fa49bef1b6..0df3ef4c85 100644 --- a/internal/locate/replica_selector_test.go +++ b/internal/locate/replica_selector_test.go @@ -1156,14 +1156,16 @@ func testReplicaReadAccessPathByBasicCase(s *testReplicaSelectorSuite) { case kv.ReplicaReadLeader: accessPath = []string{"{addr: store1, replica-read: false, stale-read: false}"} case kv.ReplicaReadFollower: - // For RegionNotFoundErr from follower, it will retry on leader + // For RegionNotFoundErr from follower, it will retry on leader. + // The region is hard-invalidated to stop concurrent requests, + // but the current selector bypasses the validity check. if tp == RegionNotFoundErr { accessPath = []string{ "{addr: store2, replica-read: true, stale-read: false}", "{addr: store1, replica-read: false, stale-read: false}", } respRegionError = nil - regionIsValid = true + regionIsValid = false } else { accessPath = []string{"{addr: store2, replica-read: true, stale-read: false}"} } From 7895e73be6a19386b68b4feaf31298b472f106b4 Mon Sep 17 00:00:00 2001 From: ClamChowderTiDB Date: Wed, 13 May 2026 22:44:51 -0700 Subject: [PATCH 5/6] replace deprecated reflect.Ptr with reflect.Pointer X-Original-Commit: ed0ce441e96e9871c7e4a23d22238e8c8d6403b1 --- internal/logutil/hex.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/logutil/hex.go b/internal/logutil/hex.go index 1accd622ce..066a496357 100644 --- a/internal/logutil/hex.go +++ b/internal/logutil/hex.go @@ -87,7 +87,7 @@ func prettyPrint(w io.Writer, val reflect.Value) { prettyPrint(w, fv) } fmt.Fprintf(w, "}") - case reflect.Ptr: + case reflect.Pointer: if val.IsNil() { fmt.Fprintf(w, "%v", val.Interface()) } else { From 4a4daf23c789fa89c6300fc6b0aace5ba4303061 Mon Sep 17 00:00:00 2001 From: fzzf678 <108643977+fzzf678@users.noreply.github.com> Date: Thu, 29 Jan 2026 16:36:23 +0800 Subject: [PATCH 6/6] tikv: close region cache before PD client (#1853) X-Original-Commit: 6de135a0571dc0c0fc9ff313bf2b240819f2c86a --- internal/locate/region_cache.go | 7 +++++++ tikv/kv.go | 17 +++++++++++------ tikv/kv_test.go | 17 +++++++++++++++++ 3 files changed, 35 insertions(+), 6 deletions(-) diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go index 6deb0e9c36..9403adacf9 100644 --- a/internal/locate/region_cache.go +++ b/internal/locate/region_cache.go @@ -780,6 +780,13 @@ func (c *RegionCache) Close() { c.bg.shutdown(true) } +// IsBackgroundRunnerClosed returns whether RegionCache's background runner context has been canceled. +// +// It is intended for tests/debugging only. +func (c *RegionCache) IsBackgroundRunnerClosed() bool { + return c.bg.closed() +} + // checkAndResolve checks and resolve addr of failed stores. // this method isn't thread-safe and only be used by one goroutine. func (c *RegionCache) checkAndResolve(needCheckStores []*Store, needCheck func(*Store) bool) []*Store { diff --git a/tikv/kv.go b/tikv/kv.go index a1f1e2d861..15f6a9eee7 100644 --- a/tikv/kv.go +++ b/tikv/kv.go @@ -448,20 +448,25 @@ func (s *KVStore) Close() error { s.wg.Wait() s.oracle.Close() - s.pdClient.Close() - if s.pdHttpClient != nil { - s.pdHttpClient.Close() + if s.txnLatches != nil { + s.txnLatches.Close() } s.lockResolver.Close() + // Close region cache before closing clients it depends on. Otherwise, its background tasks might still try + // to access PD/TiKV and encounter "grpc: the client connection is closing". + s.regionCache.Close() if err := s.GetTiKVClient().Close(); err != nil { return err } - if s.txnLatches != nil { - s.txnLatches.Close() + if s.pdHttpClient != nil { + s.pdHttpClient.Close() } - s.regionCache.Close() + if _, err := util.EvalFailpoint("checkRegionCacheClosedBeforePDClose"); err == nil && !s.regionCache.IsBackgroundRunnerClosed() { + panic("region cache is not closed before closing pd client") + } + s.pdClient.Close() if err := s.kv.Close(); err != nil { return err diff --git a/tikv/kv_test.go b/tikv/kv_test.go index 8e609c4974..8a7042819d 100644 --- a/tikv/kv_test.go +++ b/tikv/kv_test.go @@ -22,6 +22,7 @@ import ( "testing" "time" + "github.com/pingcap/failpoint" "github.com/pingcap/kvproto/pkg/kvrpcpb" "github.com/pingcap/kvproto/pkg/metapb" "github.com/stretchr/testify/require" @@ -280,3 +281,19 @@ func (s *testKVSuite) TestErrorHalfwayInNewKVStore() { _, err := NewKVStore("TestErrorHalfwayInNewKVStore", s.store.pdClient, NewMockSafePointKV(), &mocktikv.RPCClient{}) require.Error(s.T(), err) } + +func TestKVStoreCloseCheckRegionCacheClosedBeforePDClose(t *testing.T) { + util.EnableFailpoints() + require.NoError(t, failpoint.Enable("tikvclient/checkRegionCacheClosedBeforePDClose", "return(true)")) + t.Cleanup(func() { + require.NoError(t, failpoint.Disable("tikvclient/checkRegionCacheClosedBeforePDClose")) + }) + + client, cluster, pdClient, err := testutils.NewMockTiKV("", nil) + require.NoError(t, err) + _, _, _, _ = mocktikv.BootstrapWithMultiStores(cluster, 1) + + store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0) + require.NoError(t, err) + require.NoError(t, store.Close()) +}