DNM: refine dispatcher scan priority with syncpoint refine #4123

gemini-code-assist · 2026-02-03T12:38:34Z

The logic in the default case is identical to the case heartbeatpb.Action_Pass, heartbeatpb.Action_Skip. This code duplication can be removed by combining them. You can use a default case to handle Pass, Skip, and any potential future actions with the same logic, which simplifies the code and improves maintainability.

Suggested change

case heartbeatpb.Action_Pass, heartbeatpb.Action_Skip:

failpoint.Inject("BlockOrWaitBeforePass", nil)

d.PassBlockEventToSink(pendingEvent)

failpoint.Inject("BlockAfterPass", nil)

default:

failpoint.Inject("BlockOrWaitBeforePass", nil)

d.PassBlockEventToSink(pendingEvent)

failpoint.Inject("BlockAfterPass", nil)

case heartbeatpb.Action_Pass, heartbeatpb.Action_Skip:

fallthrough

default:

failpoint.Inject("BlockOrWaitBeforePass", nil)

d.PassBlockEventToSink(pendingEvent)

failpoint.Inject("BlockAfterPass", nil)

gemini-code-assist · 2026-02-03T12:38:34Z

The logic to find the minimum value for changefeedUsedMemory and changefeedMaxMemory is duplicated. This pattern of checking existence and updating the minimum value could be extracted into a small helper function to reduce code duplication and improve readability.

-Original file line number
+Diff line change
@@ Expand Up / @@ -402,6 +402,70 @@ func TestDispatcherHandleEvents(t *testing.T) { @@
     	require.Equal(t, uint64(7), checkpointTs)
     }
+    func TestDispatcherIgnoreActionWithSameCommitTsDifferentType(t *testing.T) {
+    	sink := sink.NewMockSink(common.MysqlSinkType)
+    	tableSpan, err := getCompleteTableSpan(getTestingKeyspaceID())
+    	require.NoError(t, err)
+    	dispatcher := newDispatcherForTest(sink, tableSpan)
+    	nodeID := node.NewID()
+    	ddlEvent := &commonEvent.DDLEvent{
+    		FinishedTs: 10,
+    		BlockedTables: &commonEvent.InfluencedTables{
+    			InfluenceType: commonEvent.InfluenceTypeNormal,
+    			TableIDs:      []int64{0, 1},
+    		},
+    	}
+    	block := dispatcher.HandleEvents([]DispatcherEvent{NewDispatcherEvent(&nodeID, ddlEvent)}, callback)
+    	require.True(t, block)
+    	// Pending DDL event should be in WAITING stage.
+    	pending, stage := dispatcher.blockEventStatus.getEventAndStage()
+    	require.NotNil(t, pending)
+    	require.Equal(t, heartbeatpb.BlockStage_WAITING, stage)
+    	// ACK the DDL so the resend task is cancelled.
+    	dispatcher.HandleDispatcherStatus(&heartbeatpb.DispatcherStatus{
+    		Ack: &heartbeatpb.ACK{
+    			CommitTs:    ddlEvent.FinishedTs,
+    			IsSyncPoint: false,
+    		},
+    	})
+    	require.Equal(t, 0, dispatcher.resendTaskMap.Len())
+    	// Drain any previously sent block status messages.
+    drain:
+    	for {
+    		select {
+    		case <-dispatcher.sharedInfo.blockStatusesChan:
+    		default:
+    			break drain
+    		}
+    	}
+    	// Send an action for a syncpoint with the same commitTs. It should be ignored.
+    	dispatcher.HandleDispatcherStatus(&heartbeatpb.DispatcherStatus{
+    		Action: &heartbeatpb.DispatcherAction{
+    			Action:      heartbeatpb.Action_Skip,
+    			CommitTs:    ddlEvent.FinishedTs,
+    			IsSyncPoint: true,
+    		},
+    	})
+    	// Still waiting on the DDL.
+    	pending, stage = dispatcher.blockEventStatus.getEventAndStage()
+    	require.NotNil(t, pending)
+    	require.Equal(t, heartbeatpb.BlockStage_WAITING, stage)
+    	// Should not emit a DONE block status for the mismatched action.
+    	select {
+    	case msg := <-dispatcher.sharedInfo.blockStatusesChan:
+    		require.FailNow(t, "unexpected block status emitted", "msg=%v", msg)
+    	default:
+    	}
+    }
     // test uncompelete table span can correctly handle the ddl events
     func TestUncompeleteTableSpanDispatcherHandleEvents(t *testing.T) {
     	count.Swap(0)
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up @@
     		return false
     	}
-    	return b.blockCommitTs == action.CommitTs
+    	if b.blockCommitTs != action.CommitTs {
+    		return false
+    	}
+    	isSyncPointPending := b.blockPendingEvent.GetType() == commonEvent.TypeSyncPointEvent
+    	return isSyncPointPending == action.IsSyncPoint
     }
     func (b *BlockEventStatus) getEventCommitTs() (uint64, bool) {
@@ Expand Down @@

-Original file line number
+Diff line change
@@ Expand Up / @@ -113,6 +113,8 @@ type EventCollector struct { @@
     	dispatcherMap sync.Map // key: dispatcherID, value: dispatcherStat
     	changefeedMap sync.Map // key: changefeedID.GID, value: *changefeedStat
+    	lastCongestionLogTime int64
     	mc messaging.MessageCenter
     	logCoordinatorClient *LogCoordinatorClient
@@ Expand Down Expand Up @@
     	// collect path-level available memory and total available memory for each changefeed
     	changefeedPathMemory := make(map[common.ChangeFeedID]map[common.DispatcherID]uint64)
     	changefeedTotalMemory := make(map[common.ChangeFeedID]uint64)
+    	changefeedUsedMemory := make(map[common.ChangeFeedID]uint64)
+    	changefeedMaxMemory := make(map[common.ChangeFeedID]uint64)
+    	totalDispatchers := 0
+    	dispatchersWithoutService := 0
     	// collect from main dynamic stream
     	for _, quota := range c.ds.GetMetrics().MemoryControl.AreaMemoryMetrics {
@@ Expand All @@
     		}
     		// store total available memory from AreaMemoryMetric
     		changefeedTotalMemory[cfID] = uint64(quota.AvailableMemory())
+    		changefeedUsedMemory[cfID] = uint64(quota.MemoryUsage())
+    		changefeedMaxMemory[cfID] = uint64(quota.MaxMemory())
     	}
     	// collect from redo dynamic stream and take minimum
@@ Expand All @@
     		} else {
     			changefeedTotalMemory[cfID] = uint64(quota.AvailableMemory())
     		}
+    		if existing, exists := changefeedUsedMemory[cfID]; exists {
+    			changefeedUsedMemory[cfID] = min(existing, uint64(quota.MemoryUsage()))
+    		} else {
+    			changefeedUsedMemory[cfID] = uint64(quota.MemoryUsage())
+    		}
+    		if existing, exists := changefeedMaxMemory[cfID]; exists {
+    			changefeedMaxMemory[cfID] = min(existing, uint64(quota.MaxMemory()))
+    		} else {
+    			changefeedMaxMemory[cfID] = uint64(quota.MaxMemory())
+    		}
     	}
     	if len(changefeedPathMemory) == 0 {
@@ Expand All @@
     	c.dispatcherMap.Range(func(k, v interface{}) bool {
     		stat := v.(*dispatcherStat)
     		eventServiceID := stat.connState.getEventServiceID()
+    		totalDispatchers++
     		if eventServiceID == "" {
+    			dispatchersWithoutService++
     			return true
     		}
@@ Expand All @@
     	// build congestion control messages for each node
     	result := make(map[node.ID]*event.CongestionControl)
+    	changefeedsInMessages := 0
     	for nodeID, changefeedDispatchers := range nodeDispatcherMemory {
-    		congestionControl := event.NewCongestionControl()
+    		congestionControl := event.NewCongestionControlWithVersion(event.CongestionControlVersion2)
     		for changefeedID, dispatcherMemory := range changefeedDispatchers {
     			if len(dispatcherMemory) == 0 {
@@ Expand All @@
     			// get total available memory directly from AreaMemoryMetric
     			totalAvailable := uint64(changefeedTotalMemory[changefeedID])
     			if totalAvailable > 0 {
-    				congestionControl.AddAvailableMemoryWithDispatchers(
+    				congestionControl.AddAvailableMemoryWithDispatchersAndUsage(
     					changefeedID.ID(),
     					totalAvailable,
+    					changefeedUsedMemory[changefeedID],
+    					changefeedMaxMemory[changefeedID],
     					dispatcherMemory,
     				)
+    				changefeedsInMessages++
     			}
     		}
@@ Expand All @@
     		}
     	}
+    	nowUnix := time.Now().Unix()
+    	lastLog := atomic.LoadInt64(&c.lastCongestionLogTime)
+    	if nowUnix-lastLog >= int64(time.Minute.Seconds()) &&
+    		atomic.CompareAndSwapInt64(&c.lastCongestionLogTime, lastLog, nowUnix) {
+    		log.Info("congestion control build summary",
+    			zap.Int("dispatchers", totalDispatchers),
+    			zap.Int("dispatchersWithoutService", dispatchersWithoutService),
+    			zap.Int("changefeedsWithMetrics", len(changefeedPathMemory)),
+    			zap.Int("changefeedsInMessages", changefeedsInMessages),
+    			zap.Int("nodes", len(result)),
+    		)
+    	}
     	return result
     }
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

DNM: refine dispatcher scan priority with syncpoint refine #4123

Uh oh!

Diff view

Diff view

There are no files selected for viewing

gemini-code-assist bot Feb 3, 2026

Uh oh!

gemini-code-assist bot Feb 3, 2026

Uh oh!

Uh oh!

Uh oh!

DNM: refine dispatcher scan priority with syncpoint refine #4123

Are you sure you want to change the base?

Uh oh!

DNM: refine dispatcher scan priority with syncpoint refine #4123

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

gemini-code-assist bot Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist bot Feb 3, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!