logpuller: split subscription runtime helpers from subscription client

[lidezhu]

What problem does this PR solve?

Issue Number: close #5604

What is changed and how it works?

This pull request refactors the log puller's subscription client to improve modularity and maintainability. By decomposing the monolithic subscription client into specialized components, the system gains better separation of concerns regarding event processing, error recovery, and span lifecycle management. These changes facilitate more robust handling of region events and failures while simplifying the codebase for future development.

Highlights

• Component Refactoring: Restructured the subscription client by extracting event delivery, failure handling, and span lifecycle management into three dedicated components: regionEventSink, regionFailureHandler, and spanRegistry.
• Improved Reliability: Introduced centralized span tracking and improved recovery logic for region failures, enhancing the stability of log puller subscriptions.
• Enhanced Monitoring: Updated metrics collection to provide better visibility into dynamic stream memory usage and resolved timestamp lag.

Check List

Tests

• Unit test
• Integration test
• Manual test (add detailed scripts or steps below)
• No code

Questions

Will it cause performance regression or break compatibility?

Do you need to update user documentation, design documentation or monitoring documentation?

Release note

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

If you don't think this PR needs a release note then fill it with `None`.

Summary by CodeRabbit

• New Features

  • Added a new pipeline for handling region events and failures, improving how log puller subscriptions are managed.
  • Introduced centralized span tracking and lock-resolution handling for more reliable subscription processing.

• Bug Fixes

  • Improved recovery behavior when region-related operations fail.
  • Updated event delivery to better handle paused/resumed processing states.

• Refactor

  • Reworked internal subscription handling to use clearer, separated components for events, failures, and span management.

[ti-chi-bot]

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

[ti-chi-bot]

[APPROVALNOTIFIER] This PR is NOT APPROVED

This pull-request has been approved by:
Once this PR has been reviewed and has the lgtm label, please assign wlwilliamx for approval. For more information see the Code Review Process.
Please ensure that each of them provides their approval before proceeding.

The full list of commands accepted by this bot can be found here.

Details

Needs approval from an approver in each of these files:

• OWNERS

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[coderabbitai]

📝 Walkthrough

Walkthrough

The subscriptionClient in logpuller is restructured: dynstream event delivery, region failure handling, and subscribed-span lifecycle management (resolve-lock checking, slow-region logging, metrics) are extracted into three new components—regionEventSink, regionFailureHandler, and spanRegistry—with subscription_client.go and its tests updated to delegate to them.

Changes

Subscription client decomposition

Layer / File(s)	Summary
regionEventSink: dynstream event delivery logservice/logpuller/region_event_sink.go	New sink wraps a dynstream.DynamicStream, providing AddPath, RemovePath, Wake, pause-aware Push, feedback-driven Run, UpdateMetrics, and Close.
regionFailureHandler: error buffering and retry logservice/logpuller/region_failure_handler.go	New handler with errCache, Report, Run, and handleError that categorizes region errors and schedules retries, region/range requests, or table drain.
spanRegistry: span lifecycle and monitoring logservice/logpuller/span_registry.go	New registry defines subscribedSpan and manages Add/Get/Remove/UpdateMetrics, plus background resolve-lock checking and slow-region logging loops.
Client and handler rewiring logservice/logpuller/subscription_client.go, logservice/logpuller/region_event_handler.go	subscriptionClient removes dynstream/errCache/span-map fields and delegates Subscribe, Unsubscribe, event push, table drain, and region failure to the new components; regionEventHandler now calls eventSink.Wake and failureHandler.Report.
Test wiring updates logservice/logpuller/subscription_client_test.go, logservice/logpuller/region_request_worker_test.go	Tests construct spans via newSubscribedSpan, register them in spanRegistry, inject regionEventSink/regionFailureHandler into the client, and update assertions accordingly.

Estimated code review effort: 4 (Complex) | ~60 minutes

Possibly related PRs

• pingcap/ticdc#4954: Both touch the resolve-lock pipeline; this PR moves resolve-lock handling into spanRegistry/subscribedSpan while #4954 refines resolve-lock task execution/throttling and metrics.
• pingcap/ticdc#5419: Both touch stale-lock targeting logic in runResolveLockChecker, which this PR relocates into spanRegistry.

Suggested labels: lgtm, approved

Suggested reviewers: asddongmen, wk989898

Poem

A rabbit split the burrow wide,
Three tunnels dug side by side —
One for pushing, one for fails,
One to watch the spans and trails.
🐇 Cleaner warrens, tidy code,
Hopping down a lighter road!

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 12.50% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	The changes extract event sink, failure handling, and span lifecycle into dedicated components as requested by #5604.
Out of Scope Changes check	✅ Passed	The file additions and test updates are all aligned with the stated refactor and do not appear unrelated.
Title check	✅ Passed	The title clearly matches the refactor: splitting runtime helpers out of subscription client.
Description check	✅ Passed	The description covers the issue, changes, and test checklist, but the questions and release note fields are not fully completed.

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch ldz/refactor-puller01

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[gemini-code-assist]

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.

[coderabbitai]

🧹 Nitpick comments (1)

logservice/logpuller/span_registry.go (1)

274-312: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

logSlowRegions holds RLock across IterAll + logging for all spans.

Unlike runResolveLockChecker (which snapshots into spanAndTsCache and releases the lock before doing work), this loop performs rangeLock.IterAll(nil) and emits log.Info for every span while still holding r.RLock(). With many spans this can stall Add/Remove (i.e. Subscribe/Unsubscribe) for the duration of the scan. Consider snapshotting the spans (or their attrs) under the read lock and logging outside it, mirroring the resolve-lock checker.

🤖 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/span_registry.go` around lines 274 - 312, The
`logSlowRegions` loop in `spanRegistry` is holding `r.RLock()` while calling
`rangeLock.IterAll(nil)` and emitting multiple `log.Info` entries, which can
block `Subscribe`/`Unsubscribe` for the full scan. Update `logSlowRegions` to
snapshot the needed span/attr data under the read lock, then release the lock
before doing the time checks and logging, following the same pattern used by
`runResolveLockChecker` and its `spanAndTsCache`.

🤖 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.

Nitpick comments:
In `@logservice/logpuller/span_registry.go`:
- Around line 274-312: The `logSlowRegions` loop in `spanRegistry` is holding
`r.RLock()` while calling `rangeLock.IterAll(nil)` and emitting multiple
`log.Info` entries, which can block `Subscribe`/`Unsubscribe` for the full scan.
Update `logSlowRegions` to snapshot the needed span/attr data under the read
lock, then release the lock before doing the time checks and logging, following
the same pattern used by `runResolveLockChecker` and its `spanAndTsCache`.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 069e02a2-3506-4694-8d51-b16be08fedd8

📥 Commits

Reviewing files that changed from the base of the PR and between 040cb1e and 12232a4.

📒 Files selected for processing (7)

• logservice/logpuller/region_event_handler.go
• logservice/logpuller/region_event_sink.go
• logservice/logpuller/region_failure_handler.go
• logservice/logpuller/region_request_worker_test.go
• logservice/logpuller/span_registry.go
• logservice/logpuller/subscription_client.go
• logservice/logpuller/subscription_client_test.go

[lidezhu]

/gemini summary

[gemini-code-assist]

Summary of Changes

This pull request refactors the log puller's subscription client to improve modularity and maintainability. By decomposing the monolithic subscription client into specialized components, the system gains better separation of concerns regarding event processing, error recovery, and span lifecycle management. These changes facilitate more robust handling of region events and failures while simplifying the codebase for future development.

Highlights

• Component Refactoring: Restructured the subscription client by extracting event delivery, failure handling, and span lifecycle management into three dedicated components: regionEventSink, regionFailureHandler, and spanRegistry.
• Improved Reliability: Introduced centralized span tracking and improved recovery logic for region failures, enhancing the stability of log puller subscriptions.
• Enhanced Monitoring: Updated metrics collection to provide better visibility into dynamic stream memory usage and resolved timestamp lag.

New Features

🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Activity

• The pull request was created and marked as a draft.
• Automated code review identified several critical and medium-priority issues, including missing fields, potential nil pointer dereferences, and anti-patterns in concurrency control.
• The author requested a summary of the changes.
• Review comments provided specific suggestions for fixing the identified issues, such as using channel-based signaling instead of sync.Cond and cleaning up unused error cache fields.

[lidezhu]

/test all