Release 8.5 20260608 v8.5.6

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 {
@@ -1827,18 +1834,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)

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

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,
 		},
@@ -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"},
@@ -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,

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
 	}
@@ -182,17 +189,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 +311,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.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
+}
+
 func hasDeadlineExceededError(replicas []*replica) bool {
 	for _, replica := range replicas {
 		if replica.hasFlag(deadlineErrUsingConfTimeoutFlag) {
@@ -509,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)
@@ -546,6 +571,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)