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 description:
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 cbd265e..aabeb81 100644
--- a/internal/locate/region_cache.go
+++ b/internal/locate/region_cache.go
@@ -2456,11 +2456,22 @@ func (s *Store) reResolve(c *RegionCache) (bool, error) {
 	if s.addr != addr || !s.IsSameLabels(store.GetLabels()) {
 		newStore := &Store{storeID: s.storeID, addr: addr, saddr: store.GetStatusAddress(), storeType: storeType, labels: store.GetLabels(), state: uint64(resolved)}
 		newStore.livenessState = atomic.LoadUint32(&s.livenessState)
-		newStore.unreachableSince = s.unreachableSince
+		if newStore.getLivenessState() != reachable {
+			newStore.unreachableSince = s.unreachableSince
+			go newStore.checkUntilHealth(c, newStore.getLivenessState(), storeReResolveInterval)
+		}
 		c.storeMu.Lock()
 		c.storeMu.stores[newStore.storeID] = newStore
 		c.storeMu.Unlock()
 		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)
@@ -2598,6 +2609,8 @@ func (s livenessState) String() string {
 	}
 }
 
+var storeReResolveInterval = 30 * time.Second
+
 func (s *Store) startHealthCheckLoopIfNeeded(c *RegionCache, liveness livenessState) {
 	// This mechanism doesn't support non-TiKV stores currently.
 	if s.storeType != tikvrpc.TiKV {
@@ -2609,7 +2622,7 @@ func (s *Store) startHealthCheckLoopIfNeeded(c *RegionCache, liveness livenessSt
 	// 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
@@ -2637,6 +2650,10 @@ func (s *Store) checkUntilHealth(c *RegionCache, liveness livenessState, reResol
 		case <-c.ctx.Done():
 			return
 		case <-ticker.C:
+			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
+			}
 			if time.Since(lastCheckPDTime) > reResolveInterval {
 				lastCheckPDTime = time.Now()
 
@@ -2644,19 +2661,7 @@ func (s *Store) checkUntilHealth(c *RegionCache, liveness livenessState, reResol
 				if err != nil {
 					logutil.BgLogger().Warn("[health check] failed to re-resolve unhealthy store", zap.Error(err))
 				} else if !valid {
-					if s.getResolveState() == deleted {
-						// if the store is deleted, a new store with same id must be inserted (guaranteed by reResolve).
-						c.storeMu.RLock()
-						newStore := c.storeMu.stores[s.storeID]
-						c.storeMu.RUnlock()
-						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)))
-						go newStore.checkUntilHealth(c, liveness, reResolveInterval)
-					}
+					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
 				}
 			}


Test patch:
diff --git a/internal/locate/region_cache_test.go b/internal/locate/region_cache_test.go
index 9d79619..4e2c54c 100644
--- a/internal/locate/region_cache_test.go
+++ b/internal/locate/region_cache_test.go
@@ -1708,3 +1708,45 @@ func (s *testRegionCacheSuite) TestHealthCheckWithStoreReplace() {
 		return newStore1.getResolveState() == resolved && newStore1.getLivenessState() == reachable
 	}, 3*time.Second, time.Second)
 }
+
+func (s *testRegionCacheSuite) TestIssue1401() {
+	// init region cache
+	s.cache.LocateKey(s.bo, []byte("a"))
+
+	store1 := s.cache.getStoreByStoreID(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
+	tf := func(s *Store, bo *retry.Backoffer) livenessState {
+		return reachable
+	}
+	s.cache.testingKnobs.mockRequestLiveness.Store((*livenessFunc)(&tf))
+
+	// start health check loop
+	go store1.checkUntilHealth(s.cache, 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.getStoreByStoreID(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"))
+}