schedule, server: fix replicated state semantics

pkg/mcs/scheduling/server/apis/v1/api.go

@@ -1466,7 +1466,7 @@ func splitRegions(c *gin.Context) {
 }
 
 // @Tags     region
-// @Summary  Check if regions in the given key ranges are replicated. Returns 'REPLICATED', 'INPROGRESS', or 'PENDING'. 'PENDING' means that there is at least one region pending for scheduling. Similarly, 'INPROGRESS' means there is at least one region in scheduling.
+// @Summary  Check if regions in the given key ranges are replicated. Returns 'REPLICATED', 'INPROGRESS', or 'PENDING'. 'PENDING' means that the current Store topology cannot satisfy at least one required placement-rule peer. 'INPROGRESS' means that placement is incomplete but PD can continue or retry scheduling, including incomplete Region metadata in the requested range.
 // @Param    startKey  query  string  true  "Regions start key, hex encoded"
 // @Param    endKey    query  string  true  "Regions end key, hex encoded"
 // @Produce  plain

pkg/schedule/checker/checker_controller.go

@@ -535,7 +535,7 @@ func (c *Controller) ClearSuspectKeyRanges() {
 
 // IsPendingRegion returns true if the given region is in the pending list.
 func (c *Controller) IsPendingRegion(regionID uint64) bool {
-	_, exist := c.ruleChecker.pendingList.Get(regionID)
+	_, exist := c.ruleChecker.pendingList.Peek(regionID)
 	return exist
 }
 

pkg/schedule/checker/replica_strategy.go

@@ -30,13 +30,15 @@ import (
 // ReplicaStrategy collects some utilities to manipulate region peers. It
 // exists to allow replica_checker and rule_checker to reuse common logics.
 type ReplicaStrategy struct {
-	checkerName    string // replica-checker / rule-checker
-	cluster        sche.CheckerCluster
-	locationLabels []string
-	isolationLevel string
-	region         *core.RegionInfo
-	extraFilters   []filter.Filter
-	fastFailover   bool
+	checkerName             string // replica-checker / rule-checker
+	cluster                 sche.CheckerCluster
+	locationLabels          []string
+	isolationLevel          string
+	region                  *core.RegionInfo
+	extraFilters            []filter.Filter
+	fastFailover            bool
+	storeCandidates         []*core.StoreInfo
+	storeCandidatesPrepared bool
 }
 
 // SelectStoreToAdd returns the store to add a replica to a region.
@@ -62,25 +64,29 @@ func (s *ReplicaStrategy) SelectStoreToAdd(coLocationStores []*core.StoreInfo, e
 	if s.fastFailover {
 		level = constant.Urgent
 	}
-	filters := []filter.Filter{
-		filter.NewExcludedFilter(s.checkerName, nil, s.region.GetStoreIDs()),
-		filter.NewStorageThresholdFilter(s.checkerName),
-		filter.NewSpecialUseFilter(s.checkerName),
-		&filter.StoreStateFilter{ActionScope: s.checkerName, MoveRegion: true, AllowTemporaryStates: true, OperatorLevel: level},
+	stores := s.storeCandidates
+	filters := []filter.Filter{filter.NewExcludedFilter(s.checkerName, nil, s.region.GetStoreIDs())}
+	if !s.storeCandidatesPrepared {
+		stores = s.cluster.GetStores()
+		filters = append(filters,
+			filter.NewStorageThresholdFilter(s.checkerName),
+			filter.NewSpecialUseFilter(s.checkerName),
+			&filter.StoreStateFilter{ActionScope: s.checkerName, MoveRegion: true, AllowTemporaryStates: true, OperatorLevel: level},
+		)
 	}
 	if len(s.locationLabels) > 0 && s.isolationLevel != "" {
 		filters = append(filters, filter.NewIsolationFilter(s.checkerName, s.isolationLevel, s.locationLabels, coLocationStores))
 	}
 	if len(extraFilters) > 0 {
 		filters = append(filters, extraFilters...)
 	}
-	if len(s.extraFilters) > 0 {
+	if !s.storeCandidatesPrepared && len(s.extraFilters) > 0 {
 		filters = append(filters, s.extraFilters...)
 	}
 
 	isolationComparer := filter.IsolationComparer(s.locationLabels, coLocationStores)
 	strictStateFilter := &filter.StoreStateFilter{ActionScope: s.checkerName, MoveRegion: true, AllowFastFailover: s.fastFailover, OperatorLevel: level}
-	targetCandidate := filter.NewCandidates(s.cluster.GetStores()).
+	targetCandidate := filter.NewCandidates(stores).
 		FilterTarget(s.cluster.GetCheckerConfig(), nil, nil, filters...).
 		KeepTheTopStores(isolationComparer, false) // greater isolation score is better
 	if targetCandidate.Len() == 0 {
@@ -94,6 +100,24 @@ func (s *ReplicaStrategy) SelectStoreToAdd(coLocationStores []*core.StoreInfo, e
 	return target.GetID(), false
 }
 
+// prepareStoreCandidates applies the Region-independent target filters so the
+// result can be reused by placement-state checks using the same Rule.
+func (s *ReplicaStrategy) prepareStoreCandidates(stores []*core.StoreInfo) {
+	level := constant.High
+	if s.fastFailover {
+		level = constant.Urgent
+	}
+	filters := []filter.Filter{
+		filter.NewStorageThresholdFilter(s.checkerName),
+		filter.NewSpecialUseFilter(s.checkerName),
+		&filter.StoreStateFilter{ActionScope: s.checkerName, MoveRegion: true, AllowTemporaryStates: true, OperatorLevel: level},
+	}
+	filters = append(filters, s.extraFilters...)
+	s.storeCandidates = filter.NewCandidates(stores).
+		FilterTarget(s.cluster.GetCheckerConfig(), nil, nil, filters...).Stores
+	s.storeCandidatesPrepared = true
+}
+
 // SelectStoreToFix returns a store to replace down/offline old peer. The location
 // placement after scheduling is allowed to be worse than original.
 func (s *ReplicaStrategy) SelectStoreToFix(coLocationStores []*core.StoreInfo, old uint64) (uint64, bool) {

pkg/schedule/checker/rule_checker.go

@@ -34,11 +34,24 @@
 	"github.com/tikv/pd/pkg/schedule/operator"
 	"github.com/tikv/pd/pkg/schedule/placement"
 	"github.com/tikv/pd/pkg/schedule/types"
+	"github.com/tikv/pd/pkg/statistics"
 	"github.com/tikv/pd/pkg/utils/syncutil"
 )
 
 const maxPendingListLen = 100000
 
+// RegionPlacementState describes the current placement-rule scheduling state.
+type RegionPlacementState int
+
+const (
+	// RegionPlacementStateReplicated means no placement work remains.
+	RegionPlacementStateReplicated RegionPlacementState = iota
+	// RegionPlacementStateInProgress means placement is incomplete but can progress or retry.
+	RegionPlacementStateInProgress
+	// RegionPlacementStatePending means the current topology cannot satisfy the placement rules.
+	RegionPlacementStatePending
+)
+
 // RuleChecker fix/improve region by placement rules.
 type RuleChecker struct {
 	PauseController
@@ -67,6 +80,161 @@
 	return types.RuleChecker.String()
 }
 
+type placementStateCandidateKey struct {
+	rule         *placement.Rule
+	fastFailover bool
+}
+
+type placementStateContext struct {
+	storesLoaded bool
+	stores       []*core.StoreInfo
+	candidates   map[placementStateCandidateKey][]*core.StoreInfo
+}
+
+func (ctx *placementStateContext) getStrategy(c *RuleChecker, region *core.RegionInfo, rule *placement.Rule, fastFailover bool) *ReplicaStrategy {
+	strategy := c.strategy(region, rule, fastFailover)
+	key := placementStateCandidateKey{rule: rule, fastFailover: fastFailover}
+	if candidates, ok := ctx.candidates[key]; ok {
+		strategy.storeCandidates = candidates
+		strategy.storeCandidatesPrepared = true
+		return strategy
+	}
+	if !ctx.storesLoaded {
+		ctx.stores = c.cluster.GetStores()
+		ctx.storesLoaded = true
+	}
+	if ctx.candidates == nil {
+		ctx.candidates = make(map[placementStateCandidateKey][]*core.StoreInfo)
+	}
+	strategy.prepareStoreCandidates(ctx.stores)
+	ctx.candidates[key] = strategy.storeCandidates
+	return strategy
+}
+
+// GetRegionPlacementState revalidates the placement state without creating an
+// Operator. It is intended for Regions already recorded in RuleChecker's
+// pending list.
+func (c *RuleChecker) GetRegionPlacementState(region *core.RegionInfo) RegionPlacementState {
+	return c.getRegionPlacementState(region, &placementStateContext{})
+}
+
+func (c *RuleChecker) getRegionPlacementState(region *core.RegionInfo, context *placementStateContext) RegionPlacementState {
+	if region.GetLeader() == nil || len(region.GetPendingPeers()) > 0 {
+		return RegionPlacementStateInProgress
+	}
+	fit := c.ruleManager.FitRegionWithoutCache(c.cluster, region)
+	if len(fit.RuleFits) == 0 || len(fit.OrphanPeers) > 0 {
+		return RegionPlacementStateInProgress
+	}
+
+	hasUnfixablePlacement := false
+	for _, rf := range fit.RuleFits {
+		if len(rf.Peers) < rf.Rule.Count {
+			isWitness := rf.Rule.IsWitness && isWitnessEnabled(c.cluster)
+			strategy := context.getStrategy(c, region, rf.Rule, isWitness)
+			storeID, filteredByTemporaryState := strategy.SelectStoreToAdd(getRuleFitStores(c.cluster, rf))
+			if storeID != 0 || filteredByTemporaryState ||
+				c.hasStoreToSwapForMissingRule(region, fit, rf, context) {
+				return RegionPlacementStateInProgress
+			}
+			hasUnfixablePlacement = true
+			continue
+		}
+
+		ruleUnfixable := false
+		for _, peer := range rf.Peers {
+			store := c.cluster.GetStore(peer.GetStoreId())
+			if store == nil {
+				return RegionPlacementStateInProgress
+			}
+
+			var fastFailover bool
+			switch {
+			case c.isDownPeer(region, peer):
+				if !c.isStoreDownTimeHitMaxDownTime(peer.GetStoreId()) {
+					return RegionPlacementStateInProgress
+				}
+				fastFailover = isWitnessEnabled(c.cluster) && store.IsTiKV()
+			case c.isOfflinePeer(peer):
+				fastFailover = isWitnessEnabled(c.cluster) && store.IsTiKV() && rf.Rule.IsWitness
+			default:
+				continue
+			}
+
+			strategy := context.getStrategy(c, region, rf.Rule, fastFailover)
+			storeID, filteredByTemporaryState := strategy.SelectStoreToFix(getRuleFitStores(c.cluster, rf), peer.GetStoreId())
+			if storeID != 0 || filteredByTemporaryState {
+				return RegionPlacementStateInProgress
+			}
+			hasUnfixablePlacement = true
+			ruleUnfixable = true
+			break
+		}
+		if ruleUnfixable {
+			continue
+		}
+		if len(rf.PeersWithDifferentRole) > 0 {
+			return RegionPlacementStateInProgress
+		}
+		if len(rf.Rule.LocationLabels) == 0 {
+			continue
+		}
+		isWitness := rf.Rule.IsWitness && isWitnessEnabled(c.cluster)
+		strategy := context.getStrategy(c, region, rf.Rule, isWitness)
+		_, newStoreID, filteredByTemporaryState := strategy.getBetterLocation(c.cluster, region, fit, rf)
+		if newStoreID != 0 || filteredByTemporaryState {
+			return RegionPlacementStateInProgress
+		}
+		if !statistics.IsRegionLabelIsolationSatisfied(
+			getRuleFitStores(c.cluster, rf),
+			rf.Rule.LocationLabels,
+			rf.Rule.IsolationLevel,
+		) {
+			hasUnfixablePlacement = true
+		}
+	}
+	if hasUnfixablePlacement {
+		return RegionPlacementStatePending
+	}
+	return RegionPlacementStateReplicated
+}
+
+// GetRegionsPlacementState returns the aggregate placement state of Regions.
+func (c *RuleChecker) GetRegionsPlacementState(regions []*core.RegionInfo) RegionPlacementState {
+	state := RegionPlacementStateReplicated
+	context := &placementStateContext{}
+	for _, region := range regions {
+		switch c.getRegionPlacementState(region, context) {
+		case RegionPlacementStatePending:
+			return RegionPlacementStatePending
+		case RegionPlacementStateInProgress:
+			state = RegionPlacementStateInProgress
+		}
+	}
+	return state
+}
+
+func (c *RuleChecker) hasStoreToSwapForMissingRule(region *core.RegionInfo, fit *placement.RegionFit, missingRuleFit *placement.RuleFit, context *placementStateContext) bool {
+	for _, peer := range region.GetPeers() {
+		store := c.cluster.GetStore(peer.GetStoreId())
+		if store == nil || !placement.MatchLabelConstraints(store, missingRuleFit.Rule.LabelConstraints) {
+			continue
+		}
+		oldRuleFit := fit.GetRuleFit(peer.GetId())
+		if oldRuleFit == nil || oldRuleFit == missingRuleFit || !oldRuleFit.IsSatisfied() {
+			continue
+		}
+
+		fastFailover := isWitnessEnabled(c.cluster) && store.IsTiKV() && oldRuleFit.Rule.IsWitness
+		strategy := context.getStrategy(c, region, oldRuleFit.Rule, fastFailover)
+		storeID, filteredByTemporaryState := strategy.SelectStoreToFix(getRuleFitStores(c.cluster, oldRuleFit), peer.GetStoreId())
+		if storeID != 0 || filteredByTemporaryState {
+			return true
+		}
+	}
+	return false
+}
+
 // GetType returns RuleChecker's type.
 func (*RuleChecker) GetType() types.CheckerSchedulerType {
 	return types.RuleChecker