logpuller: split subscription runtime helpers from subscription client#5603
logpuller: split subscription runtime helpers from subscription client#5603lidezhu wants to merge 4 commits into
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
do-not-merge/needs-linked-issue
do-not-merge/work-in-progress
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
|
Important Review skippedDraft detected. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ti-chi-bot
Bot
added
size/XXL
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the logpuller package by modularizing subscriptionClient and extracting its responsibilities into dedicated components: regionEventSink for managing event streams, regionFailureHandler for error dispatching, and spanRegistry for tracking subscribed spans. Additionally, a pullerMemoryQuota is introduced to handle memory backpressure. Feedback on these changes highlights a missing memoryQuota field and initialization in subscriptionClient that would cause compilation errors and nil pointer dereferences. The review also recommends cleaning up unused fields and methods in errCache, and refactoring the sync.Cond synchronization in regionEventSink to a safer, channel-based signaling mechanism to prevent goroutine leaks during context cancellation.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| eventSink *regionEventSink | ||
| failureHandler *regionFailureHandler | ||
| spanRegistry *spanRegistry |
There was a problem hiding this comment.
The subscriptionClient struct is missing the memoryQuota field, which is referenced in puller_memory_quota.go (e.g., s.memoryQuota.acquireEvent). This will cause a compilation error. Please add the memoryQuota field to the struct.
| eventSink *regionEventSink | |
| failureHandler *regionFailureHandler | |
| spanRegistry *spanRegistry | |
| eventSink *regionEventSink | |
| failureHandler *regionFailureHandler | |
| spanRegistry *spanRegistry | |
| memoryQuota *pullerMemoryQuota |
| subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) | ||
| subClient.failureHandler = newRegionFailureHandler(subClient) | ||
| subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) |
There was a problem hiding this comment.
The memoryQuota field of subscriptionClient is never initialized in NewSubscriptionClient, which will lead to nil pointer dereferences when accessing it. Please initialize it using newPullerMemoryQuota(0).
| subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) | |
| subClient.failureHandler = newRegionFailureHandler(subClient) | |
| subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) | |
| subClient.spanRegistry = newSpanRegistry(subClient.pd, subClient.pdClock) | |
| subClient.failureHandler = newRegionFailureHandler(subClient) | |
| subClient.eventSink = newRegionEventSink(subClient.ctx, subClient) | |
| subClient.memoryQuota = newPullerMemoryQuota(0) |
| type errCache struct { | ||
| sync.Mutex | ||
| cache []regionErrorInfo | ||
| errCh chan regionErrorInfo | ||
| notify chan struct{} | ||
| } |
There was a problem hiding this comment.
The errCh field in errCache is completely unused in the new regionFailureHandler implementation, as errors are now popped directly from the cache via popBatch and handled in Run. To improve maintainability and clean up dead code, we should remove errCh and its associated unused methods (dispatchBatch and dispatch).
type errCache struct {
sync.Mutex
cache []regionErrorInfo
notify chan struct{}
}| func newErrCache() *errCache { | ||
| return &errCache{ | ||
| cache: make([]regionErrorInfo, 0, 1024), | ||
| errCh: make(chan regionErrorInfo, 4096), | ||
| notify: make(chan struct{}, 1024), | ||
| } | ||
| } |
There was a problem hiding this comment.
Since errCh is unused, we should also remove its initialization in newErrCache.
func newErrCache() *errCache {
return &errCache{
cache: make([]regionErrorInfo, 0, 1024),
notify: make(chan struct{}, 1024),
}
}| mu sync.Mutex | ||
| cond *sync.Cond | ||
| paused atomic.Bool | ||
| } |
There was a problem hiding this comment.
Using sync.Cond with select and default in Push is a known anti-pattern in Go because s.cond.Wait() cannot be interrupted by context cancellation. If the context is canceled while a goroutine is blocked in s.cond.Wait(), it will remain blocked until s.cond.Broadcast() is called, potentially leaking goroutines or delaying shutdown.
We can refactor this to use a channel-based signaling mechanism (resumeCh), which is completely lock-free on the fast path, safer, and fully interruptible by context cancellation.
| mu sync.Mutex | |
| cond *sync.Cond | |
| paused atomic.Bool | |
| } | |
| mu sync.Mutex | |
| paused bool | |
| resumeCh chan struct{} |
| func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *regionEventSink { | ||
| sink := ®ionEventSink{ctx: ctx} | ||
| sink.cond = sync.NewCond(&sink.mu) |
There was a problem hiding this comment.
Initialize the resumeCh channel in the constructor and close it initially since the sink is not paused by default.
| func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *regionEventSink { | |
| sink := ®ionEventSink{ctx: ctx} | |
| sink.cond = sync.NewCond(&sink.mu) | |
| func newRegionEventSink(ctx context.Context, subClient *subscriptionClient) *regionEventSink { | |
| sink := ®ionEventSink{ | |
| ctx: ctx, | |
| resumeCh: make(chan struct{}), | |
| } | |
| close(sink.resumeCh) // initially not paused |
| 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) | ||
| } |
There was a problem hiding this comment.
Refactor Push to use the channel-based signaling mechanism. This is lock-free on the fast path and fully interruptible by context cancellation. We also add a defensive check for resumeCh == nil to ensure compatibility with unit tests that manually construct regionEventSink without calling the constructor.
func (s *regionEventSink) Push(subID SubscriptionID, event regionEvent) {
s.mu.Lock()
resumeCh := s.resumeCh
s.mu.Unlock()
if resumeCh == nil {
s.ds.Push(subID, event)
return
}
select {
case <-s.ctx.Done():
return
case <-resumeCh:
s.ds.Push(subID, event)
}
}| 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") |
There was a problem hiding this comment.
Update the Run loop to manage the resumeCh channel when pausing and resuming.
case dynstream.PauseArea:
s.mu.Lock()
if !s.paused {
s.paused = true
s.resumeCh = make(chan struct{})
}
s.mu.Unlock()
log.Info("subscription client pause push region event")
case dynstream.ResumeArea:
s.mu.Lock()
if s.paused {
s.paused = false
close(s.resumeCh)
}
s.mu.Unlock()
log.Info("subscription client resume push region event")| func (s *regionEventSink) Close() { | ||
| s.mu.Lock() | ||
| s.paused.Store(false) | ||
| s.cond.Broadcast() | ||
| s.mu.Unlock() | ||
| s.ds.Close() | ||
| } |
There was a problem hiding this comment.
Update Close to safely close the resumeCh channel if currently paused.
| func (s *regionEventSink) Close() { | |
| s.mu.Lock() | |
| s.paused.Store(false) | |
| s.cond.Broadcast() | |
| s.mu.Unlock() | |
| s.ds.Close() | |
| } | |
| func (s *regionEventSink) Close() { | |
| s.mu.Lock() | |
| if s.paused { | |
| s.paused = false | |
| close(s.resumeCh) | |
| } | |
| s.mu.Unlock() | |
| s.ds.Close() | |
| } |
1 participant
What problem does this PR solve?
Issue Number: close #5604
What is changed and how it works?
Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note