Repository: tikv/client-go (go)

Issue:
1 tikv down may affect too long time when enable forwarding
Related Issue: https://github.com/pingcap/tidb/issues/52394

### Unexpected

Issue description: 1 tikv down will cause qps drop and did not recover for a long time, which is unexpected.

Related metrics:

TiDB:

The problem is, when 1 tikv down, TiDB doesn't invalidate the regions in the store and get region from PD, and then still use forwarding feature to send request.

<img width="1456" alt="image" src="https://github.com/tikv/client-go/assets/26020263/1977aff4-3864-4513-89b0-6a80028e0218">



TiKV:

<img width="1456" alt="image" src="https://github.com/tikv/client-go/assets/26020263/822c4e61-bfd6-454b-862c-934a7f532f42">


### Expected

The expected TiDB metrics maybe like following, when 1 tikv server down, TiDB needs to invalidate the regions in the store and get region from PD, since those regions will has new leader.

<img width="1463" alt="image" src="https://github.com/tikv/client-go/assets/26020263/f2f8f842-5378-4a71-b86d-32525237b943">


Pull request description:
fix bug of not invalidate store when tikv store is down in replica-selector-v2 with enable forwarding
close #1274

Issue description: 1 tikv down will cause qps drop and did not recover for a long time, which is unexpected.

### Manual Test

```sh
sysbench --config-file=config oltp_read_only --tables=2 --table-size=1000000 run
```

After run a while, stop 1 tikv node.

**Before This PR**

TiDB Metrics:

- 13:40:00 stop 1 tikv
- 13:44:00 restart the down tikv.

<img width="1450" alt="image" src="https://github.com/tikv/client-go/assets/26020263/62544aec-d9bb-4c32-b794-c6bacdccd4f2">

TiKV metrics:

<img width="1449" alt="image" src="https://github.com/tikv/client-go/assets/26020263/960fb59f-33a6-4432-a94d-298a0143fe11">


**This PR**

- 13:51:00 stop 1 tikv
- 13:54:00 restart the down tikv.

<img width="1451" alt="image" src="https://github.com/tikv/client-go/assets/26020263/07370f4d-1ac6-4d4f-a146-3e5ca2c7e70f">

<img width="1449" alt="image" src="https://github.com/tikv/client-go/assets/26020263/821d340b-bebb-407f-b3b3-d409ea172c7f">



Patch:
diff --git a/internal/locate/replica_selector.go b/internal/locate/replica_selector.go
index 895add0..c288861 100644
--- a/internal/locate/replica_selector.go
+++ b/internal/locate/replica_selector.go
@@ -122,10 +122,13 @@ func (s *replicaSelectorV2) next(bo *retry.Backoffer, req *tikvrpc.Request) (rpc
 func (s *replicaSelectorV2) nextForReplicaReadLeader(req *tikvrpc.Request) {
 	if s.regionCache.enableForwarding {
 		strategy := ReplicaSelectLeaderWithProxyStrategy{}
-		s.target, s.proxy = strategy.next(s.replicas, s.region)
+		s.target, s.proxy = strategy.next(s)
 		if s.target != nil && s.proxy != nil {
 			return
 		}
+		if s.target == nil && s.proxy == nil {
+			return
+		}
 	}
 	leaderIdx := s.region.getStore().workTiKVIdx
 	strategy := ReplicaSelectLeaderStrategy{leaderIdx: leaderIdx}
@@ -134,7 +137,7 @@ func (s *replicaSelectorV2) nextForReplicaReadLeader(req *tikvrpc.Request) {
 		// If the leader is busy in our estimation, try other idle replicas.
 		// If other replicas are all busy, tryIdleReplica will try the leader again without busy threshold.
 		mixedStrategy := ReplicaSelectMixedStrategy{leaderIdx: leaderIdx, busyThreshold: s.busyThreshold}
-		idleTarget := mixedStrategy.next(s, s.region)
+		idleTarget := mixedStrategy.next(s)
 		if idleTarget != nil {
 			s.target = idleTarget
 			req.ReplicaRead = true
@@ -149,7 +152,7 @@ func (s *replicaSelectorV2) nextForReplicaReadLeader(req *tikvrpc.Request) {
 		return
 	}
 	mixedStrategy := ReplicaSelectMixedStrategy{leaderIdx: leaderIdx, leaderOnly: s.option.leaderOnly}
-	s.target = mixedStrategy.next(s, s.region)
+	s.target = mixedStrategy.next(s)
 	if s.target != nil && s.isReadOnlyReq && s.replicas[leaderIdx].deadlineErrUsingConfTimeout {
 		req.ReplicaRead = true
 		req.StaleRead = false
@@ -178,7 +181,7 @@ func (s *replicaSelectorV2) nextForReplicaReadMixed(req *tikvrpc.Request) {
 		labels:       s.option.labels,
 		stores:       s.option.stores,
 	}
-	s.target = strategy.next(s, s.region)
+	s.target = strategy.next(s)
 	if s.target != nil {
 		if s.isStaleRead && s.attempts == 1 {
 			// stale-read request first access.
@@ -233,7 +236,7 @@ type ReplicaSelectMixedStrategy struct {
 	busyThreshold time.Duration
 }
 
-func (s *ReplicaSelectMixedStrategy) next(selector *replicaSelectorV2, region *Region) *replica {
+func (s *ReplicaSelectMixedStrategy) next(selector *replicaSelectorV2) *replica {
 	replicas := selector.replicas
 	maxScoreIdxes := make([]int, 0, len(replicas))
 	var maxScore storeSelectionScore = -1
@@ -396,14 +399,15 @@ func (s *ReplicaSelectMixedStrategy) calculateScore(r *replica, isLeader bool) s
 
 type ReplicaSelectLeaderWithProxyStrategy struct{}
 
-func (s ReplicaSelectLeaderWithProxyStrategy) next(replicas []*replica, region *Region) (leader *replica, proxy *replica) {
-	rs := region.getStore()
+func (s ReplicaSelectLeaderWithProxyStrategy) next(selector *replicaSelectorV2) (leader *replica, proxy *replica) {
+	rs := selector.region.getStore()
 	leaderIdx := rs.workTiKVIdx
+	replicas := selector.replicas
 	leader = replicas[leaderIdx]
 	if leader.store.getLivenessState() == reachable || leader.notLeader {
 		// if leader's store is reachable, no need use proxy.
-		rs.unsetProxyStoreIfNeeded(region)
-		return nil, nil
+		rs.unsetProxyStoreIfNeeded(selector.region)
+		return leader, nil
 	}
 	proxyIdx := rs.proxyTiKVIdx
 	if proxyIdx >= 0 && int(proxyIdx) < len(replicas) && s.isCandidate(replicas[proxyIdx], proxyIdx == leaderIdx) {
@@ -415,6 +419,10 @@ func (s ReplicaSelectLeaderWithProxyStrategy) next(replicas []*replica, region *
 			return leader, r
 		}
 	}
+	// If all followers are tried as a proxy and fail, mark the leader store invalid, then backoff and retry.
+	metrics.TiKVReplicaSelectorFailureCounter.WithLabelValues("exhausted").Inc()
+	selector.invalidateReplicaStore(leader, errors.Errorf("all followers are tried as proxy but fail"))
+	selector.region.setSyncFlags(needReloadOnAccess)
 	return nil, nil
 }
 


Test patch:
diff --git a/internal/locate/replica_selector_test.go b/internal/locate/replica_selector_test.go
index 521700d..e4d249b 100644
--- a/internal/locate/replica_selector_test.go
+++ b/internal/locate/replica_selector_test.go
@@ -95,8 +95,8 @@ type replicaSelectorAccessPathCase struct {
 	accessErrInValid bool
 	expect           *accessPathResult
 	result           accessPathResult
-	beforeRun        func() // beforeRun will be called before the test case execute, if it is nil, resetStoreState will be called.
-	afterRun         func() // afterRun will be called after the test case execute, if it is nil, invalidateRegion will be called.
+	beforeRun        func()                // beforeRun will be called before the test case execute, if it is nil, resetStoreState will be called.
+	afterRun         func(ReplicaSelector) // afterRun will be called after the test case execute, if it is nil, invalidateRegion will be called.
 }
 
 type accessPathResult struct {
@@ -1408,7 +1408,7 @@ func TestReplicaReadAccessPathByLeaderCase(t *testing.T) {
 				backoffDetail:   []string{"regionScheduling+1"},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -1426,7 +1426,7 @@ func TestReplicaReadAccessPathByLeaderCase(t *testing.T) {
 				backoffDetail:   []string{},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -1732,7 +1732,7 @@ func TestReplicaReadAccessPathByMixedAndPreferLeaderCase(t *testing.T) {
 				backoffDetail:   []string{},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -1776,7 +1776,7 @@ func TestReplicaReadAccessPathByMixedAndPreferLeaderCase(t *testing.T) {
 				backoffDetail:   []string{},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -2125,7 +2125,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) {
 				backoffDetail:   []string{"tikvRPC+2"},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -2144,7 +2144,7 @@ func TestReplicaReadAccessPathByStaleReadCase(t *testing.T) {
 				backoffDetail:   []string{"tikvServerBusy+1"},
 				regionIsValid:   false,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 	}
 	s.True(s.runMultiCaseAndCompare(cas))
@@ -2404,7 +2404,7 @@ func TestReplicaReadAccessPathByProxyCase(t *testing.T) {
 				backoffDetail:   []string{"tikvRPC+1"},
 				regionIsValid:   true,
 			},
-			afterRun: func() { /* don't invalid region */ },
+			afterRun: func(_ ReplicaSelector) { /* don't invalid region */ },
 		},
 		{
 			reqType:   tikvrpc.CmdGet,
@@ -2480,6 +2480,13 @@ func TestReplicaReadAccessPathByProxyCase(t *testing.T) {
 			backoffDetail:   []string{"tikvRPC+1", "tikvServerBusy+2"},
 			regionIsValid:   false,
 		},
+		afterRun: func(selector ReplicaSelector) {
+			base := selector.getBaseReplicaSelector()
+			s.NotNil(base)
+			s.True(base.replicas[0].isEpochStale())
+			s.True(base.replicas[0].epoch < atomic.LoadUint32(&base.replicas[0].store.epoch))
+			s.False(base.region.isValid())
+		},
 	}
 	s.True(s.runCaseAndCompare(ca))
 
@@ -2955,7 +2962,7 @@ func beforeRun(s *testReplicaSelectorSuite, ca *replicaSelectorAccessPathCase) {
 
 func afterRun(ca *replicaSelectorAccessPathCase, sender *RegionRequestSender) {
 	if ca.afterRun != nil {
-		ca.afterRun()
+		ca.afterRun(sender.replicaSelector)
 	} else {
 		sender.replicaSelector.invalidateRegion() // invalidate region to reload for next test case.
 	}