eventservice: optimize changefeed low-latency scheduling - #5900
Conversation
Keep the low-latency mode focused on setting the logpuller advance interval to zero. Remove the additional batched heap update path and its helper after the 10k-region E2E test showed comparable mean and p95 latency without it.
Replace the periodic schema-capped scan retry with applied SchemaStore notifications. Serialize dispatcher scan scheduling with a short-lock state machine and coalesce worker continuations without dropping queued work.
Keep no-DML/no-DDL resolved notifications out of the scan worker queue while preserving dispatcher scan ownership. Gate continuation and schema-blocked recovery on low-latency mode, and cover queue-full recovery with a dropped-task metric.
…level-low-latency-mode
Publish LogCoordinator resolved lag only from complete node-report rounds and calculate it at each node's report time. Preserve Maintainer watermark ordering, expose checkpoint/resolved lag atomically, and update dashboards to use the paired metrics.
Restore the unrelated Maintainer metric collector, watermark state, and dashboard changes. Keep this follow-up scoped to LogCoordinator owner resolved-ts lag calculation and its tests.
Keep the first changefeed-level low-latency PR free of pkg/eventservice changes. Move the coupled EventStore subscription isolation and EventService dropped-task metric into the follow-up part as well so both trees compile independently.
Use changefeed mode to isolate EventStore subscriptions and select immediate resolved-ts advancement. Add the EventService scan state machine, resolved-notification fast path, low-latency continuation and schema-blocked retry, deferred active-scan lifecycle, and focused coverage.
|
Skipping CI for Draft Pull Request. |
ti-chi-bot
Bot
added
release-note
|
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 (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughThe PR adds per-changefeed low-latency mode. It propagates the mode through EventService and EventStore, isolates subscriptions by mode, introduces dispatcher scan states, and adds schema-blocked retry and queue-recovery handling. ChangesLow-latency changefeed processing
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant EventBroker
participant Dispatcher
participant EventStore
participant SchemaStore
EventBroker->>Dispatcher: request scan
Dispatcher->>EventBroker: claim scan state
EventBroker->>EventStore: scan event range
EventStore-->>EventBroker: return progress
EventBroker->>SchemaStore: check schema frontier
SchemaStore-->>EventBroker: return schema state
EventBroker->>Dispatcher: complete or schedule continuation
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
🧪 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
removed
the
do-not-merge/work-in-progress
|
/test |
|
@asddongmen: The The following commands are available to trigger optional jobs: Use DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
/test all |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
pkg/eventservice/event_broker.go (1)
774-790: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winClear
interruptedwhen the scan returns an error.
scanner.scancan setinterruptedand return a non-nilerrin the same call. The function then returns at Line 789 withinterruptedstill true. The deferredfinishScantreats that as an interrupted scan and callstryEnqueueScanLocked, so the dispatcher is re-queued immediately.If the error is persistent, for example a repeated SchemaStore or EventStore failure, this forms a tight loop with no backoff: enqueue, scan, error, enqueue. The loop occupies one scan worker continuously and repeats the error log at Line 784.
Set
interruptedto false on the error path so the dispatcher returns todispatcherScanIdleand waits for the next notification.🐛 Proposed fix
if err != nil { + // A failed scan must not be treated as an interrupted scan. Otherwise + // finishScan re-queues the dispatcher immediately and a persistent + // error becomes a tight retry loop. + interrupted = false releaseQuota(available, uint64(sl.maxDMLBytes)) if task.isRemoved.Load() { return }🤖 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/eventservice/event_broker.go` around lines 774 - 790, In the error branch after scanner.scan returns, explicitly set interrupted to false before returning so deferred finishScan does not re-enqueue the dispatcher. Update the err handling block in the surrounding scan flow while preserving the existing quota release, removal check, and error logging behavior.
🧹 Nitpick comments (7)
pkg/eventservice/metrics_collector.go (1)
340-340: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRename the log key to match the reported state.
isScanBusy()returns true fordispatcherScanQueued,dispatcherScanRunning, anddispatcherScanRunningPending. The keyisTaskScanningstates that an EventStore scan is in progress. That is now inaccurate for a queued dispatcher.Rename the key so the log matches the scheduler state it reports.
♻️ Proposed change
- zap.Bool("isTaskScanning", dispatcher.isScanBusy()), + zap.Bool("isScanBusy", dispatcher.isScanBusy()),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/eventservice/metrics_collector.go` at line 340, Rename the zap log key in the metrics collection code from isTaskScanning to a name that accurately represents dispatcher.isScanBusy(), including queued, running, and pending scan states; keep the existing state value and logging behavior unchanged.Source: Coding guidelines
pkg/eventservice/event_broker.go (1)
1230-1234: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winLog the SchemaStore error in the retry loop.
The loop marks the bucket dirty and skips the keyspace when
GetTableDDLEventStatefails. It emits no signal. A persistent SchemaStore failure then parks every low-latency dispatcher in that keyspace silently, and the only visible symptom is rising resolved-ts lag.The ticker fires every 50 ms, so log at a rate-limited or debug level rather than on every failure.
♻️ Proposed change
ddlState, err := c.schemaStore.GetTableDDLEventState(keyspaceMeta, d.info.GetTableSpan().TableID) if err != nil { + log.Debug("get table ddl event state failed in schema-blocked retry", + zap.Uint32("keyspaceID", keyspaceMeta.ID), + zap.Int64("tableID", d.info.GetTableSpan().TableID), + zap.Error(err)) bucket.dirty.Store(true) return true }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/eventservice/event_broker.go` around lines 1230 - 1234, In the retry loop around GetTableDDLEventState, record the returned SchemaStore error before marking bucket.dirty and returning, using the repository’s approved rate-limited or debug logging mechanism so failures are observable without emitting a log on every 50 ms retry.Source: Coding guidelines
logservice/eventstore/event_store_test.go (1)
282-286: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDerive the throughput interval from the server config.
The test hardcodes
100as the throughput advance interval. That value comes from the defaultKVClient.AdvanceIntervalInMs. If the default changes, this test fails with an unclear reason. Read the configured value instead.♻️ Proposed change
intervals := make(map[int64]int) for _, subscription := range mockSubClient.subscriptions { intervals[subscription.advanceInterval]++ } - require.Equal(t, map[int64]int{0: 1, 100: 1}, intervals) + throughputInterval := int64(config.GetGlobalServerConfig().KVClient.AdvanceIntervalInMs) + require.Equal(t, map[int64]int{0: 1, throughputInterval: 1}, intervals)As per path instructions for
**/*_test.go: "Prefer focused deterministic 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/eventstore/event_store_test.go` around lines 282 - 286, Replace the hardcoded 100 in the intervals assertion with the configured throughput advance interval from the server or KVClient configuration used by this test. Keep the expected interval counts unchanged and reference the existing configuration symbol so the test remains deterministic when the default changes.Source: Path instructions
pkg/metrics/event_service.go (1)
159-165: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider a
reasonlabel for the dropped-scan counter.
tryEnqueueScanLockedis called from four distinct paths inpkg/eventservice/event_broker.go: the internal request path at Line 1124, the interrupted-scan path at Line 1161, the low-latency continuation path at Line 1174, and the schema-retry path at Line 1256. The counter aggregates all four into one number.A
reasonlabel would show which scheduling path saturates the worker queue. That distinction matters during triage, because the schema-retry path self-recovers on the next tick while the continuation path waits for the next EventStore notification.The metric name follows the existing
_countconvention used by the neighboring counters in this file, so no rename is needed.🤖 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/metrics/event_service.go` around lines 159 - 165, Add a reason label to EventServiceDroppedScanTaskCount and update tryEnqueueScanLocked call sites to pass distinct labels for the internal request, interrupted-scan, low-latency continuation, and schema-retry paths. Ensure each dropped-task increment uses the corresponding reason while preserving the existing metric name and count behavior.pkg/eventservice/event_broker_test.go (2)
385-385: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSet
scanStateunderscanMuin tests.These lines write
disp.scanStatedirectly without holdingdisp.scanMu. Production code always mutatesscanStateunder that mutex. The writes are currently safe because no other goroutine touches the dispatcher at these points, so-racestays clean.Adding the lock keeps the test setup consistent with the invariant and stays correct if a future test adds a concurrent goroutine. Alternatively, add a small test helper such as
setScanStateForTest.As per path instructions for
**/*_test.go: "Prefer focused deterministic tests".Also applies to: 680-680, 713-713, 1055-1055
🤖 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/eventservice/event_broker_test.go` at line 385, Protect every direct test assignment to disp.scanState with disp.scanMu, including the occurrences near lines 385, 680, 713, and 1055. Use the existing mutex consistently around each setup mutation, or introduce and reuse a focused setScanStateForTest helper that performs the locked update.Source: Path instructions
469-501: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a test for a scan that fails with an error.
TestInterruptedScanQueueFullRecoversOnNextNotifycovers the interrupted path. No test coversscanner.scanreturning bothinterrupted == trueand a non-nilerr. That combination drives the re-queue loop described in thepkg/eventservice/event_broker.gocomment on Lines 774-790.Add a case that injects a scan error and asserts the dispatcher settles in
dispatcherScanIdleinstead of being re-queued. I can generate this test if you want.As per path instructions for
**/*_test.go: "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 `@pkg/eventservice/event_broker_test.go` around lines 469 - 501, Add a focused deterministic test alongside TestInterruptedScanQueueFullRecoversOnNextNotify that makes scanner.scan return interrupted == true with a non-nil error. Assert the dispatcher’s scan state settles at dispatcherScanIdle and that the failed task is not re-queued, covering the error-handling path in the broker scan flow.Source: Path instructions
logservice/eventstore/event_store.go (1)
695-704: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd the mode to the new-subscription log.
A table can now hold one throughput subscription and one low-latency subscription at the same time. The
new subscription createdlog does not show which mode created the subscription. Add the mode and the applied advance interval. This makes duplicate per-table subscriptions diagnosable from logs alone.♻️ Proposed log fields
log.Info("new subscription created", zap.Stringer("dispatcherID", dispatcherID), zap.Uint64("startTs", startTs), zap.Uint64("subscriptionID", uint64(subStat.subID)), + zap.Bool("lowLatencyMode", lowLatencyMode), + zap.Int64("resolvedTsAdvanceInterval", resolvedTsAdvanceInterval), zap.String("subSpan", common.FormatTableSpan(subStat.tableSpan)))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/eventstore/event_store.go` around lines 695 - 704, Update the “new subscription created” log in the subscription setup flow to include the subscription mode, distinguishing throughput from low-latency, and the applied resolved timestamp advance interval from resolvedTsAdvanceInterval. Preserve the existing fields and use the established mode symbol or representation already available in this flow.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 `@pkg/eventservice/event_broker.go`:
- Around line 1106-1108: Update the enqueue in prepareScanFromNotify, used by
advanceResolvedTs, so advancing resolved timestamps never blocks on a saturated
scan worker taskChan. Use a non-blocking enqueue or an intermediate channel
while preserving delivery of resolved-ts work and the existing requestScan
behavior for scan workers.
---
Outside diff comments:
In `@pkg/eventservice/event_broker.go`:
- Around line 774-790: In the error branch after scanner.scan returns,
explicitly set interrupted to false before returning so deferred finishScan does
not re-enqueue the dispatcher. Update the err handling block in the surrounding
scan flow while preserving the existing quota release, removal check, and error
logging behavior.
---
Nitpick comments:
In `@logservice/eventstore/event_store_test.go`:
- Around line 282-286: Replace the hardcoded 100 in the intervals assertion with
the configured throughput advance interval from the server or KVClient
configuration used by this test. Keep the expected interval counts unchanged and
reference the existing configuration symbol so the test remains deterministic
when the default changes.
In `@logservice/eventstore/event_store.go`:
- Around line 695-704: Update the “new subscription created” log in the
subscription setup flow to include the subscription mode, distinguishing
throughput from low-latency, and the applied resolved timestamp advance interval
from resolvedTsAdvanceInterval. Preserve the existing fields and use the
established mode symbol or representation already available in this flow.
In `@pkg/eventservice/event_broker_test.go`:
- Line 385: Protect every direct test assignment to disp.scanState with
disp.scanMu, including the occurrences near lines 385, 680, 713, and 1055. Use
the existing mutex consistently around each setup mutation, or introduce and
reuse a focused setScanStateForTest helper that performs the locked update.
- Around line 469-501: Add a focused deterministic test alongside
TestInterruptedScanQueueFullRecoversOnNextNotify that makes scanner.scan return
interrupted == true with a non-nil error. Assert the dispatcher’s scan state
settles at dispatcherScanIdle and that the failed task is not re-queued,
covering the error-handling path in the broker scan flow.
In `@pkg/eventservice/event_broker.go`:
- Around line 1230-1234: In the retry loop around GetTableDDLEventState, record
the returned SchemaStore error before marking bucket.dirty and returning, using
the repository’s approved rate-limited or debug logging mechanism so failures
are observable without emitting a log on every 50 ms retry.
In `@pkg/eventservice/metrics_collector.go`:
- Line 340: Rename the zap log key in the metrics collection code from
isTaskScanning to a name that accurately represents dispatcher.isScanBusy(),
including queued, running, and pending scan states; keep the existing state
value and logging behavior unchanged.
In `@pkg/metrics/event_service.go`:
- Around line 159-165: Add a reason label to EventServiceDroppedScanTaskCount
and update tryEnqueueScanLocked call sites to pass distinct labels for the
internal request, interrupted-scan, low-latency continuation, and schema-retry
paths. Ensure each dropped-task increment uses the corresponding reason while
preserving the existing metric name and count behavior.
🪄 Autofix
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: 250f1807-45d5-4dbf-9032-394fe0f3a198
📒 Files selected for processing (10)
docs/design/2026-08-07-changefeed-low-latency-eventservice.mdlogservice/eventstore/event_store.gologservice/eventstore/event_store_test.gopkg/eventservice/dispatcher_stat.gopkg/eventservice/event_broker.gopkg/eventservice/event_broker_test.gopkg/eventservice/event_service.gopkg/eventservice/event_service_test.gopkg/eventservice/metrics_collector.gopkg/metrics/event_service.go
ti-chi-bot
Bot
added
needs-1-more-lgtm
|
/test nex-gen |
|
@lidezhu: The specified target(s) for The following commands are available to trigger optional jobs: Use DetailsIn response to this:
Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: lidezhu, wk989898 The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
ti-chi-bot
Bot
added
lgtm
and removed
needs-1-more-lgtm
…level-low-latency-eventservice
|
/test all |
|
/retest |
3 participants
What problem does this PR solve?
Issue Number: close #5705
Latency-sensitive changefeeds need EventService to advance and deliver resolved
timestamps without periodic scheduling delays. This is Part 2 of the
changefeed-level low-latency feature; #5862 provides the persisted mode and
control-plane propagation consumed here.
What is changed and how it works?
LogPuller resolved-ts advancement for low-latency subscriptions.
no-event resolved notifications, and coalesced continuation after active
scans.
bounded worker queues and eventual recovery.
expose dropped scheduling attempts through an EventService metric.
Check List
Tests
low-latency/throughput changefeeds with 100k tables each, 200k isolated
subscriptions, about 20 MB/s shared traffic for 30 minutes; both feeds
remained normal)
Questions
Will it cause performance regression or break compatibility?
Throughput behavior remains unchanged. Low-latency mode performs more eager
frontier and scan scheduling; opposite modes intentionally do not share
EventStore subscriptions.
Do you need to update user documentation, design documentation or monitoring documentation?
Yes. The changefeed performance-mode option introduced by #5862 needs user
documentation. The dropped-task metric is diagnostic and does not change
existing metric names.
Release note
Summary by CodeRabbit
New Features
Bug Fixes
Documentation