eventservice: simplify syncpoint handling with checkpoint bounds

- cap scan upper bound by checkpointTs + 2*syncPointInterval when syncpoint is enabled
- suppress syncpoint emission when dispatcher lag exceeds threshold, while still advancing nextSyncPoint
- resume syncpoint emission with hysteresis to avoid flapping
- apply checkpoint cap to normal scan path, pending-DDL local advance, and table-trigger DDL path
- add metrics for syncpoint lag, suppression count, and checkpoint-cap hits
- add unit tests for checkpoint cap and suppress/resume behavior

Signed-off-by: dongmen <414110582@qq.com>

Author: asddongmen

pkg/config/debug.go

@@ -118,6 +118,14 @@ type EventServiceConfig struct {
 	ScanTaskQueueSize int `toml:"scan-task-queue-size" json:"scan_task_queue_size"`
 	ScanLimitInBytes  int `toml:"scan-limit-in-bytes" json:"scan_limit_in_bytes"`
 
+	// SyncPointCheckpointCapMultiplier controls how far scan can lead checkpoint when syncpoint is enabled.
+	// The scan upper bound is capped by checkpointTs + multiplier*syncPointInterval.
+	SyncPointCheckpointCapMultiplier int `toml:"sync-point-checkpoint-cap-multiplier" json:"sync_point_checkpoint_cap_multiplier"`
+	// SyncPointLagSuppressThreshold controls when to suppress syncpoint emission for lagging dispatchers.
+	SyncPointLagSuppressThreshold time.Duration `toml:"sync-point-lag-suppress-threshold" json:"sync_point_lag_suppress_threshold"`
+	// SyncPointLagResumeThreshold controls when to resume syncpoint emission after suppression.
+	SyncPointLagResumeThreshold time.Duration `toml:"sync-point-lag-resume-threshold" json:"sync_point_lag_resume_threshold"`
+
 	// DMLEventMaxRows is the maximum number of rows in a DML event when split txn is enabled.
 	DMLEventMaxRows int32 `toml:"dml-event-max-rows" json:"dml_event_max_rows"`
 	// DMLEventMaxBytes is the maximum size of a DML event in bytes when split txn is enabled.
@@ -133,10 +141,13 @@ type EventServiceConfig struct {
 // NewDefaultEventServiceConfig return the default event service configuration
 func NewDefaultEventServiceConfig() *EventServiceConfig {
 	return &EventServiceConfig{
-		ScanTaskQueueSize:        1024 * 8,
-		ScanLimitInBytes:         1024 * 1024 * 256, // 256MB
-		DMLEventMaxRows:          256,
-		DMLEventMaxBytes:         1024 * 1024 * 1, // 1MB
-		EnableRemoteEventService: true,
+		ScanTaskQueueSize:                1024 * 8,
+		ScanLimitInBytes:                 1024 * 1024 * 256, // 256MB
+		SyncPointCheckpointCapMultiplier: 2,
+		SyncPointLagSuppressThreshold:    20 * time.Minute,
+		SyncPointLagResumeThreshold:      15 * time.Minute,
+		DMLEventMaxRows:                  256,
+		DMLEventMaxBytes:                 1024 * 1024 * 1, // 1MB
+		EnableRemoteEventService:         true,
 	}
 }

pkg/eventservice/dispatcher_stat.go

@@ -70,7 +70,9 @@ type dispatcherStat struct {
 	enableSyncPoint   bool
 	nextSyncPoint     atomic.Uint64
 	syncPointInterval time.Duration
-	txnAtomicity      config.AtomicityLevel
+	// syncPointSendSuppressed tracks whether syncpoint emission is temporarily suppressed due to lag.
+	syncPointSendSuppressed atomic.Bool
+	txnAtomicity            config.AtomicityLevel
 
 	// =============================================================================
 	// ================== below are fields need copied when reset ==================

pkg/eventservice/event_broker.go

@@ -52,6 +52,9 @@ const (
 	// defaultSendResolvedTsInterval use to control whether to send a resolvedTs event to the dispatcher when its scan is skipped.
 	defaultSendResolvedTsInterval           = time.Second * 2
 	defaultRefreshMinSentResolvedTsInterval = time.Second * 1
+	defaultSyncPointCheckpointCapMultiplier = 2
+	defaultSyncPointLagSuppressThreshold    = 20 * time.Minute
+	defaultSyncPointLagResumeThreshold      = 15 * time.Minute
 )
 
 // eventBroker get event from the eventStore, and send the event to the dispatchers.
@@ -92,8 +95,11 @@ type eventBroker struct {
 	// metricsCollector handles all metrics collection and reporting
 	metricsCollector *metricsCollector
 
-	scanRateLimiter  *rate.Limiter
-	scanLimitInBytes uint64
+	scanRateLimiter                  *rate.Limiter
+	scanLimitInBytes                 uint64
+	syncPointCheckpointCapMultiplier uint64
+	syncPointLagSuppressThreshold    time.Duration
+	syncPointLagResumeThreshold      time.Duration
 }
 
 func newEventBroker(
@@ -113,10 +119,26 @@ func newEventBroker(
 	sendMessageWorkerCount := config.DefaultBasicEventHandlerConcurrency
 	scanWorkerCount := config.DefaultBasicEventHandlerConcurrency * 4
 
-	scanTaskQueueSize := config.GetGlobalServerConfig().Debug.EventService.ScanTaskQueueSize / scanWorkerCount
+	eventServiceConfig := config.GetGlobalServerConfig().Debug.EventService
+	scanTaskQueueSize := eventServiceConfig.ScanTaskQueueSize / scanWorkerCount
 	sendMessageQueueSize := basicChannelSize * 4
 
-	scanLimitInBytes := config.GetGlobalServerConfig().Debug.EventService.ScanLimitInBytes
+	scanLimitInBytes := eventServiceConfig.ScanLimitInBytes
+	syncPointCheckpointCapMultiplier := eventServiceConfig.SyncPointCheckpointCapMultiplier
+	if syncPointCheckpointCapMultiplier <= 0 {
+		syncPointCheckpointCapMultiplier = defaultSyncPointCheckpointCapMultiplier
+	}
+	syncPointLagSuppressThreshold := eventServiceConfig.SyncPointLagSuppressThreshold
+	if syncPointLagSuppressThreshold <= 0 {
+		syncPointLagSuppressThreshold = defaultSyncPointLagSuppressThreshold
+	}
+	syncPointLagResumeThreshold := eventServiceConfig.SyncPointLagResumeThreshold
+	if syncPointLagResumeThreshold <= 0 {
+		syncPointLagResumeThreshold = defaultSyncPointLagResumeThreshold
+	}
+	if syncPointLagResumeThreshold > syncPointLagSuppressThreshold {
+		syncPointLagResumeThreshold = syncPointLagSuppressThreshold
+	}
 
 	g, ctx := errgroup.WithContext(ctx)
 	ctx, cancel := context.WithCancel(ctx)
@@ -125,22 +147,25 @@ func newEventBroker(
 	// For now, since there is only one upstream, using the default pdClock is sufficient.
 	pdClock := appcontext.GetService[pdutil.Clock](appcontext.DefaultPDClock)
 	c := &eventBroker{
-		tidbClusterID:           id,
-		eventStore:              eventStore,
-		pdClock:                 pdClock,
-		mounter:                 event.NewMounter(tz, integrity),
-		schemaStore:             schemaStore,
-		changefeedMap:           sync.Map{},
-		dispatchers:             sync.Map{},
-		tableTriggerDispatchers: sync.Map{},
-		msgSender:               mc,
-		taskChan:                make([]chan scanTask, scanWorkerCount),
-		messageCh:               make([]chan *wrapEvent, sendMessageWorkerCount),
-		redoMessageCh:           make([]chan *wrapEvent, sendMessageWorkerCount),
-		cancel:                  cancel,
-		g:                       g,
-		scanRateLimiter:         rate.NewLimiter(rate.Limit(scanLimitInBytes), scanLimitInBytes),
-		scanLimitInBytes:        uint64(scanLimitInBytes),
+		tidbClusterID:                    id,
+		eventStore:                       eventStore,
+		pdClock:                          pdClock,
+		mounter:                          event.NewMounter(tz, integrity),
+		schemaStore:                      schemaStore,
+		changefeedMap:                    sync.Map{},
+		dispatchers:                      sync.Map{},
+		tableTriggerDispatchers:          sync.Map{},
+		msgSender:                        mc,
+		taskChan:                         make([]chan scanTask, scanWorkerCount),
+		messageCh:                        make([]chan *wrapEvent, sendMessageWorkerCount),
+		redoMessageCh:                    make([]chan *wrapEvent, sendMessageWorkerCount),
+		cancel:                           cancel,
+		g:                                g,
+		scanRateLimiter:                  rate.NewLimiter(rate.Limit(scanLimitInBytes), scanLimitInBytes),
+		scanLimitInBytes:                 uint64(scanLimitInBytes),
+		syncPointCheckpointCapMultiplier: uint64(syncPointCheckpointCapMultiplier),
+		syncPointLagSuppressThreshold:    syncPointLagSuppressThreshold,
+		syncPointLagResumeThreshold:      syncPointLagResumeThreshold,
 	}
 
 	// Initialize metrics collector
@@ -185,7 +210,12 @@ func newEventBroker(
 		return c.refreshMinSentResolvedTs(ctx)
 	})
 
-	log.Info("new event broker created", zap.Uint64("id", id), zap.Uint64("scanLimitInBytes", c.scanLimitInBytes))
+	log.Info("new event broker created",
+		zap.Uint64("id", id),
+		zap.Uint64("scanLimitInBytes", c.scanLimitInBytes),
+		zap.Uint64("syncPointCheckpointCapMultiplier", c.syncPointCheckpointCapMultiplier),
+		zap.Duration("syncPointLagSuppressThreshold", c.syncPointLagSuppressThreshold),
+		zap.Duration("syncPointLagResumeThreshold", c.syncPointLagResumeThreshold))
 	return c
 }
 
@@ -355,13 +385,17 @@ func (c *eventBroker) tickTableTriggerDispatchers(ctx context.Context) error {
 					return true
 				}
 				stat.receivedResolvedTs.Store(endTs)
+				boundedEndTs := c.capCommitTsEndByCheckpoint(stat, endTs)
 				for _, e := range ddlEvents {
+					if e.FinishedTs > boundedEndTs {
+						break
+					}
 					ep := &e
 					c.sendDDL(ctx, remoteID, ep, stat)
 				}
-				if endTs > startTs {
+				if boundedEndTs > startTs {
 					// After all the events are sent, we send the watermark to the dispatcher.
-					c.sendResolvedTs(stat, endTs)
+					c.sendResolvedTs(stat, boundedEndTs)
 				} else {
 					// If there is no new ddl event, we still need to send a signal resolved-ts event to keep downstream responsive,
 					// but do not advance the watermark here.
@@ -406,6 +440,33 @@ func (c *eventBroker) logUninitializedDispatchers(ctx context.Context) error {
 	}
 }
 
+func (c *eventBroker) capCommitTsEndByCheckpoint(task scanTask, commitTsEnd uint64) uint64 {
+	if !task.enableSyncPoint || task.syncPointInterval <= 0 || c.syncPointCheckpointCapMultiplier == 0 {
+		return commitTsEnd
+	}
+	checkpointTs := task.checkpointTs.Load()
+	if checkpointTs == 0 {
+		return commitTsEnd
+	}
+
+	capDuration := time.Duration(c.syncPointCheckpointCapMultiplier) * task.syncPointInterval
+	checkpointCapTs := oracle.GoTimeToTS(oracle.GetTimeFromTS(checkpointTs).Add(capDuration))
+	if checkpointCapTs >= commitTsEnd {
+		return commitTsEnd
+	}
+
+	log.Debug("scan range commitTsEnd capped by checkpoint bound",
+		zap.Stringer("changefeedID", task.changefeedStat.changefeedID),
+		zap.Stringer("dispatcherID", task.id),
+		zap.Uint64("oldCommitTsEnd", commitTsEnd),
+		zap.Uint64("newCommitTsEnd", checkpointCapTs),
+		zap.Uint64("checkpointTs", checkpointTs),
+		zap.Duration("syncPointInterval", task.syncPointInterval),
+		zap.Uint64("multiplier", c.syncPointCheckpointCapMultiplier))
+	metrics.EventServiceScanCappedByCheckpointCount.WithLabelValues(task.changefeedStat.changefeedID.String()).Inc()
+	return checkpointCapTs
+}
+
 // getScanTaskDataRange determines the valid data range for scanning a given task.
 // It checks various conditions (dispatcher status, DDL state, max commit ts of dml event)
 // to decide whether scanning is needed and returns the appropriate time range.
@@ -450,6 +511,7 @@ func (c *eventBroker) getScanTaskDataRange(task scanTask) (bool, common.DataRang
 			)
 		}
 	}
+	dataRange.CommitTsEnd = c.capCommitTsEndByCheckpoint(task, dataRange.CommitTsEnd)
 
 	if dataRange.CommitTsEnd <= dataRange.CommitTsStart && hasPendingDDLEventInCurrentRange {
 		// Global scan window base can be pinned by other lagging dispatchers.
@@ -461,6 +523,7 @@ func (c *eventBroker) getScanTaskDataRange(task scanTask) (bool, common.DataRang
 		}
 		localScanMaxTs := oracle.GoTimeToTS(oracle.GetTimeFromTS(dataRange.CommitTsStart).Add(interval))
 		dataRange.CommitTsEnd = min(commitTsEndBeforeWindow, localScanMaxTs)
+		dataRange.CommitTsEnd = c.capCommitTsEndByCheckpoint(task, dataRange.CommitTsEnd)
 		if dataRange.CommitTsEnd > dataRange.CommitTsStart {
 			log.Info("scan window local advance due to pending ddl",
 				zap.Stringer("changefeedID", task.changefeedStat.changefeedID),
@@ -587,22 +650,74 @@ func (c *eventBroker) hasSyncPointEventsBeforeTs(ts uint64, d *dispatcherStat) b
 	return d.enableSyncPoint && ts > d.nextSyncPoint.Load()
 }
 
+func syncPointLagDuration(sentResolvedTs, checkpointTs uint64) time.Duration {
+	if sentResolvedTs <= checkpointTs {
+		return 0
+	}
+	return oracle.GetTimeFromTS(sentResolvedTs).Sub(oracle.GetTimeFromTS(checkpointTs))
+}
+
+func (c *eventBroker) shouldSuppressSyncPointEmission(d *dispatcherStat) bool {
+	if d == nil || c.syncPointLagSuppressThreshold <= 0 {
+		return false
+	}
+
+	sentResolvedTs := d.sentResolvedTs.Load()
+	checkpointTs := d.checkpointTs.Load()
+	lag := syncPointLagDuration(sentResolvedTs, checkpointTs)
+	metrics.EventServiceSyncPointLagGaugeVec.WithLabelValues(d.changefeedStat.changefeedID.String()).Set(lag.Seconds())
+
+	if d.syncPointSendSuppressed.Load() {
+		if lag <= c.syncPointLagResumeThreshold {
+			if d.syncPointSendSuppressed.CompareAndSwap(true, false) {
+				log.Info("syncpoint emission resumed",
+					zap.Stringer("changefeedID", d.changefeedStat.changefeedID),
+					zap.Stringer("dispatcherID", d.id),
+					zap.Uint64("sentResolvedTs", sentResolvedTs),
+					zap.Uint64("checkpointTs", checkpointTs),
+					zap.Duration("lag", lag),
+					zap.Duration("resumeThreshold", c.syncPointLagResumeThreshold))
+			}
+			return false
+		}
+		return true
+	}
+
+	if lag > c.syncPointLagSuppressThreshold {
+		if d.syncPointSendSuppressed.CompareAndSwap(false, true) {
+			log.Info("syncpoint emission suppressed due to lag",
+				zap.Stringer("changefeedID", d.changefeedStat.changefeedID),
+				zap.Stringer("dispatcherID", d.id),
+				zap.Uint64("sentResolvedTs", sentResolvedTs),
+				zap.Uint64("checkpointTs", checkpointTs),
+				zap.Duration("lag", lag),
+				zap.Duration("suppressThreshold", c.syncPointLagSuppressThreshold),
+				zap.Duration("resumeThreshold", c.syncPointLagResumeThreshold))
+		}
+		return true
+	}
+	return false
+}
+
 // emitSyncPointEventIfNeeded emits a sync point event if the current ts is greater than the next sync point, and updates the next sync point.
 // We need call this function every time we send a event(whether dml/ddl/resolvedTs),
 // thus to ensure the sync point event is in correct order for each dispatcher.
 func (c *eventBroker) emitSyncPointEventIfNeeded(ts uint64, d *dispatcherStat, remoteID node.ID) {
 	for d.enableSyncPoint && ts > d.nextSyncPoint.Load() {
 		commitTs := d.nextSyncPoint.Load()
 		d.nextSyncPoint.Store(oracle.GoTimeToTS(oracle.GetTimeFromTS(commitTs).Add(d.syncPointInterval)))
+		if c.shouldSuppressSyncPointEmission(d) {
+			metrics.EventServiceSyncPointSuppressedCount.WithLabelValues(d.changefeedStat.changefeedID.String()).Inc()
+			continue
+		}
 
 		e := event.NewSyncPointEvent(d.id, commitTs, d.seq.Add(1), d.epoch)
+		syncPointEvent := newWrapSyncPointEvent(remoteID, e)
+		c.getMessageCh(d.messageWorkerIndex, common.IsRedoMode(d.info.GetMode())) <- syncPointEvent
 		log.Debug("send syncpoint event to dispatcher",
 			zap.Stringer("changefeedID", d.changefeedStat.changefeedID),
 			zap.Stringer("dispatcherID", d.id), zap.Int64("tableID", d.info.GetTableSpan().GetTableID()),
 			zap.Uint64("commitTs", e.GetCommitTs()), zap.Uint64("seq", e.GetSeq()))
-
-		syncPointEvent := newWrapSyncPointEvent(remoteID, e)
-		c.getMessageCh(d.messageWorkerIndex, common.IsRedoMode(d.info.GetMode())) <- syncPointEvent
 	}
 }