From 41d10817c630ccda264a556c101931637a800133 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 14:36:44 +0800 Subject: [PATCH 1/9] logpuller: split subscription runtime helpers from subscription client --- logservice/logpuller/puller_memory_quota.go | 214 +++++++ .../logpuller/puller_memory_quota_test.go | 107 ++++ logservice/logpuller/region_event_sink.go | 135 ++++ .../logpuller/region_failure_handler.go | 282 +++++++++ .../logpuller/region_request_worker_test.go | 2 +- logservice/logpuller/span_registry.go | 321 ++++++++++ logservice/logpuller/subscription_client.go | 579 ++---------------- .../logpuller/subscription_client_test.go | 29 +- 8 files changed, 1115 insertions(+), 554 deletions(-) create mode 100644 logservice/logpuller/puller_memory_quota.go create mode 100644 logservice/logpuller/puller_memory_quota_test.go create mode 100644 logservice/logpuller/region_event_sink.go create mode 100644 logservice/logpuller/region_failure_handler.go create mode 100644 logservice/logpuller/span_registry.go diff --git a/logservice/logpuller/puller_memory_quota.go b/logservice/logpuller/puller_memory_quota.go new file mode 100644 index 0000000000..e9f8147290 --- /dev/null +++ b/logservice/logpuller/puller_memory_quota.go @@ -0,0 +1,214 @@ +// 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" +) + +const ( + defaultPullerMemoryQuotaBytes = 1 * 1024 * 1024 * 1024 + pullerMemoryQuotaRequestRatio = 0.8 + minPullerEventMemoryBytes = 256 +) + +type pullerMemoryQuotaSnapshot struct { + MaxBytes uint64 + UsedBytes uint64 + RegionRequestLimit uint64 + EventWaiterCount int + RegionReqWaiterCount int + Closed bool +} + +// pullerMemoryQuota provides coarse backpressure for two puller-owned +// boundaries: region events entering dynamic stream and normal region feed +// requests being sent to TiKV. It does not reorder anything; callers wait at +// their current position and continue when memory is released. +type pullerMemoryQuota struct { + mu sync.Mutex + + maxBytes uint64 + regionRequestLimit uint64 + usedBytes uint64 + + eventWaiterCount int + regionReqWaiterCount int + + closed bool + notify chan struct{} +} + +func newPullerMemoryQuota(maxBytes uint64) *pullerMemoryQuota { + if maxBytes == 0 { + maxBytes = defaultPullerMemoryQuotaBytes + } + regionRequestLimit := uint64(float64(maxBytes) * pullerMemoryQuotaRequestRatio) + if regionRequestLimit == 0 { + regionRequestLimit = 1 + } + return &pullerMemoryQuota{ + maxBytes: maxBytes, + regionRequestLimit: regionRequestLimit, + notify: make(chan struct{}), + } +} + +func (q *pullerMemoryQuota) acquireEvent(ctx context.Context, size uint64) (uint64, bool) { + if size < minPullerEventMemoryBytes { + size = minPullerEventMemoryBytes + } + + q.mu.Lock() + for { + if q.closed { + q.mu.Unlock() + return 0, false + } + if q.canAcquireEventLocked(size) { + q.usedBytes += size + q.mu.Unlock() + return size, true + } + + notify := q.notify + q.eventWaiterCount++ + q.mu.Unlock() + select { + case <-ctx.Done(): + q.mu.Lock() + q.eventWaiterCount-- + q.mu.Unlock() + return 0, false + case <-notify: + q.mu.Lock() + q.eventWaiterCount-- + } + } +} + +func (q *pullerMemoryQuota) canAcquireEventLocked(size uint64) bool { + if q.usedBytes == 0 && size > q.maxBytes { + return true + } + if q.usedBytes > q.maxBytes { + return false + } + return size <= q.maxBytes-q.usedBytes +} + +func (q *pullerMemoryQuota) release(size uint64) { + if size == 0 { + return + } + + q.mu.Lock() + if size >= q.usedBytes { + q.usedBytes = 0 + } else { + q.usedBytes -= size + } + q.notifyWaitersLocked() + q.mu.Unlock() +} + +func (q *pullerMemoryQuota) waitRegionRequest(ctx context.Context) bool { + q.mu.Lock() + for { + if q.closed { + q.mu.Unlock() + return false + } + if q.usedBytes < q.regionRequestLimit { + q.mu.Unlock() + return true + } + + notify := q.notify + q.regionReqWaiterCount++ + q.mu.Unlock() + select { + case <-ctx.Done(): + q.mu.Lock() + q.regionReqWaiterCount-- + q.mu.Unlock() + return false + case <-notify: + q.mu.Lock() + q.regionReqWaiterCount-- + } + } +} + +func (q *pullerMemoryQuota) close() { + q.mu.Lock() + q.closed = true + q.notifyWaitersLocked() + q.mu.Unlock() +} + +func (q *pullerMemoryQuota) snapshot() pullerMemoryQuotaSnapshot { + q.mu.Lock() + defer q.mu.Unlock() + return pullerMemoryQuotaSnapshot{ + MaxBytes: q.maxBytes, + UsedBytes: q.usedBytes, + RegionRequestLimit: q.regionRequestLimit, + EventWaiterCount: q.eventWaiterCount, + RegionReqWaiterCount: q.regionReqWaiterCount, + Closed: q.closed, + } +} + +func (q *pullerMemoryQuota) notifyWaitersLocked() { + close(q.notify) + q.notify = make(chan struct{}) +} + +func (s *subscriptionClient) acquireRegionEventQuota(event *regionEvent) bool { + if s == nil || s.memoryQuota == nil { + return true + } + ctx := s.ctx + if ctx == nil { + ctx = context.Background() + } + chargedBytes, ok := s.memoryQuota.acquireEvent(ctx, uint64(event.getSize())) + if !ok { + return false + } + event.quotaBytes = chargedBytes + return true +} + +func (s *subscriptionClient) releaseRegionEventQuota(event regionEvent) { + if s == nil || s.memoryQuota == nil { + return + } + s.memoryQuota.release(event.quotaBytes) +} + +func (s *subscriptionClient) waitRegionRequestQuota(ctx context.Context) error { + if s == nil || s.memoryQuota == nil { + return nil + } + if s.memoryQuota.waitRegionRequest(ctx) { + return nil + } + if ctx.Err() != nil { + return ctx.Err() + } + return context.Canceled +} diff --git a/logservice/logpuller/puller_memory_quota_test.go b/logservice/logpuller/puller_memory_quota_test.go new file mode 100644 index 0000000000..f03b4e17e3 --- /dev/null +++ b/logservice/logpuller/puller_memory_quota_test.go @@ -0,0 +1,107 @@ +// 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/stretchr/testify/require" +) + +func TestPullerMemoryQuotaEventWaitsForRelease(t *testing.T) { + quota := newPullerMemoryQuota(minPullerEventMemoryBytes) + chargedBytes, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) + require.True(t, ok) + + done := make(chan struct{}) + go func() { + _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) + require.True(t, ok) + close(done) + }() + + select { + case <-done: + t.Fatal("event acquire should wait when quota is full") + case <-time.After(100 * time.Millisecond): + } + + quota.release(chargedBytes) + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("event acquire should resume after release") + } +} + +func TestPullerMemoryQuotaCloseWakesEventWaiter(t *testing.T) { + quota := newPullerMemoryQuota(minPullerEventMemoryBytes) + _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) + require.True(t, ok) + + done := make(chan bool) + go func() { + _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) + done <- ok + }() + + select { + case <-done: + t.Fatal("event acquire should wait before quota is closed") + case <-time.After(100 * time.Millisecond): + } + + quota.close() + select { + case ok := <-done: + require.False(t, ok) + case <-time.After(5 * time.Second): + t.Fatal("event acquire should resume after quota close") + } +} + +func TestPullerMemoryQuotaAllowsSingleOversizedEvent(t *testing.T) { + quota := newPullerMemoryQuota(minPullerEventMemoryBytes) + chargedBytes, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes*2) + require.True(t, ok) + require.Equal(t, uint64(minPullerEventMemoryBytes*2), chargedBytes) + require.Equal(t, uint64(minPullerEventMemoryBytes*2), quota.snapshot().UsedBytes) +} + +func TestPullerMemoryQuotaRegionRequestWaitsAtHighWatermark(t *testing.T) { + quota := newPullerMemoryQuota(1000) + chargedBytes, ok := quota.acquireEvent(context.Background(), 800) + require.True(t, ok) + + done := make(chan struct{}) + go func() { + require.True(t, quota.waitRegionRequest(context.Background())) + close(done) + }() + + select { + case <-done: + t.Fatal("region request should wait when usage reaches high watermark") + case <-time.After(100 * time.Millisecond): + } + + quota.release(chargedBytes) + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("region request should resume after release") + } +} diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go new file mode 100644 index 0000000000..ef6024299d --- /dev/null +++ b/logservice/logpuller/region_event_sink.go @@ -0,0 +1,135 @@ +// 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, subClient *subscriptionClient) *regionEventSink { + sink := ®ionEventSink{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", + ®ionEventHandler{subClient: subClient}, + 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 + } + + ctx := s.ctx + if ctx == nil { + ctx = context.Background() + } + + // slow path: wait until paused is false + s.mu.Lock() + for s.paused.Load() { + select { + case <-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 { + 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() +} diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go new file mode 100644 index 0000000000..12fd2f0fb7 --- /dev/null +++ b/logservice/logpuller/region_failure_handler.go @@ -0,0 +1,282 @@ +// 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" + "time" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/tikv/client-go/v2/tikv" + "go.uber.org/zap" +) + +var ( + metricFeedNotLeaderCounter = metrics.EventFeedErrorCounter.WithLabelValues("NotLeader") + metricFeedEpochNotMatchCounter = metrics.EventFeedErrorCounter.WithLabelValues("EpochNotMatch") + metricFeedRegionNotFoundCounter = metrics.EventFeedErrorCounter.WithLabelValues("RegionNotFound") + metricFeedDuplicateRequestCounter = metrics.EventFeedErrorCounter.WithLabelValues("DuplicateRequest") + metricFeedUnknownErrorCounter = metrics.EventFeedErrorCounter.WithLabelValues("Unknown") + metricFeedRPCCtxUnavailable = metrics.EventFeedErrorCounter.WithLabelValues("RPCCtxUnavailable") + metricGetStoreErr = metrics.EventFeedErrorCounter.WithLabelValues("GetStoreErr") + metricStoreSendRequestErr = metrics.EventFeedErrorCounter.WithLabelValues("SendRequestToStore") + metricKvIsBusyCounter = metrics.EventFeedErrorCounter.WithLabelValues("KvIsBusy") + metricKvCongestedCounter = metrics.EventFeedErrorCounter.WithLabelValues("KvCongested") +) + +type regionFailureHandler struct { + cache *errCache + client *subscriptionClient +} + +func newRegionFailureHandler(client *subscriptionClient) *regionFailureHandler { + return ®ionFailureHandler{ + cache: newErrCache(), + client: client, + } +} + +// Report admits a region failure into the recovery pipeline. It releases the +// corresponding range lock before enqueueing the failure so new range tasks are +// not blocked by stale region ownership. +func (r *regionFailureHandler) Report(errInfo regionErrorInfo) { + if errInfo.subscribedSpan.rangeLock.UnlockRange( + errInfo.span.StartKey, errInfo.span.EndKey, + errInfo.verID.GetID(), errInfo.verID.GetVer(), errInfo.resolvedTs()) { + r.client.onTableDrained(errInfo.subscribedSpan) + return + } + r.cache.add(errInfo) +} + +func (r *regionFailureHandler) Run(ctx context.Context) error { + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + + handlePendingFailures := func() error { + for { + batch := r.cache.popBatch(errCacheDispatchBatchSize) + for _, errInfo := range batch { + select { + case <-ctx.Done(): + log.Info("subscription client handle errors and exit") + return ctx.Err() + default: + } + if err := r.handleError(ctx, errInfo); err != nil { + return err + } + } + if len(batch) < errCacheDispatchBatchSize { + return nil + } + } + } + + for { + select { + case <-ctx.Done(): + log.Info("subscription client handle errors and exit") + return ctx.Err() + case <-ticker.C: + if err := handlePendingFailures(); err != nil { + return err + } + case <-r.cache.notify: + if err := handlePendingFailures(); err != nil { + return err + } + } + } +} + +func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionErrorInfo) error { + err := errors.Cause(errInfo.err) + //nolint:errorlint // converting large type switch to errors.As is a significant refactor + if _, requestCancelled := err.(*requestCancelledErr); !requestCancelled { + log.Debug("cdc region error", + zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), + zap.Uint64("regionID", errInfo.verID.GetID()), + zap.Error(err)) + } + + //nolint:errorlint // converting large type switch to errors.As is a significant refactor + switch eerr := err.(type) { + case *eventError: + innerErr := eerr.err + 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) + return nil + } + if innerErr.GetEpochNotMatch() != nil { + metricFeedEpochNotMatchCounter.Inc() + r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + return nil + } + if innerErr.GetRegionNotFound() != nil { + metricFeedRegionNotFoundCounter.Inc() + r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + return nil + } + if innerErr.GetCongested() != nil { + metricKvCongestedCounter.Inc() + r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) + return nil + } + if innerErr.GetServerIsBusy() != nil { + metricKvIsBusyCounter.Inc() + r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) + return nil + } + if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil { + // TODO(qupeng): It's better to add a new machanism to deregister one region. + metricFeedDuplicateRequestCounter.Inc() + return errors.New("duplicate request") + } + if compatibility := innerErr.GetCompatibility(); compatibility != nil { + return errors.ErrVersionIncompatible.GenWithStackByArgs(compatibility) + } + if mismatch := innerErr.GetClusterIdMismatch(); mismatch != nil { + return errors.ErrClusterIDMismatch.GenWithStackByArgs(mismatch.Current, mismatch.Request) + } + + log.Warn("empty or unknown cdc error", + zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), + zap.Stringer("error", innerErr)) + metricFeedUnknownErrorCounter.Inc() + r.client.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) + return nil + case *rpcCtxUnavailableErr: + metricFeedRPCCtxUnavailable.Inc() + r.client.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) + 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) + 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) + return nil + case *requestCancelledErr: + // the corresponding subscription has been unsubscribed, just ignore. + return nil + default: + // TODO(qupeng): for some errors it's better to just deregister the region from TiKVs. + log.Warn("subscription client meets an internal error, fail the changefeed", + zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), + zap.Error(err)) + return err + } +} + +type errCache struct { + sync.Mutex + cache []regionErrorInfo + errCh chan regionErrorInfo + notify chan struct{} +} + +const errCacheDispatchBatchSize = 1024 + +func newErrCache() *errCache { + return &errCache{ + cache: make([]regionErrorInfo, 0, 1024), + errCh: make(chan regionErrorInfo, 4096), + notify: make(chan struct{}, 1024), + } +} + +func (e *errCache) add(errInfo regionErrorInfo) { + e.Lock() + defer e.Unlock() + e.cache = append(e.cache, errInfo) + select { + case e.notify <- struct{}{}: + default: + } +} + +func (e *errCache) popBatch(limit int) []regionErrorInfo { + e.Lock() + defer e.Unlock() + if len(e.cache) == 0 { + return nil + } + if limit <= 0 || limit > len(e.cache) { + limit = len(e.cache) + } + batch := make([]regionErrorInfo, limit) + copy(batch, e.cache[:limit]) + clear(e.cache[:limit]) + if limit == len(e.cache) { + e.cache = e.cache[:0] + } else { + e.cache = e.cache[limit:] + } + 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 + } + } + } +} diff --git a/logservice/logpuller/region_request_worker_test.go b/logservice/logpuller/region_request_worker_test.go index f9752a0eb3..268045c492 100644 --- a/logservice/logpuller/region_request_worker_test.go +++ b/logservice/logpuller/region_request_worker_test.go @@ -157,7 +157,7 @@ func newDispatchResolvedTsTestWorker(regionCount int) (*regionRequestWorker, *mo metrics: sharedClientMetrics{ batchResolvedSize: prometheus.ObserverFunc(func(float64) {}), }, - ds: ds, + eventSink: ®ionEventSink{ds: ds}, }, } worker.requestedRegions.subscriptions = map[SubscriptionID]regionFeedStates{ diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go new file mode 100644 index 0000000000..7e855e474e --- /dev/null +++ b/logservice/logpuller/span_registry.go @@ -0,0 +1,321 @@ +// 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" + "time" + + "github.com/pingcap/log" + "github.com/pingcap/ticdc/heartbeatpb" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" + "github.com/pingcap/ticdc/pkg/common" + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/pdutil" + "github.com/pingcap/ticdc/pkg/util" + "github.com/tikv/client-go/v2/oracle" + pd "github.com/tikv/pd/client" + "go.uber.org/zap" + "golang.org/x/sync/errgroup" +) + +const kvEventsCacheMaxSize = 32 + +// subscribedSpan represents a span to subscribe. +// It contains a sub span of a table(or the total span of a table), +// the startTs of the table, and the output event channel. +type subscribedSpan struct { + subID SubscriptionID + startTs uint64 + // Whether to filter out the value written by TiCDC itself. + // It should be `true` in BDR mode. + filterLoop bool + + // The target span + span heartbeatpb.TableSpan + // The range lock of the span, + // it is used to prevent duplicate requests to the same region range, + // and it also used to calculate this table's resolvedTs. + rangeLock *regionlock.RangeLock + + consumeKVEvents func(events []common.RawKVEntry, wakeCallback func()) bool + + advanceResolvedTs func(ts uint64) + + advanceInterval int64 + + kvEventsCache []common.RawKVEntry + + // To handle span removing. + stopped atomic.Bool + + // To handle stale lock resolvings. + tryResolveLock func(regionID uint64, state *regionlock.LockedRangeState) + staleLocksTargetTs atomic.Uint64 + + lastAdvanceTime atomic.Int64 + + initialized atomic.Bool + resolvedTsUpdated atomic.Int64 + resolvedTs atomic.Uint64 +} + +type spanRegistry struct { + sync.RWMutex + spans map[SubscriptionID]*subscribedSpan + + pd pd.Client + pdClock pdutil.Clock +} + +type spanAndTargetTs struct { + span *subscribedSpan + targetTs uint64 +} + +func newSubscribedSpan( + ctx context.Context, + resolveLockRateLimiter *resolveLockRateLimiter, + resolveLockTaskCh chan resolveLockTask, + subID SubscriptionID, + span heartbeatpb.TableSpan, + startTs uint64, + consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, + advanceResolvedTs func(ts uint64), + advanceInterval int64, + filterLoop bool, +) *subscribedSpan { + rangeLock := regionlock.NewRangeLock(uint64(subID), span.StartKey, span.EndKey, startTs) + + rt := &subscribedSpan{ + subID: subID, + span: span, + startTs: startTs, + filterLoop: filterLoop, + rangeLock: rangeLock, + + consumeKVEvents: consumeKVEvents, + advanceResolvedTs: advanceResolvedTs, + advanceInterval: advanceInterval, + } + rt.initialized.Store(false) + rt.resolvedTsUpdated.Store(time.Now().Unix()) + rt.resolvedTs.Store(startTs) + + rt.tryResolveLock = func(regionID uint64, state *regionlock.LockedRangeState) { + targetTs := rt.staleLocksTargetTs.Load() + if !state.Initialized.Load() || state.ResolvedTs.Load() >= targetTs { + return + } + key := resolveLockKey{keyspaceID: span.KeyspaceID, regionID: regionID} + if !resolveLockRateLimiter.trySchedule(key, time.Now()) { + return + } + select { + case <-ctx.Done(): + resolveLockRateLimiter.cancel(key) + case resolveLockTaskCh <- resolveLockTask{ + keyspaceID: span.KeyspaceID, + regionID: regionID, + targetTs: targetTs, + state: state, + }: + // it is ok to ignore resolve lock task when the channel is full + default: + resolveLockRateLimiter.cancel(key) + metrics.SubscriptionClientResolveLockTaskDropCounter.Inc() + } + } + return rt +} + +func (span *subscribedSpan) clearKVEventsCache() { + if cap(span.kvEventsCache) > kvEventsCacheMaxSize { + span.kvEventsCache = nil + } else { + span.kvEventsCache = span.kvEventsCache[:0] + } +} + +func (span *subscribedSpan) resolveStaleLocks(targetTs uint64) { + util.MustCompareAndMonotonicIncrease(&span.staleLocksTargetTs, targetTs) + res := span.rangeLock.IterAll(span.tryResolveLock) + log.Debug("subscription client finds slow locked ranges", + zap.Uint64("subscriptionID", uint64(span.subID)), + zap.Any("ranges", res)) +} + +func newSpanRegistry(pd pd.Client, pdClock pdutil.Clock) *spanRegistry { + return &spanRegistry{ + spans: make(map[SubscriptionID]*subscribedSpan), + pd: pd, + pdClock: pdClock, + } +} + +func (r *spanRegistry) Add(span *subscribedSpan) { + r.Lock() + defer r.Unlock() + r.spans[span.subID] = span +} + +func (r *spanRegistry) Get(subID SubscriptionID) *subscribedSpan { + r.RLock() + defer r.RUnlock() + return r.spans[subID] +} + +func (r *spanRegistry) Remove(subID SubscriptionID) { + r.Lock() + defer r.Unlock() + delete(r.spans, subID) +} + +func (r *spanRegistry) GetResolvedTsLag() float64 { + pullerMinResolvedTs := uint64(0) + r.RLock() + for _, span := range r.spans { + resolvedTs := span.resolvedTs.Load() + if pullerMinResolvedTs == 0 || resolvedTs < pullerMinResolvedTs { + pullerMinResolvedTs = resolvedTs + } + } + r.RUnlock() + if pullerMinResolvedTs == 0 { + return 0 + } + pdTime := r.pdClock.CurrentTime() + phyResolvedTs := oracle.ExtractPhysical(pullerMinResolvedTs) + return float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3 +} + +func (r *spanRegistry) UpdateMetrics() { + count := 0 + r.RLock() + for _, span := range r.spans { + count += span.rangeLock.Len() + } + r.RUnlock() + metrics.SubscriptionClientSubscribedRegionCount.Set(float64(count)) + + resolvedTsLag := r.GetResolvedTsLag() + if resolvedTsLag > 0 { + metrics.LogPullerResolvedTsLag.Set(resolvedTsLag) + } +} + +func (r *spanRegistry) Run(ctx context.Context) error { + g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { return r.runResolveLockChecker(ctx) }) + g.Go(func() error { return r.logSlowRegions(ctx) }) + return g.Wait() +} + +func (r *spanRegistry) runResolveLockChecker(ctx context.Context) error { + resolveLockTicker := time.NewTicker(resolveLockTickInterval) + defer resolveLockTicker.Stop() + maxCacheSize := 1024 + spanAndTsCache := make([]spanAndTargetTs, 0, maxCacheSize) + // getResolvedTargetTs returns the targetTs to resolve stale locks. 0 means no need to resolve. + getResolvedTargetTs := func(subSpan *subscribedSpan, currentTime time.Time, currentTs uint64) uint64 { + resolvedTsUpdated := time.Unix(subSpan.resolvedTsUpdated.Load(), 0) + if !subSpan.initialized.Load() || time.Since(resolvedTsUpdated) < resolveLockFence { + return 0 + } + resolvedTs := subSpan.resolvedTs.Load() + resolvedTime := oracle.GetTimeFromTS(resolvedTs) + if currentTime.Sub(resolvedTime) < resolveLockFence { + return 0 + } + return min(currentTs, oracle.GoTimeToTS(resolvedTime.Add(resolveLockFence))) + } + + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-resolveLockTicker.C: + } + + physical, logic, err := r.pd.GetTS(ctx) + if err != nil { + log.Warn("get ts from pd failed", zap.Error(err)) + continue + } + currentTs := oracle.ComposeTS(physical, logic) + currentTime := r.pdClock.CurrentTime() + r.RLock() + for _, subSpan := range r.spans { + if subSpan != nil { + targetTs := getResolvedTargetTs(subSpan, currentTime, currentTs) + if targetTs > 0 { + spanAndTsCache = append(spanAndTsCache, spanAndTargetTs{ + span: subSpan, + targetTs: targetTs, + }) + } + } + } + r.RUnlock() + for _, spanAndTs := range spanAndTsCache { + spanAndTs.span.resolveStaleLocks(spanAndTs.targetTs) + } + spanAndTsCache = spanAndTsCache[:0] + if cap(spanAndTsCache) > maxCacheSize { + spanAndTsCache = make([]spanAndTargetTs, 0, maxCacheSize) + } + } +} + +func (r *spanRegistry) logSlowRegions(ctx context.Context) error { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + } + + currTime := r.pdClock.CurrentTime() + r.RLock() + for subscriptionID, rt := range r.spans { + attr := rt.rangeLock.IterAll(nil) + ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs) + if attr.SlowestRegion.Initialized { + if currTime.Sub(ckptTime) > 6*resolveLockMinInterval { + log.Info("subscription client finds a initialized slow region", + zap.Uint64("subscriptionID", uint64(subscriptionID)), + zap.Any("slowRegion", attr.SlowestRegion)) + } + } else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute { + log.Info("subscription client initializes a region too slow", + zap.Uint64("subscriptionID", uint64(subscriptionID)), + zap.Any("slowRegion", attr.SlowestRegion)) + } else if currTime.Sub(ckptTime) > 10*time.Minute { + log.Info("subscription client finds a uninitialized slow region", + zap.Uint64("subscriptionID", uint64(subscriptionID)), + zap.Any("slowRegion", attr.SlowestRegion)) + } + if len(attr.UnLockedRanges) > 0 { + log.Info("subscription client holes exist", + zap.Uint64("subscriptionID", uint64(subscriptionID)), + zap.Any("holes", attr.UnLockedRanges)) + } + } + r.RUnlock() + } +} diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index fbcffdc4f1..75ecfb34ae 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -33,11 +33,9 @@ import ( "github.com/pingcap/ticdc/pkg/security" "github.com/pingcap/ticdc/pkg/spanz" "github.com/pingcap/ticdc/pkg/util" - "github.com/pingcap/ticdc/utils/dynstream" "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" @@ -60,18 +58,8 @@ const ( ) var ( - metricFeedNotLeaderCounter = metrics.EventFeedErrorCounter.WithLabelValues("NotLeader") - metricFeedEpochNotMatchCounter = metrics.EventFeedErrorCounter.WithLabelValues("EpochNotMatch") - metricFeedRegionNotFoundCounter = metrics.EventFeedErrorCounter.WithLabelValues("RegionNotFound") - metricFeedDuplicateRequestCounter = metrics.EventFeedErrorCounter.WithLabelValues("DuplicateRequest") - metricFeedUnknownErrorCounter = metrics.EventFeedErrorCounter.WithLabelValues("Unknown") - metricFeedRPCCtxUnavailable = metrics.EventFeedErrorCounter.WithLabelValues("RPCCtxUnavailable") - metricGetStoreErr = metrics.EventFeedErrorCounter.WithLabelValues("GetStoreErr") - metricStoreSendRequestErr = metrics.EventFeedErrorCounter.WithLabelValues("SendRequestToStore") - metricKvIsBusyCounter = metrics.EventFeedErrorCounter.WithLabelValues("KvIsBusy") - metricKvCongestedCounter = metrics.EventFeedErrorCounter.WithLabelValues("KvCongested") - metricResolveLockSuccessCounter = metrics.SubscriptionClientResolveLockCounter.WithLabelValues("success") - metricResolveLockFailureCounter = metrics.SubscriptionClientResolveLockCounter.WithLabelValues("failure") + metricResolveLockSuccessCounter = metrics.SubscriptionClientResolveLockCounter.WithLabelValues("success") + metricResolveLockFailureCounter = metrics.SubscriptionClientResolveLockCounter.WithLabelValues("failure") metricSubscriptionClientDSChannelSize = metrics.DynamicStreamEventChanSize.WithLabelValues("event-store") metricSubscriptionClientDSPendingQueueLen = metrics.DynamicStreamPendingQueueLen.WithLabelValues("event-store") @@ -102,55 +90,6 @@ type rangeTask struct { priority TaskType } -const kvEventsCacheMaxSize = 32 - -// subscribedSpan represents a span to subscribe. -// It contains a sub span of a table(or the total span of a table), -// the startTs of the table, and the output event channel. -type subscribedSpan struct { - subID SubscriptionID - startTs uint64 - // Whether to filter out the value written by TiCDC itself. - // It should be `true` in BDR mode. - filterLoop bool - - // The target span - span heartbeatpb.TableSpan - // The range lock of the span, - // it is used to prevent duplicate requests to the same region range, - // and it also used to calculate this table's resolvedTs. - rangeLock *regionlock.RangeLock - - consumeKVEvents func(events []common.RawKVEntry, wakeCallback func()) bool - - advanceResolvedTs func(ts uint64) - - advanceInterval int64 - - kvEventsCache []common.RawKVEntry - - // To handle span removing. - stopped atomic.Bool - - // To handle stale lock resolvings. - tryResolveLock func(regionID uint64, state *regionlock.LockedRangeState) - staleLocksTargetTs atomic.Uint64 - - lastAdvanceTime atomic.Int64 - - initialized atomic.Bool - resolvedTsUpdated atomic.Int64 - resolvedTs atomic.Uint64 -} - -func (span *subscribedSpan) clearKVEventsCache() { - if cap(span.kvEventsCache) > kvEventsCacheMaxSize { - span.kvEventsCache = nil - } else { - span.kvEventsCache = span.kvEventsCache[:0] - } -} - type SubscriptionClientConfig struct { // The number of region request workers to send region task for every tikv store RegionRequestWorkerPerStore uint @@ -194,19 +133,12 @@ type subscriptionClient struct { stores sync.Map - ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] - // the following three fields are used to manage feedback from ds and notify other goroutines - mu sync.Mutex - cond *sync.Cond - paused atomic.Bool - // the credential to connect tikv credential *security.Credential - totalSpans struct { - sync.RWMutex - spanMap map[SubscriptionID]*subscribedSpan - } + eventSink *regionEventSink + failureHandler *regionFailureHandler + spanRegistry *spanRegistry // rangeTaskCh is used to receive range tasks. // The tasks will be handled in `handleRangeTask` goroutine. @@ -218,9 +150,6 @@ type subscriptionClient struct { // The tasks will be handled in `handleResolveLockTasks` goroutine. resolveLockTaskCh chan resolveLockTask resolveLockRateLimiter *resolveLockRateLimiter - // errCh is used to receive region errors. - // The errors will be handled in `handleErrors` goroutine. - errCache *errCache } // NewSubscriptionClient creates a client. @@ -245,26 +174,11 @@ func NewSubscriptionClient( regionTaskQueue: priorityqueue.New[PriorityTask](), resolveLockTaskCh: make(chan resolveLockTask, 1024), resolveLockRateLimiter: newResolveLockRateLimiter(), - errCache: newErrCache(), } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) - subClient.totalSpans.spanMap = make(map[SubscriptionID]*subscribedSpan) - - 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", - ®ionEventHandler{subClient: subClient}, - option, - ) - ds.Start() - subClient.ds = ds - subClient.cond = sync.NewCond(&subClient.mu) + subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) + subClient.failureHandler = newRegionFailureHandler(subClient) + subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) subClient.initMetrics() return subClient @@ -292,11 +206,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - resolvedTsLag := s.GetResolvedTsLag() - if resolvedTsLag > 0 { - metrics.LogPullerResolvedTsLag.Set(resolvedTsLag) - } - dsMetrics := s.ds.GetMetrics() + dsMetrics := s.eventSink.Metrics() metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 1 { @@ -331,21 +241,14 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { }) metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("pending").Set(float64(pendingRegionReqCount)) - - count := 0 - s.totalSpans.RLock() - for _, rt := range s.totalSpans.spanMap { - count += rt.rangeLock.Len() - } - s.totalSpans.RUnlock() - metrics.SubscriptionClientSubscribedRegionCount.Set(float64(count)) + s.spanRegistry.UpdateMetrics() } } } // Subscribe the given table span. // NOTE: `span.TableID` must be set correctly. -// It new a subscribedSpan and store it in `s.totalSpans`, +// It new a subscribedSpan and store it in `s.spanRegistry`, // and send a rangeTask to `s.rangeTaskCh`. // The rangeTask will be handled in `handleRangeTasks` goroutine. func (s *subscriptionClient) Subscribe( @@ -363,12 +266,8 @@ func (s *subscriptionClient) Subscribe( } rt := s.newSubscribedSpan(subID, span, startTs, consumeKVEvents, advanceResolvedTs, advanceInterval, bdrMode) - s.totalSpans.Lock() - s.totalSpans.spanMap[subID] = rt - s.totalSpans.Unlock() - - areaSetting := dynstream.NewAreaSettingsWithMaxPendingSize(1*1024*1024*1024, dynstream.MemoryControlForPuller, "logPuller") // 1GB - s.ds.AddPath(rt.subID, rt, areaSetting) + s.spanRegistry.Add(rt) + s.eventSink.AddPath(rt) select { case <-s.ctx.Done(): @@ -383,10 +282,8 @@ func (s *subscriptionClient) Subscribe( // 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) { - // NOTE: `subID` is cleared from `s.totalSpans` in `onTableDrained`. - s.totalSpans.Lock() - rt := s.totalSpans.spanMap[subID] - s.totalSpans.Unlock() + // NOTE: `subID` is cleared from `s.spanRegistry` in `onTableDrained`. + rt := s.spanRegistry.Get(subID) if rt == nil { log.Warn("unknown subscription", zap.Uint64("subscriptionID", uint64(subID))) return @@ -399,53 +296,11 @@ func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) { } func (s *subscriptionClient) wakeSubscription(subID SubscriptionID) { - s.ds.Wake(subID) + s.eventSink.Wake(subID) } func (s *subscriptionClient) pushRegionEventToDS(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 *subscriptionClient) handleDSFeedBack(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. - } - } - } + s.eventSink.Push(subID, event) } func (s *subscriptionClient) Run(ctx context.Context) error { @@ -459,14 +314,12 @@ func (s *subscriptionClient) Run(ctx context.Context) error { g, ctx := errgroup.WithContext(ctx) g.Go(func() error { return s.updateMetrics(ctx) }) - g.Go(func() error { return s.handleDSFeedBack(ctx) }) + g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.handleRangeTasks(ctx) }) g.Go(func() error { return s.handleRegions(ctx, g) }) - g.Go(func() error { return s.handleErrors(ctx) }) - g.Go(func() error { return s.runResolveLockChecker(ctx) }) + g.Go(func() error { return s.failureHandler.Run(ctx) }) g.Go(func() error { return s.handleResolveLockTasks(ctx) }) - g.Go(func() error { return s.logSlowRegions(ctx) }) - g.Go(func() error { return s.errCache.dispatch(ctx) }) + g.Go(func() error { return s.spanRegistry.Run(ctx) }) log.Info("subscription client starts") defer log.Info("subscription client exits") @@ -476,11 +329,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { // Close closes the client. Must be called after `Run` returns. func (s *subscriptionClient) Close(ctx context.Context) error { s.cancel() - s.mu.Lock() - s.paused.Store(false) - s.cond.Broadcast() - s.mu.Unlock() - s.ds.Close() + s.eventSink.Close() s.regionTaskQueue.Close() return nil } @@ -504,27 +353,18 @@ func (s *subscriptionClient) onTableDrained(rt *subscribedSpan) { log.Info("subscription client stop span is finished", zap.Uint64("subscriptionID", uint64(rt.subID))) - err := s.ds.RemovePath(rt.subID) + err := s.eventSink.RemovePath(rt.subID) if err != nil { log.Warn("subscription client remove path failed", zap.Uint64("subscriptionID", uint64(rt.subID)), zap.Error(err)) } - s.totalSpans.Lock() - defer s.totalSpans.Unlock() - delete(s.totalSpans.spanMap, rt.subID) + s.spanRegistry.Remove(rt.subID) } // Note: don't block the caller, otherwise there may be deadlock func (s *subscriptionClient) onRegionFail(errInfo regionErrorInfo) { - // unlock the range early to prevent blocking the range. - if errInfo.subscribedSpan.rangeLock.UnlockRange( - errInfo.span.StartKey, errInfo.span.EndKey, - errInfo.verID.GetID(), errInfo.verID.GetVer(), errInfo.resolvedTs()) { - s.onTableDrained(errInfo.subscribedSpan) - return - } - s.errCache.add(errInfo) + s.failureHandler.Report(errInfo) } // requestedStore represents a store that has been connected. @@ -835,167 +675,6 @@ func (s *subscriptionClient) scheduleRangeRequest( } } -func (s *subscriptionClient) handleErrors(ctx context.Context) error { - for { - select { - case <-ctx.Done(): - log.Info("subscription client handle errors and exit") - return ctx.Err() - case errInfo := <-s.errCache.errCh: - if err := s.doHandleError(ctx, errInfo); err != nil { - return err - } - } - } -} - -func (s *subscriptionClient) doHandleError(ctx context.Context, errInfo regionErrorInfo) error { - err := errors.Cause(errInfo.err) - if _, requestCancelled := err.(*requestCancelledErr); !requestCancelled { - log.Debug("cdc region error", - zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), - zap.Uint64("regionID", errInfo.verID.GetID()), - zap.Error(err)) - } - - //nolint:errorlint // converting large type switch to errors.As is a significant refactor - switch eerr := err.(type) { - case *eventError: - innerErr := eerr.err - if notLeader := innerErr.GetNotLeader(); notLeader != nil { - metricFeedNotLeaderCounter.Inc() - s.regionCache.UpdateLeader(errInfo.verID, notLeader.GetLeader(), errInfo.rpcCtx.AccessIdx) - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) - return nil - } - if innerErr.GetEpochNotMatch() != nil { - metricFeedEpochNotMatchCounter.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) - return nil - } - if innerErr.GetRegionNotFound() != nil { - metricFeedRegionNotFoundCounter.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) - return nil - } - if innerErr.GetCongested() != nil { - metricKvCongestedCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) - return nil - } - if innerErr.GetServerIsBusy() != nil { - metricKvIsBusyCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskLowPrior) - return nil - } - if duplicated := innerErr.GetDuplicateRequest(); duplicated != nil { - // TODO(qupeng): It's better to add a new machanism to deregister one region. - metricFeedDuplicateRequestCounter.Inc() - return errors.New("duplicate request") - } - if compatibility := innerErr.GetCompatibility(); compatibility != nil { - return errors.ErrVersionIncompatible.GenWithStackByArgs(compatibility) - } - if mismatch := innerErr.GetClusterIdMismatch(); mismatch != nil { - return errors.ErrClusterIDMismatch.GenWithStackByArgs(mismatch.Current, mismatch.Request) - } - - log.Warn("empty or unknown cdc error", - zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), - zap.Stringer("error", innerErr)) - metricFeedUnknownErrorCounter.Inc() - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) - return nil - case *rpcCtxUnavailableErr: - metricFeedRPCCtxUnavailable.Inc() - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) - 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. - s.regionCache.OnSendFail(bo, errInfo.rpcCtx, true, err) - s.scheduleRangeRequest(ctx, errInfo.span, errInfo.subscribedSpan, errInfo.filterLoop, TaskHighPrior) - return nil - case *storeStreamErr: - metricStoreSendRequestErr.Inc() - bo := tikv.NewBackoffer(ctx, tikvRequestMaxBackoff) - s.regionCache.OnSendFail(bo, errInfo.rpcCtx, regionScheduleReload, err) - s.scheduleRegionRequest(ctx, errInfo.regionInfo, TaskHighPrior) - return nil - case *requestCancelledErr: - // the corresponding subscription has been unsubscribed, just ignore. - return nil - default: - // TODO(qupeng): for some errors it's better to just deregister the region from TiKVs. - log.Warn("subscription client meets an internal error, fail the changefeed", - zap.Uint64("subscriptionID", uint64(errInfo.subscribedSpan.subID)), - zap.Error(err)) - return err - } -} - -type subscriptionAndTargetTs struct { - subSpan *subscribedSpan - targetTs uint64 -} - -func (s *subscriptionClient) runResolveLockChecker(ctx context.Context) error { - resolveLockTicker := time.NewTicker(resolveLockTickInterval) - defer resolveLockTicker.Stop() - maxCacheSize := 1024 - subSpanAndTsCache := make([]subscriptionAndTargetTs, 0, maxCacheSize) - // getResolvedTargetTs returns the targetTs to resolve stale locks. 0 means no need to resolve. - getResolvedTargetTs := func(subSpan *subscribedSpan, currentTime time.Time, currentTs uint64) uint64 { - resolvedTsUpdated := time.Unix(subSpan.resolvedTsUpdated.Load(), 0) - if !subSpan.initialized.Load() || time.Since(resolvedTsUpdated) < resolveLockFence { - return 0 - } - resolvedTs := subSpan.resolvedTs.Load() - resolvedTime := oracle.GetTimeFromTS(resolvedTs) - if currentTime.Sub(resolvedTime) < resolveLockFence { - return 0 - } - return min(currentTs, oracle.GoTimeToTS(resolvedTime.Add(resolveLockFence))) - } - - for { - select { - case <-ctx.Done(): - return ctx.Err() - case <-resolveLockTicker.C: - } - - physical, logic, err := s.pd.GetTS(ctx) - if err != nil { - log.Warn("get ts from pd failed", zap.Error(err)) - continue - } - currentTs := oracle.ComposeTS(physical, logic) - currentTime := s.pdClock.CurrentTime() - s.totalSpans.Lock() - for _, subSpan := range s.totalSpans.spanMap { - if subSpan != nil { - targetTs := getResolvedTargetTs(subSpan, currentTime, currentTs) - if targetTs > 0 { - subSpanAndTsCache = append(subSpanAndTsCache, subscriptionAndTargetTs{ - subSpan: subSpan, - targetTs: targetTs, - }) - } - } - } - s.totalSpans.Unlock() - for _, subSpanAndTs := range subSpanAndTsCache { - subSpanAndTs.subSpan.resolveStaleLocks(subSpanAndTs.targetTs) - } - subSpanAndTsCache = subSpanAndTsCache[:0] - if cap(subSpanAndTsCache) > maxCacheSize { - subSpanAndTsCache = make([]subscriptionAndTargetTs, 0, maxCacheSize) - } - } -} - func (s *subscriptionClient) handleResolveLockTasks(ctx context.Context) error { doResolve := func(task resolveLockTask) { keyspaceID := task.keyspaceID @@ -1034,48 +713,6 @@ func (s *subscriptionClient) handleResolveLockTasks(ctx context.Context) error { } } -func (s *subscriptionClient) logSlowRegions(ctx context.Context) error { - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - for { - select { - case <-ctx.Done(): - return ctx.Err() - case <-ticker.C: - } - - currTime := s.pdClock.CurrentTime() - s.totalSpans.RLock() - slowInitializeRegion := 0 - for subscriptionID, rt := range s.totalSpans.spanMap { - attr := rt.rangeLock.IterAll(nil) - ckptTime := oracle.GetTimeFromTS(attr.SlowestRegion.ResolvedTs) - if attr.SlowestRegion.Initialized { - if currTime.Sub(ckptTime) > 6*resolveLockMinInterval { - log.Info("subscription client finds a initialized slow region", - zap.Uint64("subscriptionID", uint64(subscriptionID)), - zap.Any("slowRegion", attr.SlowestRegion)) - } - } else if currTime.Sub(attr.SlowestRegion.Created) > 10*time.Minute { - slowInitializeRegion++ - log.Info("subscription client initializes a region too slow", - zap.Uint64("subscriptionID", uint64(subscriptionID)), - zap.Any("slowRegion", attr.SlowestRegion)) - } else if currTime.Sub(ckptTime) > 10*time.Minute { - log.Info("subscription client finds a uninitialized slow region", - zap.Uint64("subscriptionID", uint64(subscriptionID)), - zap.Any("slowRegion", attr.SlowestRegion)) - } - if len(attr.UnLockedRanges) > 0 { - log.Info("subscription client holes exist", - zap.Uint64("subscriptionID", uint64(subscriptionID)), - zap.Any("holes", attr.UnLockedRanges)) - } - } - s.totalSpans.RUnlock() - } -} - func (s *subscriptionClient) newSubscribedSpan( subID SubscriptionID, span heartbeatpb.TableSpan, @@ -1085,163 +722,23 @@ func (s *subscriptionClient) newSubscribedSpan( advanceInterval int64, filterLoop bool, ) *subscribedSpan { - rangeLock := regionlock.NewRangeLock(uint64(subID), span.StartKey, span.EndKey, startTs) - - rt := &subscribedSpan{ - subID: subID, - span: span, - startTs: startTs, - filterLoop: filterLoop, - rangeLock: rangeLock, - - consumeKVEvents: consumeKVEvents, - advanceResolvedTs: advanceResolvedTs, - advanceInterval: advanceInterval, - } - rt.initialized.Store(false) - rt.resolvedTsUpdated.Store(time.Now().Unix()) - rt.resolvedTs.Store(startTs) - - rt.tryResolveLock = func(regionID uint64, state *regionlock.LockedRangeState) { - targetTs := rt.staleLocksTargetTs.Load() - if !state.Initialized.Load() || state.ResolvedTs.Load() >= targetTs { - return - } - key := resolveLockKey{keyspaceID: span.KeyspaceID, regionID: regionID} - if !s.resolveLockRateLimiter.trySchedule(key, time.Now()) { - return - } - select { - case <-s.ctx.Done(): - s.resolveLockRateLimiter.cancel(key) - case s.resolveLockTaskCh <- resolveLockTask{ - keyspaceID: span.KeyspaceID, - regionID: regionID, - targetTs: targetTs, - state: state, - }: - // it is ok to ignore resolve lock task when the channel is full - default: - s.resolveLockRateLimiter.cancel(key) - metrics.SubscriptionClientResolveLockTaskDropCounter.Inc() - } - } - return rt + return newSubscribedSpan( + s.ctx, + s.resolveLockRateLimiter, + s.resolveLockTaskCh, + subID, + span, + startTs, + consumeKVEvents, + advanceResolvedTs, + advanceInterval, + filterLoop, + ) } func (s *subscriptionClient) GetResolvedTsLag() float64 { - pullerMinResolvedTs := uint64(0) - s.totalSpans.RLock() - for _, rt := range s.totalSpans.spanMap { - resolvedTs := rt.resolvedTs.Load() - if pullerMinResolvedTs == 0 || resolvedTs < pullerMinResolvedTs { - pullerMinResolvedTs = resolvedTs - } - } - s.totalSpans.RUnlock() - if pullerMinResolvedTs == 0 { + if s.spanRegistry == nil { return 0 } - pdTime := s.pdClock.CurrentTime() - phyResolvedTs := oracle.ExtractPhysical(pullerMinResolvedTs) - lag := float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3 - return lag -} - -func (r *subscribedSpan) resolveStaleLocks(targetTs uint64) { - util.MustCompareAndMonotonicIncrease(&r.staleLocksTargetTs, targetTs) - res := r.rangeLock.IterAll(r.tryResolveLock) - log.Debug("subscription client finds slow locked ranges", - zap.Uint64("subscriptionID", uint64(r.subID)), - zap.Any("ranges", res)) -} - -type errCache struct { - sync.Mutex - cache []regionErrorInfo - errCh chan regionErrorInfo - notify chan struct{} -} - -const errCacheDispatchBatchSize = 1024 - -func newErrCache() *errCache { - return &errCache{ - cache: make([]regionErrorInfo, 0, 1024), - errCh: make(chan regionErrorInfo, 4096), - notify: make(chan struct{}, 1024), - } -} - -func (e *errCache) add(errInfo regionErrorInfo) { - e.Lock() - defer e.Unlock() - e.cache = append(e.cache, errInfo) - select { - case e.notify <- struct{}{}: - default: - } -} - -func (e *errCache) popBatch(limit int) []regionErrorInfo { - e.Lock() - defer e.Unlock() - if len(e.cache) == 0 { - return nil - } - if limit <= 0 || limit > len(e.cache) { - limit = len(e.cache) - } - batch := make([]regionErrorInfo, limit) - copy(batch, e.cache[:limit]) - clear(e.cache[:limit]) - if limit == len(e.cache) { - e.cache = e.cache[:0] - } else { - e.cache = e.cache[limit:] - } - 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 - } - } - } + return s.spanRegistry.GetResolvedTsLag() } diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index b502c4d7d1..941a14c78e 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -66,10 +66,10 @@ func TestGenerateResolveLockTask(t *testing.T) { } consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span := client.newSubscribedSpan(SubscriptionID(1), rawSpan, 100, consumeKVEvents, advanceResolvedTs, 0, false) - client.totalSpans.spanMap = make(map[SubscriptionID]*subscribedSpan) - client.totalSpans.spanMap[SubscriptionID(1)] = span client.pdClock = pdutil.NewClock4Test() + client.spanRegistry = newSpanRegistry(nil, client.pdClock) + span := client.newSubscribedSpan(SubscriptionID(1), rawSpan, 100, consumeKVEvents, advanceResolvedTs, 0, false) + client.spanRegistry.Add(span) // Lock a range, and then ResolveLock will trigger a task for it. res := span.rangeLock.LockRange(context.Background(), []byte{'b'}, []byte{'c'}, 1, 100) @@ -310,9 +310,10 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { client := &subscriptionClient{ - errCache: newErrCache(), - ds: &mockDynamicStream{}, + eventSink: ®ionEventSink{ds: &mockDynamicStream{}}, } + client.spanRegistry = newSpanRegistry(nil, nil) + client.failureHandler = newRegionFailureHandler(client) rawSpan := heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte("a"), @@ -323,7 +324,7 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { span: rawSpan, rangeLock: regionlock.NewRangeLock(1, rawSpan.StartKey, rawSpan.EndKey, 100), } - client.totalSpans.spanMap = map[SubscriptionID]*subscribedSpan{span.subID: span} + client.spanRegistry.Add(span) res1 := span.rangeLock.LockRange(context.Background(), []byte("a"), []byte("m"), 1, 1) require.Equal(t, regionlock.LockRangeStatusSuccess, res1.Status) @@ -338,7 +339,7 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { lockedRangeState: res1.LockedRangeState, }, &requestCancelledErr{})) - require.Len(t, client.errCache.cache, 1) + require.Len(t, client.failureHandler.cache.cache, 1) require.Len(t, span.rangeLock.IterAll(nil).UnLockedRanges, 1) client.onRegionFail(newRegionErrorInfo(regionInfo{ @@ -348,8 +349,8 @@ func TestOnRegionFailQueuesCanceledErrorCache(t *testing.T) { lockedRangeState: res2.LockedRangeState, }, &requestCancelledErr{})) - require.Len(t, client.errCache.cache, 1) - require.NotContains(t, client.totalSpans.spanMap, span.subID) + require.Len(t, client.failureHandler.cache.cache, 1) + require.Nil(t, client.spanRegistry.Get(span.subID)) } type mockDynamicStream struct{} @@ -383,14 +384,18 @@ func (s *mockDynamicStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] } func TestPushRegionEventToDSUnblocksOnClose(t *testing.T) { + sink := ®ionEventSink{ + ctx: context.Background(), + ds: &mockDynamicStream{}, + } + sink.cond = sync.NewCond(&sink.mu) client := &subscriptionClient{ - ds: &mockDynamicStream{}, + eventSink: sink, regionTaskQueue: priorityqueue.New[PriorityTask](), } client.ctx, client.cancel = context.WithCancel(context.Background()) - client.cond = sync.NewCond(&client.mu) - client.paused.Store(true) + sink.paused.Store(true) done := make(chan struct{}) go func() { From c27f91d99f256c07c423d5ba44122cbccb636a1b Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 15:32:03 +0800 Subject: [PATCH 2/9] remove unnecessary file --- logservice/logpuller/puller_memory_quota.go | 214 ------------------ .../logpuller/puller_memory_quota_test.go | 107 --------- 2 files changed, 321 deletions(-) delete mode 100644 logservice/logpuller/puller_memory_quota.go delete mode 100644 logservice/logpuller/puller_memory_quota_test.go diff --git a/logservice/logpuller/puller_memory_quota.go b/logservice/logpuller/puller_memory_quota.go deleted file mode 100644 index e9f8147290..0000000000 --- a/logservice/logpuller/puller_memory_quota.go +++ /dev/null @@ -1,214 +0,0 @@ -// 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" -) - -const ( - defaultPullerMemoryQuotaBytes = 1 * 1024 * 1024 * 1024 - pullerMemoryQuotaRequestRatio = 0.8 - minPullerEventMemoryBytes = 256 -) - -type pullerMemoryQuotaSnapshot struct { - MaxBytes uint64 - UsedBytes uint64 - RegionRequestLimit uint64 - EventWaiterCount int - RegionReqWaiterCount int - Closed bool -} - -// pullerMemoryQuota provides coarse backpressure for two puller-owned -// boundaries: region events entering dynamic stream and normal region feed -// requests being sent to TiKV. It does not reorder anything; callers wait at -// their current position and continue when memory is released. -type pullerMemoryQuota struct { - mu sync.Mutex - - maxBytes uint64 - regionRequestLimit uint64 - usedBytes uint64 - - eventWaiterCount int - regionReqWaiterCount int - - closed bool - notify chan struct{} -} - -func newPullerMemoryQuota(maxBytes uint64) *pullerMemoryQuota { - if maxBytes == 0 { - maxBytes = defaultPullerMemoryQuotaBytes - } - regionRequestLimit := uint64(float64(maxBytes) * pullerMemoryQuotaRequestRatio) - if regionRequestLimit == 0 { - regionRequestLimit = 1 - } - return &pullerMemoryQuota{ - maxBytes: maxBytes, - regionRequestLimit: regionRequestLimit, - notify: make(chan struct{}), - } -} - -func (q *pullerMemoryQuota) acquireEvent(ctx context.Context, size uint64) (uint64, bool) { - if size < minPullerEventMemoryBytes { - size = minPullerEventMemoryBytes - } - - q.mu.Lock() - for { - if q.closed { - q.mu.Unlock() - return 0, false - } - if q.canAcquireEventLocked(size) { - q.usedBytes += size - q.mu.Unlock() - return size, true - } - - notify := q.notify - q.eventWaiterCount++ - q.mu.Unlock() - select { - case <-ctx.Done(): - q.mu.Lock() - q.eventWaiterCount-- - q.mu.Unlock() - return 0, false - case <-notify: - q.mu.Lock() - q.eventWaiterCount-- - } - } -} - -func (q *pullerMemoryQuota) canAcquireEventLocked(size uint64) bool { - if q.usedBytes == 0 && size > q.maxBytes { - return true - } - if q.usedBytes > q.maxBytes { - return false - } - return size <= q.maxBytes-q.usedBytes -} - -func (q *pullerMemoryQuota) release(size uint64) { - if size == 0 { - return - } - - q.mu.Lock() - if size >= q.usedBytes { - q.usedBytes = 0 - } else { - q.usedBytes -= size - } - q.notifyWaitersLocked() - q.mu.Unlock() -} - -func (q *pullerMemoryQuota) waitRegionRequest(ctx context.Context) bool { - q.mu.Lock() - for { - if q.closed { - q.mu.Unlock() - return false - } - if q.usedBytes < q.regionRequestLimit { - q.mu.Unlock() - return true - } - - notify := q.notify - q.regionReqWaiterCount++ - q.mu.Unlock() - select { - case <-ctx.Done(): - q.mu.Lock() - q.regionReqWaiterCount-- - q.mu.Unlock() - return false - case <-notify: - q.mu.Lock() - q.regionReqWaiterCount-- - } - } -} - -func (q *pullerMemoryQuota) close() { - q.mu.Lock() - q.closed = true - q.notifyWaitersLocked() - q.mu.Unlock() -} - -func (q *pullerMemoryQuota) snapshot() pullerMemoryQuotaSnapshot { - q.mu.Lock() - defer q.mu.Unlock() - return pullerMemoryQuotaSnapshot{ - MaxBytes: q.maxBytes, - UsedBytes: q.usedBytes, - RegionRequestLimit: q.regionRequestLimit, - EventWaiterCount: q.eventWaiterCount, - RegionReqWaiterCount: q.regionReqWaiterCount, - Closed: q.closed, - } -} - -func (q *pullerMemoryQuota) notifyWaitersLocked() { - close(q.notify) - q.notify = make(chan struct{}) -} - -func (s *subscriptionClient) acquireRegionEventQuota(event *regionEvent) bool { - if s == nil || s.memoryQuota == nil { - return true - } - ctx := s.ctx - if ctx == nil { - ctx = context.Background() - } - chargedBytes, ok := s.memoryQuota.acquireEvent(ctx, uint64(event.getSize())) - if !ok { - return false - } - event.quotaBytes = chargedBytes - return true -} - -func (s *subscriptionClient) releaseRegionEventQuota(event regionEvent) { - if s == nil || s.memoryQuota == nil { - return - } - s.memoryQuota.release(event.quotaBytes) -} - -func (s *subscriptionClient) waitRegionRequestQuota(ctx context.Context) error { - if s == nil || s.memoryQuota == nil { - return nil - } - if s.memoryQuota.waitRegionRequest(ctx) { - return nil - } - if ctx.Err() != nil { - return ctx.Err() - } - return context.Canceled -} diff --git a/logservice/logpuller/puller_memory_quota_test.go b/logservice/logpuller/puller_memory_quota_test.go deleted file mode 100644 index f03b4e17e3..0000000000 --- a/logservice/logpuller/puller_memory_quota_test.go +++ /dev/null @@ -1,107 +0,0 @@ -// 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/stretchr/testify/require" -) - -func TestPullerMemoryQuotaEventWaitsForRelease(t *testing.T) { - quota := newPullerMemoryQuota(minPullerEventMemoryBytes) - chargedBytes, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) - require.True(t, ok) - - done := make(chan struct{}) - go func() { - _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) - require.True(t, ok) - close(done) - }() - - select { - case <-done: - t.Fatal("event acquire should wait when quota is full") - case <-time.After(100 * time.Millisecond): - } - - quota.release(chargedBytes) - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("event acquire should resume after release") - } -} - -func TestPullerMemoryQuotaCloseWakesEventWaiter(t *testing.T) { - quota := newPullerMemoryQuota(minPullerEventMemoryBytes) - _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) - require.True(t, ok) - - done := make(chan bool) - go func() { - _, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes) - done <- ok - }() - - select { - case <-done: - t.Fatal("event acquire should wait before quota is closed") - case <-time.After(100 * time.Millisecond): - } - - quota.close() - select { - case ok := <-done: - require.False(t, ok) - case <-time.After(5 * time.Second): - t.Fatal("event acquire should resume after quota close") - } -} - -func TestPullerMemoryQuotaAllowsSingleOversizedEvent(t *testing.T) { - quota := newPullerMemoryQuota(minPullerEventMemoryBytes) - chargedBytes, ok := quota.acquireEvent(context.Background(), minPullerEventMemoryBytes*2) - require.True(t, ok) - require.Equal(t, uint64(minPullerEventMemoryBytes*2), chargedBytes) - require.Equal(t, uint64(minPullerEventMemoryBytes*2), quota.snapshot().UsedBytes) -} - -func TestPullerMemoryQuotaRegionRequestWaitsAtHighWatermark(t *testing.T) { - quota := newPullerMemoryQuota(1000) - chargedBytes, ok := quota.acquireEvent(context.Background(), 800) - require.True(t, ok) - - done := make(chan struct{}) - go func() { - require.True(t, quota.waitRegionRequest(context.Background())) - close(done) - }() - - select { - case <-done: - t.Fatal("region request should wait when usage reaches high watermark") - case <-time.After(100 * time.Millisecond): - } - - quota.release(chargedBytes) - select { - case <-done: - case <-time.After(5 * time.Second): - t.Fatal("region request should resume after release") - } -} From f83181f70328cbab877473d71b900807445c8752 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 15:58:47 +0800 Subject: [PATCH 3/9] small refactor --- logservice/logpuller/region_event_sink.go | 31 ++++++++++++++++++--- logservice/logpuller/subscription_client.go | 27 ++---------------- 2 files changed, 30 insertions(+), 28 deletions(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index ef6024299d..f4a8e96274 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -19,6 +19,7 @@ import ( "sync/atomic" "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/utils/dynstream" "go.uber.org/zap" ) @@ -26,7 +27,7 @@ import ( type regionEventSink struct { ctx context.Context ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] - + // the following three fields are used to manage feedback from ds and notify other goroutines mu sync.Mutex cond *sync.Cond paused atomic.Bool @@ -34,7 +35,6 @@ type regionEventSink struct { func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *regionEventSink { sink := ®ionEventSink{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) @@ -50,6 +50,7 @@ func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *reg ) ds.Start() sink.ds = ds + sink.cond = sync.NewCond(&sink.mu) return sink } @@ -122,8 +123,30 @@ func (s *regionEventSink) Run(ctx context.Context) error { } } -func (s *regionEventSink) Metrics() dynstream.Metrics[int, SubscriptionID] { - return s.ds.GetMetrics() +func (s *regionEventSink) UpdateMetrics() { + dsMetrics := s.ds.GetMetrics() + metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) + metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) + if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 1 { + log.Panic("subscription client should have only one area") + } + if len(dsMetrics.MemoryControl.AreaMemoryMetrics) == 0 { + return + } + + areaMetric := dsMetrics.MemoryControl.AreaMemoryMetrics[0] + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "max", + "default", + "default", + ).Set(float64(areaMetric.MaxMemory())) + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "used", + "default", + "default", + ).Set(float64(areaMetric.MemoryUsage())) } func (s *regionEventSink) Close() { diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 75ecfb34ae..f7195d0e13 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -176,9 +176,9 @@ func NewSubscriptionClient( resolveLockRateLimiter: newResolveLockRateLimiter(), } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) - subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) - subClient.failureHandler = newRegionFailureHandler(subClient) subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) + subClient.failureHandler = newRegionFailureHandler(subClient) + subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) subClient.initMetrics() return subClient @@ -206,28 +206,6 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { case <-ctx.Done(): return ctx.Err() case <-ticker.C: - dsMetrics := s.eventSink.Metrics() - metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) - metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 1 { - log.Panic("subscription client should have only one area") - } - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 0 { - areaMetric := dsMetrics.MemoryControl.AreaMemoryMetrics[0] - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "max", - "default", - "default", - ).Set(float64(areaMetric.MaxMemory())) - metrics.DynamicStreamMemoryUsage.WithLabelValues( - "log-puller", - "used", - "default", - "default", - ).Set(float64(areaMetric.MemoryUsage())) - } - pendingRegionReqCount := 0 s.stores.Range(func(_, value any) bool { store := value.(*requestedStore) @@ -241,6 +219,7 @@ func (s *subscriptionClient) updateMetrics(ctx context.Context) error { }) metrics.SubscriptionClientRequestedRegionCount.WithLabelValues("pending").Set(float64(pendingRegionReqCount)) + s.eventSink.UpdateMetrics() s.spanRegistry.UpdateMetrics() } } From cf031f8cb381c7a7fb6dca2ac299a88613a09d04 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 16:30:16 +0800 Subject: [PATCH 4/9] more refactor --- logservice/logpuller/region_event_handler.go | 7 ++++--- logservice/logpuller/region_event_sink.go | 10 ++++------ logservice/logpuller/subscription_client.go | 6 +----- 3 files changed, 9 insertions(+), 14 deletions(-) diff --git a/logservice/logpuller/region_event_handler.go b/logservice/logpuller/region_event_handler.go index 7e96008c77..777af088bd 100644 --- a/logservice/logpuller/region_event_handler.go +++ b/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. @@ -255,7 +256,7 @@ func (h *regionEventHandler) handleRegionError(state *regionFeedState) { } if stepsToRemoved { worker.takeRegionState(SubscriptionID(state.requestID), state.getRegionID()) - h.subClient.onRegionFail(newRegionErrorInfo(state.getRegionInfo(), err)) + h.failureHandler.Report(newRegionErrorInfo(state.getRegionInfo(), err)) } } diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index f4a8e96274..380ca2ab9e 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -33,7 +33,7 @@ type regionEventSink struct { paused atomic.Bool } -func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *regionEventSink { +func newRegionEventSink(ctx context.Context, failureHandler *regionFailureHandler) *regionEventSink { sink := ®ionEventSink{ctx: ctx} option := dynstream.NewOption() @@ -45,7 +45,7 @@ func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *reg option.EnableMemoryControl = true ds := dynstream.NewParallelDynamicStream( "log-puller", - ®ionEventHandler{subClient: subClient}, + ®ionEventHandler{eventSink: sink, failureHandler: failureHandler}, option, ) ds.Start() @@ -127,10 +127,8 @@ func (s *regionEventSink) UpdateMetrics() { dsMetrics := s.ds.GetMetrics() metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) > 1 { - log.Panic("subscription client should have only one area") - } - if len(dsMetrics.MemoryControl.AreaMemoryMetrics) == 0 { + if len(dsMetrics.MemoryControl.AreaMemoryMetrics) != 1 { + log.Warn("subscription client should have exactly one area") return } diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index f7195d0e13..b8c91f69a5 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -176,8 +176,8 @@ func NewSubscriptionClient( resolveLockRateLimiter: newResolveLockRateLimiter(), } subClient.ctx, subClient.cancel = context.WithCancel(context.Background()) - subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) subClient.failureHandler = newRegionFailureHandler(subClient) + subClient.eventSink = newRegionEventSink(subClient.ctx, subClient.failureHandler) subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) subClient.initMetrics() @@ -274,10 +274,6 @@ func (s *subscriptionClient) Unsubscribe(subID SubscriptionID) { zap.Bool("exists", rt != nil)) } -func (s *subscriptionClient) wakeSubscription(subID SubscriptionID) { - s.eventSink.Wake(subID) -} - func (s *subscriptionClient) pushRegionEventToDS(subID SubscriptionID, event regionEvent) { s.eventSink.Push(subID, event) } From 13d997fd7a26a0e08dd1bd54384d56897725e8c3 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 17:26:10 +0800 Subject: [PATCH 5/9] refactor --- logservice/logpuller/region_event_sink.go | 7 +-- logservice/logpuller/span_registry.go | 43 +++++++--------- logservice/logpuller/subscription_client.go | 50 +++++++------------ .../logpuller/subscription_client_test.go | 45 ++++++++++++++--- 4 files changed, 74 insertions(+), 71 deletions(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 380ca2ab9e..207dd0f96f 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -78,16 +78,11 @@ func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) { return } - ctx := s.ctx - if ctx == nil { - ctx = context.Background() - } - // slow path: wait until paused is false s.mu.Lock() for s.paused.Load() { select { - case <-ctx.Done(): + case <-s.ctx.Done(): s.mu.Unlock() return default: diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go index 7e855e474e..291b44c4ef 100644 --- a/logservice/logpuller/span_registry.go +++ b/logservice/logpuller/span_registry.go @@ -166,6 +166,13 @@ func newSpanRegistry(pd pd.Client, pdClock pdutil.Clock) *spanRegistry { } } +func (r *spanRegistry) Run(ctx context.Context) error { + g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { return r.runResolveLockChecker(ctx) }) + g.Go(func() error { return r.logSlowRegions(ctx) }) + return g.Wait() +} + func (r *spanRegistry) Add(span *subscribedSpan) { r.Lock() defer r.Unlock() @@ -184,46 +191,30 @@ func (r *spanRegistry) Remove(subID SubscriptionID) { delete(r.spans, subID) } -func (r *spanRegistry) GetResolvedTsLag() float64 { +func (r *spanRegistry) UpdateMetrics() { + count := 0 pullerMinResolvedTs := uint64(0) r.RLock() for _, span := range r.spans { + count += span.rangeLock.Len() resolvedTs := span.resolvedTs.Load() if pullerMinResolvedTs == 0 || resolvedTs < pullerMinResolvedTs { pullerMinResolvedTs = resolvedTs } } r.RUnlock() - if pullerMinResolvedTs == 0 { - return 0 - } - pdTime := r.pdClock.CurrentTime() - phyResolvedTs := oracle.ExtractPhysical(pullerMinResolvedTs) - return float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3 -} - -func (r *spanRegistry) UpdateMetrics() { - count := 0 - r.RLock() - for _, span := range r.spans { - count += span.rangeLock.Len() - } - r.RUnlock() metrics.SubscriptionClientSubscribedRegionCount.Set(float64(count)) - resolvedTsLag := r.GetResolvedTsLag() - if resolvedTsLag > 0 { - metrics.LogPullerResolvedTsLag.Set(resolvedTsLag) + if pullerMinResolvedTs != 0 { + pdTime := r.pdClock.CurrentTime() + phyResolvedTs := oracle.ExtractPhysical(pullerMinResolvedTs) + resolvedTsLag := float64(oracle.GetPhysical(pdTime)-phyResolvedTs) / 1e3 + if resolvedTsLag > 0 { + metrics.LogPullerResolvedTsLag.Set(resolvedTsLag) + } } } -func (r *spanRegistry) Run(ctx context.Context) error { - g, ctx := errgroup.WithContext(ctx) - g.Go(func() error { return r.runResolveLockChecker(ctx) }) - g.Go(func() error { return r.logSlowRegions(ctx) }) - return g.Wait() -} - func (r *spanRegistry) runResolveLockChecker(ctx context.Context) error { resolveLockTicker := time.NewTicker(resolveLockTickInterval) defer resolveLockTicker.Stop() diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index b8c91f69a5..15d5002108 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -136,9 +136,12 @@ type subscriptionClient struct { // the credential to connect tikv credential *security.Credential - eventSink *regionEventSink + // failureHandler handles failed regions and owns reschedule/retry decisions. failureHandler *regionFailureHandler - spanRegistry *spanRegistry + // eventSink delivers region events and owns dynstream interaction. + eventSink *regionEventSink + // spanRegistry tracks subscribed spans and owns span-level background tasks. + spanRegistry *spanRegistry // rangeTaskCh is used to receive range tasks. // The tasks will be handled in `handleRangeTask` goroutine. @@ -244,7 +247,18 @@ func (s *subscriptionClient) Subscribe( return } - rt := s.newSubscribedSpan(subID, span, startTs, consumeKVEvents, advanceResolvedTs, advanceInterval, bdrMode) + rt := newSubscribedSpan( + s.ctx, + s.resolveLockRateLimiter, + s.resolveLockTaskCh, + subID, + span, + startTs, + consumeKVEvents, + advanceResolvedTs, + advanceInterval, + bdrMode, + ) s.spanRegistry.Add(rt) s.eventSink.AddPath(rt) @@ -687,33 +701,3 @@ func (s *subscriptionClient) handleResolveLockTasks(ctx context.Context) error { } } } - -func (s *subscriptionClient) newSubscribedSpan( - subID SubscriptionID, - span heartbeatpb.TableSpan, - startTs uint64, - consumeKVEvents func(raw []common.RawKVEntry, wakeCallback func()) bool, - advanceResolvedTs func(ts uint64), - advanceInterval int64, - filterLoop bool, -) *subscribedSpan { - return newSubscribedSpan( - s.ctx, - s.resolveLockRateLimiter, - s.resolveLockTaskCh, - subID, - span, - startTs, - consumeKVEvents, - advanceResolvedTs, - advanceInterval, - filterLoop, - ) -} - -func (s *subscriptionClient) GetResolvedTsLag() float64 { - if s.spanRegistry == nil { - return 0 - } - return s.spanRegistry.GetResolvedTsLag() -} diff --git a/logservice/logpuller/subscription_client_test.go b/logservice/logpuller/subscription_client_test.go index 941a14c78e..95d0921f4d 100644 --- a/logservice/logpuller/subscription_client_test.go +++ b/logservice/logpuller/subscription_client_test.go @@ -68,7 +68,18 @@ func TestGenerateResolveLockTask(t *testing.T) { advanceResolvedTs := func(ts uint64) {} client.pdClock = pdutil.NewClock4Test() client.spanRegistry = newSpanRegistry(nil, client.pdClock) - span := client.newSubscribedSpan(SubscriptionID(1), rawSpan, 100, consumeKVEvents, advanceResolvedTs, 0, false) + span := newSubscribedSpan( + client.ctx, + client.resolveLockRateLimiter, + client.resolveLockTaskCh, + SubscriptionID(1), + rawSpan, + 100, + consumeKVEvents, + advanceResolvedTs, + 0, + false, + ) client.spanRegistry.Add(span) // Lock a range, and then ResolveLock will trigger a task for it. @@ -136,12 +147,12 @@ func TestResolveLockTaskDeduplicatedAcrossSubscribedSpans(t *testing.T) { consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span1 := client.newSubscribedSpan(SubscriptionID(1), heartbeatpb.TableSpan{ + span1 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte{'a'}, EndKey: []byte{'z'}, }, 100, consumeKVEvents, advanceResolvedTs, 0, false) - span2 := client.newSubscribedSpan(SubscriptionID(2), heartbeatpb.TableSpan{ + span2 := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(2), heartbeatpb.TableSpan{ TableID: 2, StartKey: []byte{'a'}, EndKey: []byte{'z'}, @@ -244,7 +255,18 @@ func TestResolveLockTaskDroppedWhenChannelFull(t *testing.T) { } consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span := client.newSubscribedSpan(SubscriptionID(1), rawSpan, 100, consumeKVEvents, advanceResolvedTs, 0, false) + span := newSubscribedSpan( + client.ctx, + client.resolveLockRateLimiter, + client.resolveLockTaskCh, + SubscriptionID(1), + rawSpan, + 100, + consumeKVEvents, + advanceResolvedTs, + 0, + false, + ) res := span.rangeLock.LockRange(context.Background(), []byte{'b'}, []byte{'c'}, 1, 100) require.Equal(t, regionlock.LockRangeStatusSuccess, res.Status) @@ -292,7 +314,18 @@ func TestStopTaskUsesSubscribedSpanFilterLoop(t *testing.T) { } consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span := client.newSubscribedSpan(SubscriptionID(1), rawSpan, 100, consumeKVEvents, advanceResolvedTs, 0, true) + span := newSubscribedSpan( + client.ctx, + client.resolveLockRateLimiter, + client.resolveLockTaskCh, + SubscriptionID(1), + rawSpan, + 100, + consumeKVEvents, + advanceResolvedTs, + 0, + true, + ) res := span.rangeLock.LockRange(context.Background(), rawSpan.StartKey, rawSpan.EndKey, 1, 1) require.Equal(t, regionlock.LockRangeStatusSuccess, res.Status) @@ -709,7 +742,7 @@ func TestGetResolvedTargetTs(t *testing.T) { consumeKVEvents := func(_ []common.RawKVEntry, _ func()) bool { return false } advanceResolvedTs := func(ts uint64) {} - span := client.newSubscribedSpan(SubscriptionID(1), heartbeatpb.TableSpan{ + span := newSubscribedSpan(client.ctx, client.resolveLockRateLimiter, client.resolveLockTaskCh, SubscriptionID(1), heartbeatpb.TableSpan{ TableID: 1, StartKey: []byte{'a'}, EndKey: []byte{'z'}, From 8e0ba23128fc7b58e006622572e429694b1fd1d0 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 17:39:15 +0800 Subject: [PATCH 6/9] fix --- .../logpuller/region_failure_handler.go | 35 ++++--------------- logservice/logpuller/subscription_client.go | 1 + 2 files changed, 7 insertions(+), 29 deletions(-) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index 12fd2f0fb7..4871824b41 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -64,46 +64,23 @@ func (r *regionFailureHandler) Report(errInfo regionErrorInfo) { } func (r *regionFailureHandler) Run(ctx context.Context) error { - ticker := time.NewTicker(10 * time.Millisecond) - defer ticker.Stop() - - handlePendingFailures := func() error { - for { - batch := r.cache.popBatch(errCacheDispatchBatchSize) - for _, errInfo := range batch { - select { - case <-ctx.Done(): - log.Info("subscription client handle errors and exit") - return ctx.Err() - default: - } - if err := r.handleError(ctx, errInfo); err != nil { - return err - } - } - if len(batch) < errCacheDispatchBatchSize { - return nil - } - } - } - for { select { case <-ctx.Done(): log.Info("subscription client handle errors and exit") return ctx.Err() - case <-ticker.C: - if err := handlePendingFailures(); err != nil { - return err - } - case <-r.cache.notify: - if err := handlePendingFailures(); err != nil { + case errInfo := <-r.cache.errCh: + if err := r.handleError(ctx, errInfo); err != nil { return err } } } } +func (r *regionFailureHandler) Dispatch(ctx context.Context) error { + return r.cache.dispatch(ctx) +} + func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionErrorInfo) error { err := errors.Cause(errInfo.err) //nolint:errorlint // converting large type switch to errors.As is a significant refactor diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 15d5002108..180cc11029 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -306,6 +306,7 @@ func (s *subscriptionClient) Run(ctx context.Context) error { g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.handleRangeTasks(ctx) }) g.Go(func() error { return s.handleRegions(ctx, g) }) + g.Go(func() error { return s.failureHandler.Dispatch(ctx) }) g.Go(func() error { return s.failureHandler.Run(ctx) }) g.Go(func() error { return s.handleResolveLockTasks(ctx) }) g.Go(func() error { return s.spanRegistry.Run(ctx) }) From 12232a4c0882a8843b5703f43801c29621f95c38 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 17:45:56 +0800 Subject: [PATCH 7/9] refactor --- .../logpuller/region_failure_handler.go | 28 ++++++++++--------- logservice/logpuller/subscription_client.go | 1 - 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index 4871824b41..e216fef714 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -23,6 +23,7 @@ import ( "github.com/pingcap/ticdc/pkg/metrics" "github.com/tikv/client-go/v2/tikv" "go.uber.org/zap" + "golang.org/x/sync/errgroup" ) var ( @@ -64,21 +65,22 @@ func (r *regionFailureHandler) Report(errInfo regionErrorInfo) { } func (r *regionFailureHandler) Run(ctx context.Context) error { - for { - select { - case <-ctx.Done(): - log.Info("subscription client handle errors and exit") - return ctx.Err() - case errInfo := <-r.cache.errCh: - if err := r.handleError(ctx, errInfo); err != nil { - return err + g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { return r.cache.dispatch(ctx) }) + g.Go(func() error { + for { + select { + case <-ctx.Done(): + log.Info("subscription client handle errors and exit") + return ctx.Err() + case errInfo := <-r.cache.errCh: + if err := r.handleError(ctx, errInfo); err != nil { + return err + } } } - } -} - -func (r *regionFailureHandler) Dispatch(ctx context.Context) error { - return r.cache.dispatch(ctx) + }) + return g.Wait() } func (r *regionFailureHandler) handleError(ctx context.Context, errInfo regionErrorInfo) error { diff --git a/logservice/logpuller/subscription_client.go b/logservice/logpuller/subscription_client.go index 180cc11029..15d5002108 100644 --- a/logservice/logpuller/subscription_client.go +++ b/logservice/logpuller/subscription_client.go @@ -306,7 +306,6 @@ func (s *subscriptionClient) Run(ctx context.Context) error { g.Go(func() error { return s.eventSink.Run(ctx) }) g.Go(func() error { return s.handleRangeTasks(ctx) }) g.Go(func() error { return s.handleRegions(ctx, g) }) - g.Go(func() error { return s.failureHandler.Dispatch(ctx) }) g.Go(func() error { return s.failureHandler.Run(ctx) }) g.Go(func() error { return s.handleResolveLockTasks(ctx) }) g.Go(func() error { return s.spanRegistry.Run(ctx) }) From f2167b360b37d4c2b87339ae8fe3ab14e348a054 Mon Sep 17 00:00:00 2001 From: lidezhu Date: Thu, 9 Jul 2026 18:36:24 +0800 Subject: [PATCH 8/9] add some test --- .../logpuller/region_event_sink_test.go | 128 ++++++++++++++++++ logservice/logpuller/span_registry.go | 10 +- logservice/logpuller/span_registry_test.go | 65 +++++++++ 3 files changed, 198 insertions(+), 5 deletions(-) create mode 100644 logservice/logpuller/region_event_sink_test.go create mode 100644 logservice/logpuller/span_registry_test.go diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go new file mode 100644 index 0000000000..2ce7a10baf --- /dev/null +++ b/logservice/logpuller/region_event_sink_test.go @@ -0,0 +1,128 @@ +// 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" + "testing" + "time" + + "github.com/pingcap/ticdc/utils/dynstream" + "github.com/stretchr/testify/require" +) + +type mockRegionEventSinkStream struct { + feedbackCh chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan] + pushCount atomic.Int32 + pushCh chan struct{} +} + +func newMockRegionEventSinkStream() *mockRegionEventSinkStream { + return &mockRegionEventSinkStream{ + feedbackCh: make(chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan], 2), + pushCh: make(chan struct{}, 1), + } +} + +func (s *mockRegionEventSinkStream) Start() {} + +func (s *mockRegionEventSinkStream) Close() {} + +func (s *mockRegionEventSinkStream) Push(_ SubscriptionID, _ regionEvent) { + s.pushCount.Add(1) + s.pushCh <- struct{}{} +} + +func (s *mockRegionEventSinkStream) Wake(_ SubscriptionID) {} + +func (s *mockRegionEventSinkStream) Feedback() <-chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan] { + return s.feedbackCh +} + +func (s *mockRegionEventSinkStream) AddPath(_ SubscriptionID, _ *subscribedSpan, _ ...dynstream.AreaSettings) error { + return nil +} + +func (s *mockRegionEventSinkStream) RemovePath(_ SubscriptionID) error { + return nil +} + +func (s *mockRegionEventSinkStream) Release(_ SubscriptionID) {} + +func (s *mockRegionEventSinkStream) SetAreaSettings(_ int, _ dynstream.AreaSettings) {} + +func (s *mockRegionEventSinkStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] { + return dynstream.Metrics[int, SubscriptionID]{} +} + +func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + ds := newMockRegionEventSinkStream() + sink := ®ionEventSink{ + ctx: ctx, + ds: ds, + } + sink.cond = sync.NewCond(&sink.mu) + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- sink.Run(ctx) + }() + + ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ + FeedbackType: dynstream.PauseArea, + } + require.Eventually(t, sink.paused.Load, time.Second, 10*time.Millisecond) + + pushDone := make(chan struct{}) + go func() { + sink.Push(SubscriptionID(1), regionEvent{resolvedTs: 100}) + close(pushDone) + }() + + select { + case <-pushDone: + t.Fatal("Push should block while the sink is paused") + case <-time.After(100 * time.Millisecond): + } + require.Equal(t, int32(0), ds.pushCount.Load()) + + ds.feedbackCh <- dynstream.Feedback[int, SubscriptionID, *subscribedSpan]{ + FeedbackType: dynstream.ResumeArea, + } + require.Eventually(t, func() bool { return !sink.paused.Load() }, time.Second, 10*time.Millisecond) + + select { + case <-ds.pushCh: + case <-time.After(time.Second): + t.Fatal("Push should resume after ResumeArea feedback") + } + select { + case <-pushDone: + case <-time.After(time.Second): + t.Fatal("Push should return after ResumeArea feedback") + } + + cancel() + select { + case err := <-runErrCh: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("Run should exit after context cancellation") + } +} diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go index 291b44c4ef..f1cd8930a6 100644 --- a/logservice/logpuller/span_registry.go +++ b/logservice/logpuller/span_registry.go @@ -81,11 +81,6 @@ type spanRegistry struct { pdClock pdutil.Clock } -type spanAndTargetTs struct { - span *subscribedSpan - targetTs uint64 -} - func newSubscribedSpan( ctx context.Context, resolveLockRateLimiter *resolveLockRateLimiter, @@ -215,6 +210,11 @@ func (r *spanRegistry) UpdateMetrics() { } } +type spanAndTargetTs struct { + span *subscribedSpan + targetTs uint64 +} + func (r *spanRegistry) runResolveLockChecker(ctx context.Context) error { resolveLockTicker := time.NewTicker(resolveLockTickInterval) defer resolveLockTicker.Stop() diff --git a/logservice/logpuller/span_registry_test.go b/logservice/logpuller/span_registry_test.go new file mode 100644 index 0000000000..e41df10887 --- /dev/null +++ b/logservice/logpuller/span_registry_test.go @@ -0,0 +1,65 @@ +// 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/ticdc/heartbeatpb" + "github.com/pingcap/ticdc/logservice/logpuller/regionlock" + "github.com/pingcap/ticdc/pkg/metrics" + "github.com/pingcap/ticdc/pkg/pdutil" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/oracle" +) + +func TestSpanRegistryUpdateMetrics(t *testing.T) { + now := time.Unix(1_700_000_000, 0) + registry := newSpanRegistry(nil, pdutil.NewClockWithValue4Test(now)) + + span1 := &subscribedSpan{ + subID: SubscriptionID(1), + rangeLock: regionlock.NewRangeLock(1, []byte("a"), []byte("m"), 100), + span: heartbeatpb.TableSpan{TableID: 1, StartKey: []byte("a"), EndKey: []byte("m")}, + } + span1.resolvedTs.Store(oracle.GoTimeToTS(now.Add(-5 * time.Second))) + + span2 := &subscribedSpan{ + subID: SubscriptionID(2), + rangeLock: regionlock.NewRangeLock(2, []byte("m"), []byte("z"), 100), + span: heartbeatpb.TableSpan{TableID: 2, StartKey: []byte("m"), EndKey: []byte("z")}, + } + span2.resolvedTs.Store(oracle.GoTimeToTS(now.Add(-2 * time.Second))) + + lockRange := func(t *testing.T, span *subscribedSpan, start, end string, regionID uint64) { + t.Helper() + res := span.rangeLock.LockRange(context.Background(), []byte(start), []byte(end), regionID, 100) + require.Equal(t, regionlock.LockRangeStatusSuccess, res.Status) + } + + lockRange(t, span1, "a", "g", 11) + lockRange(t, span1, "g", "m", 12) + lockRange(t, span2, "m", "z", 21) + + registry.Add(span1) + registry.Add(span2) + + registry.UpdateMetrics() + + require.Equal(t, float64(3), testutil.ToFloat64(metrics.SubscriptionClientSubscribedRegionCount)) + require.InDelta(t, 5.0, testutil.ToFloat64(metrics.LogPullerResolvedTsLag), 0.001) +} From 125b52b951fef23ae010950ed85664e5216546df Mon Sep 17 00:00:00 2001 From: lidezhu Date: Fri, 10 Jul 2026 16:49:10 +0800 Subject: [PATCH 9/9] address comment --- logservice/logpuller/region_event_sink.go | 5 ++ .../logpuller/region_event_sink_test.go | 89 ++++++++++++++++++- .../logpuller/region_failure_handler.go | 1 + logservice/logpuller/span_registry.go | 1 + 4 files changed, 95 insertions(+), 1 deletion(-) diff --git a/logservice/logpuller/region_event_sink.go b/logservice/logpuller/region_event_sink.go index 207dd0f96f..3fb1771461 100644 --- a/logservice/logpuller/region_event_sink.go +++ b/logservice/logpuller/region_event_sink.go @@ -24,6 +24,7 @@ import ( "go.uber.org/zap" ) +// regionEventSink delivers region events to dynstream and owns push-side flow control. type regionEventSink struct { ctx context.Context ds dynstream.DynamicStream[int, SubscriptionID, regionEvent, *subscribedSpan, *regionEventHandler] @@ -122,6 +123,10 @@ func (s *regionEventSink) UpdateMetrics() { dsMetrics := s.ds.GetMetrics() metricSubscriptionClientDSChannelSize.Set(float64(dsMetrics.EventChanSize)) metricSubscriptionClientDSPendingQueueLen.Set(float64(dsMetrics.PendingQueueLen)) + + if len(dsMetrics.MemoryControl.AreaMemoryMetrics) == 0 { + return + } if len(dsMetrics.MemoryControl.AreaMemoryMetrics) != 1 { log.Warn("subscription client should have exactly one area") return diff --git a/logservice/logpuller/region_event_sink_test.go b/logservice/logpuller/region_event_sink_test.go index 2ce7a10baf..5928a1525a 100644 --- a/logservice/logpuller/region_event_sink_test.go +++ b/logservice/logpuller/region_event_sink_test.go @@ -20,7 +20,9 @@ import ( "testing" "time" + "github.com/pingcap/ticdc/pkg/metrics" "github.com/pingcap/ticdc/utils/dynstream" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" ) @@ -28,6 +30,7 @@ type mockRegionEventSinkStream struct { feedbackCh chan dynstream.Feedback[int, SubscriptionID, *subscribedSpan] pushCount atomic.Int32 pushCh chan struct{} + metrics dynstream.Metrics[int, SubscriptionID] } func newMockRegionEventSinkStream() *mockRegionEventSinkStream { @@ -65,7 +68,7 @@ func (s *mockRegionEventSinkStream) Release(_ SubscriptionID) {} func (s *mockRegionEventSinkStream) SetAreaSettings(_ int, _ dynstream.AreaSettings) {} func (s *mockRegionEventSinkStream) GetMetrics() dynstream.Metrics[int, SubscriptionID] { - return dynstream.Metrics[int, SubscriptionID]{} + return s.metrics } func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { @@ -126,3 +129,87 @@ func TestRegionEventSinkRunPausesAndResumesPush(t *testing.T) { t.Fatal("Run should exit after context cancellation") } } + +func TestRegionEventSinkUpdateMetrics(t *testing.T) { + t.Run("empty area metrics returns after queue gauges", func(t *testing.T) { + ds := newMockRegionEventSinkStream() + ds.metrics = dynstream.Metrics[int, SubscriptionID]{ + EventChanSize: 11, + PendingQueueLen: 22, + } + + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "max", + "default", + "default", + ).Set(123) + metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "used", + "default", + "default", + ).Set(456) + + sink := ®ionEventSink{ + ctx: context.Background(), + ds: ds, + } + sink.cond = sync.NewCond(&sink.mu) + sink.UpdateMetrics() + + require.Equal(t, float64(11), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) + require.Equal(t, float64(22), testutil.ToFloat64(metricSubscriptionClientDSPendingQueueLen)) + require.Equal(t, float64(123), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "max", + "default", + "default", + ))) + require.Equal(t, float64(456), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "used", + "default", + "default", + ))) + }) + + t.Run("single area metrics updates memory gauges", func(t *testing.T) { + ds := newMockRegionEventSinkStream() + ds.metrics = dynstream.Metrics[int, SubscriptionID]{ + EventChanSize: 33, + PendingQueueLen: 44, + MemoryControl: dynstream.MemoryMetric[int, SubscriptionID]{ + AreaMemoryMetrics: []dynstream.AreaMemoryMetric[int, SubscriptionID]{ + { + UsedMemoryValue: 55, + MaxMemoryValue: 66, + PathMaxMemoryValue: 66, + }, + }, + }, + } + + sink := ®ionEventSink{ + ctx: context.Background(), + ds: ds, + } + sink.cond = sync.NewCond(&sink.mu) + sink.UpdateMetrics() + + require.Equal(t, float64(33), testutil.ToFloat64(metricSubscriptionClientDSChannelSize)) + require.Equal(t, float64(44), testutil.ToFloat64(metricSubscriptionClientDSPendingQueueLen)) + require.Equal(t, float64(66), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "max", + "default", + "default", + ))) + require.Equal(t, float64(55), testutil.ToFloat64(metrics.DynamicStreamMemoryUsage.WithLabelValues( + "log-puller", + "used", + "default", + "default", + ))) + }) +} diff --git a/logservice/logpuller/region_failure_handler.go b/logservice/logpuller/region_failure_handler.go index e216fef714..ef86f55876 100644 --- a/logservice/logpuller/region_failure_handler.go +++ b/logservice/logpuller/region_failure_handler.go @@ -39,6 +39,7 @@ var ( metricKvCongestedCounter = metrics.EventFeedErrorCounter.WithLabelValues("KvCongested") ) +// regionFailureHandler handles failed regions and owns retry and reschedule decisions. type regionFailureHandler struct { cache *errCache client *subscriptionClient diff --git a/logservice/logpuller/span_registry.go b/logservice/logpuller/span_registry.go index f1cd8930a6..d69a06d831 100644 --- a/logservice/logpuller/span_registry.go +++ b/logservice/logpuller/span_registry.go @@ -73,6 +73,7 @@ type subscribedSpan struct { resolvedTs atomic.Uint64 } +// spanRegistry tracks subscribed spans and owns span-level background maintenance. type spanRegistry struct { sync.RWMutex spans map[SubscriptionID]*subscribedSpan