diff --git a/maintainer/barrier.go b/maintainer/barrier.go index 9948fca20e..a2b9b363cc 100644 --- a/maintainer/barrier.go +++ b/maintainer/barrier.go @@ -88,16 +88,27 @@ 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() { - 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 - } if !b.spanController.IsReplicating(task) { log.Info("Get block status from unreplicating dispatcher, ignore it", zap.String("changefeed", request.ChangefeedID.GetName()), diff --git a/maintainer/barrier_test.go b/maintainer/barrier_test.go index ee4ac4e783..2cbd0516f1 100644 --- a/maintainer/barrier_test.go +++ b/maintainer/barrier_test.go @@ -168,6 +168,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(t) tableTriggerEventDispatcherID := common.NewDispatcherID() @@ -1101,7 +1144,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{ { @@ -1253,7 +1296,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{ { @@ -1293,6 +1336,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{ @@ -1328,6 +1376,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 46287450a5..e4d87769a9 100644 --- a/maintainer/operator/operator_controller.go +++ b/maintainer/operator/operator_controller.go @@ -53,7 +53,8 @@ type Controller struct { nodeManager *watcher.NodeManager maintainerEpoch atomic.Uint64 - // admissionMu serializes removing-mode quiesce with normal operator side effects. + // 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. @@ -259,7 +260,7 @@ func (oc *Controller) AddOperator(op operator.Operator[common.DispatcherID, *hea zap.String("operator", op.String())) return false } - return oc.pushOperatorWithAdmission(op) + return oc.pushOperatorWithAdmission(op, false) } func (oc *Controller) UpdateOperatorStatus(id common.DispatcherID, from node.ID, status *heartbeatpb.TableSpanStatus) { @@ -465,8 +466,8 @@ func (oc *Controller) cancelOperator(opID common.DispatcherID) { } func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) { - oc.admissionMu.RLock() - defer oc.admissionMu.RUnlock() + oc.admissionMu.Lock() + defer oc.admissionMu.Unlock() if !oc.isOperatorAllowed(op.ID()) { log.Info("skip remove operator while controller is quiescing", @@ -488,35 +489,32 @@ func (oc *Controller) removeReplicaSet(op *removeDispatcherOperator) { old.OP.OnTaskRemoved() oc.finalizeOperator(old, op.ID()) } - oc.pushOperatorWithAdmission(op) + oc.pushOperatorWithAdmission(op, true) } -// pushOperator add an operator to the controller queue. -func (oc *Controller) pushOperator(op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]) bool { - oc.admissionMu.RLock() - defer oc.admissionMu.RUnlock() +func (oc *Controller) pushOperatorWithAdmission( + op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus], + replaceExisting bool, +) bool { + withTime := operator.NewOperatorWithTime(op, time.Now()) + opID := op.ID() - if !oc.isOperatorAllowed(op.ID()) { - log.Info("skip operator while controller is quiescing", + oc.mu.Lock() + 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("dispatcherID", op.ID().String()), - zap.String("operator", op.String())) + zap.String("operator", op.String()), + zap.String("oldOperator", old.OP.String())) return false } - return oc.pushOperatorWithAdmission(op) -} - -func (oc *Controller) pushOperatorWithAdmission(op operator.Operator[common.DispatcherID, *heartbeatpb.TableSpanStatus]) bool { + oc.operators[opID] = withTime + oc.mu.Unlock() log.Info("add operator to running queue", zap.String("role", oc.role), zap.Stringer("changefeedID", oc.changefeedID), zap.String("operator", op.String())) - withTime := operator.NewOperatorWithTime(op, time.Now()) - - oc.mu.Lock() - oc.operators[op.ID()] = withTime - oc.mu.Unlock() op.Start() // Check affected nodes after Start to avoid operators being forced into terminal states diff --git a/maintainer/operator/operator_controller_test.go b/maintainer/operator/operator_controller_test.go index c299af907a..9aa2627c5d 100644 --- a/maintainer/operator/operator_controller_test.go +++ b/maintainer/operator/operator_controller_test.go @@ -146,6 +146,7 @@ type countingOperator struct { id common.DispatcherID targetNode node.ID blockTsForward bool + startCount syncatomic.Int32 scheduleCount syncatomic.Int32 checkCount syncatomic.Int32 nodeRemovedCount syncatomic.Int32 @@ -153,7 +154,7 @@ type countingOperator struct { func (o *countingOperator) ID() common.DispatcherID { return o.id } func (o *countingOperator) Type() string { return "add" } -func (o *countingOperator) Start() {} +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{}) @@ -172,6 +173,35 @@ 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 @@ -286,6 +316,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) @@ -405,6 +516,61 @@ func TestController_RemoveReplicaSet_ReplacesRemoveOperatorOnTaskRemoved(t *test require.NotNil(t, oc.GetOperator(replicaSet.ID)) } +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 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