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())
+}
diff --git a/logservice/coordinator/coordinator.go b/logservice/coordinator/coordinator.go
index d557b2fb12..cb7335fcab 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()),
 			}
 		}
-		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 {
diff --git a/logservice/coordinator/coordinator_test.go b/logservice/coordinator/coordinator_test.go
index 448eb0585c..7f2b204254 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,101 @@ 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, lag time.Duration) uint64 {
+		mockPDClock.SetTS(oracle.GoTimeToTS(reportTime))
+		resolvedTs := oracle.GoTimeToTS(reportTime.Add(-lag))
+		c.updateChangefeedStates(nodeID, &logservicepb.ChangefeedStates{
+			States: []*logservicepb.ChangefeedStateEntry{{
+				ChangefeedID: cfID.ToPB(),
+				ResolvedTs:   resolvedTs,
+			}},
+		})
+		return resolvedTs
+	}
+
+	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)
+
+	report("node-3", pdTime.Add(600*time.Millisecond), 120*time.Millisecond)
+	require.Equal(t, newResolvedTs, state.minLogServiceResolvedTs)
+	require.InDelta(t, 0.18, testutil.ToFloat64(state.resolvedTsLagGauge), 1e-9)
+	require.Empty(t, state.nodesReportedSinceLastUpdate)
+}
+
 func TestReportMetricsForAffectedChangefeeds(t *testing.T) {
 	c := newLogCoordinatorForTest()
 	mockPDClock := pdutil.NewClock4Test()
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/maintainer/maintainer.go b/maintainer/maintainer.go
index 1465e9326e..2ede34b127 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,33 @@ 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)
+			watermarkChanged := m.setWatermark(*newWatermark)
+			if watermarkChanged && config.GetGlobalServerConfig().IsLowLatencyMode() && m.statusChanged != nil {
+				m.statusChanged.Store(true)
 			}
+			m.updateMetrics()
 		}
 	}
 }
@@ -899,6 +906,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 +915,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 +953,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) {
@@ -1384,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 242c25af2f..e39772dead 100644
--- a/maintainer/maintainer_test.go
+++ b/maintainer/maintainer_test.go
@@ -738,6 +738,48 @@ 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 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)
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()