*: reset prepare checker once the cache is reset

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

@@ -383,7 +383,7 @@ func deleteAllRegionCache(c *gin.Context) {
 		c.String(http.StatusInternalServerError, errs.ErrNotBootstrapped.GenWithStackByArgs().Error())
 		return
 	}
-	cluster.ResetRegionCache()
+	cluster.ResetPreparedAndResetRegionCache()
 	c.String(http.StatusOK, "All regions are removed from server cache.")
 }
 

pkg/mcs/scheduling/server/cluster.go

@@ -939,6 +939,17 @@ func (c *Cluster) SetPrepared() {
 	c.coordinator.GetPrepareChecker().SetPrepared()
 }
 
+// ResetPrepared reset the prepare checker.
+func (c *Cluster) ResetPrepared() {
+	c.coordinator.GetPrepareChecker().ResetPrepared()
+}
+
+// ResetPreparedAndResetRegionCache atomically pauses scheduling and clears the
+// region cache.
+func (c *Cluster) ResetPreparedAndResetRegionCache() {
+	c.coordinator.GetPrepareChecker().ResetPreparedAndRun(c.ResetRegionCache)
+}
+
 // IsSchedulingHalted returns whether the scheduling is halted.
 // Currently, the microservice scheduling is halted when:
 //   - The `HaltScheduling` persist option is set to true.

pkg/mcs/scheduling/server/server.go

@@ -222,7 +222,7 @@ func (s *Server) updatePDMemberLoop() {
 				if cluster != nil {
 					if cluster.SwitchPDLeader(pdpb.NewPDClient(cc)) {
 						if status.Leader != curLeader {
-							log.Info("switch PD leader", zap.String("leader-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0]))
+							log.Info("switch PD leader", zap.String("current-leader", strconv.FormatUint(curLeader, 16)), zap.String("new-leader-id", strconv.FormatUint(ep.ID, 16)), zap.String("endpoint", ep.ClientURLs[0]))
 						}
 						curLeader = ep.ID
 						break

pkg/mock/mockcluster/mockcluster.go

@@ -91,6 +91,11 @@ func NewCluster(ctx context.Context, opts *config.PersistOptions) *Cluster {
 	return c
 }
 
+// Context returns the cluster context.
+func (mc *Cluster) Context() context.Context {
+	return mc.ctx
+}
+
 // GetStoreConfig returns the store config.
 func (mc *Cluster) GetStoreConfig() sc.StoreConfigProvider {
 	return mc.PersistOptions.GetStoreConfig()
@@ -937,3 +942,12 @@ func (mc *Cluster) ObserveRegionsStats() {
 	storeIDs, writeBytesRates, writeKeysRates := mc.GetStoresWriteRate()
 	mc.HotStat.ObserveRegionsStats(storeIDs, writeBytesRates, writeKeysRates)
 }
+
+// ResetPrepared mocks method.
+func (*Cluster) ResetPrepared() {}
+
+// ResetPreparedAndResetRegionCache mocks method.
+func (mc *Cluster) ResetPreparedAndResetRegionCache(context.Context) error {
+	mc.ResetRegionCache()
+	return nil
+}

pkg/schedule/checker/checker_controller.go

@@ -35,6 +35,7 @@ import (
 	sche "github.com/tikv/pd/pkg/schedule/core"
 	"github.com/tikv/pd/pkg/schedule/labeler"
 	"github.com/tikv/pd/pkg/schedule/operator"
+	"github.com/tikv/pd/pkg/schedule/preparecheck"
 	"github.com/tikv/pd/pkg/utils/keyutil"
 	"github.com/tikv/pd/pkg/utils/logutil"
 )
@@ -94,13 +95,20 @@ type Controller struct {
 	// patrolRegionScanLimit is the limit of regions to scan.
 	// It is calculated by the number of regions.
 	patrolRegionScanLimit int
+	prepareChecker        *preparecheck.Checker
 	metrics               *checkerControllerMetrics
 }
 
 // NewController create a new Controller.
-func NewController(ctx context.Context, cluster sche.CheckerCluster, conf config.CheckerConfigProvider, opController *operator.Controller) *Controller {
+func NewController(ctx context.Context, cluster sche.CheckerCluster, conf config.CheckerConfigProvider, opController *operator.Controller, prepareCheckers ...*preparecheck.Checker) *Controller {
 	pendingProcessedRegions := cache.NewIDTTL(ctx, time.Minute, 3*time.Minute)
 	ruleManager := cluster.GetRuleManager()
+	prepareChecker := preparecheck.NewChecker(cluster.GetPrepareRegionCount)
+	if len(prepareCheckers) > 0 && prepareCheckers[0] != nil {
+		prepareChecker = prepareCheckers[0]
+	} else {
+		prepareChecker.SetPrepared()
+	}
 	c := &Controller{
 		ctx:                     ctx,
 		cluster:                 cluster,
@@ -119,6 +127,7 @@ func NewController(ctx context.Context, cluster sche.CheckerCluster, conf config
 		patrolRegionContext:     &PatrolRegionContext{},
 		interval:                cluster.GetCheckerConfig().GetPatrolRegionInterval(),
 		patrolRegionScanLimit:   calculateScanLimit(cluster),
+		prepareChecker:          prepareChecker,
 		metrics:                 newCheckerControllerMetrics(),
 	}
 	c.splitScatter = newSplitScatterController(ctx, cluster, opController, c.AddPendingProcessedRegions)
@@ -148,6 +157,9 @@ func (c *Controller) PatrolRegions() {
 			})
 			c.updateTickerIfNeeded(ticker)
 			c.updatePatrolWorkersIfNeeded()
+			if !c.prepareChecker.IsPrepared() {
+				continue
+			}
 			if c.cluster.IsSchedulingHalted() {
 				for len(c.patrolRegionContext.regionChan) > 0 {
 					<-c.patrolRegionContext.regionChan
@@ -174,7 +186,7 @@ func (c *Controller) PatrolRegions() {
 			})
 
 			measure(c.metrics.patrolPhaseHistograms[phaseDispatchSplitScatter], func() {
-				c.splitScatter.dispatchSplitScatterRegions()
+				c.dispatchSplitScatterRegions()
 			})
 
 			measure(c.metrics.patrolPhaseHistograms[phaseScanRegions], func() {
@@ -211,6 +223,10 @@ func (c *Controller) PatrolRegions() {
 	}
 }
 
+func (c *Controller) dispatchSplitScatterRegions() {
+	c.prepareChecker.RunIfPrepared(c.splitScatter.dispatchSplitScatterRegions)
+}
+
 func (c *Controller) updateTickerIfNeeded(ticker *time.Ticker) {
 	// Note: we reset the ticker here to support updating configuration dynamically.
 	newInterval := c.cluster.GetCheckerConfig().GetPatrolRegionInterval()
@@ -272,27 +288,29 @@ func (c *Controller) checkPendingProcessedRegions() {
 
 // checkPriorityRegions checks priority regions
 func (c *Controller) checkPriorityRegions() {
-	items := c.GetPriorityRegions()
-	removes := make([]uint64, 0)
-	priorityListGauge.Set(float64(len(items)))
-	for _, id := range items {
-		region := c.cluster.GetRegion(id)
-		if region == nil {
-			removes = append(removes, id)
-			continue
-		}
-		ops := c.CheckRegion(region)
-		// it should skip if region needs to merge
-		if len(ops) == 0 || ops[0].HasRelatedMergeRegion() {
-			continue
+	c.prepareChecker.RunIfPrepared(func() {
+		items := c.GetPriorityRegions()
+		removes := make([]uint64, 0)
+		priorityListGauge.Set(float64(len(items)))
+		for _, id := range items {
+			region := c.cluster.GetRegion(id)
+			if region == nil {
+				removes = append(removes, id)
+				continue
+			}
+			ops := c.CheckRegion(region)
+			// it should skip if region needs to merge
+			if len(ops) == 0 || ops[0].HasRelatedMergeRegion() {
+				continue
+			}
+			if !c.opController.ExceedStoreLimit(ops...) {
+				c.opController.AddWaitingOperator(ops...)
+			}
 		}
-		if !c.opController.ExceedStoreLimit(ops...) {
-			c.opController.AddWaitingOperator(ops...)
+		for _, v := range removes {
+			c.RemovePriorityRegions(v)
 		}
-	}
-	for _, v := range removes {
-		c.RemovePriorityRegions(v)
-	}
+	})
 }
 
 // CheckRegion will check the region and add a new operator if needed.
@@ -394,6 +412,12 @@ func (c *Controller) CheckRegion(region *core.RegionInfo) []*operator.Operator {
 }
 
 func (c *Controller) tryAddOperators(region *core.RegionInfo) {
+	c.prepareChecker.RunIfPrepared(func() {
+		c.tryAddOperatorsLocked(region)
+	})
+}
+
+func (c *Controller) tryAddOperatorsLocked(region *core.RegionInfo) {
 	if region == nil {
 		// the region could be recent split, continue to wait.
 		return

pkg/schedule/checker/checker_controller_test.go

@@ -0,0 +1,57 @@
+// Copyright 2026 TiKV Project Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package checker
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/tikv/pd/pkg/mock/mockcluster"
+	"github.com/tikv/pd/pkg/mock/mockconfig"
+	"github.com/tikv/pd/pkg/schedule/hbstream"
+	"github.com/tikv/pd/pkg/schedule/operator"
+	"github.com/tikv/pd/pkg/schedule/preparecheck"
+)
+
+func TestTryAddOperatorsRespectsPrepareChecker(t *testing.T) {
+	re := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	opts := mockconfig.NewTestOptions()
+	cluster := mockcluster.NewCluster(ctx, opts)
+	cluster.SetEnablePlacementRules(false)
+	cluster.SetMaxReplicas(3)
+	cluster.AddRegionStore(1, 1)
+	cluster.AddRegionStore(2, 0)
+	cluster.AddRegionStore(3, 0)
+	region := cluster.AddLeaderRegion(1, 1)
+
+	stream := hbstream.NewTestHeartbeatStreams(ctx, cluster, false /* no need to run */)
+	defer stream.Close()
+	opController := operator.NewController(ctx, cluster.GetBasicCluster(), cluster.GetSharedConfig(), stream)
+	prepareChecker := preparecheck.NewChecker(cluster.GetPrepareRegionCount)
+	controller := NewController(ctx, cluster, cluster.GetCheckerConfig(), opController, prepareChecker)
+
+	controller.tryAddOperators(region)
+	re.Empty(opController.GetWaitingOperators())
+	re.Empty(opController.GetOperators())
+
+	prepareChecker.SetPrepared()
+	controller.tryAddOperators(region)
+	re.NotEmpty(append(opController.GetWaitingOperators(), opController.GetOperators()...))
+}

pkg/schedule/checker/split_scatter_test.go

@@ -54,10 +54,6 @@ func (c *Controller) collectTopPendingSplitScatter(limit int) []splitScatterPend
 	return c.splitScatter.collectTopPendingSplitScatter(limit)
 }
 
-func (c *Controller) dispatchSplitScatterRegions() {
-	c.splitScatter.dispatchSplitScatterRegions()
-}
-
 func TestSplitScatterControllerCleanupResetsPendingGauge(t *testing.T) {
 	re := require.New(t)
 	splitScatterPendingGauge.Set(7)
@@ -122,6 +118,33 @@ func TestCheckSplitScatterRegionsCreatesScatterOperator(t *testing.T) {
 	re.Equal(group, batchGroup)
 }
 
+func TestDispatchSplitScatterRegionsRequiresPrepared(t *testing.T) {
+	re := require.New(t)
+	controller, tc, oc, cleanup := newTestSplitScatterController(t)
+	defer cleanup()
+
+	controller.RecordSplitScatterBatch(100, splitScatterTestSourceWaitVersion, []uint64{101, 102})
+	putSplitScatterRegion(tc, 101, "m", "t", splitScatterReportedCPUUsage)
+	putSplitScatterRegion(tc, 102, "t", "", splitScatterReportedCPUUsage)
+	advanceSplitScatterSourceVersion(t, tc)
+
+	controller.prepareChecker.ResetPrepared()
+	controller.dispatchSplitScatterRegions()
+	for _, regionID := range []uint64{100, 101, 102} {
+		re.Nil(oc.GetOperator(regionID))
+	}
+
+	controller.prepareChecker.SetPrepared()
+	controller.dispatchSplitScatterRegions()
+	var op *operator.Operator
+	for _, regionID := range []uint64{100, 101, 102} {
+		if op = oc.GetOperator(regionID); op != nil {
+			break
+		}
+	}
+	re.NotNil(op)
+}
+
 func TestDispatchSplitScatterKeepsPendingUntilSplitHeartbeat(t *testing.T) {
 	re := require.New(t)
 	controller, tc, oc, cleanup := newTestSplitScatterController(t)

pkg/schedule/coordinator.go

@@ -34,6 +34,7 @@ import (
 	"github.com/tikv/pd/pkg/schedule/diagnostic"
 	"github.com/tikv/pd/pkg/schedule/hbstream"
 	"github.com/tikv/pd/pkg/schedule/operator"
+	"github.com/tikv/pd/pkg/schedule/preparecheck"
 	"github.com/tikv/pd/pkg/schedule/scatter"
 	"github.com/tikv/pd/pkg/schedule/schedulers"
 	"github.com/tikv/pd/pkg/schedule/splitter"
@@ -46,9 +47,7 @@ import (
 
 const (
 	runSchedulerCheckInterval = 3 * time.Second
-	// CollectTimeout is the timeout for collecting regions.
-	CollectTimeout       = 5 * time.Minute
-	maxLoadConfigRetries = 10
+	maxLoadConfigRetries      = 10
 	// pushOperatorTickInterval is the interval try to push the operator.
 	pushOperatorTickInterval = 500 * time.Millisecond
 
@@ -69,7 +68,7 @@ type Coordinator struct {
 	schedulersInitialized bool
 
 	cluster           sche.ClusterInformer
-	prepareChecker    *prepareChecker
+	prepareChecker    *preparecheck.Checker
 	checkers          *checker.Controller
 	regionScatterer   *scatter.RegionScatterer
 	regionSplitter    *splitter.RegionSplitter
@@ -83,9 +82,10 @@ type Coordinator struct {
 // NewCoordinator creates a new Coordinator.
 func NewCoordinator(parentCtx context.Context, cluster sche.ClusterInformer, hbStreams *hbstream.HeartbeatStreams) *Coordinator {
 	ctx, cancel := context.WithCancel(parentCtx)
+	prepareChecker := preparecheck.NewChecker(cluster.GetPrepareRegionCount)
 	opController := operator.NewController(ctx, cluster.GetBasicCluster(), cluster.GetSharedConfig(), hbStreams)
-	schedulers := schedulers.NewController(ctx, cluster, cluster.GetStorage(), opController)
-	checkers := checker.NewController(ctx, cluster, cluster.GetCheckerConfig(), opController)
+	schedulers := schedulers.NewController(ctx, cluster, cluster.GetStorage(), opController, prepareChecker)
+	checkers := checker.NewController(ctx, cluster, cluster.GetCheckerConfig(), opController, prepareChecker)
 
 	// Set the callbacks for operator success
 	opController.SetSuccessCallbacks(
@@ -101,7 +101,7 @@ func NewCoordinator(parentCtx context.Context, cluster sche.ClusterInformer, hbS
 		cancel:                cancel,
 		schedulersInitialized: false,
 		cluster:               cluster,
-		prepareChecker:        newPrepareChecker(cluster.GetPrepareRegionCount),
+		prepareChecker:        prepareChecker,
 		checkers:              checkers,
 		regionScatterer:       scatter.NewRegionScatterer(ctx, cluster, opController, checkers.AddPendingProcessedRegions),
 		regionSplitter:        splitter.NewRegionSplitter(cluster, splitter.NewSplitRegionsHandler(cluster, opController), checkers.AddPendingProcessedRegions),
@@ -216,6 +216,26 @@ func (c *Coordinator) driveSlowNodeScheduler() {
 	}
 }
 
+func (c *Coordinator) runPrepareChecker() {
+	defer logutil.LogPanic()
+	defer c.wg.Done()
+
+	ticker := time.NewTicker(3 * time.Second)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-c.ctx.Done():
+			return
+		case <-ticker.C:
+			if !c.prepareChecker.IsPrepared() {
+				if c.prepareChecker.Check(c.cluster.GetBasicCluster()) {
+					log.Info("prepare checker is ready")
+				}
+			}
+		}
+	}
+}
+
 // RunUntilStop runs the coordinator until receiving the stop signal.
 func (c *Coordinator) RunUntilStop() {
 	c.Run()
@@ -249,7 +269,8 @@ func (c *Coordinator) Run() {
 	log.Info("coordinator starts to run schedulers")
 	c.InitSchedulers(true)
 
-	c.wg.Add(4)
+	c.wg.Add(5)
+	go c.runPrepareChecker()
 	// Starts to patrol regions.
 	go c.PatrolRegions()
 	// Checks suspect key ranges
@@ -581,7 +602,7 @@ func ResetHotSpotMetrics() {
 
 // ShouldRun returns true if the coordinator should run.
 func (c *Coordinator) ShouldRun() bool {
-	return c.prepareChecker.check(c.cluster.GetBasicCluster())
+	return c.prepareChecker.Check(c.cluster.GetBasicCluster())
 }
 
 // GetSchedulersController returns the schedulers controller.
@@ -649,7 +670,7 @@ func (c *Coordinator) GetRuleChecker() *checker.RuleChecker {
 }
 
 // GetPrepareChecker returns the prepare checker.
-func (c *Coordinator) GetPrepareChecker() *prepareChecker {
+func (c *Coordinator) GetPrepareChecker() *preparecheck.Checker {
 	return c.prepareChecker
 }