diff --git a/maintainer/barrier.go b/maintainer/barrier.go index b9ba504ed2..c5326a5be3 100644 --- a/maintainer/barrier.go +++ b/maintainer/barrier.go @@ -80,10 +80,54 @@ func (b *Barrier) HandleStatus(from node.ID, for _, status := range request.BlockStatuses { // only receive block status from the replicating dispatcher dispatcherID := common.NewDispatcherIDFromPB(status.ID) + task := b.spanController.GetTaskByID(dispatcherID) + if task == nil { + log.Info("Get block status from unexisted dispatcher, ignore it", + zap.String("changefeed", request.ChangefeedID.GetName()), + zap.String("dispatcher", dispatcherID.String()), + zap.Uint64("commitTs", status.State.BlockTs), + zap.Int64("mode", b.mode)) + continue + } + ownerNodeID := task.GetNodeID() + if ownerNodeID != from { + log.Warn("ignore block status from non-owner dispatcher", + zap.String("changefeed", request.ChangefeedID.GetName()), + zap.String("dispatcherID", dispatcherID.String()), + zap.String("ownerNodeID", ownerNodeID.String()), + zap.String("fromNodeID", from.String()), + zap.Uint64("commitTs", status.State.BlockTs), + zap.Int64("mode", b.mode)) + continue + } if dispatcherID != b.spanController.GetDDLDispatcherID() { +<<<<<<< HEAD task := b.spanController.GetTaskByID(dispatcherID) if task == nil { log.Info("Get block status from unexisted dispatcher, ignore it", zap.String("changefeed", request.ChangefeedID.GetName()), zap.String("dispatcher", dispatcherID.String()), zap.Uint64("commitTs", status.State.BlockTs), zap.Int64("mode", b.mode)) +======= + if !b.spanController.IsReplicating(task) { + log.Info("Get block status from unreplicating dispatcher, ignore it", + zap.String("changefeed", request.ChangefeedID.GetName()), + zap.String("dispatcher", dispatcherID.String()), + zap.Uint64("commitTs", status.State.BlockTs), + zap.Int64("mode", b.mode)) + // A newly added dispatcher may report its first WAITING barrier before the add + // operator moves it from scheduling to replicating. We still cannot admit that + // status into barrier, but silently dropping it would leave dispatcher waiting + // for the slow 5s resend timer. Return IgnoredBlockStatus so it keeps the live + // WAITING state locally and schedules a fast retry instead. + dispatcherStatus = append(dispatcherStatus, &heartbeatpb.DispatcherStatus{ + InfluencedDispatchers: &heartbeatpb.InfluencedDispatchers{ + InfluenceType: heartbeatpb.InfluenceType_Normal, + DispatcherIDs: []*heartbeatpb.DispatcherID{status.ID}, + }, + IgnoredBlockStatus: &heartbeatpb.IgnoredBlockStatus{ + CommitTs: status.State.BlockTs, + IsSyncPoint: status.State.IsSyncPoint, + }, + }) +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) continue } else { if !b.spanController.IsReplicating(task) { diff --git a/maintainer/barrier_test.go b/maintainer/barrier_test.go index a1ebf37323..38574f997a 100644 --- a/maintainer/barrier_test.go +++ b/maintainer/barrier_test.go @@ -165,6 +165,49 @@ func TestOneBlockEvent(t *testing.T) { require.Len(t, resp.DispatcherStatuses, 0) } +func TestBarrierIgnoresBlockStatusFromNonOwner(t *testing.T) { + testutil.SetUpTestServices(t) + tableTriggerEventDispatcherID := common.NewDispatcherID() + cfID := common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceName) + ddlSpan := replica.NewWorkingSpanReplication(cfID, tableTriggerEventDispatcherID, + common.DDLSpanSchemaID, + common.KeyspaceDDLSpan(common.DefaultKeyspaceID), &heartbeatpb.TableSpanStatus{ + ID: tableTriggerEventDispatcherID.ToPB(), + ComponentStatus: heartbeatpb.ComponentState_Working, + CheckpointTs: 1, + }, "node1", false) + spanController := span.NewController(cfID, ddlSpan, nil, nil, nil, common.DefaultKeyspaceID, common.DefaultMode) + operatorController := operator.NewOperatorController(cfID, spanController, 1000, common.DefaultMode) + spanController.AddNewTable(commonEvent.Table{SchemaID: 1, TableID: 1}, 10) + stm := spanController.GetTasksByTableID(1)[0] + spanController.BindSpanToNode("", "node1", stm) + spanController.MarkSpanReplicating(stm) + + barrier := NewBarrier(spanController, operatorController, false, nil, common.DefaultMode, nil) + msgs := barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{ + ChangefeedID: cfID.ToPB(), + BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{ + { + ID: stm.ID.ToPB(), + State: &heartbeatpb.State{ + IsBlocked: true, + BlockTs: 10, + BlockTables: &heartbeatpb.InfluencedTables{ + InfluenceType: heartbeatpb.InfluenceType_Normal, + TableIDs: []int64{1}, + }, + }, + }, + }, + }) + + require.Len(t, msgs, 1) + resp := msgs[0].Message[0].(*heartbeatpb.HeartBeatResponse) + require.Empty(t, resp.DispatcherStatuses) + require.Empty(t, barrier.blockedEvents.m) + require.Equal(t, uint64(10), stm.GetStatus().CheckpointTs) +} + func TestNormalBlock(t *testing.T) { testutil.SetUpTestServices() tableTriggerEventDispatcherID := common.NewDispatcherID() @@ -620,7 +663,7 @@ func TestSchemaBlock(t *testing.T) { require.Len(t, resp.DispatcherStatuses, 1) // selected node write done - _ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{ + _ = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{ ChangefeedID: cfID.ToPB(), BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{ { @@ -772,7 +815,7 @@ func TestSyncPointBlock(t *testing.T) { require.Equal(t, event.writerDispatcher, spanController.GetDDLDispatcherID()) // selected node write done - _ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{ + _ = barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{ ChangefeedID: cfID.ToPB(), BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{ { @@ -812,6 +855,11 @@ func TestSyncPointBlock(t *testing.T) { IsSyncPoint: true, }, }, + }, + }) + _ = barrier.HandleStatus("node2", &heartbeatpb.BlockStatusRequest{ + ChangefeedID: cfID.ToPB(), + BlockStatuses: []*heartbeatpb.TableSpanBlockStatus{ { ID: dispatcherIDs[2], State: &heartbeatpb.State{ @@ -847,6 +895,7 @@ func TestNonBlocked(t *testing.T) { stm := spanController.GetTasksByTableID(int64(id))[0] dispatcherID := stm.ID blockedDispatcherIDS = append(blockedDispatcherIDS, dispatcherID.ToPB()) + spanController.BindSpanToNode("", "node1", stm) spanController.MarkSpanReplicating(stm) } msgs := barrier.HandleStatus("node1", &heartbeatpb.BlockStatusRequest{ diff --git a/maintainer/operator/operator_controller.go b/maintainer/operator/operator_controller.go index 5684611aea..9fb67c33bf 100644 --- a/maintainer/operator/operator_controller.go +++ b/maintainer/operator/operator_controller.go @@ -53,6 +53,15 @@ type Controller struct { nodeManager *watcher.NodeManager splitter *split.Splitter +<<<<<<< HEAD +======= + // admissionMu serializes removing-mode quiesce and remove-operator replacement + // with normal operator side effects. + // A normal operator must hold the read side from its final allow check through + // Start or Schedule/SendCommand so it cannot cross the handoff boundary after + // QuiesceExcept has made the controller quiescing. + admissionMu sync.RWMutex +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) mu sync.RWMutex // protect the following fields operators map[common.DispatcherID]*operator.OperatorWithTime[common.DispatcherID, *heartbeatpb.TableSpanStatus] runningQueue operator.OperatorQueue[common.DispatcherID, *heartbeatpb.TableSpanStatus] @@ -161,8 +170,12 @@ func (oc *Controller) AddOperator(op operator.Operator[common.DispatcherID, *hea zap.String("operator", op.String())) return false } +<<<<<<< HEAD oc.pushOperator(op) return true +======= + return oc.pushOperatorWithAdmission(op, false) +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) } func (oc *Controller) UpdateOperatorStatus(id common.DispatcherID, from node.ID, status *heartbeatpb.TableSpanStatus) { @@ -315,18 +328,40 @@ func (oc *Controller) finalizeOperator( zap.String("operator", op.String())) } -func (oc *Controller) cancelOperator(opID common.DispatcherID) { +func (oc *Controller) cancelOperator( + expected operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], +) { + // Serialize rollback with remove-operator replacement. Otherwise a stale rollback + // could resolve the dispatcher ID after the replacement and cancel the new operator. + oc.admissionMu.RLock() + defer oc.admissionMu.RUnlock() + + opID := expected.ID() oc.mu.RLock() item, ok := oc.operators[opID] oc.mu.RUnlock() - if !ok { + if !ok || item.OP != expected { return } - item.OP.OnTaskRemoved() + expected.OnTaskRemoved() oc.finalizeOperator(item, opID) } func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) { +<<<<<<< HEAD +======= + oc.admissionMu.Lock() + defer oc.admissionMu.Unlock() + + if !oc.isOperatorAllowed(op.ID()) { + log.Info("skip remove operator while controller is quiescing", + zap.String("role", oc.role), + zap.Stringer("changefeedID", oc.changefeedID), + zap.String("dispatcherID", op.ID().String()), + zap.String("operator", op.String())) + return + } +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) oc.mu.RLock() old, ok := oc.operators[op.ID()] oc.mu.RUnlock() @@ -339,20 +374,40 @@ func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) { old.OP.OnTaskRemoved() oc.finalizeOperator(old, op.ID()) } +<<<<<<< HEAD oc.pushOperator(op) } // pushOperator add an operator to the controller queue. func (oc *Controller) pushOperator(op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]) { - log.Info("add operator to running queue", - zap.String("role", oc.role), - zap.Stringer("changefeedID", oc.changefeedID), - zap.String("operator", op.String())) +======= + oc.pushOperatorWithAdmission(op, true) +} + +func (oc *Controller) pushOperatorWithAdmission( + op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], + replaceExisting bool, +) bool { withTime := operator.NewOperatorWithTime(op, time.Now()) + opID := op.ID() oc.mu.Lock() - oc.operators[op.ID()] = withTime + if old, ok := oc.operators[opID]; ok && !replaceExisting { + oc.mu.Unlock() + log.Info("add operator failed, operator already exists", + zap.String("role", oc.role), + zap.Stringer("changefeedID", oc.changefeedID), + zap.String("operator", op.String()), + zap.String("oldOperator", old.OP.String())) + return false + } + oc.operators[opID] = withTime oc.mu.Unlock() +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) + log.Info("add operator to running queue", + zap.String("role", oc.role), + zap.Stringer("changefeedID", oc.changefeedID), + zap.String("operator", op.String())) op.Start() // Check affected nodes after Start to avoid operators being forced into terminal states @@ -411,6 +466,39 @@ func checkMergeOperator(affectedReplicaSets []*replica.SpanReplication) bool { return true } +<<<<<<< HEAD +======= +// addMergeOccupyOperators reserves every source replica or rolls back the partial reservation. +func (oc *Controller) addMergeOccupyOperators( + affectedReplicaSets []*replica.SpanReplication, +) ([]operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], bool) { + operators := make([]operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], 0, len(affectedReplicaSets)) + for _, replicaSet := range affectedReplicaSets { + occupyOperator := NewOccupyDispatcherOperator(oc.spanController, replicaSet) + if oc.AddOperator(occupyOperator) { + operators = append(operators, occupyOperator) + continue + } + log.Error("failed to add occupy dispatcher operator", + zap.Stringer("changefeedID", oc.changefeedID), + zap.Int64("group", replicaSet.GetGroupID()), + zap.String("span", common.FormatTableSpan(replicaSet.Span)), + zap.String("operator", occupyOperator.String())) + oc.cancelMergeOccupyOperators(operators) + return nil, false + } + return operators, true +} + +func (oc *Controller) cancelMergeOccupyOperators( + operators []operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], +) { + for _, op := range operators { + oc.cancelOperator(op) + } +} + +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) // AddMergeOperator creates a merge operator, which merge consecutive replica sets. // We need create a mergeOperator for the new replicaset, and create len(affectedReplicaSets) empty operator // to occupy these replica set not evolve other scheduling among merging. diff --git a/maintainer/operator/operator_controller_test.go b/maintainer/operator/operator_controller_test.go index 4909d5dc06..bc3360eeb7 100644 --- a/maintainer/operator/operator_controller_test.go +++ b/maintainer/operator/operator_controller_test.go @@ -25,6 +25,7 @@ import ( appcontext "github.com/pingcap/ticdc/pkg/common/context" "github.com/pingcap/ticdc/pkg/messaging" "github.com/pingcap/ticdc/pkg/node" + scheduleroperator "github.com/pingcap/ticdc/pkg/scheduler/operator" "github.com/pingcap/ticdc/server/watcher" "github.com/stretchr/testify/require" ) @@ -101,6 +102,150 @@ func (o *neverFinishOperator) OnTaskRemoved() {} func (o *neverFinishOperator) String() string { return "never-finish" } func (o *neverFinishOperator) BlockTsForward() bool { return false } +<<<<<<< HEAD +======= +type countingOperator struct { + id common.DispatcherID + targetNode node.ID + blockTsForward bool + startCount syncatomic.Int32 + scheduleCount syncatomic.Int32 + checkCount syncatomic.Int32 + nodeRemovedCount syncatomic.Int32 +} + +func (o *countingOperator) ID() common.DispatcherID { return o.id } +func (o *countingOperator) Type() string { return "add" } +func (o *countingOperator) Start() { o.startCount.Add(1) } +func (o *countingOperator) Schedule() *messaging.TargetMessage { + o.scheduleCount.Add(1) + return messaging.NewSingleTargetMessage(o.targetNode, messaging.MaintainerManagerTopic, &heartbeatpb.RemoveMaintainerRequest{}) +} +func (o *countingOperator) IsFinished() bool { return false } +func (o *countingOperator) PostFinish() {} +func (o *countingOperator) Check(node.ID, *heartbeatpb.TableSpanStatus) { + o.checkCount.Add(1) +} + +func (o *countingOperator) OnNodeRemove(node.ID) { + o.nodeRemovedCount.Add(1) +} +func (o *countingOperator) AffectedNodes() []node.ID { return []node.ID{o.targetNode} } +func (o *countingOperator) OnTaskRemoved() {} +func (o *countingOperator) String() string { return "counting-operator" } +func (o *countingOperator) BlockTsForward() bool { return o.blockTsForward } + +type blockingTaskRemovedOperator struct { + *countingOperator + taskRemovedEntered chan struct{} + releaseTaskRemoved chan struct{} +} + +func (o *blockingTaskRemovedOperator) OnTaskRemoved() { + close(o.taskRemovedEntered) + <-o.releaseTaskRemoved +} + +type synchronizedAdmissionOperator struct { + *countingOperator + idCalls syncatomic.Int32 + ready *sync.WaitGroup + release <-chan struct{} +} + +func (o *synchronizedAdmissionOperator) ID() common.DispatcherID { + // The second ID lookup happens after AddOperator's initial duplicate check + // and before the operator is registered. Hold both callers in that window + // to deterministically exercise concurrent admission for the same ID. + if o.idCalls.Add(1) == 2 { + o.ready.Done() + <-o.release + } + return o.countingOperator.ID() +} + +type blockingScheduleOperator struct { + id common.DispatcherID + targetNode node.ID + + scheduleEntered chan struct{} + releaseSchedule chan struct{} + scheduleOnce sync.Once + scheduleCount syncatomic.Int32 +} + +func newBlockingScheduleOperator(id common.DispatcherID, targetNode node.ID) *blockingScheduleOperator { + return &blockingScheduleOperator{ + id: id, + targetNode: targetNode, + scheduleEntered: make(chan struct{}), + releaseSchedule: make(chan struct{}), + } +} + +func (o *blockingScheduleOperator) ID() common.DispatcherID { return o.id } +func (o *blockingScheduleOperator) Type() string { return "add" } +func (o *blockingScheduleOperator) Start() {} +func (o *blockingScheduleOperator) Schedule() *messaging.TargetMessage { + o.scheduleCount.Add(1) + o.scheduleOnce.Do(func() { close(o.scheduleEntered) }) + <-o.releaseSchedule + return messaging.NewSingleTargetMessage(o.targetNode, messaging.MaintainerManagerTopic, &heartbeatpb.RemoveMaintainerRequest{}) +} + +func (o *blockingScheduleOperator) IsFinished() bool { return false } +func (o *blockingScheduleOperator) PostFinish() {} +func (o *blockingScheduleOperator) Check(node.ID, *heartbeatpb.TableSpanStatus) { +} + +func (o *blockingScheduleOperator) OnNodeRemove(node.ID) { +} +func (o *blockingScheduleOperator) AffectedNodes() []node.ID { return []node.ID{o.targetNode} } +func (o *blockingScheduleOperator) OnTaskRemoved() {} +func (o *blockingScheduleOperator) String() string { return "blocking-schedule" } +func (o *blockingScheduleOperator) BlockTsForward() bool { return false } + +type blockingStartOperator struct { + id common.DispatcherID + targetNode node.ID + + startEntered chan struct{} + releaseStart chan struct{} + startOnce sync.Once + startCount syncatomic.Int32 +} + +func newBlockingStartOperator(id common.DispatcherID, targetNode node.ID) *blockingStartOperator { + return &blockingStartOperator{ + id: id, + targetNode: targetNode, + startEntered: make(chan struct{}), + releaseStart: make(chan struct{}), + } +} + +func (o *blockingStartOperator) ID() common.DispatcherID { return o.id } +func (o *blockingStartOperator) Type() string { return "add" } +func (o *blockingStartOperator) Start() { + o.startCount.Add(1) + o.startOnce.Do(func() { close(o.startEntered) }) + <-o.releaseStart +} + +func (o *blockingStartOperator) Schedule() *messaging.TargetMessage { return nil } +func (o *blockingStartOperator) IsFinished() bool { return false } +func (o *blockingStartOperator) PostFinish() {} +func (o *blockingStartOperator) Check(node.ID, *heartbeatpb.TableSpanStatus) { +} + +func (o *blockingStartOperator) OnNodeRemove(node.ID) { +} +func (o *blockingStartOperator) AffectedNodes() []node.ID { return []node.ID{o.targetNode} } +func (o *blockingStartOperator) OnTaskRemoved() {} +func (o *blockingStartOperator) String() string { return "blocking-start" } +func (o *blockingStartOperator) BlockTsForward() bool { return false } + +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070)) func setAliveNodes(nodeManager *watcher.NodeManager, alive map[node.ID]*node.Info) { type nodeMap = map[node.ID]*node.Info v := reflect.ValueOf(nodeManager).Elem().FieldByName("nodes") @@ -134,6 +279,87 @@ func TestController_PostFinishCalledOnceOnReplace(t *testing.T) { require.Equal(t, int32(1), op.postFinishCount.Load()) } +func TestController_AddOperatorAtomicallyRejectsConcurrentDuplicate(t *testing.T) { + messageCenter, _, _ := messaging.NewMessageCenterForTest(t) + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 1, common.DefaultMode) + var ready sync.WaitGroup + ready.Add(2) + release := make(chan struct{}) + operators := []*synchronizedAdmissionOperator{ + { + countingOperator: &countingOperator{id: replicaSet.ID, targetNode: nodeA}, + ready: &ready, + release: release, + }, + { + countingOperator: &countingOperator{id: replicaSet.ID, targetNode: nodeA}, + ready: &ready, + release: release, + }, + } + + results := make(chan bool, len(operators)) + for _, op := range operators { + go func() { + results <- oc.AddOperator(op) + }() + } + ready.Wait() + close(release) + + successes := 0 + for range operators { + if <-results { + successes++ + } + } + require.Equal(t, 1, successes) + require.Equal(t, int32(1), operators[0].startCount.Load()+operators[1].startCount.Load()) + require.Equal(t, 1, oc.OperatorSize()) + require.Len(t, oc.runningQueue, 1) +} + +func TestController_AddOperatorAllowsMoveWithEmptyOrigin(t *testing.T) { + messageCenter, _, _ := messaging.NewMessageCenterForTest(t) + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, _, nodeB := setupTestEnvironment(t) + absentReplica := replica.NewSpanReplication( + changefeedID, + replicaSet.ID, + replicaSet.GetSchemaID(), + replicaSet.Span, + replicaSet.GetStatus().CheckpointTs, + common.DefaultMode, + false, + ) + spanController.AddAbsentReplicaSet(absentReplica) + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeB: {ID: nodeB}}) + + oc := NewOperatorController(changefeedID, spanController, 1, common.DefaultMode) + op := NewMoveDispatcherOperator(spanController, absentReplica, "", nodeB, 7) + require.True(t, oc.AddOperator(op)) + require.Same(t, op, oc.GetOperator(absentReplica.ID)) + require.Equal(t, nodeB, absentReplica.GetNodeID()) + require.Equal(t, 0, spanController.GetAbsentSize()) + require.Equal(t, 1, spanController.GetSchedulingSize()) + require.Len(t, oc.runningQueue, 1) + + msg := op.Schedule() + require.NotNil(t, msg) + require.Equal(t, nodeB, msg.To) + require.Equal(t, heartbeatpb.ScheduleAction_Create, + msg.Message[0].(*heartbeatpb.ScheduleDispatcherRequest).ScheduleAction) +} + func TestController_OnNodeRemoved_WithOccupyOperatorMarksSpanAbsent(t *testing.T) { messageCenter, _, _ := messaging.NewMessageCenterForTest(t) appcontext.SetService(appcontext.MessageCenter, messageCenter) @@ -202,3 +428,331 @@ func TestController_RemoveReplicaSet_ReplacesRemoveOperatorOnTaskRemoved(t *test require.Equal(t, int32(0), postFinishCount.Load()) require.NotNil(t, oc.GetOperator(replicaSet.ID)) } +<<<<<<< HEAD +======= + +func TestController_RemoveReplicaSetBlocksNormalAdmissionUntilReplacement(t *testing.T) { + messageCenter, _, _ := messaging.NewMessageCenterForTest(t) + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 1, common.DefaultMode) + old := &blockingTaskRemovedOperator{ + countingOperator: &countingOperator{id: replicaSet.ID, targetNode: nodeA}, + taskRemovedEntered: make(chan struct{}), + releaseTaskRemoved: make(chan struct{}), + } + require.True(t, oc.AddOperator(old)) + + replacement := newRemoveDispatcherOperator( + spanController, + replicaSet, + heartbeatpb.OperatorType_O_Remove, + 7, + ) + replacementDone := make(chan struct{}) + go func() { + oc.removeReplicaSet(replacement) + close(replacementDone) + }() + <-old.taskRemovedEntered + + concurrent := &countingOperator{id: replicaSet.ID, targetNode: nodeA} + addStarted := make(chan struct{}) + addResult := make(chan bool, 1) + go func() { + close(addStarted) + addResult <- oc.AddOperator(concurrent) + }() + <-addStarted + require.Never(t, func() bool { return len(addResult) != 0 }, 100*time.Millisecond, 10*time.Millisecond) + + close(old.releaseTaskRemoved) + require.Eventually(t, func() bool { + select { + case <-replacementDone: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + require.Eventually(t, func() bool { return len(addResult) == 1 }, time.Second, 10*time.Millisecond) + require.False(t, <-addResult) + require.Equal(t, int32(0), concurrent.startCount.Load()) + require.Same(t, replacement, oc.GetOperator(replicaSet.ID)) +} + +func TestControllerStaleMergeRollbackDoesNotCancelReplacementRemove(t *testing.T) { + messageCenter, _, _ := messaging.NewMessageCenterForTest(t) + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 1, common.DefaultMode) + occupy := NewOccupyDispatcherOperator(spanController, replicaSet) + require.True(t, oc.AddOperator(occupy)) + + replacement := newRemoveDispatcherOperator( + spanController, + replicaSet, + heartbeatpb.OperatorType_O_Remove, + 7, + ) + oc.removeReplicaSet(replacement) + require.Same(t, replacement, oc.GetOperator(replicaSet.ID)) + + // Simulate a delayed merge rollback that still holds the replaced occupy operator. + oc.cancelMergeOccupyOperators( + []scheduleroperator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]{occupy}, + ) + + require.Same(t, replacement, oc.GetOperator(replicaSet.ID)) + require.False(t, replacement.IsFinished()) + msg := replacement.Schedule() + require.NotNil(t, msg) + require.Equal(t, nodeA, msg.To) + require.Equal(t, heartbeatpb.ScheduleAction_Remove, + msg.Message[0].(*heartbeatpb.ScheduleDispatcherRequest).ScheduleAction) +} + +func TestController_QuiesceExceptFreezesNonAllowedOperators(t *testing.T) { + // Scenario: removing mode allows only the DDL close operator to keep running. + // Steps: quiesce the controller with one allowed dispatcher, then verify the + // allowed operator still accepts status and schedules, while the frozen operator + // does not run but still blocks checkpoint advancement. + messageCenter := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + + allowedID := common.NewDispatcherID() + allowedReplica := setupReplicaSetWithID(t, changefeedID, allowedID, nodeA) + allowedReplica.UpdateStatus(&heartbeatpb.TableSpanStatus{ + ID: allowedID.ToPB(), + ComponentStatus: heartbeatpb.ComponentState_Working, + CheckpointTs: 20, + Mode: common.DefaultMode, + }) + spanController.AddReplicatingSpan(allowedReplica) + + blockedID := common.NewDispatcherID() + blockedReplica := setupReplicaSetWithID(t, changefeedID, blockedID, nodeA) + spanController.AddReplicatingSpan(blockedReplica) + + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 10, common.DefaultMode) + allowedOp := &countingOperator{id: allowedID, targetNode: nodeA, blockTsForward: true} + blockedOp := &countingOperator{id: blockedID, targetNode: nodeA, blockTsForward: true} + require.True(t, oc.AddOperator(allowedOp)) + require.True(t, oc.AddOperator(blockedOp)) + + oc.QuiesceExcept(allowedID) + + oc.UpdateOperatorStatus(allowedID, nodeA, &heartbeatpb.TableSpanStatus{ID: allowedID.ToPB()}) + oc.UpdateOperatorStatus(blockedID, nodeA, &heartbeatpb.TableSpanStatus{ID: blockedID.ToPB()}) + require.Equal(t, int32(1), allowedOp.checkCount.Load()) + require.Equal(t, int32(0), blockedOp.checkCount.Load()) + + oc.OnNodeRemoved(nodeA) + require.Equal(t, int32(0), allowedOp.nodeRemovedCount.Load()) + require.Equal(t, int32(0), blockedOp.nodeRemovedCount.Load()) + require.Equal(t, 0, spanController.GetAbsentSize()) + + newBlockedID := common.NewDispatcherID() + newBlockedReplica := setupReplicaSetWithID(t, changefeedID, newBlockedID, nodeA) + spanController.AddReplicatingSpan(newBlockedReplica) + require.False(t, oc.AddOperator(&countingOperator{id: newBlockedID, targetNode: nodeA})) + + require.Equal(t, uint64(10), oc.GetMinCheckpointTs(^uint64(0))) + + next := oc.Execute() + require.False(t, next.IsZero()) + require.Equal(t, int32(1), allowedOp.scheduleCount.Load()) + require.Equal(t, int32(0), blockedOp.scheduleCount.Load()) + require.Len(t, messageCenter.GetMessageChannel(), 1) + require.Equal(t, 2, oc.OperatorSize()) +} + +func TestController_QuiesceExceptDropsBlockedOnlyQueueFromExecution(t *testing.T) { + // Scenario: after removing starts, the running queue can contain only frozen ordinary operators. + // Steps: poll a quiesced controller with one non-allowed operator and verify it leaves the heap, + // remains in the operator map for checkpoint safety, and the next poll terminates the Execute loop. + messageCenter := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 10, common.DefaultMode) + blockedOp := &countingOperator{id: replicaSet.ID, targetNode: nodeA, blockTsForward: true} + require.True(t, oc.AddOperator(blockedOp)) + + oc.QuiesceExcept(common.NewDispatcherID()) + + op, next := oc.pollQueueingOperator() + require.Nil(t, op) + require.True(t, next) + require.Equal(t, 0, oc.runningQueue.Len()) + require.Equal(t, 1, oc.OperatorSize()) + require.Equal(t, uint64(1000), oc.GetMinCheckpointTs(^uint64(0))) + + op, next = oc.pollQueueingOperator() + require.Nil(t, op) + require.False(t, next) + require.Equal(t, int32(0), blockedOp.scheduleCount.Load()) +} + +func TestController_QuiesceExceptWaitsForInFlightSchedule(t *testing.T) { + // Scenario: Execute has already passed the queue poll and is inside a normal operator's Schedule. + // Steps: block Schedule with a channel, start QuiesceExcept, verify quiesce cannot return until + // Schedule/SendCommand leaves the admission boundary, then verify later Execute calls do not reschedule it. + messageCenter := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 10, common.DefaultMode) + op := newBlockingScheduleOperator(replicaSet.ID, nodeA) + require.True(t, oc.AddOperator(op)) + + executeDone := make(chan struct{}) + go func() { + defer close(executeDone) + oc.Execute() + }() + <-op.scheduleEntered + + quiesceDone := make(chan struct{}) + go func() { + defer close(quiesceDone) + oc.QuiesceExcept(common.NewDispatcherID()) + }() + + require.Never(t, func() bool { + select { + case <-quiesceDone: + return true + default: + return false + } + }, 100*time.Millisecond, 10*time.Millisecond) + + close(op.releaseSchedule) + require.Eventually(t, func() bool { + select { + case <-executeDone: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + require.Eventually(t, func() bool { + select { + case <-quiesceDone: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + require.Equal(t, int32(1), op.scheduleCount.Load()) + require.Len(t, messageCenter.GetMessageChannel(), 1) + + oc.Execute() + require.Equal(t, int32(1), op.scheduleCount.Load()) + require.Len(t, messageCenter.GetMessageChannel(), 1) +} + +func TestController_QuiesceExceptWaitsForInFlightPush(t *testing.T) { + // Scenario: a normal operator has passed admission and is inside Start while removing mode begins. + // Steps: block Start with a channel, start QuiesceExcept, verify quiesce cannot return until Start + // finishes, then verify a later ordinary operator is rejected without being started. + messageCenter := messaging.NewMockMessageCenter() + appcontext.SetService(appcontext.MessageCenter, messageCenter) + + spanController, changefeedID, replicaSet, nodeA, _ := setupTestEnvironment(t) + spanController.AddReplicatingSpan(replicaSet) + + nodeManager := appcontext.GetService[*watcher.NodeManager](watcher.NodeManagerName) + setAliveNodes(nodeManager, map[node.ID]*node.Info{nodeA: {ID: nodeA}}) + + oc := NewOperatorController(changefeedID, spanController, 10, common.DefaultMode) + op := newBlockingStartOperator(replicaSet.ID, nodeA) + + addResult := make(chan bool, 1) + go func() { + addResult <- oc.AddOperator(op) + }() + <-op.startEntered + + quiesceDone := make(chan struct{}) + go func() { + defer close(quiesceDone) + oc.QuiesceExcept(common.NewDispatcherID()) + }() + + require.Never(t, func() bool { + select { + case <-quiesceDone: + return true + default: + return false + } + }, 100*time.Millisecond, 10*time.Millisecond) + + close(op.releaseStart) + require.Eventually(t, func() bool { return len(addResult) == 1 }, time.Second, 10*time.Millisecond) + require.True(t, <-addResult) + require.Eventually(t, func() bool { + select { + case <-quiesceDone: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + require.Equal(t, int32(1), op.startCount.Load()) + + blockedID := common.NewDispatcherID() + blockedReplica := setupReplicaSetWithID(t, changefeedID, blockedID, nodeA) + spanController.AddReplicatingSpan(blockedReplica) + blockedOp := newBlockingStartOperator(blockedID, nodeA) + require.False(t, oc.AddOperator(blockedOp)) + require.Equal(t, int32(0), blockedOp.startCount.Load()) +} + +func setupReplicaSetWithID( + t *testing.T, + changefeedID common.ChangeFeedID, + dispatcherID common.DispatcherID, + nodeID node.ID, +) *replica.SpanReplication { + t.Helper() + + tableID := int64(dispatcherID.Low + 100) + span := testutil.GetTableSpanByID(tableID) + return replica.NewWorkingSpanReplication(changefeedID, dispatcherID, 1, span, &heartbeatpb.TableSpanStatus{ + ID: dispatcherID.ToPB(), + ComponentStatus: heartbeatpb.ComponentState_Working, + CheckpointTs: 10, + Mode: common.DefaultMode, + }, nodeID, false) +} +>>>>>>> 83a45498b (maintainer: make dispatcher operator admission atomic (#6070))