replication mode: fix wrong available store list

fix tikv#7221

Signed-off-by: disksing <i@disksing.com>

server/replication/replication_mode.go

@@ -483,7 +483,7 @@ func (m *ModeManager) tickUpdateState() {
 			m.drSwitchToSyncRecover()
 			break
 		}
-		if !reflect.DeepEqual(m.drGetAvailableStores(), stores[primaryUp]) && m.drCheckStoreStateUpdated(storeIDs(stores[primaryUp])) {
+		if !reflect.DeepEqual(m.drGetAvailableStores(), storeIDs(stores[primaryUp])) && m.drCheckStoreStateUpdated(storeIDs(stores[primaryUp])) {
 			m.drSwitchToAsync(storeIDs(stores[primaryUp]))
 		}
 	case drStateSyncRecover:
@@ -571,10 +571,6 @@ func (m *ModeManager) checkStoreStatus() [][]*core.StoreInfo {
 		if s.IsRemoved() {
 			continue
 		}
-		// learner peers do not participate in major commit or vote, so it should not count in primary/dr as a normal store.
-		if s.GetRegionCount() == s.GetLearnerCount() {
-			continue
-		}
 		down := s.DownTime() >= m.config.DRAutoSync.WaitStoreTimeout.Duration
 		labelValue := s.GetLabelValue(m.config.DRAutoSync.LabelKey)
 		if labelValue == m.config.DRAutoSync.Primary {

server/replication/replication_mode_test.go

@@ -635,6 +635,8 @@ func TestComplexPlacementRules(t *testing.T) {
 	setStoreState(cluster, "up", "up", "up", "up", "up", "up", "up", "down", "up", "down")
 	rep.tickUpdateState()
 	re.Equal(drStateAsyncWait, rep.drGetState())
+	rep.tickReplicateStatus()
+	re.Equal(fmt.Sprintf(`{"state":"async_wait","state_id":%d,"available_stores":[1,2,3,4,5,6]}`, rep.drAutoSync.StateID), replicator.lastData[1])
 
 	// reset to sync
 	setStoreState(cluster, "up", "up", "up", "up", "up", "up", "up", "up", "up", "up")
@@ -694,6 +696,47 @@ func TestComplexPlacementRules2(t *testing.T) {
 	setStoreState(cluster, "up", "up", "up", "up", "down", "down", "up")
 	rep.tickUpdateState()
 	re.Equal(drStateAsyncWait, rep.drGetState())
+	rep.tickReplicateStatus()
+	re.Equal(fmt.Sprintf(`{"state":"async_wait","state_id":%d,"available_stores":[1,2,3,4]}`, rep.drAutoSync.StateID), replicator.lastData[1])
+}
+
+func TestComplexPlacementRules3(t *testing.T) {
+	re := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	store := storage.NewStorageWithMemoryBackend()
+	conf := config.ReplicationModeConfig{ReplicationMode: modeDRAutoSync, DRAutoSync: config.DRAutoSyncReplicationConfig{
+		LabelKey:         "zone",
+		Primary:          "zone1",
+		DR:               "zone2",
+		WaitStoreTimeout: typeutil.Duration{Duration: time.Minute},
+	}}
+	cluster := mockcluster.NewCluster(ctx, config.NewTestOptions())
+	replicator := newMockReplicator([]uint64{1})
+	rep, err := NewReplicationModeManager(conf, store, cluster, replicator)
+	re.NoError(err)
+	cluster.GetRuleManager().SetAllGroupBundles(
+		genPlacementRuleConfig([]ruleConfig{
+			{key: "logic", value: "logic1", role: placement.Voter, count: 2},
+			{key: "logic", value: "logic2", role: placement.Learner, count: 1},
+			{key: "logic", value: "logic3", role: placement.Voter, count: 1},
+		}), true)
+
+	cluster.AddLabelsStore(1, 1, map[string]string{"zone": "zone1", "logic": "logic1"})
+	cluster.AddLabelsStore(2, 1, map[string]string{"zone": "zone1", "logic": "logic1"})
+	cluster.AddLabelsStore(3, 1, map[string]string{"zone": "zone1", "logic": "logic2"})
+	cluster.AddLabelsStore(4, 1, map[string]string{"zone": "zone1", "logic": "logic2"})
+	cluster.AddLabelsStore(5, 1, map[string]string{"zone": "zone2", "logic": "logic3"})
+
+	// initial state is sync
+	re.Equal(drStateSync, rep.drGetState())
+
+	// zone2 down, switch state, available stores should contain logic2 (learner)
+	setStoreState(cluster, "up", "up", "up", "up", "down")
+	rep.tickUpdateState()
+	re.Equal(drStateAsyncWait, rep.drGetState())
+	rep.tickReplicateStatus()
+	re.Equal(fmt.Sprintf(`{"state":"async_wait","state_id":%d,"available_stores":[1,2,3,4]}`, rep.drAutoSync.StateID), replicator.lastData[1])
 }
 
 func genRegions(cluster *mockcluster.Cluster, stateID uint64, n int) []*core.RegionInfo {