cockroachdb / cockroachdb/cockroach
changefeedccl: make checkpoint flushing linger-based instead of edge-triggered
- Dominant language
- Go
- Stars
- 32.5k
- Forks
- 4.1k
- PR merge metrics
- PR metrics pending
Description
## Summary
All of the changefeed's checkpoint flush decisions are **edge-triggered**:
we only decide whether to flush at the moment a resolved span (checkpoint)
is delivered. If no checkpoint arrives, no flush decision is ever made.
We should switch to a **linger-based** model: once something flushable
arrives, start a timer for the flush interval and flush whatever has
accumulated when the timer fires. This decouples flush timing from
checkpoint delivery so we never get stuck waiting on a checkpoint to
trigger a flush, and it lets us tightly bound flush latency.
## Problem
The flush logic reacts only to incoming events. The decision to flush (or
not) is computed synchronously inside the handler for a delivered resolved
span. The rate limiter (`saveRateLimiter`) only gates *how soon* a flush is
allowed; it never *schedules* one. Consequences:
- If checkpoints stop arriving (or arrive slowly/irregularly), buffered
progress can sit unflushed well past the configured interval, because
nothing wakes us up to flush it.
- Flush latency is a function of checkpoint cadence, not the configured
flush interval, so we cannot give a tight upper bound on it.
- The interaction between the rate limiter and edge triggering means a
flush that's denied because "not enough time elapsed" is simply dropped;
it relies on a *future* checkpoint to retry, rather than being deferred
and executed when the interval expires.
## Where flush/no-flush decisions are made today (all edge-triggered)
All references are in `pkg/ccl/changefeedccl/changefeed_processors.go`.
### changeAggregator (sink flush + emit resolved spans to frontier)
- `tick()` (~line 765) — the workhorse loop. Flush decisions only happen as
a side effect of consuming an event:
- `TypeResolved` → `noteResolvedSpan()`
- `TypeFlush` → `flushBufferedEvents()` (kvfeed-driven flush request)
- `noteResolvedSpan()` (~line 815) — the core edge-triggered decision.
Computed only when a resolved span is delivered:
- `ShouldFlushFrontier` testing knob (~847)
- `forceFlush`: frontier advanced past a schema-change boundary (~852)
- `shouldFlush`: `(frontierAdvanced || (periodicFlushEnabled &&
unflushedSpans)) && frontierFlushLimiter.canSave()` (~855)
- `checkpointSpans`: backfill / lagging-span span-level checkpoint, gated
by `canCheckpointSpans()` and `lastSpanFlush` (~870)
- `flushFrontier()` (~884) / `flushBufferedEvents()` (~803) — the actual
flush mechanics (flush sink, emit resolved batch). Fine to keep; only the
triggering should change.
### changeFrontier (persist coordinator frontier / highwater)
- `noteAggregatorProgress()` (~line 1670) — only entered when an aggregator
progress row arrives via `Next()` (~1595). Calls:
- `forwardFrontier()` (~1694)
- `maybePersistFrontier()` (~1706)
- `maybePersistFrontier()` (~1928) — gated solely by
`frontierPersistenceLimiter.canSave()`; again only runs when an
aggregator progress row is delivered.
### Rate limiter (gates, does not schedule)
- `saveRateLimiter` (~line 2436), `canSave()` (~2471), `doneSave()` (~2506)
— enforces a minimum interval and an avg-save-duration backoff. It only
answers "may I flush now?"; it never causes a deferred flush to happen.
`frontierFlushLimiter` (aggregator) and `frontierPersistenceLimiter`
(frontier) are both instances of this.
## Proposed approach
Replace the edge-triggered checks with a linger timer:
1. When a flushable change first arrives (e.g. `unflushedSpans` becomes
true / frontier advances), arm a timer for the flush interval if one
isn't already armed.
2. Select on that timer alongside event consumption. When it fires, flush
whatever has accumulated, then disarm.
3. Keep `forceFlush` semantics (schema-change boundaries) as an immediate
flush that also resets the linger timer.
4. Fold the rate limiter's interval into the linger interval so the timer
*is* the rate limit, rather than a separate gate that drops flushes.
This bounds flush latency to roughly `flush interval + flush duration`
regardless of checkpoint cadence, and applies to both the aggregator's sink
flush and the frontier's persistence.
## Notes / open questions
- The aggregator `tick()` loop is currently a blocking `eventProducer.Get`;
a linger timer needs that loop to wake on a timer as well as on events.
- Need to preserve the avg-save-duration backoff behavior (don't flush more
often than a flush takes) — likely as a floor on the linger interval.
- Backfill / lagging-span span-level checkpoints (`checkpointSpans`) have
their own cadence (`canCheckpointSpans`) and may want a separate linger.
Jira issue: CRDB-64868
Contributor guide
Research direction
Start in pkg/ccl/changefeedccl/changefeed_processors.go by reading changeAggregator.tick(), noteResolvedSpan(), and changeFrontier.noteAggregatorProgress()/maybePersistFrontier(), then inspect saveRateLimiter. Trace how event consumption currently triggers sink flushes and frontier persistence. Done means both paths use linger-based wakeups, preserve force-flush and backoff behavior, and no longer depend on future checkpoints.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- databases, distributed-systems
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100