logpuller: keep bootstrap requests flow-controlled

logservice/logpuller/region_failure_handler.go

@@ -138,12 +138,12 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr
 		}
 		if innerErr.GetCongested() != nil {
 			metricKvCongestedCounter.Inc()
-			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
+			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior)
 			return nil
 		}
 		if innerErr.GetServerIsBusy() != nil {
 			metricKvIsBusyCounter.Inc()
-			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, retryPriority)
+			r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior)
 			return nil
 		}
 		if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil {

logservice/logpuller/region_req_cache.go

@@ -61,8 +61,9 @@ type requestCache struct {
 	}
 
 	// pendingCount is a flow control slot counter.
-	// A slot is acquired when a request is successfully enqueued into pendingQueue (see add),
+	// A slot is acquired before publishing a request into pendingQueue (see add),
 	// and is released when the request is finished/removed (resolve/markStopped/markDone/clear).
+	// add releases the slot itself if queue publication fails.
 	// pop and markSent don't change it. If markSent overwrites an existing request for the same region,
 	// it will release a slot for the replaced request to avoid leaking pendingCount.
 	pendingCount atomic.Int64
@@ -92,27 +93,32 @@ func newRequestCache(maxPendingCount int) *requestCache {
 }
 
 // add adds a new region request to the cache
-// It blocks if pendingCount >= maxPendingCount until there's space or ctx is cancelled
+// Normal data requests are limited to maxPendingCount. Forced data requests can
+// use one additional slot, while stop/control requests keep their existing bypass.
 func (c *requestCache) add(ctx context.Context, region regionInfo, force bool) (bool, error) {
 	start := time.Now()
 	ticker := time.NewTicker(addReqRetryInterval)
 	defer ticker.Stop()
 	addReqRetryLimit := addReqRetryLimit
 
 	for {
-		current := c.pendingCount.Load()
-		if current < c.maxPendingCount || force {
-			// Try to add the request
+		limit := c.maxPendingCount
+		if force {
+			limit++
+		}
+		if c.tryAcquireSlot(limit, region.isStopped()) {
+			// Try to publish the request after reserving its slot.
 			req := newRegionReq(region)
 			select {
 			case <-ctx.Done():
+				c.markDone()
 				return false, ctx.Err()
 			case c.pendingQueue <- req:
-				c.pendingCount.Inc()
 				cost := time.Since(start)
 				metrics.SubscriptionClientAddRegionRequestDuration.Observe(cost.Seconds())
 				return true, nil
 			case <-ticker.C:
+				c.markDone()
 				addReqRetryLimit--
 				if addReqRetryLimit <= 0 {
 					return false, nil
@@ -137,6 +143,18 @@ func (c *requestCache) add(ctx context.Context, region regionInfo, force bool) (
 	}
 }
 
+func (c *requestCache) tryAcquireSlot(limit int64, bypassLimit bool) bool {
+	for {
+		current := c.pendingCount.Load()
+		if !bypassLimit && current >= limit {
+			return false
+		}
+		if c.pendingCount.CompareAndSwap(current, current+1) {
+			return true
+		}
+	}
+}
+
 // pop gets the next pending request.
 // Note: it doesn't change pendingCount. The slot acquired in add() should be released later
 // (e.g. resolve/markStopped/markDone).

logservice/logpuller/region_req_cache_test.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/ticdc/heartbeatpb"
+	"github.com/pingcap/ticdc/logservice/logpuller/regionlock"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/tikv"
 )
@@ -38,7 +39,9 @@ func createTestRegionInfo(subID SubscriptionID, regionID uint64) regionInfo {
 		span:    span,
 	}
 
-	return newRegionInfo(verID, span, nil, subscribedSpan, false)
+	region := newRegionInfo(verID, span, nil, subscribedSpan, false)
+	region.lockedRangeState = &regionlock.LockedRangeState{}
+	return region
 }
 
 func TestRequestCacheAdd_NormalCase(t *testing.T) {
@@ -77,14 +80,7 @@ func TestRequestCacheAdd_ForceFlag(t *testing.T) {
 	require.False(t, ok)
 	require.NoError(t, err)
 
-	// With force=true, it should still fail because the channel is full
-	// The force flag only bypasses the pendingCount check, not the channel capacity
-	region3 := createTestRegionInfo(1, 3)
-	ok, err = cache.add(ctx, region3, true)
-	require.False(t, ok)
-	require.NoError(t, err)
-
-	// consume the pending queue ann add with force
+	// Move the normal request to sentRequests so the pending queue has room.
 	req, err := cache.pop(ctx)
 	require.NoError(t, err)
 	require.NotNil(t, req)
@@ -93,15 +89,91 @@ func TestRequestCacheAdd_ForceFlag(t *testing.T) {
 	cache.markSent(req)
 	require.Equal(t, 1, cache.getPendingCount())
 
+	// A forced data request can use one extra slot.
+	region3 := createTestRegionInfo(1, 3)
 	ok, err = cache.add(ctx, region3, true)
 	require.True(t, ok)
 	require.NoError(t, err)
-	// It is 2 since region1 is unresolved
 	require.Equal(t, 2, cache.getPendingCount())
 
-	// resolve region1
-	cache.resolve(region1.subscribedSpan.subID, region1.verID.GetID())
+	// No additional forced data request can exceed the N+1 ceiling.
+	req, err = cache.pop(ctx)
+	require.NoError(t, err)
+	cache.markSent(req)
+	region4 := createTestRegionInfo(1, 4)
+	ok, err = cache.add(ctx, region4, true)
+	require.False(t, ok)
+	require.NoError(t, err)
+	require.Equal(t, 2, cache.getPendingCount())
+
+	// Stop/control requests keep their existing bypass and remain accounted.
+	stopRegion := createTestRegionInfo(2, 5)
+	stopRegion.lockedRangeState = nil
+	ok, err = cache.add(ctx, stopRegion, true)
+	require.True(t, ok)
+	require.NoError(t, err)
+	require.Equal(t, 3, cache.getPendingCount())
+
+	stopReq, err := cache.pop(ctx)
+	require.NoError(t, err)
+	cache.markSent(stopReq)
+	cache.markStopped(stopReq.regionInfo.subscribedSpan.subID, stopReq.regionInfo.verID.GetID())
+	require.Equal(t, 2, cache.getPendingCount())
+
+	require.True(t, cache.resolve(region1.subscribedSpan.subID, region1.verID.GetID()))
 	require.Equal(t, 1, cache.getPendingCount())
+	ok, err = cache.add(ctx, region4, true)
+	require.True(t, ok)
+	require.NoError(t, err)
+	require.Equal(t, 2, cache.getPendingCount())
+}
+
+func TestRequestCacheAddRollsBackReservedSlot(t *testing.T) {
+	cache := newRequestCache(1)
+	cache.pendingQueue <- newRegionReq(createTestRegionInfo(1, 1))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	ok, err := cache.add(ctx, createTestRegionInfo(1, 2), false)
+	require.False(t, ok)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Equal(t, 0, cache.getPendingCount())
+}
+
+func TestRequestCacheConcurrentForcedAddsStayWithinCeiling(t *testing.T) {
+	const normalLimit = 10
+	cache := newRequestCache(normalLimit)
+	ctx := context.Background()
+
+	for i := range normalLimit {
+		ok, err := cache.add(ctx, createTestRegionInfo(1, uint64(i+1)), false)
+		require.True(t, ok)
+		require.NoError(t, err)
+	}
+	for range normalLimit {
+		req, err := cache.pop(ctx)
+		require.NoError(t, err)
+		cache.markSent(req)
+	}
+
+	const addCount = 20
+	results := make(chan bool, addCount)
+	for i := range addCount {
+		go func(regionID uint64) {
+			ok, err := cache.add(ctx, createTestRegionInfo(1, regionID), true)
+			require.NoError(t, err)
+			results <- ok
+		}(uint64(normalLimit + i + 1))
+	}
+
+	successes := 0
+	for range addCount {
+		if <-results {
+			successes++
+		}
+	}
+	require.Equal(t, 1, successes)
+	require.Equal(t, normalLimit+1, cache.getPendingCount())
 }
 
 func TestRequestCacheAdd_ContextCancellation(t *testing.T) {

logservice/logpuller/scan_priority_test.go

@@ -141,11 +141,86 @@ func TestScanPriorityUsesRestoredRegionProgress(t *testing.T) {
 	retryRegion := newRegionInfo(tikv.NewRegionVerID(1, 1, 2), rawSpan, nil, span, false)
 	client.scheduleRegionRequest(context.Background(), retryRegion, TaskLowPrior)
 	retryTask := popRegionPriorityTask(t, client.regionTaskQueue)
-	require.Equal(t, TaskHighPrior, retryTask.taskType)
+	require.Equal(t, TaskLowPrior, retryTask.taskType)
 	require.Equal(t, cdcpb.ScanPriority_SCAN_PRIORITY_HIGH, retryTask.GetRegionInfo().scanPriority)
 	require.False(t, span.priorityPolicy.everCaughtUp.Load())
 }
 
+func TestScheduleRegionRequestSeparatesRemoteAndLocalPriority(t *testing.T) {
+	currentTime := time.Date(2026, time.June, 27, 12, 0, 0, 0, time.UTC)
+	currentTs := oracle.GoTimeToTS(currentTime)
+
+	for _, tc := range []struct {
+		name           string
+		priorRemote    cdcpb.ScanPriority
+		inheritedLocal TaskType
+		startTs        uint64
+		expectedRemote cdcpb.ScanPriority
+		expectedLocal  TaskType
+	}{
+		{
+			name:           "busy retry preserves remote high",
+			priorRemote:    cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			inheritedLocal: TaskLowPrior,
+			startTs:        oracle.GoTimeToTS(currentTime.Add(-time.Hour)),
+			expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			expectedLocal:  TaskLowPrior,
+		},
+		{
+			name:           "repair is high locally and remotely",
+			priorRemote:    cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			inheritedLocal: TaskHighPrior,
+			startTs:        oracle.GoTimeToTS(currentTime.Add(-time.Hour)),
+			expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			expectedLocal:  TaskHighPrior,
+		},
+		{
+			name:           "recent bootstrap is only high remotely",
+			priorRemote:    cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			inheritedLocal: TaskLowPrior,
+			startTs:        oracle.GoTimeToTS(currentTime.Add(-time.Minute)),
+			expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_HIGH,
+			expectedLocal:  TaskLowPrior,
+		},
+		{
+			name:           "old bootstrap stays low",
+			priorRemote:    cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			inheritedLocal: TaskLowPrior,
+			startTs:        oracle.GoTimeToTS(currentTime.Add(-time.Hour)),
+			expectedRemote: cdcpb.ScanPriority_SCAN_PRIORITY_LOW,
+			expectedLocal:  TaskLowPrior,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			pdClock := pdutil.NewClock4Test()
+			pdClock.(*pdutil.Clock4Test).SetTS(currentTs)
+			client := &subscriptionClient{
+				pdClock:         pdClock,
+				regionTaskQueue: priorityqueue.New[PriorityTask](),
+			}
+			rawSpan := heartbeatpb.TableSpan{
+				TableID:  1,
+				StartKey: []byte("a"),
+				EndKey:   []byte("z"),
+			}
+			span := &subscribedSpan{
+				subID:          SubscriptionID(1),
+				span:           rawSpan,
+				startTs:        tc.startTs,
+				rangeLock:      regionlock.NewRangeLock(1, rawSpan.StartKey, rawSpan.EndKey, tc.startTs),
+				priorityPolicy: newScanPriorityPolicy(pdClock, 30*time.Minute),
+			}
+			region := newRegionInfo(tikv.NewRegionVerID(1, 1, 1), rawSpan, nil, span, false)
+			region.scanPriority = tc.priorRemote
+
+			client.scheduleRegionRequest(context.Background(), region, tc.inheritedLocal)
+			task := popRegionPriorityTask(t, client.regionTaskQueue)
+			require.Equal(t, tc.expectedLocal, task.taskType)
+			require.Equal(t, tc.expectedRemote, task.GetRegionInfo().scanPriority)
+		})
+	}
+}
+
 func popRegionPriorityTask(
 	t *testing.T,
 	queue *priorityqueue.PriorityQueue[PriorityTask],

logservice/logpuller/subscription_client.go

@@ -19,6 +19,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/kvproto/pkg/cdcpb"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/heartbeatpb"
@@ -652,13 +653,17 @@ func (s *subscriptionClient) scheduleRegionRequest(
 	case regionlock.LockRangeStatusSuccess:
 		region.lockedRangeState = lockRangeResult.LockedRangeState
 		currentTs := s.pdClock.CurrentTS()
-		priority := region.subscribedSpan.priorityPolicy.resolve(
-			inheritedPriority,
+		remoteBase := inheritedPriority
+		if region.scanPriority == cdcpb.ScanPriority_SCAN_PRIORITY_HIGH {
+			remoteBase = TaskHighPrior
+		}
+		remotePriority := region.subscribedSpan.priorityPolicy.resolve(
+			remoteBase,
 			region.resolvedTs(),
 			oracle.GetTimeFromTS(currentTs),
 		)
-		region.scanPriority = priority.scanPriority()
-		s.regionTaskQueue.Push(NewRegionPriorityTask(priority, region, currentTs))
+		region.scanPriority = remotePriority.scanPriority()
+		s.regionTaskQueue.Push(NewRegionPriorityTask(inheritedPriority, region, currentTs))
 		if log.GetLevel() <= zapcore.DebugLevel {
 			log.Debug("cdc region scan task enqueued",
 				zap.Uint64("subscriptionID", uint64(region.subscribedSpan.subID)),
@@ -667,7 +672,7 @@ func (s *subscriptionClient) scheduleRegionRequest(
 				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("priority", inheritedPriority.String()),
 				zap.String("scanPriority", region.scanPriority.String()),
 				zap.String("span", common.FormatTableSpan(&region.span)))
 		}