logpuller: simplify region failure handling pipeline

logservice/logpuller/region_failure_handler.go

@@ -23,7 +23,6 @@ import (
 	"github.com/pingcap/ticdc/pkg/metrics"
 	"github.com/tikv/client-go/v2/tikv"
 	"go.uber.org/zap"
-	"golang.org/x/sync/errgroup"
 )
 
 var (
@@ -65,22 +64,44 @@ func (r *regionFailureHandler) Report(errInfo regionErrorInfo) {
 }
 
 func (r *regionFailureHandler) Run(ctx context.Context) error {
-	g, ctx := errgroup.WithContext(ctx)
-	g.Go(func() error { return r.cache.dispatch(ctx) })
-	g.Go(func() error {
+	handleCachedErrors := func() error {
 		for {
-			select {
-			case <-ctx.Done():
-				log.Info("subscription client handle errors and exit")
-				return ctx.Err()
-			case errInfo := <-r.cache.errCh:
+			batch := r.cache.popBatch(errCacheBatchSize)
+			for _, errInfo := range batch {
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				default:
+				}
 				if err := r.handleError(ctx, errInfo); err != nil {
 					return err
 				}
 			}
+			if len(batch) < errCacheBatchSize {
+				return nil
+			}
 		}
-	})
-	return g.Wait()
+	}
+
+	// r.cache.ready() should handle failures promptly in normal flow. The ticker is only a
+	// fallback scan and is not expected to be needed in practice.
+	ticker := time.NewTicker(200 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Info("subscription client handle errors and exit")
+			return ctx.Err()
+		case <-ticker.C:
+			if err := handleCachedErrors(); err != nil {
+				return err
+			}
+		case <-r.cache.ready():
+			if err := handleCachedErrors(); err != nil {
+				return err
+			}
+		}
+	}
 }
 
 func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionErrorInfo) error {
@@ -173,17 +194,15 @@ func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionEr
 type errCache struct {
 	sync.Mutex
 	cache  []regionErrorInfo
-	errCh  chan regionErrorInfo
 	notify chan struct{}
 }
 
-const errCacheDispatchBatchSize = 1024
+const errCacheBatchSize = 1024
 
 func newErrCache() *errCache {
 	return &errCache{
 		cache:  make([]regionErrorInfo, 0, 1024),
-		errCh:  make(chan regionErrorInfo, 4096),
-		notify: make(chan struct{}, 1024),
+		notify: make(chan struct{}, 1),
 	}
 }
 
@@ -217,45 +236,6 @@ func (e *errCache) popBatch(limit int) []regionErrorInfo {
 	return batch
 }
 
-func (e *errCache) dispatchBatch(ctx context.Context, limit int) (int, error) {
-	batch := e.popBatch(limit)
-	for _, errInfo := range batch {
-		select {
-		case <-ctx.Done():
-			log.Info("subscription client dispatch err cache done")
-			return 0, ctx.Err()
-		case e.errCh <- errInfo:
-		}
-	}
-	return len(batch), nil
-}
-
-func (e *errCache) dispatch(ctx context.Context) error {
-	ticker := time.NewTicker(10 * time.Millisecond)
-	defer ticker.Stop()
-	sendToErrCh := func() error {
-		for {
-			n, err := e.dispatchBatch(ctx, errCacheDispatchBatchSize)
-			if err != nil {
-				return err
-			}
-			if n < errCacheDispatchBatchSize {
-				return nil
-			}
-		}
-	}
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-ticker.C:
-			if err := sendToErrCh(); err != nil {
-				return err
-			}
-		case <-e.notify:
-			if err := sendToErrCh(); err != nil {
-				return err
-			}
-		}
-	}
+func (e *errCache) ready() <-chan struct{} {
+	return e.notify
 }

logservice/logpuller/region_failure_handler_test.go

@@ -0,0 +1,144 @@
+// 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"
+	"testing"
+	"time"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/ticdc/heartbeatpb"
+	"github.com/stretchr/testify/require"
+	"github.com/tikv/client-go/v2/tikv"
+)
+
+func newTestRegionErrorInfo(err error) regionErrorInfo {
+	return regionErrorInfo{
+		regionInfo: regionInfo{
+			verID: tikv.NewRegionVerID(1, 1, 1),
+			span:  heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("b")},
+		},
+		err: err,
+	}
+}
+
+func TestErrCachePopBatch(t *testing.T) {
+	mockErrInfo := newTestRegionErrorInfo(errors.New("test error"))
+
+	tests := []struct {
+		name          string
+		cacheLen      int
+		limit         int
+		expectedN     int
+		expectedCache int
+		expectedCalls int
+	}{
+		{
+			name:          "handle all when limit equals cache length",
+			cacheLen:      5,
+			limit:         5,
+			expectedN:     5,
+			expectedCache: 0,
+			expectedCalls: 5,
+		},
+		{
+			name:          "keep remaining cache when limit is smaller",
+			cacheLen:      5,
+			limit:         2,
+			expectedN:     2,
+			expectedCache: 3,
+			expectedCalls: 2,
+		},
+		{
+			name:          "handle all when limit is larger",
+			cacheLen:      5,
+			limit:         10,
+			expectedN:     5,
+			expectedCache: 0,
+			expectedCalls: 5,
+		},
+		{
+			name:          "handle all when limit is zero",
+			cacheLen:      5,
+			limit:         0,
+			expectedN:     5,
+			expectedCache: 0,
+			expectedCalls: 5,
+		},
+		{
+			name:          "handle all when limit is negative",
+			cacheLen:      5,
+			limit:         -1,
+			expectedN:     5,
+			expectedCache: 0,
+			expectedCalls: 5,
+		},
+		{
+			name:          "empty cache",
+			cacheLen:      0,
+			limit:         5,
+			expectedN:     0,
+			expectedCache: 0,
+			expectedCalls: 0,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			errCache := &errCache{
+				cache:  make([]regionErrorInfo, 0, 10),
+				notify: make(chan struct{}, 1),
+			}
+			for i := 0; i < tc.cacheLen; i++ {
+				errCache.add(mockErrInfo)
+			}
+
+			batch := errCache.popBatch(tc.limit)
+			n := len(batch)
+			require.Equal(t, tc.expectedN, n)
+			require.Len(t, batch, tc.expectedCalls)
+			require.Len(t, errCache.cache, tc.expectedCache)
+		})
+	}
+}
+
+func TestRegionFailureHandlerRunDrainsErrCacheWithoutDispatcher(t *testing.T) {
+	handler := newRegionFailureHandler(&subscriptionClient{})
+	for i := 0; i < errCacheBatchSize+5; i++ {
+		handler.cache.add(newTestRegionErrorInfo(&requestCancelledErr{}))
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	runDone := make(chan error, 1)
+	go func() {
+		runDone <- handler.Run(ctx)
+	}()
+
+	require.Eventually(t, func() bool {
+		handler.cache.Lock()
+		defer handler.cache.Unlock()
+		return len(handler.cache.cache) == 0
+	}, time.Second, 10*time.Millisecond)
+
+	cancel()
+	select {
+	case err := <-runDone:
+		require.Equal(t, context.Canceled, errors.Cause(err))
+	case <-time.After(time.Second):
+		t.Fatal("failure handler did not exit after context cancellation")
+	}
+}