logpuller: send CDC scan priority

go.mod

@@ -52,7 +52,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.26
 	github.com/pingcap/errors v0.11.5-0.20260508054701-306e305bcf41
 	github.com/pingcap/failpoint v0.0.0-20240528011301-b51a646c7c86
-	github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278
+	github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c
 	github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9
 	github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5
 	github.com/pingcap/tidb v1.1.0-beta.0.20260604031706-f9faeaf4828f

go.sum

@@ -750,8 +750,8 @@ github.com/pingcap/fn v1.0.0/go.mod h1:u9WZ1ZiOD1RpNhcI42RucFh/lBuzTu6rw88a+oF2Z
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989 h1:surzm05a8C9dN8dIUmo4Be2+pMRb6f55i+UIYrluu2E=
 github.com/pingcap/goleveldb v0.0.0-20191226122134-f82aafb29989/go.mod h1:O17XtbryoCJhkKGbT62+L2OlrniwqiGLSqrmdHCMzZw=
 github.com/pingcap/kvproto v0.0.0-20191211054548-3c6b38ea5107/go.mod h1:WWLmULLO7l8IOcQG+t+ItJ3fEcrL5FxF0Wu+HrMy26w=
-github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278 h1:VV03wSSG2XhOwhF79lvT88Z83lwzMT+aZHXrcCIN9B0=
-github.com/pingcap/kvproto v0.0.0-20260601035955-b2b3bb492278/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE=
+github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c h1:jeuQge1aBQp9VRboYbh+k7CCMlLI+8WT1Zustk5jpHU=
+github.com/pingcap/kvproto v0.0.0-20260707053407-18552472891c/go.mod h1:z6+aAHB7dBkA+LyinEX+48/ImRJ3jag0Hg0c7wkhEvE=
 github.com/pingcap/log v0.0.0-20191012051959-b742a5d432e9/go.mod h1:4rbK1p9ILyIfb6hU7OG2CiWSqMXnp3JMbiaVJ6mvoY8=
 github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuRurdGxZXBz0At+9avep+ub7U1AGYLIMM=
 github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=

logservice/logpuller/priority_task.go

@@ -14,9 +14,9 @@
 package logpuller
 
 import (
-	"fmt"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/tikv/client-go/v2/oracle"
 )
 
@@ -39,7 +39,36 @@ const (
 )
 
 func (t TaskType) String() string {
-	return fmt.Sprintf("%d", t)
+	switch t {
+	case TaskHighPrior:
+		return "high"
+	case TaskLowPrior:
+		return "low"
+	default:
+		return "unknown"
+	}
+}
+
+func (t TaskType) scanPriority() cdcpb.ScanPriority {
+	switch t {
+	case TaskHighPrior:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_HIGH
+	case TaskLowPrior:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_LOW
+	default:
+		return cdcpb.ScanPriority_SCAN_PRIORITY_LOW
+	}
+}
+
+func taskTypeFromScanPriority(priority cdcpb.ScanPriority) TaskType {
+	if priority == cdcpb.ScanPriority_SCAN_PRIORITY_HIGH {
+		return TaskHighPrior
+	}
+	return TaskLowPrior
+}
+
+func normalizeScanPriority(priority cdcpb.ScanPriority) cdcpb.ScanPriority {
+	return taskTypeFromScanPriority(priority).scanPriority()
 }
 
 // PriorityTask is the interface for priority-based tasks

logservice/logpuller/priority_task_test.go

@@ -18,13 +18,23 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/utils/priorityqueue"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 )
 
+func TestTaskTypeScanPriorityMapping(t *testing.T) {
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, TaskHighPrior.scanPriority())
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, TaskLowPrior.scanPriority())
+	require.Equal(t, TaskHighPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_HIGH))
+	require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_LOW))
+	require.Equal(t, TaskLowPrior, taskTypeFromScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN))
+	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_LOW, normalizeScanPriority(cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN))
+}
+
 // TestPriorityCalculationLogic tests the priority calculation logic in isolation
 func TestPriorityCalculationLogic(t *testing.T) {
 	currentTime := time.Now()

logservice/logpuller/region_event_handler.go

@@ -143,6 +143,9 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent)
 			log.Panic("should not reach", zap.Any("event", event), zap.Any("events", events))
 		}
 	}
+	if newResolvedTs > 0 && h.failureHandler != nil && h.failureHandler.client != nil {
+		h.failureHandler.client.maybeEnableRealtimeScanPriority(span, newResolvedTs)
+	}
 	tryAdvanceResolvedTs := func() {
 		if newResolvedTs != 0 {
 			span.advanceResolvedTs(newResolvedTs)

logservice/logpuller/region_failure_handler.go

@@ -86,6 +86,7 @@ func (r *regionFailureHandler) Run(ctx context.Context) error {
 
 func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionErrorInfo) error {
 	err := errors.Cause(errInfo.err)
+	retryPriority := taskTypeFromScanPriority(errInfo.scanPriority)
 	//nolint:errorlint // converting large type switch to errors.As is a significant refactor
 	if _, requestCancelled := err.(*requestCancelledErr); !requestCancelled {
 		log.Debug("cdc region error",
@@ -101,27 +102,27 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr
 		if notLeader := innerErr.GetNotLeader(); notLeader != nil {
 			metricFeedNotLeaderCounter.Inc()
 			r.client.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx)
-			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior)
+			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
 			return nil
 		}
 		if innerErr.GetEpochNotMatch() != nil {
 			metricFeedEpochNotMatchCounter.Inc()
-			r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior)
+			r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority)
 			return nil
 		}
 		if innerErr.GetRegionNotFound() != nil {
 			metricFeedRegionNotFoundCounter.Inc()
-			r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior)
+			r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority)
 			return nil
 		}
 		if innerErr.GetCongested() != nil {
 			metricKvCongestedCounter.Inc()
-			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior)
+			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
 			return nil
 		}
 		if innerErr.GetServerIsBusy() != nil {
 			metricKvIsBusyCounter.Inc()
-			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior)
+			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
 			return nil
 		}
 		if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil {
@@ -140,24 +141,24 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr
 			zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)),
 			zap.Stringer("error", innerErr))
 		metricFeedUnknownErrorCounter.Inc()
-		r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior)
+		r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
 		return nil
 	case *rpcCtxUnavailableErr:
 		metricFeedRPCCtxUnavailable.Inc()
-		r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior)
+		r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority)
 		return nil
 	case *getStoreErr:
 		metricGetStoreErr.Inc()
 		bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff)
 		// cannot get the store the region belongs to, so we need to reload the region.
 		r.client.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err)
-		r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior)
+		r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, retryPriority)
 		return nil
 	case *storeStreamErr:
 		metricStoreSendRequestErr.Inc()
 		bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff)
 		r.client.regionCache.OnSendFail(bo, errInfo.rpcCtx, regionScheduleReload, err)
-		r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior)
+		r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
 		return nil
 	case *requestCancelledErr:
 		// the corresponding subscription has been unsubscribed, just ignore.

logservice/logpuller/region_request_worker.go

@@ -448,6 +448,7 @@ func (s *regionRequestWorker) createRegionRequest(region regionInfo) *cdcpb.Chan
 		EndKey:       region.span.EndKey,
 		ExtraOp:      kvrpcpb.ExtraOp_ReadOldValue,
 		FilterLoop:   region.filterLoop,
+		ScanPriority: normalizeScanPriority(region.scanPriority),
 	}
 }
 

logservice/logpuller/region_request_worker_test.go

@@ -57,6 +57,42 @@ func prepareRegionForSendTest(region regionInfo) regionInfo {
 	return region
 }
 
+func TestCreateRegionRequestScanPriority(t *testing.T) {
+	worker := &regionRequestWorker{
+		client: &subscriptionClient{clusterID: 1},
+	}
+
+	for _, tc := range []struct {
+		name     string
+		priority cdcpb.ScanPriority
+		expected cdcpb.ScanPriority
+	}{
+		{
+			name:     "high",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+		},
+		{
+			name:     "low",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+		},
+		{
+			name:     "unknown defaults to low",
+			priority: cdcpb.ScanPriority_SCAN_PRIORITY_UNKNOWN,
+			expected: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			region := prepareRegionForSendTest(createTestRegionInfo(1, 1))
+			region.scanPriority = tc.priority
+
+			req := worker.createRegionRequest(region)
+			require.Equal(t, tc.expected, req.GetScanPriority())
+		})
+	}
+}
+
 func TestRegionStatesOperation(t *testing.T) {
 	worker := &regionRequestWorker{}
 	worker.requestedRegions.subscriptions = make(map[SubscriptionID]regionFeedStates)

logservice/logpuller/region_state.go

@@ -16,6 +16,7 @@ package logpuller
 import (
 	"sync"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/ticdc/heartbeatpb"
 	"github.com/pingcap/ticdc/logservice/logpuller/regionlock"
 	"github.com/tikv/client-go/v2/tikv"
@@ -46,6 +47,9 @@ type regionInfo struct {
 	// Whether to filter out the value write by cdc itself.
 	// It should be `true` in BDR mode
 	filterLoop bool
+	// scanPriority is sent to TiKV/CSE so remote incremental scan admission can
+	// preserve TiCDC's business priority across retries.
+	scanPriority cdcpb.ScanPriority
 }
 
 func (s *regionInfo) isStopped() bool {
@@ -66,6 +70,7 @@ func newRegionInfo(
 		rpcCtx:         rpcCtx,
 		subscribedSpan: subscribedSpan,
 		filterLoop:     filterLoop,
+		scanPriority:   TaskLowPrior.scanPriority(),
 	}
 }
 

logservice/logpuller/span_registry.go

@@ -71,6 +71,10 @@ type subscribedSpan struct {
 	initialized       atomic.Bool
 	resolvedTsUpdated atomic.Int64
 	resolvedTs        atomic.Uint64
+	// realtimeScanPriority is set after this subscription catches up once.
+	// It is sticky so later recovery scans can protect realtime changefeeds
+	// even if historical catch-up scans temporarily push their lag up again.
+	realtimeScanPriority atomic.Bool
 }
 
 // spanRegistry tracks subscribed spans and owns span-level background maintenance.

logservice/logpuller/subscription_client.go

@@ -36,9 +36,11 @@ import (
 	"github.com/pingcap/ticdc/utils/priorityqueue"
 	"github.com/prometheus/client_golang/prometheus"
 	kvclientv2 "github.com/tikv/client-go/v2/kv"
+	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -262,16 +264,63 @@ func (s *subscriptionClient) Subscribe(
 	s.spanRegistry.Add(rt)
 	s.eventSink.AddPath(rt)
 
+	initialPriority := s.initialScanTaskPriority(startTs)
 	select {
 	case <-s.ctx.Done():
 		log.Warn("subscribes span failed, the subscription client has closed")
-	case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt, filterLoop: rt.filterLoop, priority: TaskLowPrior}:
-		log.Info("subscribes span done", zap.Uint64("subscriptionID", uint64(subID)),
+	case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: rt, filterLoop: rt.filterLoop, priority: initialPriority}:
+		log.Info("subscribes span done",
+			zap.Uint64("subscriptionID", uint64(subID)),
 			zap.Int64("tableID", span.TableID), zap.Uint64("startTs", startTs),
+			zap.String("initialScanPriority", initialPriority.String()),
 			zap.String("startKey", spanz.HexKey(span.StartKey)), zap.String("endKey", spanz.HexKey(span.EndKey)))
 	}
 }
 
+func (s *subscriptionClient) initialScanTaskPriority(startTs uint64) TaskType {
+	if s.isTsCloseToCurrent(startTs) {
+		return TaskHighPrior
+	}
+	return TaskLowPrior
+}
+
+func (s *subscriptionClient) oldStartTsScanLowPriorityThreshold() time.Duration {
+	threshold := time.Duration(config.GetGlobalServerConfig().Debug.Puller.OldStartTsScanLowPriorityThreshold)
+	if threshold > 0 {
+		return threshold
+	}
+	return config.DefaultOldStartTsScanLowPriorityThreshold
+}
+
+func (s *subscriptionClient) isTsCloseToCurrent(ts uint64) bool {
+	if ts == 0 {
+		return false
+	}
+	return s.pdClock.CurrentTime().Sub(oracle.GetTimeFromTS(ts)) <= s.oldStartTsScanLowPriorityThreshold()
+}
+
+func (s *subscriptionClient) maybeEnableRealtimeScanPriority(span *subscribedSpan, resolvedTs uint64) {
+	if span == nil || !span.initialized.Load() || span.realtimeScanPriority.Load() {
+		return
+	}
+	if !s.isTsCloseToCurrent(resolvedTs) {
+		return
+	}
+	if span.realtimeScanPriority.CompareAndSwap(false, true) {
+		log.Info("subscription client enables realtime scan priority",
+			zap.Uint64("subscriptionID", uint64(span.subID)),
+			zap.Uint64("resolvedTs", resolvedTs),
+			zap.Duration("threshold", s.oldStartTsScanLowPriorityThreshold()))
+	}
+}
+
+func (s *subscriptionClient) effectiveScanTaskPriority(subscribedSpan *subscribedSpan, priority TaskType) TaskType {
+	if subscribedSpan != nil && subscribedSpan.realtimeScanPriority.Load() {
+		return TaskHighPrior
+	}
+	return priority
+}
+
 // Unsubscribe the given table span. All covered regions will be deregistered asynchronously.
 // NOTE: `span.TableID` must be set correctly.
 func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) {
@@ -632,6 +681,8 @@ func (s *subscriptionClient) divideSpanAndScheduleRegionRequests(
 // scheduleRegionRequest locks the region's range and send the region to regionTaskQueue,
 // which will be handled by handleRegions.
 func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region regionInfo, priority TaskType) {
+	priority = s.effectiveScanTaskPriority(region.subscribedSpan, priority)
+	region.scanPriority = priority.scanPriority()
 	lockRangeResult := region.subscribedSpan.rangeLock.LockRange(
 		ctx, region.span.StartKey, region.span.EndKey, region.verID.GetID(), region.verID.GetVer())
 
@@ -643,6 +694,18 @@ func (s *subscriptionClient) scheduleRegionRequest(ctx context.Context, region r
 	case regionlock.LockRangeStatusSuccess:
 		region.lockedRangeState = lockRangeResult.LockedRangeState
 		s.regionTaskQueue.Push(NewRegionPriorityTask(priority, region, s.pdClock.CurrentTS()))
+		if log.GetLevel() <= zapcore.DebugLevel {
+			log.Debug("cdc region scan task enqueued",
+				zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)),
+				zap.Int64("tableID", region.subscribedSpan.span.TableID),
+				zap.Uint64("startTs", region.subscribedSpan.startTs),
+				zap.Uint64("regionID", region.verID.GetID()),
+				zap.Uint64("regionEpochVersion", region.verID.GetVer()),
+				zap.Uint64("regionEpochConfVer", region.verID.GetConfVer()),
+				zap.String("priority", priority.String()),
+				zap.String("scanPriority", region.scanPriority.String()),
+				zap.String("span", common.FormatTableSpan(&region.span)))
+		}
 	case regionlock.LockRangeStatusStale:
 		for _, r := range lockRangeResult.RetryRanges {
 			s.scheduleRangeRequest(ctx, r, region.subscribedSpan, region.filterLoop, priority)
@@ -658,6 +721,7 @@ func (s *subscriptionClient) scheduleRangeRequest(
 	filterLoop bool,
 	priority TaskType,
 ) {
+	priority = s.effectiveScanTaskPriority(subscribedSpan, priority)
 	select {
 	case <-ctx.Done():
 	case s.rangeTaskCh <- rangeTask{span: span, subscribedSpan: subscribedSpan, filterLoop: filterLoop, priority: priority}: