Repository: tikv/client-go (go)

Issue:
store's liveness may incorrectly marked as unreachable when the store restarts with label changed
### Introduce

store's liveness may incorrectly marked as unreachable when the store restarts with label changed, then RPC with leader strategy on that store will always got `Region is unavailable` error.


### How it happened?

When store stopped, the request on this sore will failed and then start health check by [checkUntilHealth](https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2622) in a separate goroutine.

https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2665-L2669 

And there is another background worker [asyncCheckAndResolveLoop](https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L458) to do [store.reResolve](https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2428) timely in a separate goroutine with following logic:

https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2456-L2463

When the store **restarts with label changes**, and if the execution order is the following:

1. A new store with **same id but different label** is added, and the new store's liveness is derived from  the old one which is still `unreachable`

https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2456-L2462

2. [checkUntilHealth](https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2622) check the old store become reachable and return:

https://github.com/tikv/client-go/blob/f9fbc4c8f5785442a686230bf453f9292e1acd09/internal/locate/region_cache.go#L2665-L2669

3. The old store's liveness became reachable, but the new store's liveness is still unreachable, and it will always be unreachable. Actually,  the new store is just restarted and available, while the client-go will not send request to this store since it's liveness is unreachable, then all region RPC request which leader on this store will failed and return `Region is unavailable` to user.


### Affetc Branch

master: fce0abfd82b6dbc28f8950e62bec74d971839877
tidb-8.1

tidb-6.5-with-kv-timeout-feature

Pull request:
fix issue that store's liveness may incorrectly marked as unreachable when the store restarts with label changed 
close #1401

Patch:
diff --git a/internal/locate/region_cache.go b/internal/locate/region_cache.go
index 151de21..4b3afdf 100644
--- a/internal/locate/region_cache.go
+++ b/internal/locate/region_cache.go
@@ -767,7 +767,7 @@ func (c *RegionCache) checkAndResolve(needCheckStores []*Store, needCheck func(*
 
 	needCheckStores = c.stores.filter(needCheckStores, needCheck)
 	for _, store := range needCheckStores {
-		_, err := store.reResolve(c.stores)
+		_, err := store.reResolve(c.stores, c.bg)
 		tikverr.Log(err)
 	}
 	return needCheckStores
@@ -1020,7 +1020,7 @@ func (c *RegionCache) GetTiFlashRPCContext(bo *retry.Backoffer, id RegionVerID,
 			return nil, nil
 		}
 		if store.getResolveState() == needCheck {
-			_, err := store.reResolve(c.stores)
+			_, err := store.reResolve(c.stores, c.bg)
 			tikverr.Log(err)
 		}
 		regionStore.workTiFlashIdx.Store(int32(accessIdx))
diff --git a/internal/locate/store_cache.go b/internal/locate/store_cache.go
index 2eba262..d0bea45 100644
--- a/internal/locate/store_cache.go
+++ b/internal/locate/store_cache.go
@@ -436,7 +436,7 @@ func (s *Store) initResolve(bo *retry.Backoffer, c storeCache) (addr string, err
 
 // reResolve try to resolve addr for store that need check. Returns false if the region is in tombstone state or is
 // deleted.
-func (s *Store) reResolve(c storeCache) (bool, error) {
+func (s *Store) reResolve(c storeCache, scheduler *bgRunner) (bool, error) {
 	var addr string
 	store, err := c.fetchStore(context.Background(), s.storeID)
 	if err != nil {
@@ -475,12 +475,23 @@ func (s *Store) reResolve(c storeCache) (bool, error) {
 			store.GetLabels(),
 		)
 		newStore.livenessState = atomic.LoadUint32(&s.livenessState)
-		newStore.unreachableSince = s.unreachableSince
+		if newStore.getLivenessState() != reachable {
+			newStore.unreachableSince = s.unreachableSince
+			startHealthCheckLoop(scheduler, c, newStore, newStore.getLivenessState(), storeReResolveInterval)
+		}
 		if s.addr == addr {
 			newStore.healthStatus = s.healthStatus
 		}
 		c.put(newStore)
 		s.setResolveState(deleted)
+		logutil.BgLogger().Info("store address or labels changed, add new store and mark old store deleted",
+			zap.Uint64("store", s.storeID),
+			zap.String("old-addr", s.addr),
+			zap.Any("old-labels", s.labels),
+			zap.String("old-liveness", s.getLivenessState().String()),
+			zap.String("new-addr", newStore.addr),
+			zap.Any("new-labels", newStore.labels),
+			zap.String("new-liveness", newStore.getLivenessState().String()))
 		return false, nil
 	}
 	s.changeResolveStateTo(needCheck, resolved)
@@ -580,6 +591,8 @@ func (s *Store) getLivenessState() livenessState {
 	return livenessState(atomic.LoadUint32(&s.livenessState))
 }
 
+var storeReResolveInterval = 30 * time.Second
+
 func (s *Store) requestLivenessAndStartHealthCheckLoopIfNeeded(bo *retry.Backoffer, scheduler *bgRunner, c storeCache) (liveness livenessState) {
 	liveness = requestLiveness(bo.GetCtx(), s, c)
 	if liveness == reachable {
@@ -595,7 +608,7 @@ func (s *Store) requestLivenessAndStartHealthCheckLoopIfNeeded(bo *retry.Backoff
 	// It may be already started by another thread.
 	if atomic.CompareAndSwapUint32(&s.livenessState, uint32(reachable), uint32(liveness)) {
 		s.unreachableSince = time.Now()
-		reResolveInterval := 30 * time.Second
+		reResolveInterval := storeReResolveInterval
 		if val, err := util.EvalFailpoint("injectReResolveInterval"); err == nil {
 			if dur, err := time.ParseDuration(val.(string)); err == nil {
 				reResolveInterval = dur
@@ -613,29 +626,19 @@ func startHealthCheckLoop(scheduler *bgRunner, c storeCache, s *Store, liveness
 	lastCheckPDTime := time.Now()
 
 	scheduler.schedule(func(ctx context.Context, t time.Time) bool {
+		if s.getResolveState() == deleted {
+			logutil.BgLogger().Info("[health check] store meta deleted, stop checking", zap.Uint64("storeID", s.storeID), zap.String("addr", s.addr), zap.String("state", s.getResolveState().String()))
+			return true
+		}
 		if t.Sub(lastCheckPDTime) > reResolveInterval {
 			lastCheckPDTime = t
 
-			valid, err := s.reResolve(c)
+			valid, err := s.reResolve(c, scheduler)
 			if err != nil {
 				logutil.BgLogger().Warn("[health check] failed to re-resolve unhealthy store", zap.Error(err))
 			} else if !valid {
-				if s.getResolveState() != deleted {
-					logutil.BgLogger().Warn("[health check] store was still unhealthy at the end of health check loop",
-						zap.Uint64("storeID", s.storeID),
-						zap.String("state", s.getResolveState().String()),
-						zap.String("liveness", s.getLivenessState().String()))
-					return true
-				}
-				// if the store is deleted, a new store with same id must be inserted (guaranteed by reResolve).
-				newStore, _ := c.get(s.storeID)
-				logutil.BgLogger().Info("[health check] store meta changed",
-					zap.Uint64("storeID", s.storeID),
-					zap.String("oldAddr", s.addr),
-					zap.String("oldLabels", fmt.Sprintf("%v", s.labels)),
-					zap.String("newAddr", newStore.addr),
-					zap.String("newLabels", fmt.Sprintf("%v", newStore.labels)))
-				s = newStore
+				logutil.BgLogger().Info("[health check] store meta deleted, stop checking", zap.Uint64("storeID", s.storeID), zap.String("addr", s.addr), zap.String("state", s.getResolveState().String()))
+				return true
 			}
 		}
 


Test patch:
diff --git a/internal/locate/region_cache_test.go b/internal/locate/region_cache_test.go
index 888b96d..7d9ff89 100644
--- a/internal/locate/region_cache_test.go
+++ b/internal/locate/region_cache_test.go
@@ -2833,3 +2833,44 @@ func (s *testRegionCacheSuite) TestScanRegionsWithGaps() {
 	})
 	s.Equal(batchScanRegionRes, regions)
 }
+
+func (s *testRegionCacheSuite) TestIssue1401() {
+	// init region cache
+	s.cache.LocateKey(s.bo, []byte("a"))
+
+	store1, _ := s.cache.stores.get(s.store1)
+	s.Require().NotNil(store1)
+	s.Require().Equal(resolved, store1.getResolveState())
+	// change store1 label.
+	labels := store1.labels
+	labels = append(labels, &metapb.StoreLabel{Key: "host", Value: "0.0.0.0:20161"})
+	s.cluster.UpdateStoreAddr(store1.storeID, store1.addr, labels...)
+
+	// mark the store is unreachable and need check.
+	atomic.StoreUint32(&store1.livenessState, uint32(unreachable))
+	store1.setResolveState(needCheck)
+
+	// setup mock liveness func
+	s.cache.stores.setMockRequestLiveness(func(ctx context.Context, s *Store) livenessState {
+		return reachable
+	})
+
+	// start health check loop
+	startHealthCheckLoop(s.cache.bg, s.cache.stores, store1, unreachable, time.Second*30)
+
+	// mock asyncCheckAndResolveLoop worker to check and resolve store.
+	s.cache.checkAndResolve(nil, func(s *Store) bool {
+		return s.getResolveState() == needCheck
+	})
+
+	// assert that the old store should be deleted.
+	s.Eventually(func() bool {
+		return store1.getResolveState() == deleted
+	}, 3*time.Second, time.Second)
+	// assert the new store should be added and it should be resolved and reachable.
+	newStore1, _ := s.cache.stores.get(s.store1)
+	s.Eventually(func() bool {
+		return newStore1.getResolveState() == resolved && newStore1.getLivenessState() == reachable
+	}, 3*time.Second, time.Second)
+	s.Require().True(isStoreContainLabel(newStore1.labels, "host", "0.0.0.0:20161"))
+}