cockroachdb / cockroachdb/cockroach

changefeedccl: make checkpoint flushing linger-based instead of edge-triggered

Open
#171,733 3 comments 0 reactions 0 assignees View on GitHub
A-cdc C-enhancement T-cdc
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.