maintainer: make dispatcher operator admission atomic (#6070)

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) {

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{

maintainer/operator/operator_controller.go

@@ -53,6 +53,15 @@
 	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 @@
 			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 @@
 		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 @@
 		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 @@
 	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.