maintainer: harden drain target add path

Author: hongyunyan

maintainer/maintainer_manager_maintainers.go

@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/liveness"
 	"github.com/pingcap/ticdc/pkg/messaging"
 	"github.com/pingcap/ticdc/pkg/node"
@@ -71,8 +72,7 @@ func (m *Manager) onAddMaintainerRequest(req *heartbeatpb.AddMaintainerRequest)
 		return nil
 	}
 
-	target, epoch := m.getDispatcherDrainTarget()
-	return m.maintainers.handleAddMaintainer(req, target, epoch)
+	return m.maintainers.handleAddMaintainer(req, m.getDispatcherDrainTarget)
 }
 
 // onRemoveMaintainerRequest delegates changefeed removal to the maintainer part.
@@ -86,10 +86,20 @@ func (m *Manager) onDispatchMaintainerRequest(
 	msg *messaging.TargetMessage,
 ) *heartbeatpb.MaintainerStatus {
 	if m.coordinatorID != msg.From {
-		log.Warn("ignore invalid coordinator id",
-			zap.Any("request", msg),
-			zap.Any("coordinatorID", m.coordinatorID),
-			zap.Stringer("from", msg.From))
+		fields := []zap.Field{
+			zap.String("type", msg.Type.String()),
+			zap.Stringer("coordinatorID", m.coordinatorID),
+			zap.Stringer("from", msg.From),
+		}
+		switch msg.Type {
+		case messaging.TypeAddMaintainerRequest:
+			changefeedID := common.NewChangefeedIDFromPB(msg.Message[0].(*heartbeatpb.AddMaintainerRequest).Id)
+			fields = append(fields, zap.Stringer("changefeedID", changefeedID))
+		case messaging.TypeRemoveMaintainerRequest:
+			changefeedID := common.NewChangefeedIDFromPB(msg.Message[0].(*heartbeatpb.RemoveMaintainerRequest).Id)
+			fields = append(fields, zap.Stringer("changefeedID", changefeedID))
+		}
+		log.Warn("ignore invalid coordinator id", fields...)
 		return nil
 	}
 	switch msg.Type {
@@ -155,33 +165,42 @@ func (p *managerMaintainerSet) buildBootstrapResponse() *heartbeatpb.Coordinator
 // with the latest node-scoped dispatcher drain target.
 func (p *managerMaintainerSet) handleAddMaintainer(
 	req *heartbeatpb.AddMaintainerRequest,
-	target node.ID,
-	epoch uint64,
+	getDrainTarget func() (node.ID, uint64),
 ) *heartbeatpb.MaintainerStatus {
 	changefeedID := common.NewChangefeedIDFromPB(req.Id)
-	_, ok := p.registry.Load(changefeedID)
-	if ok {
+	if _, ok := p.registry.Load(changefeedID); ok {
 		return nil
 	}
 
 	info := &config.ChangeFeedInfo{}
-	err := json.Unmarshal(req.Config, info)
-	if err != nil {
-		log.Panic("decode changefeed fail", zap.Error(err))
+	if err := json.Unmarshal(req.Config, info); err != nil {
+		log.Error("ignore add maintainer request with invalid config",
+			zap.Stringer("changefeedID", changefeedID),
+			zap.Int("configBytes", len(req.Config)),
+			zap.Error(err))
+		return nil
 	}
 	if req.CheckpointTs == 0 {
-		log.Panic("add maintainer with invalid checkpointTs",
+		log.Error("ignore add maintainer request with invalid checkpointTs",
 			zap.Stringer("changefeedID", changefeedID),
-			zap.Uint64("checkpointTs", req.CheckpointTs),
-			zap.Any("info", info))
+			zap.Uint64("checkpointTs", req.CheckpointTs))
+		return nil
 	}
-
 	maintainer := NewMaintainer(changefeedID, p.conf, info, p.nodeInfo, p.taskScheduler, req.CheckpointTs, req.IsNewChangefeed, req.KeyspaceId)
-	// Seed the maintainer with the manager-level drain snapshot before its event
-	// loop starts so late additions still honor an already-active drain target.
-	maintainer.SetDispatcherDrainTarget(target, epoch)
-	p.registry.Store(changefeedID, maintainer)
-	maintainer.pushEvent(&Event{changefeedID: changefeedID, eventType: EventInit})
+	registered, loaded := p.registry.LoadOrStore(changefeedID, maintainer)
+	if loaded {
+		// Duplicate add requests can race on the same changefeed. Drop the loser and
+		// stop the redundant maintainer immediately so background goroutines do not leak.
+		maintainer.Close()
+		return nil
+	}
+
+	registeredMaintainer := registered.(*Maintainer)
+	// Register the maintainer before seeding the drain snapshot so concurrent
+	// manager-level drain fanout can always observe it in the registry.
+	target, epoch := getDrainTarget()
+	registeredMaintainer.SetDispatcherDrainTarget(target, epoch)
+	registeredMaintainer.pushEvent(&Event{changefeedID: changefeedID, eventType: EventInit})
 	return nil
 }
 
@@ -276,7 +295,7 @@ func (p *managerMaintainerSet) dispatchMaintainerMessage(
 	}
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return errors.Trace(ctx.Err())
 	default:
 		maintainer := c.(*Maintainer)
 		maintainer.pushEvent(&Event{

maintainer/maintainer_manager_node.go

@@ -163,16 +163,15 @@ func (m *Manager) onSetDispatcherDrainTargetRequest(msg *messaging.TargetMessage
 
 	req := msg.Message[0].(*heartbeatpb.SetDispatcherDrainTargetRequest)
 	target := node.ID(req.TargetNodeId)
-	if !m.node.tryUpdateDispatcherDrainTarget(target, req.TargetEpoch) {
-		return
-	}
-
-	log.Info("dispatcher drain target updated",
-		zap.Stringer("targetNodeID", target),
-		zap.Uint64("targetEpoch", req.TargetEpoch))
-	m.maintainers.applyDispatcherDrainTarget(target, req.TargetEpoch)
-	// A manager-level heartbeat is the authoritative acknowledgement that this
-	// node has applied the latest drain target, even when no maintainers exist.
+	if m.node.tryUpdateDispatcherDrainTarget(target, req.TargetEpoch) {
+		log.Info("dispatcher drain target updated",
+			zap.Stringer("targetNodeID", target),
+			zap.Uint64("targetEpoch", req.TargetEpoch))
+		m.maintainers.applyDispatcherDrainTarget(target, req.TargetEpoch)
+	}
+	// A manager-level heartbeat is the authoritative acknowledgement of the
+	// latest local drain snapshot, even when this request is a retry or stale
+	// duplicate and no maintainer update is needed.
 	m.sendNodeHeartbeat(true)
 }
 

maintainer/node_liveness_test.go

@@ -13,9 +13,11 @@
 package maintainer
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/pingcap/ticdc/heartbeatpb"
+	"github.com/pingcap/ticdc/pkg/common"
 	appcontext "github.com/pingcap/ticdc/pkg/common/context"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/liveness"
@@ -187,4 +189,55 @@ func TestSetDispatcherDrainTargetSendsNodeHeartbeatAck(t *testing.T) {
 	hb = apply("", 1)
 	require.Equal(t, "", hb.DispatcherDrainTargetNodeId)
 	require.Equal(t, uint64(1), hb.DispatcherDrainTargetEpoch)
+
+	// Same-epoch reactivation is rejected locally, but retries should still get
+	// an immediate heartbeat that reflects the latest applied snapshot.
+	hb = apply("n2", 1)
+	require.Equal(t, "", hb.DispatcherDrainTargetNodeId)
+	require.Equal(t, uint64(1), hb.DispatcherDrainTargetEpoch)
+}
+
+func TestAddMaintainerIgnoreInvalidConfig(t *testing.T) {
+	mc := messaging.NewMockMessageCenter()
+	appcontext.SetService(appcontext.MessageCenter, mc)
+
+	var nodeLiveness liveness.Liveness
+	m := NewMaintainerManager(&node.Info{ID: node.ID("n1")}, &config.SchedulerConfig{}, &nodeLiveness)
+
+	changefeedID := common.NewChangeFeedIDWithName("cf-invalid-config", common.DefaultKeyspaceName)
+	status := m.onAddMaintainerRequest(&heartbeatpb.AddMaintainerRequest{
+		Id:           changefeedID.ToPB(),
+		Config:       []byte("not-json"),
+		CheckpointTs: 10,
+	})
+	require.Nil(t, status)
+
+	_, ok := m.GetMaintainerForChangefeed(changefeedID)
+	require.False(t, ok)
+}
+
+func TestAddMaintainerIgnoreInvalidCheckpointTs(t *testing.T) {
+	mc := messaging.NewMockMessageCenter()
+	appcontext.SetService(appcontext.MessageCenter, mc)
+
+	var nodeLiveness liveness.Liveness
+	m := NewMaintainerManager(&node.Info{ID: node.ID("n1")}, &config.SchedulerConfig{}, &nodeLiveness)
+
+	changefeedID := common.NewChangeFeedIDWithName("cf-invalid-checkpoint", common.DefaultKeyspaceName)
+	info := &config.ChangeFeedInfo{
+		ChangefeedID: changefeedID,
+		Config:       config.GetDefaultReplicaConfig(),
+	}
+	data, err := json.Marshal(info)
+	require.NoError(t, err)
+
+	status := m.onAddMaintainerRequest(&heartbeatpb.AddMaintainerRequest{
+		Id:           changefeedID.ToPB(),
+		Config:       data,
+		CheckpointTs: 0,
+	})
+	require.Nil(t, status)
+
+	_, ok := m.GetMaintainerForChangefeed(changefeedID)
+	require.False(t, ok)
 }

maintainer/scheduler/drain_common.go

@@ -58,7 +58,9 @@ func (s *DrainState) SetSelfNodeID(id node.ID) {
 }
 
 // SetDispatcherDrainTarget applies the newest drain target visible to the
-// changefeed. Older epochs are ignored so scheduler state does not regress.
+// changefeed. Older epochs are ignored so scheduler state does not regress, and
+// same-epoch updates follow the manager-level monotonic rule: only clearing a
+// non-empty target is allowed.
 func (s *DrainState) SetDispatcherDrainTarget(target node.ID, epoch uint64) {
 	if s == nil {
 		return
@@ -68,6 +70,15 @@ func (s *DrainState) SetDispatcherDrainTarget(target node.ID, epoch uint64) {
 	if epoch < s.targetEpoch {
 		return
 	}
+	if epoch == s.targetEpoch {
+		if target == s.targetNodeID {
+			return
+		}
+		if target.IsEmpty() && !s.targetNodeID.IsEmpty() {
+			s.targetNodeID = target
+		}
+		return
+	}
 	s.targetNodeID = target
 	s.targetEpoch = epoch
 }

maintainer/scheduler/drain_common_test.go

@@ -0,0 +1,48 @@
+// Copyright 2026 PingCAP, Inc.
+//
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package scheduler
+
+import (
+	"testing"
+
+	"github.com/pingcap/ticdc/pkg/node"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDrainStateRejectSameEpochReactivation(t *testing.T) {
+	state := NewDrainState()
+
+	state.SetDispatcherDrainTarget(node.ID("n1"), 1)
+	target, epoch := state.DispatcherDrainTarget()
+	require.Equal(t, node.ID("n1"), target)
+	require.Equal(t, uint64(1), epoch)
+
+	state.SetDispatcherDrainTarget("", 1)
+	target, epoch = state.DispatcherDrainTarget()
+	require.Equal(t, node.ID(""), target)
+	require.Equal(t, uint64(1), epoch)
+
+	// A stale add path snapshot must not reactivate the target after the same
+	// epoch has already been cleared.
+	state.SetDispatcherDrainTarget(node.ID("n1"), 1)
+	target, epoch = state.DispatcherDrainTarget()
+	require.Equal(t, node.ID(""), target)
+	require.Equal(t, uint64(1), epoch)
+
+	// A newer epoch still wins.
+	state.SetDispatcherDrainTarget(node.ID("n2"), 2)
+	target, epoch = state.DispatcherDrainTarget()
+	require.Equal(t, node.ID("n2"), target)
+	require.Equal(t, uint64(2), epoch)
+}