*: add low-latency watermark propagation

downstreamadapter/dispatchermanager/task.go

@@ -24,10 +24,17 @@ import (
 	"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/utils/threadpool"
 	"go.uber.org/zap"
 )
 
+const (
+	defaultHeartbeatInterval     = 200 * time.Millisecond
+	lowLatencyHeartbeatInterval  = 50 * time.Millisecond
+	defaultHeartbeatInitialDelay = time.Second
+)
+
 // HeartbeatTask is a perioic task to collect the heartbeat status from event dispatcher manager and push to heartbeatRequestQueue
 type HeartBeatTask struct {
 	taskHandle *threadpool.TaskHandle
@@ -42,16 +49,15 @@ func newHeartBeatTask(manager *DispatcherManager) *HeartBeatTask {
 		manager:    manager,
 		statusTick: 0,
 	}
-	t.taskHandle = taskScheduler.Submit(t, time.Now().Add(time.Second*1))
+	t.taskHandle = taskScheduler.Submit(t, time.Now().Add(heartbeatInitialDelay()))
 	return t
 }
 
 func (t *HeartBeatTask) Execute() time.Time {
 	if t.manager.closed.Load() {
 		return time.Time{}
 	}
-	executeInterval := time.Millisecond * 200
-	// 10s / 200ms = 50
+	executeInterval := heartbeatInterval()
 	completeStatusInterval := int(time.Second * 10 / executeInterval)
 	t.statusTick++
 	needCompleteStatus := (t.statusTick)%completeStatusInterval == 0
@@ -60,6 +66,20 @@ func (t *HeartBeatTask) Execute() time.Time {
 	return time.Now().Add(executeInterval)
 }
 
+func heartbeatInterval() time.Duration {
+	if config.GetGlobalServerConfig().IsLowLatencyMode() {
+		return lowLatencyHeartbeatInterval
+	}
+	return defaultHeartbeatInterval
+}
+
+func heartbeatInitialDelay() time.Duration {
+	if config.GetGlobalServerConfig().IsLowLatencyMode() {
+		return 0
+	}
+	return defaultHeartbeatInitialDelay
+}
+
 func (t *HeartBeatTask) Cancel() {
 	t.taskHandle.Cancel()
 }

downstreamadapter/dispatchermanager/task_test.go

@@ -0,0 +1,38 @@
+// 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 dispatchermanager
+
+import (
+	"testing"
+
+	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/stretchr/testify/require"
+)
+
+func TestHeartbeatIntervalsByPerformanceMode(t *testing.T) {
+	original := config.GetGlobalServerConfig()
+	t.Cleanup(func() {
+		config.StoreGlobalServerConfig(original)
+	})
+
+	cfg := original.Clone()
+	config.StoreGlobalServerConfig(cfg)
+	require.Equal(t, defaultHeartbeatInterval, heartbeatInterval())
+	require.Equal(t, defaultHeartbeatInitialDelay, heartbeatInitialDelay())
+
+	cfg.PerformanceMode = config.PerformanceModeLowLatency
+	config.StoreGlobalServerConfig(cfg)
+	require.Equal(t, lowLatencyHeartbeatInterval, heartbeatInterval())
+	require.Zero(t, heartbeatInitialDelay())
+}

logservice/coordinator/coordinator.go

@@ -55,11 +55,18 @@ type requestAndTarget struct {
 type changefeedState struct {
 	cfID       common.ChangeFeedID
 	nodeStates map[node.ID]uint64
+	// nodesReportedSinceLastUpdate tracks a complete reporting round. Publishing
+	// the global minimum only after every current node has reported avoids
+	// exposing intermediate minima from staggered node reports.
+	nodesReportedSinceLastUpdate map[node.ID]struct{}
+	nodeReportPhyTs              map[node.ID]int64
 
 	// equal to min puller resolved ts
-	minLogServiceResolvedTs uint64
-	resolvedTsGauge         prometheus.Gauge
-	resolvedTsLagGauge      prometheus.Gauge
+	minLogServiceResolvedTs     uint64
+	metricsInitialized          bool
+	metricsUpdatedSinceLastTick bool
+	resolvedTsGauge             prometheus.Gauge
+	resolvedTsLagGauge          prometheus.Gauge
 }
 
 type logCoordinator struct {
@@ -207,6 +214,8 @@ func (c *logCoordinator) handleNodeChange(allNodes map[node.ID]*node.Info) {
 			c.changefeedStates.Lock()
 			for _, state := range c.changefeedStates.m {
 				delete(state.nodeStates, id)
+				delete(state.nodesReportedSinceLastUpdate, id)
+				delete(state.nodeReportPhyTs, id)
 			}
 			c.changefeedStates.Unlock()
 		}
@@ -229,9 +238,11 @@ func (c *logCoordinator) updateEventStoreState(nodeID node.ID, newState *logserv
 func (c *logCoordinator) updateChangefeedStates(from node.ID, states *logservicepb.ChangefeedStates) {
 	c.changefeedStates.Lock()
 	defer c.changefeedStates.Unlock()
+	pdPhyTs := oracle.GetPhysical(c.pdClock.CurrentTime())
 
 	// Create a set of incoming changefeed GIDs for efficient lookup.
 	incomingGIDs := make(map[common.GID]struct{})
+	affectedGIDs := make(map[common.GID]struct{})
 	for _, state := range states.States {
 		cfID := common.NewChangefeedIDFromPB(state.GetChangefeedID())
 		incomingGIDs[cfID.ID()] = struct{}{}
@@ -244,6 +255,9 @@ func (c *logCoordinator) updateChangefeedStates(from node.ID, states *logservice
 			// ...but is no longer in the incoming message, it means the changefeed was removed from this node.
 			if _, incoming := incomingGIDs[gid]; !incoming {
 				delete(state.nodeStates, from)
+				delete(state.nodesReportedSinceLastUpdate, from)
+				delete(state.nodeReportPhyTs, from)
+				affectedGIDs[gid] = struct{}{}
 				log.Info("changefeed removed from node",
 					zap.Stringer("changefeedID", state.cfID),
 					zap.String("nodeID", string(from)),
@@ -274,13 +288,34 @@ func (c *logCoordinator) updateChangefeedStates(from node.ID, states *logservice
 				zap.Uint64("changefeedGIDHigh", gid.High))
 			// Initialize metrics for the new changefeed.
 			c.changefeedStates.m[gid] = &changefeedState{
-				cfID:               cfID,
-				nodeStates:         make(map[node.ID]uint64),
-				resolvedTsGauge:    metrics.ChangefeedResolvedTsGauge.WithLabelValues(cfID.Keyspace(), cfID.Name()),
-				resolvedTsLagGauge: metrics.ChangefeedResolvedTsLagGauge.WithLabelValues(cfID.Keyspace(), cfID.Name()),
+				cfID:                         cfID,
+				nodeStates:                   make(map[node.ID]uint64),
+				nodesReportedSinceLastUpdate: make(map[node.ID]struct{}),
+				nodeReportPhyTs:              make(map[node.ID]int64),
+				resolvedTsGauge:              metrics.ChangefeedResolvedTsGauge.WithLabelValues(cfID.Keyspace(), cfID.Name()),
+				resolvedTsLagGauge:           metrics.ChangefeedResolvedTsLagGauge.WithLabelValues(cfID.Keyspace(), cfID.Name()),
 			}
 		}
-		c.changefeedStates.m[gid].nodeStates[from] = state.GetResolvedTs()
+		changefeedState := c.changefeedStates.m[gid]
+		changefeedState.nodeStates[from] = state.GetResolvedTs()
+		changefeedState.nodesReportedSinceLastUpdate[from] = struct{}{}
+		changefeedState.nodeReportPhyTs[from] = pdPhyTs
+		affectedGIDs[gid] = struct{}{}
+	}
+
+	for gid := range affectedGIDs {
+		state, ok := c.changefeedStates.m[gid]
+		if !ok {
+			continue
+		}
+		if len(state.nodeStates) == 0 ||
+			len(state.nodesReportedSinceLastUpdate) != len(state.nodeStates) {
+			continue
+		}
+		if c.updateChangefeedMetrics(state, pdPhyTs, false) {
+			state.metricsUpdatedSinceLastTick = true
+		}
+		clear(state.nodesReportedSinceLastUpdate)
 	}
 }
 
@@ -292,29 +327,61 @@ func (c *logCoordinator) reportChangefeedMetrics() {
 	defer c.changefeedStates.Unlock()
 
 	for _, state := range c.changefeedStates.m {
-		if len(state.nodeStates) == 0 {
+		if state.metricsUpdatedSinceLastTick {
+			state.metricsUpdatedSinceLastTick = false
 			continue
 		}
+		c.updateChangefeedMetrics(state, pdPhyTs, true)
+	}
+}
 
-		minResolvedTs := uint64(math.MaxUint64)
-		for _, resolvedTs := range state.nodeStates {
-			if resolvedTs < minResolvedTs {
-				minResolvedTs = resolvedTs
+// updateChangefeedMetrics publishes the global resolved-ts and its lag.
+// A fresh complete reporting round uses each node's own report time and
+// publishes the maximum per-node lag, avoiding inflation from staggered reports.
+// A forced periodic refresh uses the current PD time and cached global minimum,
+// so the lag grows when reports stop and the cached state becomes stale.
+func (c *logCoordinator) updateChangefeedMetrics(state *changefeedState, pdPhyTs int64, force bool) bool {
+	if len(state.nodeStates) == 0 {
+		return false
+	}
+
+	minResolvedTs := uint64(math.MaxUint64)
+	var maxNodeLag float64
+	hasNodeLag := false
+	for nodeID, resolvedTs := range state.nodeStates {
+		if resolvedTs < minResolvedTs {
+			minResolvedTs = resolvedTs
+		}
+		if !force {
+			if reportPhyTs, ok := state.nodeReportPhyTs[nodeID]; ok {
+				nodeLag := float64(reportPhyTs-oracle.ExtractPhysical(resolvedTs)) / 1e3
+				if !hasNodeLag || nodeLag > maxNodeLag {
+					maxNodeLag = nodeLag
+					hasNodeLag = true
+				}
 			}
 		}
+	}
 
-		if minResolvedTs == math.MaxUint64 {
-			log.Warn("minResolvedTs is MaxUint64, this should not happen",
-				zap.Stringer("changefeedID", state.cfID))
-			continue
-		}
+	if minResolvedTs == math.MaxUint64 {
+		log.Warn("minResolvedTs is MaxUint64, this should not happen",
+			zap.Stringer("changefeedID", state.cfID))
+		return false
+	}
+	if !force && state.metricsInitialized && minResolvedTs == state.minLogServiceResolvedTs {
+		return false
+	}
 
-		phyResolvedTs := oracle.ExtractPhysical(minResolvedTs)
-		state.minLogServiceResolvedTs = minResolvedTs
-		state.resolvedTsGauge.Set(float64(phyResolvedTs))
-		lag := float64(pdPhyTs-phyResolvedTs) / 1e3
-		state.resolvedTsLagGauge.Set(lag)
+	phyResolvedTs := oracle.ExtractPhysical(minResolvedTs)
+	state.minLogServiceResolvedTs = minResolvedTs
+	state.metricsInitialized = true
+	state.resolvedTsGauge.Set(float64(phyResolvedTs))
+	lag := float64(pdPhyTs-phyResolvedTs) / 1e3
+	if !force && hasNodeLag {
+		lag = maxNodeLag
 	}
+	state.resolvedTsLagGauge.Set(lag)
+	return true
 }
 
 func (c *logCoordinator) getMinLogServiceResolvedTs(cfID common.ChangeFeedID) uint64 {