From 754651e09351fcb177104606702bebfce297d3ad Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 16:54:29 +0800 Subject: [PATCH 1/8] logservice,config: add low latency resolved ts mode --- logservice/eventstore/event_store.go | 3 ++ logservice/logpuller/region_event_handler.go | 47 ++++++++++--------- .../logpuller/regionlock/region_range_lock.go | 14 ++++++ .../regionlock/region_range_lock_test.go | 22 +++++++++ pkg/config/server.go | 32 +++++++++---- pkg/config/server_config_test.go | 19 ++++++++ 6 files changed, 108 insertions(+), 29 deletions(-) diff --git a/logservice/eventstore/event_store.go b/logservice/eventstore/event_store.go index 07ec16e5e2..060fc52c66 100644 --- a/logservice/eventstore/event_store.go +++ b/logservice/eventstore/event_store.go @@ -685,6 +685,9 @@ func (e *eventStore) RegisterDispatcher( serverConfig := config.GetGlobalServerConfig() resolvedTsAdvanceInterval := int64(serverConfig.KVClient.AdvanceIntervalInMs) + if serverConfig.IsLowLatencyMode() { + resolvedTsAdvanceInterval = 0 + } // Note: don't hold any lock when call Subscribe e.subClient.Subscribe(subStat.subID, *dispatcherSpan, startTs, consumeKVEvents, advanceResolvedTs, resolvedTsAdvanceInterval, bdrMode) log.Info("new subscription created", diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 0973b1df15..cbe7293d02 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -19,6 +19,7 @@ import ( "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/log" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/spanz" @@ -133,10 +134,28 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) handleEventEntries(span, event.mustFirstState(), event.entries) } else if event.resolvedTs != 0 { hasResolved = true - for _, state := range event.states { - resolvedTs := handleResolvedTs(span, state, event.resolvedTs) - if resolvedTs > newResolvedTs { - newResolvedTs = resolvedTs + if span.advanceInterval == 0 { + updatedStates := make([]*regionlock.LockedRangeState, 0, len(event.states)) + regionID := uint64(0) + for _, state := range event.states { + if updateRegionResolvedTs(state, event.resolvedTs) { + updatedStates = append(updatedStates, state.region.lockedRangeState) + regionID = state.getRegionID() + } + } + if len(updatedStates) > 0 { + ts := span.rangeLock.UpdateLockedRangeStateHeapBatch(updatedStates) + resolvedTs := advanceSpanResolvedTs(span, regionID, ts) + if resolvedTs > newResolvedTs { + newResolvedTs = resolvedTs + } + } + } else { + for _, state := range event.states { + resolvedTs := handleResolvedTs(span, state, event.resolvedTs) + if resolvedTs > newResolvedTs { + newResolvedTs = resolvedTs + } } } } else { @@ -358,30 +377,17 @@ func handleEventEntries(span *subscribedSpan, state *regionFeedState, entries *c } func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs uint64) uint64 { - if state.isStale() || !state.isInitialized() { + if !updateRegionResolvedTs(state, resolvedTs) { return 0 } - state.matcher.tryCleanUnmatchedValue() - regionID := state.getRegionID() - lastResolvedTs := state.getLastResolvedTs() - if resolvedTs < lastResolvedTs { - log.Debug("The resolvedTs is fallen back in subscription client", - zap.Uint64("subscriptionID", uint64(state.region.subscribedSpan.subID)), - zap.Uint64("regionID", regionID), - zap.Uint64("resolvedTs", resolvedTs), - zap.Uint64("lastResolvedTs", lastResolvedTs)) - return 0 - } - - state.updateResolvedTs(resolvedTs) + regionID := state.getRegionID() ts := uint64(0) shouldAdvance := false // advanceInterval defaults to 100ms; setting it to 0 means resolving the timestamp as soon as possible. // Note: If a single span contains an extremely large number of regions (e.g., 500k), advanceInterval = 0 may cause performance issues. if span.advanceInterval == 0 { - span.rangeLock.UpdateLockedRangeStateHeap(state.region.lockedRangeState) - ts = span.rangeLock.GetHeapMinTs() + ts = span.rangeLock.UpdateLockedRangeStateHeapBatch([]*regionlock.LockedRangeState{state.region.lockedRangeState}) shouldAdvance = true } else { now := time.Now().UnixMilli() @@ -426,6 +432,5 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u return ts } } - return 0 } diff --git a/logservice/logpuller/regionlock/region_range_lock.go b/logservice/logpuller/regionlock/region_range_lock.go index fbd67e2bd3..6bd899263f 100644 --- a/logservice/logpuller/regionlock/region_range_lock.go +++ b/logservice/logpuller/regionlock/region_range_lock.go @@ -420,11 +420,25 @@ func (l *RangeLock) UpdateLockedRangeStateHeap(lockedRangeState *LockedRangeStat l.lockedRangeStateHeap.AddOrUpdate(lockedRangeState) } +// UpdateLockedRangeStateHeapBatch updates multiple locked ranges and returns the minimum resolvedTs. +func (l *RangeLock) UpdateLockedRangeStateHeapBatch(lockedRangeStates []*LockedRangeState) uint64 { + l.mu.Lock() + defer l.mu.Unlock() + + for _, lockedRangeState := range lockedRangeStates { + l.lockedRangeStateHeap.AddOrUpdate(lockedRangeState) + } + return l.getHeapMinTs() +} + // GetHeapMinTs returns the minimum ResolvedTs from the heap. func (l *RangeLock) GetHeapMinTs() uint64 { l.mu.RLock() defer l.mu.RUnlock() + return l.getHeapMinTs() +} +func (l *RangeLock) getHeapMinTs() uint64 { minTs := uint64(math.MaxUint64) if minEntry, ok := l.lockedRangeStateHeap.PeekTop(); ok { minTs = minEntry.ResolvedTs.Load() diff --git a/logservice/logpuller/regionlock/region_range_lock_test.go b/logservice/logpuller/regionlock/region_range_lock_test.go index 8325b89356..4d3c941cef 100644 --- a/logservice/logpuller/regionlock/region_range_lock_test.go +++ b/logservice/logpuller/regionlock/region_range_lock_test.go @@ -332,3 +332,25 @@ func TestRangeLockGetHeapMinTs(t *testing.T) { require.Equal(t, uint64(50), l.GetHeapMinTs()) require.Equal(t, l.ResolvedTs(), l.GetHeapMinTs()) } + +func TestRangeLockUpdateLockedRangeStateHeapBatch(t *testing.T) { + t.Parallel() + + l := NewRangeLock(1, []byte("a"), []byte("z"), 1) + first := l.LockRange(context.Background(), []byte("a"), []byte("m"), 1, 1) + second := l.LockRange(context.Background(), []byte("m"), []byte("z"), 2, 1) + require.Equal(t, LockRangeStatusSuccess, first.Status) + require.Equal(t, LockRangeStatusSuccess, second.Status) + + first.LockedRangeState.ResolvedTs.Store(200) + second.LockedRangeState.ResolvedTs.Store(100) + require.Equal(t, uint64(100), l.UpdateLockedRangeStateHeapBatch([]*LockedRangeState{ + first.LockedRangeState, + second.LockedRangeState, + })) + + second.LockedRangeState.ResolvedTs.Store(300) + require.Equal(t, uint64(200), l.UpdateLockedRangeStateHeapBatch([]*LockedRangeState{ + second.LockedRangeState, + })) +} diff --git a/pkg/config/server.go b/pkg/config/server.go index 8606f203a0..855635ec9c 100644 --- a/pkg/config/server.go +++ b/pkg/config/server.go @@ -56,6 +56,9 @@ const ( // DefaultBasicEventHandlerConcurrency is used to calculate the number of workers for // eventService and eventCollector. DefaultBasicEventHandlerConcurrency = 32 + + PerformanceModeThroughput = "throughput" + PerformanceModeLowLatency = "low-latency" ) var ( @@ -88,11 +91,12 @@ type LogConfig struct { } var defaultServerConfig = &ServerConfig{ - Newarch: false, - Addr: "127.0.0.1:8300", - AdvertiseAddr: "", - LogFile: "", - LogLevel: "info", + Newarch: false, + PerformanceMode: PerformanceModeThroughput, + Addr: "127.0.0.1:8300", + AdvertiseAddr: "", + LogFile: "", + LogLevel: "info", Log: &LogConfig{ File: &LogFileConfig{ MaxSize: 300, @@ -138,9 +142,10 @@ var defaultServerConfig = &ServerConfig{ // ServerConfig represents a config for server type ServerConfig struct { - Newarch bool `toml:"newarch" json:"newarch"` - Addr string `toml:"addr" json:"addr"` - AdvertiseAddr string `toml:"advertise-addr" json:"advertise-addr"` + Newarch bool `toml:"newarch" json:"newarch"` + PerformanceMode string `toml:"performance-mode" json:"performance-mode"` + Addr string `toml:"addr" json:"addr"` + AdvertiseAddr string `toml:"advertise-addr" json:"advertise-addr"` LogFile string `toml:"log-file" json:"log-file"` LogLevel string `toml:"log-level" json:"log-level"` @@ -241,6 +246,13 @@ func (c *ServerConfig) ValidateAndAdjust() error { if c.GcTTL == 0 { return cerror.ErrInvalidServerOption.GenWithStack("empty GC TTL is not allowed") } + if c.PerformanceMode == "" { + c.PerformanceMode = PerformanceModeThroughput + } + if c.PerformanceMode != PerformanceModeThroughput && c.PerformanceMode != PerformanceModeLowLatency { + return cerror.ErrInvalidServerOption.GenWithStackByArgs( + fmt.Sprintf("unknown performance mode: %s", c.PerformanceMode)) + } // 5s is minimum lease ttl in etcd(PD) if c.CaptureSessionTTL < 5 { log.Warn("capture session ttl too small, set to default value 10s") @@ -301,6 +313,10 @@ func (c *ServerConfig) ValidateAndAdjust() error { return nil } +func (c *ServerConfig) IsLowLatencyMode() bool { + return c.PerformanceMode == PerformanceModeLowLatency +} + // GetDefaultServerConfig returns the default server config func GetDefaultServerConfig() *ServerConfig { return defaultServerConfig.Clone() diff --git a/pkg/config/server_config_test.go b/pkg/config/server_config_test.go index 8cb56e5158..c6ab424a48 100644 --- a/pkg/config/server_config_test.go +++ b/pkg/config/server_config_test.go @@ -55,6 +55,25 @@ enable-legacy-safepoint = true require.True(t, cfg.EnableLegacySafePoint) } +func TestServerConfigPerformanceMode(t *testing.T) { + t.Parallel() + + cfg := GetDefaultServerConfig() + require.Equal(t, PerformanceModeThroughput, cfg.PerformanceMode) + require.False(t, cfg.IsLowLatencyMode()) + + configPath := filepath.Join(t.TempDir(), "server.toml") + require.NoError(t, os.WriteFile(configPath, []byte(`performance-mode = "low-latency"`), 0o644)) + metaData, err := toml.DecodeFile(configPath, cfg) + require.NoError(t, err) + require.Empty(t, metaData.Undecoded()) + require.NoError(t, cfg.ValidateAndAdjust()) + require.True(t, cfg.IsLowLatencyMode()) + + cfg.PerformanceMode = "invalid" + require.Error(t, cfg.ValidateAndAdjust()) +} + func TestServerConfigClone(t *testing.T) { t.Parallel() conf := GetDefaultServerConfig() From 9041ffad1b1d29ee81e28a424deed7e8da07f6f0 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 16:54:33 +0800 Subject: [PATCH 2/8] downstreamadapter: reduce low latency heartbeat interval --- downstreamadapter/dispatchermanager/task.go | 26 +++++++++++-- .../dispatchermanager/task_test.go | 38 +++++++++++++++++++ 2 files changed, 61 insertions(+), 3 deletions(-) create mode 100644 downstreamadapter/dispatchermanager/task_test.go diff --git a/downstreamadapter/dispatchermanager/task.go b/downstreamadapter/dispatchermanager/task.go index 2d2ec92a29..ad4b85e43b 100644 --- a/downstreamadapter/dispatchermanager/task.go +++ b/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,7 +49,7 @@ 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 } @@ -50,8 +57,7 @@ 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() } diff --git a/downstreamadapter/dispatchermanager/task_test.go b/downstreamadapter/dispatchermanager/task_test.go new file mode 100644 index 0000000000..777bab2cb4 --- /dev/null +++ b/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()) +} From ec603153649bbae1c2c47f68e89d746da10d0366 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 16:54:58 +0800 Subject: [PATCH 3/8] maintainer: trigger checkpoint calculation on watermark update --- maintainer/maintainer.go | 74 +++++++++++++++++++++++------------ maintainer/maintainer_test.go | 19 +++++++++ 2 files changed, 67 insertions(+), 26 deletions(-) diff --git a/maintainer/maintainer.go b/maintainer/maintainer.go index 1465e9326e..60f73cdb84 100644 --- a/maintainer/maintainer.go +++ b/maintainer/maintainer.go @@ -69,8 +69,9 @@ type Maintainer struct { selfNode *node.Info controller *Controller - pdClock pdutil.Clock - eventCh *chann.DrainableChann[*Event] + pdClock pdutil.Clock + eventCh *chann.DrainableChann[*Event] + checkpointUpdateCh chan struct{} // blockStatusPending keeps the dedupe window local to the maintainer event // queue so duplicate block-status resends do not pile up while an earlier // equivalent event is still pending or being handled. @@ -200,10 +201,11 @@ func NewMaintainer(cfID common.ChangeFeedID, Name: keyspaceName, } m := &Maintainer{ - changefeedID: cfID, - selfNode: selfNode, - eventCh: chann.NewAutoDrainChann[*Event](), - startCheckpointTs: checkpointTs, + changefeedID: cfID, + selfNode: selfNode, + eventCh: chann.NewAutoDrainChann[*Event](), + checkpointUpdateCh: make(chan struct{}, 1), + startCheckpointTs: checkpointTs, controller: NewController(cfID, checkpointTs, taskScheduler, info.Config, ddlSpan, redoDDLSpan, conf.AddTableBatchSize, time.Duration(conf.CheckBalanceInterval), refresher, keyspaceMeta, enableRedo, conf.BalanceMoveBatchSize, info.Epoch), mc: mc, @@ -699,28 +701,30 @@ func (m *Maintainer) calCheckpointTs(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - if !m.initialized.Load() { - log.Warn("can not advance checkpointTs since not bootstrapped", - zap.Stringer("changefeedID", m.changefeedID), - zap.Uint64("checkpointTs", m.getWatermark().CheckpointTs), - zap.Uint64("resolvedTs", m.getWatermark().ResolvedTs)) - break - } + case <-m.checkpointUpdateCh: + } - // first check the online/offline nodes - // we need to check node changed before calculating checkpointTs - // to avoid the case when a node is offline, the node's heartbeat is missing - // while the span in this node still not set to absent, which may cause - // the checkpointTs be advanced incorrectly - m.checkNodeChanged() + if !m.initialized.Load() { + log.Warn("can not advance checkpointTs since not bootstrapped", + zap.Stringer("changefeedID", m.changefeedID), + zap.Uint64("checkpointTs", m.getWatermark().CheckpointTs), + zap.Uint64("resolvedTs", m.getWatermark().ResolvedTs)) + continue + } - // CRITICAL SECTION: Calculate checkpointTs with proper ordering to prevent race condition - newWatermark, canUpdate := m.calculateNewCheckpointTs() - if canUpdate { - m.controller.spanController.AdvanceMaintainerCommittedCheckpointTs(newWatermark.CheckpointTs) - m.setWatermark(*newWatermark) - m.updateMetrics() - } + // first check the online/offline nodes + // we need to check node changed before calculating checkpointTs + // to avoid the case when a node is offline, the node's heartbeat is missing + // while the span in this node still not set to absent, which may cause + // the checkpointTs be advanced incorrectly + m.checkNodeChanged() + + // CRITICAL SECTION: Calculate checkpointTs with proper ordering to prevent race condition + newWatermark, canUpdate := m.calculateNewCheckpointTs() + if canUpdate { + m.controller.spanController.AdvanceMaintainerCommittedCheckpointTs(newWatermark.CheckpointTs) + m.setWatermark(*newWatermark) + m.updateMetrics() } } } @@ -899,6 +903,7 @@ func (m *Maintainer) onHeartbeatRequest(msg *messaging.TargetMessage) { // ATOMIC CHECKPOINT UPDATE: Part 1 of race condition fix // Update checkpointTsByCapture BEFORE processing operator status to ensure atomicity // This works together with calCheckpointTs to prevent incorrect checkpoint advancement + watermarkUpdated := false if req.Watermark != nil { // The sequence increases when a dispatcher status changes, so accept the new watermark // even if the reported checkpoint regresses (new dispatcher might replay from @@ -907,6 +912,7 @@ func (m *Maintainer) onHeartbeatRequest(msg *messaging.TargetMessage) { old, ok := m.checkpointTsByCapture.Get(msg.From) if !ok || req.Watermark.Seq > old.Seq || (req.Watermark.Seq == old.Seq && req.Watermark.CheckpointTs > old.CheckpointTs) { m.checkpointTsByCapture.Set(msg.From, *req.Watermark) + watermarkUpdated = true } // Update last synced ts from all dispatchers. // We don't care about the checkpoint ts of scheduler or barrier here, @@ -944,9 +950,25 @@ func (m *Maintainer) onHeartbeatRequest(msg *messaging.TargetMessage) { // failover self-healing. A late Stopped/Working heartbeat from a closing dispatcher manager // would otherwise mark spans absent or remove/recreate dispatchers after shutdown has begun. m.controller.handleStatus(msg.From, req.Statuses, false) + if watermarkUpdated { + m.notifyCheckpointUpdate() + } return } m.controller.HandleStatus(msg.From, req.Statuses) + if watermarkUpdated { + m.notifyCheckpointUpdate() + } +} + +func (m *Maintainer) notifyCheckpointUpdate() { + if !config.GetGlobalServerConfig().IsLowLatencyMode() { + return + } + select { + case m.checkpointUpdateCh <- struct{}{}: + default: + } } func (m *Maintainer) onError(from node.ID, err *heartbeatpb.RunningError) { diff --git a/maintainer/maintainer_test.go b/maintainer/maintainer_test.go index 242c25af2f..036f381493 100644 --- a/maintainer/maintainer_test.go +++ b/maintainer/maintainer_test.go @@ -738,6 +738,25 @@ func TestMaintainerCalCheckpointTsSkipsInvalidGlobalCheckpoint(t *testing.T) { wg.Wait() } +func TestMaintainerCheckpointUpdateNotification(t *testing.T) { + original := config.GetGlobalServerConfig() + t.Cleanup(func() { + config.StoreGlobalServerConfig(original) + }) + + m := &Maintainer{checkpointUpdateCh: make(chan struct{}, 1)} + cfg := original.Clone() + config.StoreGlobalServerConfig(cfg) + m.notifyCheckpointUpdate() + require.Empty(t, m.checkpointUpdateCh) + + cfg.PerformanceMode = config.PerformanceModeLowLatency + config.StoreGlobalServerConfig(cfg) + m.notifyCheckpointUpdate() + m.notifyCheckpointUpdate() + require.Len(t, m.checkpointUpdateCh, 1) +} + func TestMaintainerHandleRedoMetaTsMessageUsesRedoCheckpointForRedoController(t *testing.T) { m, selfNodeID := newMaintainerForRedoCheckpointCalculationTest(t) m.initialized.Store(true) From 8daf7057d3320d695454f3f3e2ed39085d7edfbc Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 16:55:19 +0800 Subject: [PATCH 4/8] maintainer: report low latency watermarks promptly --- maintainer/maintainer.go | 15 +++++++++++---- maintainer/maintainer_manager.go | 14 +++++++++++++- maintainer/maintainer_test.go | 23 +++++++++++++++++++++++ 3 files changed, 47 insertions(+), 5 deletions(-) diff --git a/maintainer/maintainer.go b/maintainer/maintainer.go index 60f73cdb84..2ede34b127 100644 --- a/maintainer/maintainer.go +++ b/maintainer/maintainer.go @@ -723,7 +723,10 @@ func (m *Maintainer) calCheckpointTs(ctx context.Context) { newWatermark, canUpdate := m.calculateNewCheckpointTs() if canUpdate { m.controller.spanController.AdvanceMaintainerCommittedCheckpointTs(newWatermark.CheckpointTs) - m.setWatermark(*newWatermark) + watermarkChanged := m.setWatermark(*newWatermark) + if watermarkChanged && config.GetGlobalServerConfig().IsLowLatencyMode() && m.statusChanged != nil { + m.statusChanged.Store(true) + } m.updateMetrics() } } @@ -1406,13 +1409,17 @@ func (m *Maintainer) getWatermark() heartbeatpb.Watermark { return res } -func (m *Maintainer) setWatermark(newWatermark heartbeatpb.Watermark) { +func (m *Maintainer) setWatermark(newWatermark heartbeatpb.Watermark) bool { m.watermark.mu.Lock() defer m.watermark.mu.Unlock() - if newWatermark.CheckpointTs != math.MaxUint64 { + changed := false + if newWatermark.CheckpointTs != math.MaxUint64 && newWatermark.CheckpointTs != m.watermark.CheckpointTs { m.watermark.CheckpointTs = newWatermark.CheckpointTs + changed = true } - if newWatermark.ResolvedTs != math.MaxUint64 { + if newWatermark.ResolvedTs != math.MaxUint64 && newWatermark.ResolvedTs != m.watermark.ResolvedTs { m.watermark.ResolvedTs = newWatermark.ResolvedTs + changed = true } + return changed } diff --git a/maintainer/maintainer_manager.go b/maintainer/maintainer_manager.go index a4e25e0e4d..5ac0704911 100644 --- a/maintainer/maintainer_manager.go +++ b/maintainer/maintainer_manager.go @@ -28,6 +28,11 @@ import ( "go.uber.org/zap" ) +const ( + defaultManagerHeartbeatInterval = 200 * time.Millisecond + lowLatencyManagerHeartbeatInterval = 50 * time.Millisecond +) + // Manager is the manager of all changefeed maintainer in a ticdc server, each ticdc server will // start a Manager when the ticdc server is startup. It responsible for: // 1. Handle bootstrap command from coordinator and report all changefeed maintainer status. @@ -125,7 +130,7 @@ func (m *Manager) Name() string { } func (m *Manager) Run(ctx context.Context) error { - ticker := time.NewTicker(time.Millisecond * 200) + ticker := time.NewTicker(managerHeartbeatInterval()) defer ticker.Stop() for { select { @@ -141,6 +146,13 @@ func (m *Manager) Run(ctx context.Context) error { } } +func managerHeartbeatInterval() time.Duration { + if config.GetGlobalServerConfig().IsLowLatencyMode() { + return lowLatencyManagerHeartbeatInterval + } + return defaultManagerHeartbeatInterval +} + func (m *Manager) newCoordinatorTopicMessage(msg messaging.IOTypeT) *messaging.TargetMessage { return messaging.NewSingleTargetMessage( m.coordinatorID, diff --git a/maintainer/maintainer_test.go b/maintainer/maintainer_test.go index 036f381493..e39772dead 100644 --- a/maintainer/maintainer_test.go +++ b/maintainer/maintainer_test.go @@ -757,6 +757,29 @@ func TestMaintainerCheckpointUpdateNotification(t *testing.T) { require.Len(t, m.checkpointUpdateCh, 1) } +func TestManagerHeartbeatInterval(t *testing.T) { + original := config.GetGlobalServerConfig() + t.Cleanup(func() { + config.StoreGlobalServerConfig(original) + }) + + cfg := original.Clone() + config.StoreGlobalServerConfig(cfg) + require.Equal(t, defaultManagerHeartbeatInterval, managerHeartbeatInterval()) + + cfg.PerformanceMode = config.PerformanceModeLowLatency + config.StoreGlobalServerConfig(cfg) + require.Equal(t, lowLatencyManagerHeartbeatInterval, managerHeartbeatInterval()) +} + +func TestMaintainerSetWatermarkReportsChanges(t *testing.T) { + m := &Maintainer{} + m.watermark.Watermark = &heartbeatpb.Watermark{CheckpointTs: 1, ResolvedTs: 1} + require.False(t, m.setWatermark(heartbeatpb.Watermark{CheckpointTs: 1, ResolvedTs: 1})) + require.True(t, m.setWatermark(heartbeatpb.Watermark{CheckpointTs: 2, ResolvedTs: 1})) + require.True(t, m.setWatermark(heartbeatpb.Watermark{CheckpointTs: 2, ResolvedTs: 3})) +} + func TestMaintainerHandleRedoMetaTsMessageUsesRedoCheckpointForRedoController(t *testing.T) { m, selfNodeID := newMaintainerForRedoCheckpointCalculationTest(t) m.initialized.Store(true) From 56c3cc69bb11adf86c64308f9aecabc30948c245 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 18:06:06 +0800 Subject: [PATCH 5/8] logservice: reduce owner resolved ts metric lag --- logservice/coordinator/coordinator.go | 104 ++++++++++++++++----- logservice/coordinator/coordinator_test.go | 95 ++++++++++++++++++- 2 files changed, 174 insertions(+), 25 deletions(-) diff --git a/logservice/coordinator/coordinator.go b/logservice/coordinator/coordinator.go index d557b2fb12..2b0501de9e 100644 --- a/logservice/coordinator/coordinator.go +++ b/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()), + } + } + changefeedState := c.changefeedStates.m[gid] + changefeedState.nodeStates[from] = state.GetResolvedTs() + changefeedState.nodesReportedSinceLastUpdate[from] = struct{}{} + changefeedState.nodeReportPhyTs[from] = pdPhyTs + affectedGIDs[gid] = struct{}{} + } + + if len(affectedGIDs) > 0 { + for gid := range affectedGIDs { + if state, ok := c.changefeedStates.m[gid]; ok { + if len(state.nodeStates) == 0 || + len(state.nodesReportedSinceLastUpdate) != len(state.nodeStates) { + continue + } + if c.updateChangefeedMetrics(state, pdPhyTs, false) { + state.metricsUpdatedSinceLastTick = true + } + clear(state.nodesReportedSinceLastUpdate) } } - c.changefeedStates.m[gid].nodeStates[from] = state.GetResolvedTs() } } @@ -292,29 +327,50 @@ 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 - } - } +func (c *logCoordinator) updateChangefeedMetrics(state *changefeedState, pdPhyTs int64, force bool) bool { + if len(state.nodeStates) == 0 { + return false + } - if minResolvedTs == math.MaxUint64 { - log.Warn("minResolvedTs is MaxUint64, this should not happen", - zap.Stringer("changefeedID", state.cfID)) - continue + minResolvedTs := uint64(math.MaxUint64) + minResolvedTsNode := node.ID("") + for nodeID, resolvedTs := range state.nodeStates { + if resolvedTs < minResolvedTs { + minResolvedTs = resolvedTs + minResolvedTsNode = nodeID } + } - phyResolvedTs := oracle.ExtractPhysical(minResolvedTs) - state.minLogServiceResolvedTs = minResolvedTs - state.resolvedTsGauge.Set(float64(phyResolvedTs)) - lag := float64(pdPhyTs-phyResolvedTs) / 1e3 - state.resolvedTsLagGauge.Set(lag) + 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.metricsInitialized = true + state.resolvedTsGauge.Set(float64(phyResolvedTs)) + lagReferencePhyTs := pdPhyTs + if !force { + if reportPhyTs, ok := state.nodeReportPhyTs[minResolvedTsNode]; ok { + lagReferencePhyTs = reportPhyTs + } } + lag := float64(lagReferencePhyTs-phyResolvedTs) / 1e3 + state.resolvedTsLagGauge.Set(lag) + return true } func (c *logCoordinator) getMinLogServiceResolvedTs(cfID common.ChangeFeedID) uint64 { diff --git a/logservice/coordinator/coordinator_test.go b/logservice/coordinator/coordinator_test.go index 448eb0585c..e72f8bd3eb 100644 --- a/logservice/coordinator/coordinator_test.go +++ b/logservice/coordinator/coordinator_test.go @@ -29,7 +29,7 @@ import ( ) func newLogCoordinatorForTest() *logCoordinator { - c := &logCoordinator{} + c := &logCoordinator{pdClock: pdutil.NewClock4Test()} c.eventStoreStates.m = make(map[node.ID]*logservicepb.EventStoreState) c.nodes.m = make(map[node.ID]*node.Info) c.changefeedStates.m = make(map[common.GID]*changefeedState) @@ -259,6 +259,7 @@ func TestUpdateChangefeedStates(t *testing.T) { require.Equal(t, cfID1, cf1State.cfID) require.Len(t, cf1State.nodeStates, 1) require.Equal(t, uint64(100), cf1State.nodeStates[nodeID1]) + require.Equal(t, uint64(100), cf1State.minLogServiceResolvedTs) require.NotNil(t, cf1State.resolvedTsGauge) require.NotNil(t, cf1State.resolvedTsLagGauge) @@ -268,6 +269,7 @@ func TestUpdateChangefeedStates(t *testing.T) { require.Equal(t, cfID2, cf2State.cfID) require.Len(t, cf2State.nodeStates, 1) require.Equal(t, uint64(110), cf2State.nodeStates[nodeID1]) + require.Equal(t, uint64(110), cf2State.minLogServiceResolvedTs) // 2. Update from node-2 for cf1 states2 := &logservicepb.ChangefeedStates{ @@ -283,6 +285,7 @@ func TestUpdateChangefeedStates(t *testing.T) { require.Len(t, cf1State.nodeStates, 2) require.Equal(t, uint64(100), cf1State.nodeStates[nodeID1]) require.Equal(t, uint64(105), cf1State.nodeStates[nodeID2]) + require.Equal(t, uint64(100), cf1State.minLogServiceResolvedTs) // cf2 state should not change cf2State, ok = c.changefeedStates.m[cfID2.ID()] @@ -304,6 +307,7 @@ func TestUpdateChangefeedStates(t *testing.T) { require.Len(t, cf1State.nodeStates, 2) require.Equal(t, uint64(120), cf1State.nodeStates[nodeID1]) require.Equal(t, uint64(105), cf1State.nodeStates[nodeID2]) + require.Equal(t, uint64(105), cf1State.minLogServiceResolvedTs) // Check cf2 is removed from node-1, and since it's the only node for cf2, cf2 should be removed entirely. _, ok = c.changefeedStates.m[cfID2.ID()] @@ -320,10 +324,99 @@ func TestUpdateChangefeedStates(t *testing.T) { require.True(t, ok) require.Len(t, cf1State.nodeStates, 1) require.Equal(t, uint64(120), cf1State.nodeStates[nodeID1]) + require.Equal(t, uint64(105), cf1State.minLogServiceResolvedTs) _, ok = cf1State.nodeStates[nodeID2] require.False(t, ok) } +func TestUpdateChangefeedStatesRefreshesMetricsImmediately(t *testing.T) { + c := newLogCoordinatorForTest() + mockPDClock := c.pdClock.(*pdutil.Clock4Test) + pdTime := time.Now().Truncate(time.Millisecond) + mockPDClock.SetTS(oracle.GoTimeToTS(pdTime)) + + cfID := common.NewChangefeedID4Test("default", "immediate-metrics") + resolvedTs := oracle.GoTimeToTS(pdTime.Add(-500 * time.Millisecond)) + c.updateChangefeedStates(node.ID("node-1"), &logservicepb.ChangefeedStates{ + States: []*logservicepb.ChangefeedStateEntry{{ + ChangefeedID: cfID.ToPB(), + ResolvedTs: resolvedTs, + }}, + }) + + state := c.changefeedStates.m[cfID.ID()] + require.Equal(t, resolvedTs, state.minLogServiceResolvedTs) + require.Equal(t, float64(oracle.ExtractPhysical(resolvedTs)), testutil.ToFloat64(state.resolvedTsGauge)) + require.InDelta(t, 0.5, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + + mockPDClock.SetTS(oracle.GoTimeToTS(pdTime.Add(time.Second))) + c.reportChangefeedMetrics() + require.InDelta(t, 0.5, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + + newerResolvedTs := oracle.GoTimeToTS(pdTime.Add(-100 * time.Millisecond)) + c.updateChangefeedStates(node.ID("node-2"), &logservicepb.ChangefeedStates{ + States: []*logservicepb.ChangefeedStateEntry{{ + ChangefeedID: cfID.ToPB(), + ResolvedTs: newerResolvedTs, + }}, + }) + require.Equal(t, resolvedTs, state.minLogServiceResolvedTs) + require.InDelta(t, 0.5, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + + c.reportChangefeedMetrics() + require.InDelta(t, 1.5, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) +} + +func TestUpdateChangefeedStatesWaitsForCompleteReportingRound(t *testing.T) { + c := newLogCoordinatorForTest() + mockPDClock := c.pdClock.(*pdutil.Clock4Test) + pdTime := time.Now().Truncate(time.Millisecond) + mockPDClock.SetTS(oracle.GoTimeToTS(pdTime)) + + cfID := common.NewChangefeedID4Test("default", "complete-reporting-round") + oldResolvedTs := oracle.GoTimeToTS(pdTime.Add(-900 * time.Millisecond)) + state := &changefeedState{ + cfID: cfID, + nodeStates: map[node.ID]uint64{ + "node-1": oldResolvedTs, + "node-2": oldResolvedTs, + "node-3": oldResolvedTs, + }, + nodesReportedSinceLastUpdate: make(map[node.ID]struct{}), + nodeReportPhyTs: make(map[node.ID]int64), + minLogServiceResolvedTs: oldResolvedTs, + metricsInitialized: true, + resolvedTsGauge: prometheus.NewGauge(prometheus.GaugeOpts{}), + resolvedTsLagGauge: prometheus.NewGauge(prometheus.GaugeOpts{}), + } + state.resolvedTsGauge.Set(float64(oracle.ExtractPhysical(oldResolvedTs))) + state.resolvedTsLagGauge.Set(0.9) + c.changefeedStates.m[cfID.ID()] = state + + report := func(nodeID node.ID, reportTime time.Time) uint64 { + mockPDClock.SetTS(oracle.GoTimeToTS(reportTime)) + resolvedTs := oracle.GoTimeToTS(reportTime.Add(-200 * time.Millisecond)) + c.updateChangefeedStates(nodeID, &logservicepb.ChangefeedStates{ + States: []*logservicepb.ChangefeedStateEntry{{ + ChangefeedID: cfID.ToPB(), + ResolvedTs: resolvedTs, + }}, + }) + return resolvedTs + } + + newResolvedTs := report("node-1", pdTime) + report("node-1", pdTime) + report("node-2", pdTime.Add(300*time.Millisecond)) + require.Equal(t, oldResolvedTs, state.minLogServiceResolvedTs) + require.InDelta(t, 0.9, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + + report("node-3", pdTime.Add(600*time.Millisecond)) + require.Equal(t, newResolvedTs, state.minLogServiceResolvedTs) + require.InDelta(t, 0.2, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + require.Empty(t, state.nodesReportedSinceLastUpdate) +} + func TestReportMetricsForAffectedChangefeeds(t *testing.T) { c := newLogCoordinatorForTest() mockPDClock := pdutil.NewClock4Test() From 29722837bec290caae325f8b2e01699e9b0d07f6 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Mon, 13 Jul 2026 18:55:56 +0800 Subject: [PATCH 6/8] logservice: simplify low latency resolved ts advancement Keep the low-latency mode focused on setting the logpuller advance interval to zero. Remove the additional batched heap update path and its helper after the 10k-region E2E test showed comparable mean and p95 latency without it. --- logservice/logpuller/region_event_handler.go | 47 +++++++++---------- .../logpuller/regionlock/region_range_lock.go | 14 ------ .../regionlock/region_range_lock_test.go | 22 --------- 3 files changed, 21 insertions(+), 62 deletions(-) diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index cbe7293d02..0973b1df15 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/logservice/logpuller/region_event_handler.go @@ -19,7 +19,6 @@ import ( "github.com/pingcap/kvproto/pkg/cdcpb" "github.com/pingcap/log" - "github.com/pingcap/ticdc/logservice/logpuller/regionlock" "github.com/pingcap/ticdc/pkg/common" "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/pkg/spanz" @@ -134,28 +133,10 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent) handleEventEntries(span, event.mustFirstState(), event.entries) } else if event.resolvedTs != 0 { hasResolved = true - if span.advanceInterval == 0 { - updatedStates := make([]*regionlock.LockedRangeState, 0, len(event.states)) - regionID := uint64(0) - for _, state := range event.states { - if updateRegionResolvedTs(state, event.resolvedTs) { - updatedStates = append(updatedStates, state.region.lockedRangeState) - regionID = state.getRegionID() - } - } - if len(updatedStates) > 0 { - ts := span.rangeLock.UpdateLockedRangeStateHeapBatch(updatedStates) - resolvedTs := advanceSpanResolvedTs(span, regionID, ts) - if resolvedTs > newResolvedTs { - newResolvedTs = resolvedTs - } - } - } else { - for _, state := range event.states { - resolvedTs := handleResolvedTs(span, state, event.resolvedTs) - if resolvedTs > newResolvedTs { - newResolvedTs = resolvedTs - } + for _, state := range event.states { + resolvedTs := handleResolvedTs(span, state, event.resolvedTs) + if resolvedTs > newResolvedTs { + newResolvedTs = resolvedTs } } } else { @@ -377,17 +358,30 @@ func handleEventEntries(span *subscribedSpan, state *regionFeedState, entries *c } func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs uint64) uint64 { - if !updateRegionResolvedTs(state, resolvedTs) { + if state.isStale() || !state.isInitialized() { return 0 } - + state.matcher.tryCleanUnmatchedValue() regionID := state.getRegionID() + lastResolvedTs := state.getLastResolvedTs() + if resolvedTs < lastResolvedTs { + log.Debug("The resolvedTs is fallen back in subscription client", + zap.Uint64("subscriptionID", uint64(state.region.subscribedSpan.subID)), + zap.Uint64("regionID", regionID), + zap.Uint64("resolvedTs", resolvedTs), + zap.Uint64("lastResolvedTs", lastResolvedTs)) + return 0 + } + + state.updateResolvedTs(resolvedTs) + ts := uint64(0) shouldAdvance := false // advanceInterval defaults to 100ms; setting it to 0 means resolving the timestamp as soon as possible. // Note: If a single span contains an extremely large number of regions (e.g., 500k), advanceInterval = 0 may cause performance issues. if span.advanceInterval == 0 { - ts = span.rangeLock.UpdateLockedRangeStateHeapBatch([]*regionlock.LockedRangeState{state.region.lockedRangeState}) + span.rangeLock.UpdateLockedRangeStateHeap(state.region.lockedRangeState) + ts = span.rangeLock.GetHeapMinTs() shouldAdvance = true } else { now := time.Now().UnixMilli() @@ -432,5 +426,6 @@ func handleResolvedTs(span *subscribedSpan, state *regionFeedState, resolvedTs u return ts } } + return 0 } diff --git a/logservice/logpuller/regionlock/region_range_lock.go b/logservice/logpuller/regionlock/region_range_lock.go index 6bd899263f..fbd67e2bd3 100644 --- a/logservice/logpuller/regionlock/region_range_lock.go +++ b/logservice/logpuller/regionlock/region_range_lock.go @@ -420,25 +420,11 @@ func (l *RangeLock) UpdateLockedRangeStateHeap(lockedRangeState *LockedRangeStat l.lockedRangeStateHeap.AddOrUpdate(lockedRangeState) } -// UpdateLockedRangeStateHeapBatch updates multiple locked ranges and returns the minimum resolvedTs. -func (l *RangeLock) UpdateLockedRangeStateHeapBatch(lockedRangeStates []*LockedRangeState) uint64 { - l.mu.Lock() - defer l.mu.Unlock() - - for _, lockedRangeState := range lockedRangeStates { - l.lockedRangeStateHeap.AddOrUpdate(lockedRangeState) - } - return l.getHeapMinTs() -} - // GetHeapMinTs returns the minimum ResolvedTs from the heap. func (l *RangeLock) GetHeapMinTs() uint64 { l.mu.RLock() defer l.mu.RUnlock() - return l.getHeapMinTs() -} -func (l *RangeLock) getHeapMinTs() uint64 { minTs := uint64(math.MaxUint64) if minEntry, ok := l.lockedRangeStateHeap.PeekTop(); ok { minTs = minEntry.ResolvedTs.Load() diff --git a/logservice/logpuller/regionlock/region_range_lock_test.go b/logservice/logpuller/regionlock/region_range_lock_test.go index 4d3c941cef..8325b89356 100644 --- a/logservice/logpuller/regionlock/region_range_lock_test.go +++ b/logservice/logpuller/regionlock/region_range_lock_test.go @@ -332,25 +332,3 @@ func TestRangeLockGetHeapMinTs(t *testing.T) { require.Equal(t, uint64(50), l.GetHeapMinTs()) require.Equal(t, l.ResolvedTs(), l.GetHeapMinTs()) } - -func TestRangeLockUpdateLockedRangeStateHeapBatch(t *testing.T) { - t.Parallel() - - l := NewRangeLock(1, []byte("a"), []byte("z"), 1) - first := l.LockRange(context.Background(), []byte("a"), []byte("m"), 1, 1) - second := l.LockRange(context.Background(), []byte("m"), []byte("z"), 2, 1) - require.Equal(t, LockRangeStatusSuccess, first.Status) - require.Equal(t, LockRangeStatusSuccess, second.Status) - - first.LockedRangeState.ResolvedTs.Store(200) - second.LockedRangeState.ResolvedTs.Store(100) - require.Equal(t, uint64(100), l.UpdateLockedRangeStateHeapBatch([]*LockedRangeState{ - first.LockedRangeState, - second.LockedRangeState, - })) - - second.LockedRangeState.ResolvedTs.Store(300) - require.Equal(t, uint64(200), l.UpdateLockedRangeStateHeapBatch([]*LockedRangeState{ - second.LockedRangeState, - })) -} From f9a099d9933eaeecd8a8e0d85360d0857f73de8b Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Wed, 15 Jul 2026 10:40:39 +0800 Subject: [PATCH 7/8] logservice: report maximum per-node resolved ts lag --- logservice/coordinator/coordinator.go | 22 ++++++++++++++-------- logservice/coordinator/coordinator_test.go | 14 +++++++------- 2 files changed, 21 insertions(+), 15 deletions(-) diff --git a/logservice/coordinator/coordinator.go b/logservice/coordinator/coordinator.go index 2b0501de9e..550f32467d 100644 --- a/logservice/coordinator/coordinator.go +++ b/logservice/coordinator/coordinator.go @@ -341,11 +341,20 @@ func (c *logCoordinator) updateChangefeedMetrics(state *changefeedState, pdPhyTs } minResolvedTs := uint64(math.MaxUint64) - minResolvedTsNode := node.ID("") + var maxNodeLag float64 + hasNodeLag := false for nodeID, resolvedTs := range state.nodeStates { if resolvedTs < minResolvedTs { minResolvedTs = resolvedTs - minResolvedTsNode = nodeID + } + if !force { + if reportPhyTs, ok := state.nodeReportPhyTs[nodeID]; ok { + nodeLag := float64(reportPhyTs-oracle.ExtractPhysical(resolvedTs)) / 1e3 + if !hasNodeLag || nodeLag > maxNodeLag { + maxNodeLag = nodeLag + hasNodeLag = true + } + } } } @@ -362,13 +371,10 @@ func (c *logCoordinator) updateChangefeedMetrics(state *changefeedState, pdPhyTs state.minLogServiceResolvedTs = minResolvedTs state.metricsInitialized = true state.resolvedTsGauge.Set(float64(phyResolvedTs)) - lagReferencePhyTs := pdPhyTs - if !force { - if reportPhyTs, ok := state.nodeReportPhyTs[minResolvedTsNode]; ok { - lagReferencePhyTs = reportPhyTs - } + lag := float64(pdPhyTs-phyResolvedTs) / 1e3 + if !force && hasNodeLag { + lag = maxNodeLag } - lag := float64(lagReferencePhyTs-phyResolvedTs) / 1e3 state.resolvedTsLagGauge.Set(lag) return true } diff --git a/logservice/coordinator/coordinator_test.go b/logservice/coordinator/coordinator_test.go index e72f8bd3eb..2d37a501a6 100644 --- a/logservice/coordinator/coordinator_test.go +++ b/logservice/coordinator/coordinator_test.go @@ -393,9 +393,9 @@ func TestUpdateChangefeedStatesWaitsForCompleteReportingRound(t *testing.T) { state.resolvedTsLagGauge.Set(0.9) c.changefeedStates.m[cfID.ID()] = state - report := func(nodeID node.ID, reportTime time.Time) uint64 { + report := func(nodeID node.ID, reportTime time.Time, lag time.Duration) uint64 { mockPDClock.SetTS(oracle.GoTimeToTS(reportTime)) - resolvedTs := oracle.GoTimeToTS(reportTime.Add(-200 * time.Millisecond)) + resolvedTs := oracle.GoTimeToTS(reportTime.Add(-lag)) c.updateChangefeedStates(nodeID, &logservicepb.ChangefeedStates{ States: []*logservicepb.ChangefeedStateEntry{{ ChangefeedID: cfID.ToPB(), @@ -405,15 +405,15 @@ func TestUpdateChangefeedStatesWaitsForCompleteReportingRound(t *testing.T) { return resolvedTs } - newResolvedTs := report("node-1", pdTime) - report("node-1", pdTime) - report("node-2", pdTime.Add(300*time.Millisecond)) + newResolvedTs := report("node-1", pdTime, 50*time.Millisecond) + report("node-1", pdTime, 50*time.Millisecond) + report("node-2", pdTime.Add(300*time.Millisecond), 180*time.Millisecond) require.Equal(t, oldResolvedTs, state.minLogServiceResolvedTs) require.InDelta(t, 0.9, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) - report("node-3", pdTime.Add(600*time.Millisecond)) + report("node-3", pdTime.Add(600*time.Millisecond), 120*time.Millisecond) require.Equal(t, newResolvedTs, state.minLogServiceResolvedTs) - require.InDelta(t, 0.2, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) + require.InDelta(t, 0.18, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9) require.Empty(t, state.nodesReportedSinceLastUpdate) } From e245a95344ef6072ad6a99916db0924b63d38457 Mon Sep 17 00:00:00 2001 From: dongmen <414110582@qq.com> Date: Fri, 31 Jul 2026 13:11:24 +0800 Subject: [PATCH 8/8] logservice: clarify resolved ts lag reporting --- logservice/coordinator/coordinator.go | 29 +++++++++++++--------- logservice/coordinator/coordinator_test.go | 2 ++ 2 files changed, 19 insertions(+), 12 deletions(-) diff --git a/logservice/coordinator/coordinator.go b/logservice/coordinator/coordinator.go index 550f32467d..cb7335fcab 100644 --- a/logservice/coordinator/coordinator.go +++ b/logservice/coordinator/coordinator.go @@ -303,19 +303,19 @@ func (c *logCoordinator) updateChangefeedStates(from node.ID, states *logservice affectedGIDs[gid] = struct{}{} } - if len(affectedGIDs) > 0 { - for gid := range affectedGIDs { - if state, ok := c.changefeedStates.m[gid]; ok { - if len(state.nodeStates) == 0 || - len(state.nodesReportedSinceLastUpdate) != len(state.nodeStates) { - continue - } - if c.updateChangefeedMetrics(state, pdPhyTs, false) { - state.metricsUpdatedSinceLastTick = true - } - clear(state.nodesReportedSinceLastUpdate) - } + 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) } } @@ -335,6 +335,11 @@ func (c *logCoordinator) reportChangefeedMetrics() { } } +// 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 diff --git a/logservice/coordinator/coordinator_test.go b/logservice/coordinator/coordinator_test.go index 2d37a501a6..7f2b204254 100644 --- a/logservice/coordinator/coordinator_test.go +++ b/logservice/coordinator/coordinator_test.go @@ -406,7 +406,9 @@ func TestUpdateChangefeedStatesWaitsForCompleteReportingRound(t *testing.T) { } newResolvedTs := report("node-1", pdTime, 50*time.Millisecond) + // A repeated report from the same node must not complete the reporting round. report("node-1", pdTime, 50*time.Millisecond) + require.Len(t, state.nodesReportedSinceLastUpdate, 1) report("node-2", pdTime.Add(300*time.Millisecond), 180*time.Millisecond) require.Equal(t, oldResolvedTs, state.minLogServiceResolvedTs) require.InDelta(t, 0.9, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9)