cockroachdb_changefeed: checkpoint on RESOLVED timestamps (CON-504) - #4688
Conversation
The cursor was persisted from each acked row's own `updated` timestamp, but every row of a transaction — and the ENTIRE initial backfill — shares one timestamp, and CURSOR resume is exclusive. Acking a single backfill row then crashing persisted a cursor that skipped every other backfill row on restart: unbounded silent loss. When cursor_cache is set the changefeed now runs WITH RESOLVED (a user-supplied resolved='interval' is preserved) and only resolved timestamps — CockroachDB's guarantee that nothing at or below them will be emitted again, never issued before the initial scan completes — are persisted, gated through the ordered tracker so a resolved timestamp only persists once every row before it is acked. Data rows are tracked with an empty payload; nacks never resolve (logged, checkpoint pinned). Resolved records are consumed as bookkeeping rather than emitted downstream, which also fixes a latent panic: their NULL key crashed the row assertions whenever resolved was supplied via options. Restarts now redeliver at most the changes since the last resolved timestamp (duplicates, never loss).
squiidz
changed the title
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // terminal. Never resolve: the cursor stays pinned before this | ||
| // row so no resolved timestamp can be persisted past its | ||
| // undelivered data. | ||
| c.logger.Errorf("Row rejected downstream: the cursor is now pinned before this row and the input will stall once the checkpoint limit is reached, unless the row is redelivered (auto_replay_nacks) or the pipeline restarts: %v", err) |
There was a problem hiding this comment.
This introduces a new user-visible failure mode that isn't documented, which CONTRIBUTING.md §1.2.3 requires ("Known limitations and edge cases are documented").
Before this change a nack only meant that row's own cursor wasn't persisted — later rows still advanced the cursor. Now the nacked row's checkpoint slot is never released, so:
- the frontier is pinned before that row forever, and no resolved timestamp is ever persisted again for the life of the process, and
- after
checkpoint.NewCapped[string](1024)slots leak,TrackinReadblocks permanently (input_changefeed.go#L330-L341) and the input stops delivering entirely.
The comment here acknowledges the stall, but it's only reachable via a supported, user-settable option (auto_replay_nacks: false, from service.NewAutoRetryNacksToggleField() at input_changefeed.go#L66-L68), and the user-facing docs updated in this PR only state the positive guarantee ("never skips data") without mentioning it.
Suggested fix: extend the cursor_cache / options field descriptions (and the regenerated .adoc) to state that with cursor_cache set and auto_replay_nacks disabled, a nacked row pins the cursor and eventually stalls the input, so auto_replay_nacks should stay enabled.
There was a problem hiding this comment.
Fixed in e50c83e — the cursor_cache description (and regenerated docs) now state that with auto_replay_nacks disabled a rejected row pins the cursor and eventually stalls the input, that this is deliberate (advancing would silently lose the row), and recommend keeping auto_replay_nacks enabled.
With cursor_cache set and auto_replay_nacks disabled, a rejected row deliberately pins the cursor and eventually stalls the input; the field docs now state this and recommend keeping auto_replay_nacks enabled.
| var cursorReleaseFn func() *string | ||
| if c.cursorCache != "" { | ||
| // Data rows are tracked with an empty payload: they hold the | ||
| // ordered tracker's frontier (so no resolved timestamp can persist | ||
| // past an un-acked row) but never advance the cursor themselves. | ||
| // Row-level `updated` timestamps are unsafe cursors: every row of | ||
| // a transaction — and the entire initial backfill — shares one, | ||
| // and CURSOR resume is exclusive. | ||
| if cursorReleaseFn, err = c.cursorCheckpointer.Track(ctx, "", 1); err != nil { | ||
| return nil, nil, fmt.Errorf("tracking row checkpoint: %w", err) | ||
| } | ||
| } |
There was a problem hiding this comment.
Tracking data rows with an empty payload can silently swallow a pending resolved timestamp, so the cursor may stop advancing under out-of-order acks.
checkpoint.Capped's release closure returns the payload of the highest contiguously-resolved entry, not the payload of the entry being released — proven in-repo by oracledb/batcher_test.go#L75-L81 ("expected the streaming batch's SCN to survive the out-of-order snapshot ack").
Consider the tracked sequence r1(""), resolved("T1"), r2(""):
resolved("T1")is released immediately in the branch above → returnsnilbecauser1is still pending.r2is acked out of order → returnsnil.r1is acked → the contiguous run now extends throughr2, so the release returnsr2's payload"", not"T1"—input_changefeed.go#L366-L369then treats it as "nothing to persist" andT1is dropped.
Every row before T1 was acked at that point, so T1 was safe to persist. With a multi-threaded pipeline (acks routinely complete out of order) a resolved marker is persisted only in the narrow case where the row tracked immediately after it is still un-acked, so under sustained load the cursor can go long stretches without advancing and a restart replays far more than the options docs promise ("redelivers at most the changes since the last resolved timestamp").
Suggested fix: track data rows with the last-seen resolved timestamp as the payload rather than "", the way mssqlserver/oracledb carry checkpointLSN/checkpointSCN on every batch. A row tracked after resolved T is only ever contiguously resolved once T and everything before it is acked, so carrying T forward is safe and keeps the frontier payload from regressing to the empty string.
There was a problem hiding this comment.
Fixed in b335937. Data rows now carry the last resolved timestamp seen before them as their tracker payload (the mssqlserver/oracledb pattern suggested): a row tracked after resolved T only becomes contiguously resolved once T and everything before it acked, so the frontier payload can never regress to the empty string and a safe pending timestamp always persists. The same commit also serializes each release+persistCursor pair behind a persistMu — with carried payloads the concurrent ack/Read persist paths had the same out-of-order-write race just fixed in mssqlserver's batcher. Both integration tests re-verified green.
…ff is an opt-in drop)
Unwinds the nack-pinning behavior and its doc note. Per the framework's
documented contract for auto_replay_nacks ("If set to false these messages
will instead be deleted"), disabling replay is an explicit opt-in to drop
rejected messages, so the cursor advances past them; pinning produced
permanent backpressure once the checkpoint limit filled. RESOLVED-based
cursor checkpointing is unchanged.
|
The nack-handling changes from the earlier review rounds have been unwound in the latest commit. The framework's documented contract for Unwound here (134f43b): the ackFn nack pin, plus the cursor_cache doc note about pinning (docs regenerated). The RESOLVED-timestamp cursor work is unchanged. |
| var got int | ||
| require.Eventually(t, func() bool { | ||
| seenMut.Lock() | ||
| got = len(seen) | ||
| seenMut.Unlock() | ||
| return got == rowCount | ||
| }, time.Minute, time.Millisecond*100, "backfill rows were skipped after ack-one-then-crash: got %v of %v", got, rowCount) |
There was a problem hiding this comment.
The got value in the failure message is always 0. require.Eventually's msgAndArgs are variadic interface{} values evaluated at the call site, so got is copied into the args slice before the condition ever runs — the assignment inside the closure is never observed by the message. On failure this reports got 0 of 100 regardless of how many rows were actually seen, which defeats the purpose of hoisting got out of the closure.
Read the count under seenMut after Eventually returns (or use a formatted require.Equal/assert.Len follow-up) so the diagnostic reflects the real value.
Note the same pattern already exists at integration_test.go#L195-L201 — pre-existing, so only the new occurrence needs fixing here.
There was a problem hiding this comment.
Fixed in b335937 — the condition runs under assert.Eventually and the count is read under seenMut after it returns, so the require.True diagnostic reports the real value.
…rialize cursor persists Two review findings on the RESOLVED cursor path: - Data rows were tracked with an empty payload, and the ordered tracker's release returns the payload of the highest contiguously-resolved entry. Under out-of-order acks the frontier payload could therefore regress to the empty string and mask a pending resolved timestamp - the cursor only advanced in the narrow case where the row tracked right after a resolved marker was still un-acked, so restarts replayed far more than necessary. Rows now carry the last resolved timestamp seen before them: a row is only contiguously resolved once that timestamp and everything before it acked, so carrying it forward is always safe. - Releases hand out monotonic timestamps, but acks (pipeline goroutines) and resolved records (Read goroutine) persisted concurrently with no shared ordering, so two cache writes could land out of order and regress the cursor. Each release+persist pair now runs under a shared mutex - the same defect class fixed in mssqlserver's batcher on this ticket. Also fixes the new integration test's failure diagnostic (the message args were evaluated before Eventually ran, so it always reported 0 rows). Both integration tests re-verified green.
| if c.cursorCache == "" { | ||
| continue | ||
| } |
There was a problem hiding this comment.
NOTE (observability/docs): when a user supplies resolved='...' in options without cursor_cache, this branch now silently discards every resolved record — no message emitted, no log at any level. Previously those records reached the row-assertion path (a crash, per the first commit's message), so going from panic to silent drop is an improvement, but the new behaviour is undocumented and unobservable:
- The
optionsdoc note added in this PR (input_changefeed.go#L62-L63) only explains thecursor_cache-is-set case; nothing tells a user thatresolvedrecords are consumed as bookkeeping and never surface downstream. - CONTRIBUTING.md §1.2.2 — "Unexpected behavior should emit warning or error logs"; §1.2.3 — "Known limitations and edge cases are documented."
Suggested fix: emit a debug (or one-shot warn) log when a resolved record is dropped because no cursor_cache is configured, and extend the options field description to state that resolved records are internal bookkeeping and are never emitted as messages.
There was a problem hiding this comment.
Fixed in c095c91 — a one-shot warning explains the discard when resolved records arrive with no cursor_cache configured, and the options field description (and regenerated docs) now state that resolved records are internal cursor bookkeeping, never emitted as messages, and are discarded without a cursor_cache.
…without cursor_cache Review note: with resolved='...' supplied in options but no cursor_cache, resolved records were consumed as bookkeeping with no output and no log at any level. A one-shot warning now explains the discard, and the options field description states that resolved records are internal cursor bookkeeping, never emitted as messages, and are discarded without a cursor_cache. Docs regenerated.
2 participants
Part of CON-504 (CDC at-least-once / ack-gated progress).
The cursor was persisted from each acked row's own
updatedtimestamp. Every row of a transaction — and the ENTIRE initial backfill — shares one timestamp, andCURSORresume is exclusive, so acking a single backfill row and crashing persisted a cursor that skipped every other backfill row on restart: unbounded silent loss (the audit's worst finding).Changes:
cursor_cacheset, the changefeed runsWITH RESOLVED(a user-suppliedresolved='interval'is preserved). Only resolved timestamps — CockroachDB's guarantee that nothing at or below them will be emitted again, and never issued before the initial scan completes — are persisted, gated through the ordered tracker so a resolved timestamp persists only once every row before it has been acked.values[1].([]byte)assertion wheneverresolvedwas supplied viaoptions.optionsfield): the cursor granularity becomes the resolved cadence, so a restart redelivers at most the changes since the last resolved timestamp — duplicates, never loss. Tune withresolved='...'+min_checkpoint_frequency='...'options.Proof of Work
TestIntegrationCRDBBackfillAckCrash): ack exactly one backfill row of 100, crash, assert no cursor persisted, restart delivers all 100. Fails on pre-fix code with the loss signature:no cursor may be persisted while backfill rows are unacknowledged: Should be empty, but was 1786389756112691053.0000000000— a cursor that would have skipped the other 99 rows forever.-race. Docs regenerated.cockroach sqlCLI silently drops them, which is worth knowing when debugging), the first marker arrives immediately after the backfill completes, and the v26 rename of thechangefeed.min_checkpoint_frequencycluster setting does not affect the per-changefeed option.