logpuller: refactor region request worker lifecycle - #5619
Conversation
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
release-note
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (4)
💤 Files with no reviewable changes (2)
📝 WalkthroughWalkthroughThe puller replaces request-cache scheduling with typed scan-priority admission control. It adds bounded request windows, region tracking, worker lifecycle handling, configurable limits, and explicit deregistration queues. ChangesRegion scheduling and admission control
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant SubscriptionClient
participant RegionAdmissionController
participant RegionRequestWorker
participant RegionTracker
participant EventSink
SubscriptionClient->>RegionAdmissionController: submit prioritized region task
RegionAdmissionController->>RegionRequestWorker: admit region request
RegionRequestWorker->>RegionTracker: add or remove region feed state
RegionRequestWorker->>EventSink: publish region events and resolved timestamps
RegionRequestWorker->>RegionAdmissionController: finish or abort admission lease
Possibly related PRs
Suggested labels: Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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
the
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 region request and subscription lifecycle management in the logpuller service. It replaces the old channel-based requestCache and manual state-tracking map in regionRequestWorker with a simplified requestCache (using a new notifyqueue.Queue utility) and a dedicated regionTracker. Additionally, subscription deregistration is now broadcasted directly to workers' control queues rather than being routed through the main region task queue. There are no review comments provided, so I have no feedback to evaluate.
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.
|
[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 |
ti-chi-bot
Bot
removed
the
do-not-merge/work-in-progress
|
/gemini review |
|
@coderabbitai review |
✅ Action performedReview finished.
|
|
/gemini summary |
lidezhu
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
logservice/logpuller/region_event_handler.go (1)
245-260: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winGate
ReportonRemoveIfsuccess.
markRemoved()can returntrueindependently of the tracker entry for this state, whileRemoveIfreturnsfalsewhen another state owns that subscription/region. Reporting the failure in that case creates duplicate failure handling for a region that already has an owner.🐛 Proposed fix
if stepsToRemoved { - worker.tracker.RemoveIf(SubscriptionID(state.requestID), state.getRegionID(), state) - h.failureHandler.Report(newRegionErrorInfo(state.getRegionInfo(), err)) + if worker.tracker.RemoveIf(SubscriptionID(state.requestID), state.getRegionID(), state) { + h.failureHandler.Report(newRegionErrorInfo(state.getRegionInfo(), err)) + } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_event_handler.go` around lines 245 - 260, In handleRegionError, only call failureHandler.Report after worker.tracker.RemoveIf succeeds; retain the existing markRemoved check and avoid reporting when RemoveIf indicates another state owns the subscription/region.
🧹 Nitpick comments (14)
logservice/logpuller/region_request_worker_test.go (3)
165-173: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
pushCountshadows the embedded field.
countingRegionEventDynamicStreamembedsmockRegionEventDynamicStream, which already declarespushCount, and then declares its ownpushCount. Go resolvesm.pushCountto the outer field because it has the shallower depth, so the current code is correct.The type now carries two counters with the same name. If a later change calls the embedded
Push, that call increments the inner counter and the benchmark assertion reads the outer one. Rename the field to make the two counters distinct.♻️ Proposed refactor
type countingRegionEventDynamicStream struct { mockRegionEventDynamicStream - pushCount int + counted int } func (m *countingRegionEventDynamicStream) Push(_ SubscriptionID, _ regionEvent) { - m.pushCount++ + m.counted++ }Update the benchmark assertion to read
ds.counted.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_request_worker_test.go` around lines 165 - 173, Rename the outer pushCount field in countingRegionEventDynamicStream to counted, keep Push incrementing that outer counter, and update the benchmark assertion to read ds.counted. Leave the embedded mockRegionEventDynamicStream counter unchanged.
69-80: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse a fixed timestamp in
admitRegionRequest.
admitRegionRequestderivescurrentTsfromtime.Now(). Line 304 in the same file callssubmitRegionForAdmissionwith the fixed value1. If the admission controller comparescurrentTsagainst the region checkpoint to decide priority or eligibility, this helper makes those decisions depend on the wall clock and on the region fixture data.Accept the timestamp as a parameter, or use a fixed constant, so every test that uses this helper stays deterministic.
♻️ Proposed refactor
func admitRegionRequest( t *testing.T, controller *regionAdmissionController, region regionInfo, ) *regionReq { t.Helper() - currentTs := oracle.GoTimeToTS(time.Now()) - submitRegionForAdmission(t, controller, region, currentTs) + const testCurrentTs = uint64(1) + submitRegionForAdmission(t, controller, region, testCurrentTs) req, err := controller.pop(t.Context(), nil) require.NoError(t, err) return req }As per coding guidelines: "Prefer focused deterministic tests; see docs/agents/testing.md before adding or changing tests."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_request_worker_test.go` around lines 69 - 80, Make admitRegionRequest deterministic by removing its time.Now()-derived timestamp and using a caller-supplied timestamp or fixed test constant, while preserving the existing submitRegionForAdmission and pop flow. Update all callers, including the case currently passing timestamp 1, to use the chosen deterministic value.Source: Coding guidelines
397-404: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBoth stream-recovery tests duplicate
handleStreamFailureinstead of calling it.handleStreamFailureis a closure declared insideRuninlogservice/logpuller/region_request_worker.go, so no test can invoke it. Both tests work around this by copying parts of its body, which means neither test protects the production recovery path. ExtracthandleStreamFailureinto a method onregionRequestWorkerand call it from both tests.
logservice/logpuller/region_request_worker_test.go#L397-L404: replace the copied tracker-drain and event-push loop with a call to the extracted method, then keep the assertions onadmission.stats().inflight,req.abort(), andds.pushCount.logservice/logpuller/region_request_worker_test.go#L451-L453: replace the copied admission-drain andonRegionFailloop with a call to the extracted method, then keep the assertions onadmission.stats().pendingand the failure-handler cache.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_request_worker_test.go` around lines 397 - 404, Extract the closure handleStreamFailure from regionRequestWorker.Run into a regionRequestWorker method, preserving its recovery behavior. In logservice/logpuller/region_request_worker_test.go:397-404, replace the duplicated tracker-drain and event-push logic with that method call while retaining the inflight, abort, and pushCount assertions; at 451-453, replace the duplicated admission-drain and onRegionFail loop similarly while retaining the pending and failure-handler-cache assertions.Source: Coding guidelines
logservice/logpuller/subscription_client_test.go (2)
642-651: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd coverage for a deregistration that matches a pending admission task.
The test submits a task for subscription 2 and deregisters subscription 1, so the two never interact. The interesting case is a deregistration for the subscription that owns the pending task.
sendDeregisterRequestinregion_request_worker.goclears only tracker entries, so a queued admission task for a deregistered subscription survives and is sent later. It then fails thesubscribedSpan.stoppedcheck insendRegionRequest. Add a case that pins this behaviour.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/subscription_client_test.go` around lines 642 - 651, Update the deregistration test around NewRegionPriorityTask and client.broadcastDeregister to submit and deregister the same subscription, then assert the pending admission task remains queued and is later processed through the existing request path, covering the stopped subscribedSpan behavior in sendDeregisterRequest and sendRegionRequest.
404-449: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd a case for the unspecified scan priority.
The table covers
SCAN_PRIORITY_HIGHandSCAN_PRIORITY_LOWonly.cdcpb.ScanPriority_SCAN_PRIORITY_UNSPECIFIEDis the zero value, andregion_request_worker.gocallsnormalizeScanPriorityon the way out, which implies the unset value reaches this code in practice.Add a case with the zero value to pin down the retry behaviour for an unset priority.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/subscription_client_test.go` around lines 404 - 449, Extend the table-driven tests around the existing high/low cases to cover cdcpb.ScanPriority_SCAN_PRIORITY_UNSPECIFIED (the zero value), including its retry input and expected priority. Use the test’s existing fields and naming pattern to pin down normalizeScanPriority behavior for an unset priority.logservice/logpuller/region_request_worker.go (2)
247-254: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe deferred log loses the underlying stream error.
erris the named return value. Line 247 assigns theg.Wait()result to it, but line 252 overwrites it with&storeStreamErr{}before the deferred log at line 214 runs. The deferred log therefore always reportsstoreStreamErrand never the concrete cause fromRecvorSend.Log the original error before you replace it.
♻️ Proposed refactor to keep the original cause
err = g.Wait() if err != nil { + log.Warn("region request worker stream failed", + zap.Uint64("workerID", s.workerID), + zap.String("addr", s.store.storeAddr), + zap.Error(err)) if ctx.Err() != nil { return ctx.Err() } return &storeStreamErr{} } return nilConfirm the level and field set against
docs/agents/logging.mdbefore you merge. As per coding guidelines: "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_request_worker.go` around lines 247 - 254, Preserve the original g.Wait() error in the return path of the region request worker before replacing it with storeStreamErr, so the deferred log records the concrete Recv or Send failure. Keep the existing context-cancellation handling, and verify any logging changes against docs/agents/logging.md for the required level and fields.Source: Coding guidelines
417-420: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winConsider batching the deregistration events.
This loop pushes one
regionEventper state. A subscription over a large table can hold thousands of tracked regions, so a single deregistration produces thousands of sink pushes on the send goroutine.dispatchResolvedTsEventalready batches states at 1024 per push for the same reason.Batch these states the same way to bound the push count and keep the send loop responsive to the control queue.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_request_worker.go` around lines 417 - 420, The deregistration loop should batch stopped states instead of pushing one regionEvent per state. Update the flow around tracker.TakeSubscription and eventSink.Push to accumulate at most 1024 states per event, push each full batch, and push any remaining states after the loop while preserving markStopped behavior.pkg/config/debug.go (1)
102-116: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider suppressing the warning for unset fields.
PullerConfigfields default to zero when a user omits them from the TOML file. In that case both branches log a warning on every startup, even though no misconfiguration exists. Log at warn level only when the value is negative, and apply the default silently when the value is zero.♻️ Proposed adjustment
- if c.PendingRegionRequestQueueSize <= 0 { - log.Warn("pending region request queue size must be positive, use default value", - zap.Int("value", c.PendingRegionRequestQueueSize), - zap.Int("default", defaultCfg.PendingRegionRequestQueueSize)) + if c.PendingRegionRequestQueueSize <= 0 { + if c.PendingRegionRequestQueueSize < 0 { + log.Warn("pending region request queue size must be positive, use default value", + zap.Int("value", c.PendingRegionRequestQueueSize), + zap.Int("default", defaultCfg.PendingRegionRequestQueueSize)) + } c.PendingRegionRequestQueueSize = defaultCfg.PendingRegionRequestQueueSize }As per coding guidelines: "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/config/debug.go` around lines 102 - 116, Update PullerConfig.ValidateAndAdjust so negative queue-size or window-multiplier values log warnings and are replaced with defaults, while zero values are treated as unset and silently replaced without warning. Preserve the existing default assignments and warning details for negative inputs.Source: Coding guidelines
logservice/logpuller/subscription_client.go (2)
427-437: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueThe lock around the worker list creation has no effect.
Line 427 writes
rs.requestWorkers.sbefore theLockcall on Line 429, andrsis not published tos.storesuntil Line 437. No other goroutine can observersduring this window, so theLock/Unlockpair only adds noise and suggests a synchronization requirement that does not exist here. Remove the pair, or move the slice allocation inside it for consistency.♻️ Proposed cleanup
rs = &requestedStore{storeAddr: storeAddr} rs.requestWorkers.s = make([]*regionRequestWorker, 0, workerCount) - - rs.requestWorkers.Lock() for i := 0; i < workerCount; i++ { requestWorker := newRegionRequestWorker(s, rs, workerWindow, maxWindowMultiplier) rs.requestWorkers.s = append(rs.requestWorkers.s, requestWorker) } - rs.requestWorkers.Unlock() - // Publish the store only after its immutable worker list is complete. s.stores.Store(storeAddr, rs)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/subscription_client.go` around lines 427 - 437, Remove the unnecessary requestWorkers lock and unlock around worker list construction in the store initialization flow, since rs remains unpublished until s.stores.Store. Keep the slice allocation and worker appends unchanged, then publish the completed store as before.
642-659: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the duplicated priority log field.
Line 642 assigns
region.scanPriority = priority.resolvereturns onlySCAN_PRIORITY_HIGHorSCAN_PRIORITY_LOW, sonormalizeScanPriority(priority)on Line 652 equalsregion.scanPriorityon Line 653. The log record carries the same value under two keys.♻️ Proposed cleanup
- zap.String("priority", normalizeScanPriority(priority).String()), zap.String("scanPriority", region.scanPriority.String()),As per coding guidelines: "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/subscription_client.go` around lines 642 - 659, In the cdc region scan task log within the region lock handling flow, remove the redundant priority field so the record reports the priority only once. Keep region.scanPriority assignment and its existing log field unchanged, and delete the normalizeScanPriority(priority) field from the log call.Source: Coding guidelines
logservice/logpuller/priority_task.go (1)
45-70: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winKeep the priority queue ordering key stable during submission.
handleRegionspops a task and assigns a refreshedregiontoregionTask.regionInfo, butLessThan,priority, andcanUseMaxWindowstill readpt.regionInfo.scanPriority. A queued low-priority task can be re-prioritized through that live field before the same task is submitted again, so keep the normalized priority in an immutable field set at construction. Store only the resolved RPC context/task metadata inregionInfowhile enqueueing.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/priority_task.go` around lines 45 - 70, The priority queue ordering currently depends on mutable regionInfo.scanPriority, allowing queued tasks to change order after submission. Update regionPriorityTask to store the normalized scan priority in an immutable construction-time field, and make priority, canUseMaxWindow, and LessThan use that field; keep regionInfo limited to refreshed RPC context and task metadata.logservice/logpuller/region_admission_controller_test.go (2)
52-61: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove or use the unused
currentTsparameter.
submitRegionForAdmissiontakescurrentTs uint64but never uses it;NewRegionPriorityTaskis called with onlyregionandregion.verID.GetID(). Every call site threads acurrentTsvalue through for no effect. This also means none of these tests actually exercise priority derivation from checkpoint staleness (checkpointTsvscurrentTs); priority is always set explicitly viaregion.scanPriority.Either remove the unused parameter, or wire it into task creation if staleness-based priority is meant to be covered here.
♻️ Proposed fix to drop the unused parameter
func submitRegionForAdmission( t *testing.T, controller *regionAdmissionController, region regionInfo, - currentTs uint64, ) { t.Helper() task := NewRegionPriorityTask(region, region.verID.GetID()) require.True(t, controller.submit(task)) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_admission_controller_test.go` around lines 52 - 61, Remove the unused currentTs parameter from submitRegionForAdmission and update every call site to stop passing it. Leave NewRegionPriorityTask and the existing explicit region.scanPriority behavior unchanged.
88-152: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsolidate near-duplicate high-priority tests.
TestRegionAdmissionControllerHighPriorityUsesMaxWindow(lines 88-122) andTestRegionAdmissionControllerPrioritizesHighPriorityRegion(lines 124-152) exercise the same scenario: submit two normal-priority regions plus one high-priority region, and verify the high-priority region is admitted first via the max window. The second test omits only the interrupt/inflight check present in the first. Merge them into one test to reduce duplication.As per path instructions, "Prefer focused deterministic tests; see docs/agents/testing.md before adding or changing tests."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_admission_controller_test.go` around lines 88 - 152, Consolidate TestRegionAdmissionControllerPrioritizesHighPriorityRegion into TestRegionAdmissionControllerHighPriorityUsesMaxWindow, retaining the max-window admission assertions plus the interrupt/inflight verification and cleanup for all admitted requests. Remove the duplicate test and preserve coverage that the high-priority region is admitted before the remaining normal-priority region.Source: Path instructions
logservice/logpuller/region_admission_controller.go (1)
42-61: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winAvoid unconditional mutex acquisition for a debug-only log field.
r.controller.stats()locksc.muon every call tofinish(), even though it is only used to populate azap.Intfield on alog.Debugcall. Go does not defer argument evaluation for logging calls, so this lock is acquired on every completed admission lease regardless of the configured log level. Under high region churn, this adds avoidable lock contention on the admission controller's mutex.Consider dropping the
inflightCountfield from this debug log, or gating it behind an explicit level check before callingstats().As per path instructions, "Logs are operational signals; see docs/agents/logging.md before adding, removing, or rewriting logs." Please confirm this new log field is intentional per that guidance.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/region_admission_controller.go` around lines 42 - 61, Remove the inflightCount field and its unconditional r.controller.stats() call from the debug log in regionReq.finish. Preserve the remaining region request diagnostic fields and metric behavior; do not add or rewrite logging beyond eliminating this mutex acquisition.Source: Path instructions
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@logservice/logpuller/region_admission_controller.go`:
- Around line 203-233: Update requestedStore.close() to drain each
regionAdmissionController’s pending tasks before closing its worker, and
reschedule or otherwise return those tasks through the existing failure/retry
path such as onRegionFail(). Ensure shutdown removes all pending admissions
instead of leaving them in a closed controller; use
regionAdmissionController.drain() and close() while preserving normal worker
shutdown behavior.
- Around line 143-173: The regionAdmissionController.pop method currently
returns raw context cancellation errors instead of repository-defined errors.
Replace the context.Canceled and ctx.Err() returns in pop with the appropriate
TiCDC cancellation error, wrapping the underlying context error at this
boundary, and update the related test’s require.ErrorIs predicate to assert the
repository error.
In `@logservice/logpuller/region_request_worker.go`:
- Around line 217-231: Move errgroup.WithContext(ctx) in the worker flow so it
is created only after Connect succeeds, while preserving the existing gctx usage
for subsequent operations; alternatively, explicitly cancel the derived context
before every early return from the Connect failure path.
- Around line 436-453: Update sendRegionRequest so an inactive req returns nil
instead of &storeStreamErr{}, matching the stopped subscribedSpan branch and
allowing the worker stream to continue without reconnecting or draining pending
requests.
In `@logservice/logpuller/subscription_client.go`:
- Around line 339-342: Move the stopped-subscription guard into the
region-handling path, specifically around handleRequest and before it submits a
region or calls getStore() for rt.subID. Ensure pending handleRegions tasks
recheck the subscription’s stopped state after setTableStopped and skip creating
or using a worker when stopped, while preserving normal processing for active
subscriptions.
- Around line 467-485: Update the failed `store.submit(regionTask)` path in
`handleRegions` so the region task is rescheduled rather than dropped,
preserving its range lock and allowing retry when admission is unavailable.
Before returning, check `ctx.Done()` and return `ctx.Err()` only during client
shutdown; otherwise propagate the repository’s predefined admission/store error
instead of the raw `context.Canceled` sentinel. Keep successful submissions and
logging unchanged.
In `@utils/notifyqueue/notify_queue.go`:
- Around line 1-12: Both notifyqueue files have incomplete Apache license
headers. In utils/notifyqueue/notify_queue.go lines 1-12 and
utils/notifyqueue/notify_queue_test.go lines 1-12, add the standard “WITHOUT
WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.” license line
immediately after the AS IS BASIS line.
---
Outside diff comments:
In `@logservice/logpuller/region_event_handler.go`:
- Around line 245-260: In handleRegionError, only call failureHandler.Report
after worker.tracker.RemoveIf succeeds; retain the existing markRemoved check
and avoid reporting when RemoveIf indicates another state owns the
subscription/region.
---
Nitpick comments:
In `@logservice/logpuller/priority_task.go`:
- Around line 45-70: The priority queue ordering currently depends on mutable
regionInfo.scanPriority, allowing queued tasks to change order after submission.
Update regionPriorityTask to store the normalized scan priority in an immutable
construction-time field, and make priority, canUseMaxWindow, and LessThan use
that field; keep regionInfo limited to refreshed RPC context and task metadata.
In `@logservice/logpuller/region_admission_controller_test.go`:
- Around line 52-61: Remove the unused currentTs parameter from
submitRegionForAdmission and update every call site to stop passing it. Leave
NewRegionPriorityTask and the existing explicit region.scanPriority behavior
unchanged.
- Around line 88-152: Consolidate
TestRegionAdmissionControllerPrioritizesHighPriorityRegion into
TestRegionAdmissionControllerHighPriorityUsesMaxWindow, retaining the max-window
admission assertions plus the interrupt/inflight verification and cleanup for
all admitted requests. Remove the duplicate test and preserve coverage that the
high-priority region is admitted before the remaining normal-priority region.
In `@logservice/logpuller/region_admission_controller.go`:
- Around line 42-61: Remove the inflightCount field and its unconditional
r.controller.stats() call from the debug log in regionReq.finish. Preserve the
remaining region request diagnostic fields and metric behavior; do not add or
rewrite logging beyond eliminating this mutex acquisition.
In `@logservice/logpuller/region_request_worker_test.go`:
- Around line 165-173: Rename the outer pushCount field in
countingRegionEventDynamicStream to counted, keep Push incrementing that outer
counter, and update the benchmark assertion to read ds.counted. Leave the
embedded mockRegionEventDynamicStream counter unchanged.
- Around line 69-80: Make admitRegionRequest deterministic by removing its
time.Now()-derived timestamp and using a caller-supplied timestamp or fixed test
constant, while preserving the existing submitRegionForAdmission and pop flow.
Update all callers, including the case currently passing timestamp 1, to use the
chosen deterministic value.
- Around line 397-404: Extract the closure handleStreamFailure from
regionRequestWorker.Run into a regionRequestWorker method, preserving its
recovery behavior. In
logservice/logpuller/region_request_worker_test.go:397-404, replace the
duplicated tracker-drain and event-push logic with that method call while
retaining the inflight, abort, and pushCount assertions; at 451-453, replace the
duplicated admission-drain and onRegionFail loop similarly while retaining the
pending and failure-handler-cache assertions.
In `@logservice/logpuller/region_request_worker.go`:
- Around line 247-254: Preserve the original g.Wait() error in the return path
of the region request worker before replacing it with storeStreamErr, so the
deferred log records the concrete Recv or Send failure. Keep the existing
context-cancellation handling, and verify any logging changes against
docs/agents/logging.md for the required level and fields.
- Around line 417-420: The deregistration loop should batch stopped states
instead of pushing one regionEvent per state. Update the flow around
tracker.TakeSubscription and eventSink.Push to accumulate at most 1024 states
per event, push each full batch, and push any remaining states after the loop
while preserving markStopped behavior.
In `@logservice/logpuller/subscription_client_test.go`:
- Around line 642-651: Update the deregistration test around
NewRegionPriorityTask and client.broadcastDeregister to submit and deregister
the same subscription, then assert the pending admission task remains queued and
is later processed through the existing request path, covering the stopped
subscribedSpan behavior in sendDeregisterRequest and sendRegionRequest.
- Around line 404-449: Extend the table-driven tests around the existing
high/low cases to cover cdcpb.ScanPriority_SCAN_PRIORITY_UNSPECIFIED (the zero
value), including its retry input and expected priority. Use the test’s existing
fields and naming pattern to pin down normalizeScanPriority behavior for an
unset priority.
In `@logservice/logpuller/subscription_client.go`:
- Around line 427-437: Remove the unnecessary requestWorkers lock and unlock
around worker list construction in the store initialization flow, since rs
remains unpublished until s.stores.Store. Keep the slice allocation and worker
appends unchanged, then publish the completed store as before.
- Around line 642-659: In the cdc region scan task log within the region lock
handling flow, remove the redundant priority field so the record reports the
priority only once. Keep region.scanPriority assignment and its existing log
field unchanged, and delete the normalizeScanPriority(priority) field from the
log call.
In `@pkg/config/debug.go`:
- Around line 102-116: Update PullerConfig.ValidateAndAdjust so negative
queue-size or window-multiplier values log warnings and are replaced with
defaults, while zero values are treated as unset and silently replaced without
warning. Preserve the existing default assignments and warning details for
negative inputs.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 1ce0378a-a0fe-402d-adb4-e84d3a2f438f
📒 Files selected for processing (22)
logservice/logpuller/priority_task.gologservice/logpuller/priority_task_test.gologservice/logpuller/region_admission_controller.gologservice/logpuller/region_admission_controller_test.gologservice/logpuller/region_event_handler.gologservice/logpuller/region_event_handler_test.gologservice/logpuller/region_failure_handler.gologservice/logpuller/region_req_cache.gologservice/logpuller/region_req_cache_test.gologservice/logpuller/region_request_worker.gologservice/logpuller/region_request_worker_test.gologservice/logpuller/region_state.gologservice/logpuller/region_tracker.gologservice/logpuller/region_tracker_test.gologservice/logpuller/scan_priority.gologservice/logpuller/scan_priority_test.gologservice/logpuller/subscription_client.gologservice/logpuller/subscription_client_test.gopkg/config/debug.gopkg/config/debug_test.goutils/notifyqueue/notify_queue.goutils/notifyqueue/notify_queue_test.go
💤 Files with no reviewable changes (2)
- logservice/logpuller/region_req_cache.go
- logservice/logpuller/region_req_cache_test.go
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
logservice/logpuller/subscription_client_test.go (1)
858-866: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert the count after the abort.
Line 865 aborts the in-flight request but asserts nothing afterwards. The test therefore proves only that the total is 3. It does not prove that the in-flight request is the counted item. Add a second assertion after the abort to close that gap.
As per coding guidelines: "Prefer focused deterministic tests".
♻️ Proposed change
require.Equal(t, 3, store.requestedRegionCount()) require.True(t, req.abort()) + require.Equal(t, 2, store.requestedRegionCount()) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@logservice/logpuller/subscription_client_test.go` around lines 858 - 866, In the test around worker admission and request counting, add an assertion immediately after req.abort() to verify the store’s requestedRegionCount reflects removal of the aborted in-flight request. Keep the existing pre-abort count assertion and use the existing requestedRegionCount helper.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@logservice/logpuller/subscription_client_test.go`:
- Around line 452-467: Await and assert the errgroup completion in both
handleRegions tests: logservice/logpuller/subscription_client_test.go lines
452-467 and 527-547. After canceling the context and receiving the handleRegions
error, call eg.Wait() and assert its expected result before each test returns.
In `@pkg/config/debug.go`:
- Line 27: Restore DefaultOldStartTsScanLowPriorityThreshold to 30 minutes so
scan-priority classification and region request admission behavior remain
unchanged.
In `@utils/priorityqueue/priority_queue_test.go`:
- Around line 96-108: Replace the scheduler-sensitive 10-millisecond
early-return select in the priority-queue Pop test with a direct wait for the
expected context.DeadlineExceeded result and nil task, avoiding timing-based
assertions. Apply the same change to the corresponding test block around the
second referenced section; add synchronization only if the test must explicitly
verify that Pop blocked.
---
Nitpick comments:
In `@logservice/logpuller/subscription_client_test.go`:
- Around line 858-866: In the test around worker admission and request counting,
add an assertion immediately after req.abort() to verify the store’s
requestedRegionCount reflects removal of the aborted in-flight request. Keep the
existing pre-abort count assertion and use the existing requestedRegionCount
helper.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 6455fcbf-27fe-4fa2-b2c1-c5a863e0d0a9
📒 Files selected for processing (12)
logservice/logpuller/priority_task.gologservice/logpuller/priority_task_test.gologservice/logpuller/region_admission_controller_test.gologservice/logpuller/region_request_worker.gologservice/logpuller/region_request_worker_test.gologservice/logpuller/region_tracker.gologservice/logpuller/subscription_client.gologservice/logpuller/subscription_client_test.gopkg/config/debug.goutils/notifyqueue/notify_queue.goutils/notifyqueue/notify_queue_test.goutils/priorityqueue/priority_queue_test.go
🚧 Files skipped from review as they are similar to previous changes (8)
- utils/notifyqueue/notify_queue_test.go
- logservice/logpuller/region_admission_controller_test.go
- logservice/logpuller/region_tracker.go
- utils/notifyqueue/notify_queue.go
- logservice/logpuller/priority_task.go
- logservice/logpuller/priority_task_test.go
- logservice/logpuller/region_request_worker.go
- logservice/logpuller/subscription_client.go
|
/test all |
|
/retest |
|
/test all |
|
/test pull-cdc-mysql-integration-light |
1 participant
What problem does this PR solve?
Issue Number: close #5858
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
Summary by CodeRabbit
New Features
Bug Fixes