fix inconsistent store pointers between the region-cache and store-cache #1826
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -42,6 +42,7 @@ import ( | |||||
| "fmt" | ||||||
| "math/rand" | ||||||
| "reflect" | ||||||
| "slices" | ||||||
| "sync" | ||||||
| "sync/atomic" | ||||||
| "testing" | ||||||
|
|
@@ -364,7 +365,7 @@ func (s *testRegionCacheSuite) TestStoreLabels() { | |||||
| } | ||||||
| stores := s.cache.stores.filter(nil, func(s *Store) bool { return s.IsLabelsMatch(labels) }) | ||||||
| s.Equal(len(stores), 1) | ||||||
| s.Equal(stores[0].GetLabels(), labels) | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
|
|
@@ -442,14 +443,13 @@ func (s *testRegionCacheSuite) TestResolveStateTransition() { | |||||
| store = cache.stores.getOrInsertDefault(s.store1) | ||||||
| store.initResolve(bo, cache.stores) | ||||||
| s.Equal(store.getResolveState(), resolved) | ||||||
| s.cluster.UpdateStoreAddr(s.store1, store.GetAddr()+"0", &metapb.StoreLabel{Key: "k", Value: "v"}) | ||||||
| cache.stores.markStoreNeedCheck(store) | ||||||
| waitResolve(store) | ||||||
| newStore := cache.stores.getOrInsertDefault(s.store1) | ||||||
| s.Equal(newStore.getResolveState(), resolved) | ||||||
| s.Equal(newStore.GetAddr(), store.GetAddr()) | ||||||
| s.Equal(newStore.GetLabels(), []*metapb.StoreLabel{{Key: "k", Value: "v"}}) | ||||||
|
Comment on lines
449
to
+452
There was a problem hiding this comment. The assertions here no longer verify that the store address actually changed after UpdateStoreAddr. Since newStore and store are the same pointer with the new in-place update behavior, |
||||||
|
|
||||||
| // Check initResolve()ing a tombstone store. The resolve state should be tombstone. | ||||||
| cache.clear() | ||||||
|
|
@@ -2301,16 +2301,13 @@ func (s *testRegionCacheSuite) TestHealthCheckWithStoreReplace() { | |||||
| startHealthCheckLoop(s.cache.bg, s.cache.stores, store1, livenessState(store1Liveness), time.Second) | ||||||
|
|
||||||
| // update store meta | ||||||
| s.cluster.UpdateStoreAddr(store1.storeID, store1.GetAddr()+"'", store1.GetLabels()...) | ||||||
|
|
||||||
| // assert that the new store should be added and it's also not reachable | ||||||
| newStore1, _ := s.cache.stores.get(store1.storeID) | ||||||
| s.Require().NotEqual(reachable, newStore1.getLivenessState()) | ||||||
| // assert that the old store in region cache is replaced | ||||||
| s.Require().Equal(store1.GetAddr(), newStore1.GetAddr()) | ||||||
|
|
||||||
|
Comment on lines
2306
to
2311
There was a problem hiding this comment. This test still talks about a "new store" being added/replacing the old one, but the new implementation keeps the same Store pointer and updates metadata in-place. The current checks (
Contributor
There was a problem hiding this comment. +1, |
||||||
| // recover store1 | ||||||
| atomic.StoreUint32(&store1Liveness, uint32(reachable)) | ||||||
|
|
@@ -3119,9 +3116,9 @@ func (s *testRegionCacheSuite) TestIssue1401() { | |||||
| s.Require().NotNil(store1) | ||||||
| s.Require().Equal(resolved, store1.getResolveState()) | ||||||
| // change store1 label. | ||||||
| labels := slices.Clone(store1.GetLabels()) | ||||||
| labels = append(labels, &metapb.StoreLabel{Key: "host", Value: "0.0.0.0:20161"}) | ||||||
| s.cluster.UpdateStoreAddr(store1.storeID, store1.GetAddr(), labels...) | ||||||
|
|
||||||
| // mark the store is unreachable and need check. | ||||||
| atomic.StoreUint32(&store1.livenessState, uint32(unreachable)) | ||||||
|
|
@@ -3140,16 +3137,109 @@ func (s *testRegionCacheSuite) TestIssue1401() { | |||||
| return s.getResolveState() == needCheck | ||||||
| }) | ||||||
|
|
||||||
| // assert the new store should be added and it should be resolved and reachable. | ||||||
| newStore1, _ := s.cache.stores.get(s.store1) | ||||||
| // the store pointer should be updated | ||||||
| s.Equal(newStore1.GetAddr(), store1.GetAddr()) | ||||||
| s.Eventually(func() bool { | ||||||
| return newStore1.getResolveState() == resolved && store1.getLivenessState() == reachable | ||||||
| }, 3*time.Second, time.Second) | ||||||
|
Comment on lines
3140
to
3146
There was a problem hiding this comment. This test is meant to cover pointer consistency between caches (issue #1401/#1823), but it currently only compares addresses. If the goal is to ensure the store object is updated in-place (no replacement), add an assertion that |
||||||
| s.Require().True(isStoreContainLabel(newStore1.GetLabels(), "host", "0.0.0.0:20161")) | ||||||
| } | ||||||
|
|
||||||
| // TestStoreRestartWithNewLabels verifies that the store in replica selector still has | ||||||
| // reference to the old store when the store is restarted with new labels. | ||||||
| func (s *testRegionCacheSuite) TestStoreRestartWithNewLabels() { | ||||||
| // Initialize region cache | ||||||
| loc, err := s.cache.LocateKey(s.bo, []byte("a")) | ||||||
| s.Require().NoError(err) | ||||||
| s.Require().NotNil(loc) | ||||||
|
|
||||||
| // Get the leader store (store1 is typically the leader) | ||||||
| store1, exists := s.cache.stores.get(s.store1) | ||||||
| s.Require().True(exists) | ||||||
| s.Require().NotNil(store1) | ||||||
| s.Require().Equal(resolved, store1.getResolveState()) | ||||||
|
|
||||||
| // Step 1: Store starts as reachable (up) | ||||||
| atomic.StoreUint32(&store1.livenessState, uint32(reachable)) | ||||||
| s.Require().Equal(reachable, store1.getLivenessState()) | ||||||
|
|
||||||
| // Step 2: Store becomes unreachable (down) | ||||||
| store1Liveness := uint32(unreachable) | ||||||
| s.cache.stores.setMockRequestLiveness(func(ctx context.Context, st *Store) livenessState { | ||||||
| if st.storeID == store1.storeID { | ||||||
| return livenessState(atomic.LoadUint32(&store1Liveness)) | ||||||
| } | ||||||
| return reachable | ||||||
| }) | ||||||
| atomic.StoreUint32(&store1.livenessState, uint32(unreachable)) | ||||||
| s.Require().Equal(unreachable, store1.getLivenessState()) | ||||||
|
|
||||||
| // Start health check loop to monitor store | ||||||
| startHealthCheckLoop(s.cache.bg, s.cache.stores, store1, unreachable, time.Second) | ||||||
|
|
||||||
| // Create replica selector BEFORE store replacement to capture old store reference | ||||||
| region := s.cache.GetCachedRegionWithRLock(loc.Region) | ||||||
| s.Require().NotNil(region) | ||||||
| req := tikvrpc.NewRequest(tikvrpc.CmdGet, &kvrpcpb.GetRequest{Key: []byte("a")}, kvrpcpb.Context{}) | ||||||
| replicaSelector, err := newReplicaSelector(s.cache, loc.Region, req) | ||||||
| s.Require().NoError(err) | ||||||
| s.Require().NotNil(replicaSelector) | ||||||
|
|
||||||
| // Find the replica with store1 to keep reference to old store | ||||||
| var oldReplica *replica | ||||||
| var leaderPeer *metapb.Peer | ||||||
| for _, replica := range replicaSelector.replicas { | ||||||
| if replica.store.storeID == s.store1 { | ||||||
| oldReplica = replica | ||||||
| // Find the peer for this store | ||||||
| for _, peer := range region.GetMeta().Peers { | ||||||
| if peer.StoreId == s.store1 { | ||||||
| leaderPeer = peer | ||||||
| break | ||||||
| } | ||||||
| } | ||||||
| s.Require().NotNil(leaderPeer, "should find peer for store1") | ||||||
| s.Require().Equal(store1, replica.store, "replica should reference old store before replacement") | ||||||
| s.Require().Equal(unreachable, replica.store.getLivenessState(), "old store should be unreachable") | ||||||
| break | ||||||
| } | ||||||
| } | ||||||
| s.Require().NotNil(oldReplica, "should find replica with store1") | ||||||
|
|
||||||
| // Step 3: Same store_id changes label and restarts | ||||||
| // Change store1's label (address stays the same) | ||||||
| oldLabels := slices.Clone(store1.labels) | ||||||
|
There was a problem hiding this comment. This test starts a background health-check loop and then reads
Suggested change
|
||||||
| newLabels := append(oldLabels, &metapb.StoreLabel{Key: "zone", Value: "zone1"}) | ||||||
| s.cluster.UpdateStoreAddr(store1.storeID, store1.GetAddr(), newLabels...) | ||||||
|
|
||||||
| // Recover store1 (simulate restart) - set liveness back to reachable for new store | ||||||
| atomic.StoreUint32(&store1Liveness, uint32(reachable)) | ||||||
|
|
||||||
| // Trigger store metadata update by marking old store as needCheck and resolving | ||||||
| // This should trigger store_cache.go:494-497 logic: updating store metadata and resolving the store | ||||||
| store1.setResolveState(needCheck) | ||||||
| s.cache.checkAndResolve(nil, func(st *Store) bool { | ||||||
| return st.getResolveState() == needCheck | ||||||
| }) | ||||||
|
|
||||||
| // reResolve is synchronous, but add a small wait for robustness (max 500ms like TestResolveStateTransition) | ||||||
| s.Eventually(func() bool { | ||||||
| return store1.getResolveState() == resolved | ||||||
| }, 500*time.Millisecond, 50*time.Millisecond, "old store should be resolved") | ||||||
|
|
||||||
| // The store's labels should be updated | ||||||
| s.Require().True(isStoreContainLabel(store1.labels, "zone", "zone1"), "new store should have updated labels") | ||||||
|
There was a problem hiding this comment. Same issue as above: this assertion reads
Suggested change
|
||||||
| // Wait for new store to become reachable after restart | ||||||
| s.Eventually(func() bool { | ||||||
| return store1.getLivenessState() == reachable | ||||||
| }, 3*time.Second, 200*time.Millisecond, "new store should become reachable after restart") | ||||||
|
|
||||||
| // Step 4: Test the replica selector still has reference to OLD store (created before replacement) | ||||||
| s.Require().Equal(store1, oldReplica.store, "replica should still reference old store") | ||||||
| // The old store should be reachable | ||||||
| s.Require().Equal(reachable, oldReplica.store.getLivenessState(), "old store in replica should still be unreachable") | ||||||
|
There was a problem hiding this comment. The assertion message contradicts the expected value: it asserts
Suggested change
|
||||||
| } | ||||||
|
|
||||||
| func BenchmarkBatchLocateKeyRangesFromCache(t *testing.B) { | ||||||
|
|
||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The comment above this block still describes the old "deleted + new store" behavior, but the implementation now updates store metadata in-place. Please update the comment to match the new semantics so the test remains readable and doesn't mislead future debugging.