crdb stack/6 resolved buffering

[viragtripathi]

Follow-up to #4669, implementing the buffering discussed on the dedupe thread there and on Slack.

Changefeed events now wait in an in-memory buffer keyed by commit timestamp and row key. When a resolved timestamp arrives, everything it covers is emitted in commit order and the checkpoint advances to that same value, so nothing handed to the stream ever lies past the persisted checkpoint. Batch boundaries stop producing duplicates: records still buffered when a batch ends are dropped and re-read by the next batch from the cursor. Session reconnect replays land on the same buffer keys and collapse.

The buffer is bounded in bytes computed from the raw changefeed message sizes, not record count, since single events can be large. Crossing the cap (128 MiB) or a resolved stream that stalls past the batch deadline force-emits the buffer past the checkpoint with a warning, which is exactly the pre-existing at-least-once behavior, so the degraded mode is the previously reviewed semantics. Delivery overall remains at least once: a crash between emit and checkpoint persist still replays.

Validation: new unit tests for the replay-collapse, partial-emit-at-resolved and force-emit ordering paths, plus a new e2e test (Test_CDC_Exactly_Once_Across_Batches) asserting 82 events arrive exactly once across two consecutive pulls. Full suite run against a 5-node v26.2.5 cluster behind haproxy, 17/17 passing, and unit suites green.

[claude]

Code review

One issue found. Checked for bugs and CLAUDE.md compliance.

(The inline-comment tool was unavailable in this session, so this is posted as a top-level comment with permalinks.)

Grace-deadline batch cut discards the buffer, so the batch can return zero records and the mirror stalls

peerdb/flow/connectors/cockroachdb/cdc.go

Lines 478 to 484 in 3388ef7

	now := time.Now()
	if !state.graceDeadline.IsZero() && !now.Before(state.graceDeadline) {
	return true, nil
	}
	if (state.recordCount > 0 || len(state.buffer) > 0) &&
	!now.Before(state.batchDeadline.Add(2*state.resolvedInterval)) {

This branch returns true, nil without calling emitBuffered, so everything still buffered is dropped. That is fine when the buffer only holds a tail past the last resolved timestamp — but graceDeadline is now armed off recordCount + len(state.buffer):

peerdb/flow/connectors/cockroachdb/cdc.go

Lines 687 to 693 in 3388ef7

	}
	}
	if state.recordCount+uint32(len(state.buffer)) >= state.req.MaxBatchSize && state.graceDeadline.IsZero() {
	// full batch: allow a short grace period for a final resolved
	// timestamp so the checkpoint covers the pulled records
	state.graceDeadline = time.Now().Add(2 * state.resolvedInterval)
	}

Since recordCount is only incremented inside emitBuffered, a batch can reach MaxBatchSize entirely with buffered records and recordCount == 0. graceDeadline is then fillTime + 2*resolvedInterval (~2s, with changefeedResolvedInterval = 1s), while the new force-emit valve immediately below fires at batchDeadline + 2*resolvedInterval = firstBufferedRecord + IdleTimeout + 2s. Whenever the buffer fills in less than IdleTimeout, the grace branch is checked first and wins, so the valve at L483 is unreachable in exactly the case it was written for.

The result: PullRecords returns an empty batch, state.cursor never advances (it is only assigned in handleResolved), and the next batch restarts the changefeed from the same cursor and reproduces the identical state — no forward progress. Pre-PR this path returned a full MaxBatchSize of already-emitted records, so it always advanced. A realistic trigger is a resolved-timestamp stall with rows still flowing, e.g. a CockroachDB catch-up scan, which would then restart from the same cursor on every batch and never complete.

Suggested fix: either force-emit before cutting the batch here, or gate the grace deadline on state.recordCount >= state.req.MaxBatchSize so a buffered-only batch falls through to the force-emit valve at L483. Note that no current test drives the grace-deadline branch of runChangefeedSession with a non-empty buffer.

[github-actions]

🔄 Flaky Test Detected

Analysis: All four failures are fixed wall-clock deadline expiries (60s mirror-startup budget in SetupCDCFlowStatusQuery, plus one EnvWaitFor resync timeout) in MySQL/MariaDB→ClickHouse e2e tests on a CockroachDB-focused branch, with disjoint failure sets across the two failing matrix jobs and a third matrix job passing all of them — consistent with runner contention rather than a regression.
Confidence: 0.85

✅ Automatically retrying the workflow

View workflow run

[jgao54]

one small nit, otherwise lgtm!

[jgao54]

ah this is neat, since it also dedupes duplicated messages send from server-side retries

[viragtripathi]

Yeah that fell out of keying the buffer by timestamp and row key, reconnect and server side replays just land on the same slot.

[jgao54]

nit: this logline would get a bit noisy as it would be emitted every PullRecord on a busy pipe, can be removed or emit at debug level

[viragtripathi]

Moved it to debug. And thanks for the review, and for all of them across the stack, the connector came out a lot better for how thorough you've been.