diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index e216fef714..df192bcf76 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/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 } diff --git a/logservice/logpuller/region_failure_handler_test.go b/logservice/logpuller/region_failure_handler_test.go new file mode 100644 index 0000000000..ba3ab08d40 --- /dev/null +++ b/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") + } +} diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 95d0921f4d..3c9d1fa583 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -589,150 +589,6 @@ func TestSubscriptionWithFailedTiKV(t *testing.T) { } } -// TestErrCacheDispatchWithFullChannelAndCanceledContext tests that when errCh is full -// and context is canceled, the dispatch method doesn't get stuck. -func TestErrCacheDispatchWithFullChannelAndCanceledContext(t *testing.T) { - // Create errCache with a small errCh to easily fill it up - errCache := &errCache{ - cache: make([]regionErrorInfo, 0, 10), - errCh: make(chan regionErrorInfo, 2), // Small buffer to easily fill - notify: make(chan struct{}, 10), - } - - // Create a mock regionErrorInfo - mockErrInfo := regionErrorInfo{ - regionInfo: regionInfo{ - verID: tikv.NewRegionVerID(1, 1, 1), - span: heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("b")}, - }, - err: errors.New("test error"), - } - - // Fill up the errCh channel to make it full - errCache.errCh <- mockErrInfo - errCache.errCh <- mockErrInfo - - // Add some errors to the cache - for i := 0; i < 5; i++ { - errCache.add(mockErrInfo) - } - - // Create a context that will be canceled - ctx, cancel := context.WithCancel(context.Background()) - - // Channel to signal when dispatch returns - dispatchDone := make(chan error, 1) - - // Start dispatch in a goroutine - go func() { - err := errCache.dispatch(ctx) - dispatchDone <- err - }() - - // Give dispatch some time to start and potentially get stuck - time.Sleep(50 * time.Millisecond) - - // Cancel the context - cancel() - - // Wait for dispatch to return with a timeout - select { - case err := <-dispatchDone: - // Verify that dispatch returned with context.Canceled error - require.Equal(t, context.Canceled, err) - case <-time.After(5 * time.Second): - // If we timeout here, it means dispatch is stuck - t.Fatal("dispatch method is stuck and didn't return after context cancellation") - } -} - -func TestErrCacheDispatchBatch(t *testing.T) { - mockErrInfo := regionErrorInfo{ - regionInfo: regionInfo{ - verID: tikv.NewRegionVerID(1, 1, 1), - span: heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("b")}, - }, - err: errors.New("test error"), - } - - tests := []struct { - name string - cacheLen int - limit int - expectedN int - expectedCache int - expectedErrCh int - }{ - { - name: "dispatch all when limit equals cache length", - cacheLen: 5, - limit: 5, - expectedN: 5, - expectedCache: 0, - expectedErrCh: 5, - }, - { - name: "keep remaining cache when limit is smaller", - cacheLen: 5, - limit: 2, - expectedN: 2, - expectedCache: 3, - expectedErrCh: 2, - }, - { - name: "dispatch all when limit is larger", - cacheLen: 5, - limit: 10, - expectedN: 5, - expectedCache: 0, - expectedErrCh: 5, - }, - { - name: "dispatch all when limit is zero", - cacheLen: 5, - limit: 0, - expectedN: 5, - expectedCache: 0, - expectedErrCh: 5, - }, - { - name: "dispatch all when limit is negative", - cacheLen: 5, - limit: -1, - expectedN: 5, - expectedCache: 0, - expectedErrCh: 5, - }, - { - name: "empty cache", - cacheLen: 0, - limit: 5, - expectedN: 0, - expectedCache: 0, - expectedErrCh: 0, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - errCache := &errCache{ - cache: make([]regionErrorInfo, 0, 10), - errCh: make(chan regionErrorInfo, 10), - notify: make(chan struct{}, 1), - } - for i := 0; i < tc.cacheLen; i++ { - errCache.add(mockErrInfo) - } - - n, err := errCache.dispatchBatch(context.Background(), tc.limit) - require.NoError(t, err) - require.Equal(t, tc.expectedN, n) - require.Len(t, errCache.cache, tc.expectedCache) - require.Len(t, errCache.errCh, tc.expectedErrCh) - }) - } -} - func TestGetResolvedTargetTs(t *testing.T) { client := &subscriptionClient{ resolveLockTaskCh: make(chan resolveLockTask, 10),