[WIP] log puller optimization

logservice/logpuller/errors.go

@@ -38,6 +38,17 @@ func (e *rpcCtxUnavailableErr) Error() string {
 		e.verID.GetID(), e.verID.GetVer(), e.verID.GetConfVer())
 }
 
+type rpcCtxChangedError struct {
+	verID tikv.RegionVerID
+	from  string
+	to    string
+}
+
+func (e *rpcCtxChangedError) Error() string {
+	return fmt.Sprintf("rpcCtx for region %v changed from %s to %s. ver:%v, confver:%v",
+		e.verID.GetID(), e.from, e.to, e.verID.GetVer(), e.verID.GetConfVer())
+}
+
 type getStoreErr struct{}
 
 func (e *getStoreErr) Error() string { return "get store error" }

logservice/logpuller/region_event_handler.go

@@ -87,7 +87,8 @@ func (event regionEvent) mustFirstState() *regionFeedState {
 }
 
 type regionEventHandler struct {
-	subClient *subscriptionClient
+	eventSink      *regionEventSink
+	failureHandler *regionFailureHandler
 }
 
 func (h *regionEventHandler) Path(event regionEvent) SubscriptionID {
@@ -159,7 +160,7 @@ func (h *regionEventHandler) Handle(span *subscribedSpan, events ...regionEvent)
 			metricConsumeKVEventsCallbackDurationAdvanceResolvedTs.Observe(time.Since(start).Seconds())
 
 			start = time.Now()
-			h.subClient.wakeSubscription(span.subID)
+			h.eventSink.Wake(span.subID)
 			metricConsumeKVEventsCallbackDurationWakeSubscription.Observe(time.Since(start).Seconds())
 		})
 		// if not await, the wake callback will not be called, we need clear the cache manually.
@@ -254,8 +255,8 @@ func (h *regionEventHandler) handleRegionError(state *regionFeedState) {
 			zap.Error(err))
 	}
 	if stepsToRemoved {
-		worker.takeRegionState(SubscriptionID(state.requestID), state.getRegionID())
-		h.subClient.onRegionFail(newRegionErrorInfo(state.getRegionInfo(), err))
+		worker.tracker.RemoveRegion(SubscriptionID(state.requestID), state.getRegionID())
+		h.failureHandler.Report(newRegionErrorInfo(state.getRegionInfo(), err))
 	}
 }
 

logservice/logpuller/region_event_handler_test.go

@@ -79,6 +79,7 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) {
 
 	worker := &regionRequestWorker{
 		requestCache: &requestCache{},
+		tracker:      newRegionTracker(0),
 	}
 	region := newRegionInfo(
 		tikv.RegionVerID{},
@@ -88,8 +89,7 @@ func TestHandleEventEntryEventOutOfOrder(t *testing.T) {
 		false,
 	)
 	region.lockedRangeState = &regionlock.LockedRangeState{}
-	state := newRegionFeedState(region, 1, worker)
-	state.start()
+	state := newRegionFeedState(region, 1, worker, nil)
 
 	// Receive prewrite2 with empty value.
 	{
@@ -218,9 +218,9 @@ func TestHandleResolvedTs(t *testing.T) {
 	subID1 := SubscriptionID(1)
 	worker := &regionRequestWorker{
 		requestCache: &requestCache{},
+		tracker:      newRegionTracker(0),
 	}
-	state1 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(1, 1, 1)}, uint64(subID1), worker)
-	state1.start()
+	state1 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(1, 1, 1)}, uint64(subID1), worker, nil)
 	{
 		span := heartbeatpb.TableSpan{
 			TableID:  100,
@@ -243,8 +243,7 @@ func TestHandleResolvedTs(t *testing.T) {
 	}
 
 	subID2 := SubscriptionID(2)
-	state2 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(2, 2, 2)}, uint64(subID2), worker)
-	state2.start()
+	state2 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(2, 2, 2)}, uint64(subID2), worker, nil)
 	{
 		span := heartbeatpb.TableSpan{
 			TableID:  100,
@@ -267,8 +266,7 @@ func TestHandleResolvedTs(t *testing.T) {
 	}
 
 	subID3 := SubscriptionID(3)
-	state3 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(3, 3, 3)}, uint64(subID3), worker)
-	state3.start()
+	state3 := newRegionFeedState(regionInfo{verID: tikv.NewRegionVerID(3, 3, 3)}, uint64(subID3), worker, nil)
 	{
 		span := heartbeatpb.TableSpan{
 			TableID:  100,
@@ -361,16 +359,18 @@ func TestHandleResolvedTsThrottled(t *testing.T) {
 		advanceInterval: 100,
 	}
 	span.lastAdvanceTime.Store(0)
+	worker := &regionRequestWorker{}
+	worker.tracker = newRegionTracker(0)
 	state := newRegionFeedState(
 		regionInfo{
 			verID:            tikv.NewRegionVerID(1, 1, 1),
 			subscribedSpan:   span,
 			lockedRangeState: res1.LockedRangeState,
 		},
 		1,
+		worker,
 		nil,
 	)
-	state.start()
 
 	require.Equal(t, uint64(200), handleResolvedTs(span, state, 300))
 }

logservice/logpuller/region_event_sink.go

@@ -0,0 +1,133 @@
+// 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 logpuller
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/ticdc/utils/dynstream"
+	"go.uber.org/zap"
+)
+
+type regionEventSink struct {
+	ctx context.Context
+	ds  dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler]
+
+	mu     sync.Mutex
+	cond   *sync.Cond
+	paused atomic.Bool
+}
+
+func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandler) *regionEventSink {
+	sink := &regionEventSink{ctx: ctx}
+	sink.cond = sync.NewCond(&sink.mu)
+
+	option := dynstream.NewOption()
+	// Note: it is max batch size of the kv sent from tikv(not committed rows)
+	option.BatchCount = 1024
+	// TODO: Set `UseBuffer` to true until we refactor the `regionEventHandler.Handle` method so that it doesn't call any method of the dynamic stream. Currently, if `UseBuffer` is set to false, there will be a deadlock:
+	// 	ds.handleLoop fetch events from `ch` -> regionEventHandler.Handle -> ds.RemovePath -> send event to `ch`
+	option.UseBuffer = true
+	option.EnableMemoryControl = true
+	ds := dynstream.NewParallelDynamicStream(
+		"log-puller",
+		&regionEventHandler{eventSink: sink, failureHandler: failureHandler},
+		option,
+	)
+	ds.Start()
+	sink.ds = ds
+	return sink
+}
+
+func (s *regionEventSink) AddPath(rt *subscribedSpan) {
+	areaSetting := dynstream.NewAreaSettingsWithMaxPendingSize(1*1024*1024*1024, dynstream.MemoryControlForPuller, "logPuller") // 1GB
+	if err := s.ds.AddPath(rt.subID, rt, areaSetting); err != nil {
+		log.Warn("subscription client add path failed",
+			zap.Uint64("subscriptionID", uint64(rt.subID)),
+			zap.Error(err))
+	}
+}
+
+func (s *regionEventSink) RemovePath(subID SubscriptionID) error {
+	return s.ds.RemovePath(subID)
+}
+
+func (s *regionEventSink) Wake(subID SubscriptionID) {
+	s.ds.Wake(subID)
+}
+
+func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) {
+	// fast path
+	if !s.paused.Load() {
+		s.ds.Push(subID, event)
+		return
+	}
+	// slow path: wait until paused is false
+	s.mu.Lock()
+	for s.paused.Load() {
+		select {
+		case <-s.ctx.Done():
+			s.mu.Unlock()
+			return
+		default:
+			s.cond.Wait()
+		}
+	}
+	s.mu.Unlock()
+	s.ds.Push(subID, event)
+}
+
+func (s *regionEventSink) Run(ctx context.Context) error {
+	return s.handleFeedback(ctx)
+}
+
+func (s *regionEventSink) handleFeedback(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case feedback := <-s.ds.Feedback():
+			switch feedback.FeedbackType {
+			case dynstream.PauseArea:
+				s.mu.Lock()
+				s.paused.Store(true)
+				s.mu.Unlock()
+				log.Info("subscription client pause push region event")
+			case dynstream.ResumeArea:
+				s.mu.Lock()
+				s.paused.Store(false)
+				s.cond.Broadcast()
+				s.mu.Unlock()
+				log.Info("subscription client resume push region event")
+			case dynstream.ReleasePath, dynstream.ResumePath:
+				// Ignore it, because it is no need to pause and resume a path in puller.
+			}
+		}
+	}
+}
+
+func (s *regionEventSink) Metrics() dynstream.Metrics[int, SubscriptionID] {
+	return s.ds.GetMetrics()
+}
+
+func (s *regionEventSink) Close() {
+	s.mu.Lock()
+	s.paused.Store(false)
+	s.cond.Broadcast()
+	s.mu.Unlock()
+	s.ds.Close()
+}