api: backfill api poc

[josephwoodward]

The Problem

With CDC solutions there's a common requirement to snapshot (backfill) existing data before switching to realtime streaming - two logical steps with different checkpointing needs. Snapshotting data is usually one atomic operation which once completed checkpoints the current watermark before switching to realtime streaming and continually checkpointing the watermark as the cursor moves ahead through the WAL.

Currently we have a single ReadBatch model where managing the tracking of snapshot completion and gaining confidence in delivery before watermarking the current position results in horrible, error prone code that each connector has to own. Such challenges can include:

• Using batch's ack function to identifying when the batch is a snapshot batch or not
• The connector needing to flush any remaining events in the snapshot and wait for the to be completed before being able to persist the current watermark position.

Proposed Solution

This change (currently a proof-of-concept to validate its usefulness) adds a new optional API for a "backfill" stage that gets called before the main ReadBatch loop begins, creating a clearer, codified separation between the snapshot and streaming stages.

The optional API could look something like this:

type BackfillBatchInput interface {
	BatchInput

	// BackfillReadBatch must return ErrBackfillComplete once the backfill,
	// including any trailing partial batch, is fully read.
	BackfillReadBatch(context.Context) (MessageBatch, AckFunc, error)
}

// BackfillCompleter is an optional extension of BackfillBatchInput.
// BackfillComplete is called once every backfill batch has been settled
// downstream and before steady-state ReadBatch calls begin.
type BackfillCompleter interface {
	BackfillComplete(context.Context) error
}

[claude]

Note: the inline-comment service was unavailable during this review (repeated No server is currently available errors), so the findings below are reported here with file:line references instead of as inline comments.

Commits

1. api: backfillasyc poc — violates the commit message policy on three counts:
   • Typo in the scope word: backfillasyc (presumably backfill async).
   • Not imperative mood: the message is a noun phrase (poc) with no verb. Policy requires imperative mood, e.g. add, fix.
   • Vague: poc does not describe the change. Something like input: add backfill phase to async reader api would satisfy the policy.

Review

The PR adds an optional backfill phase to AsyncReader (BackfillAsync/BackfillCompleter internally, BackfillBatchInput/BackfillCompleter publicly), with an ack barrier that blocks steady-state reads until every backfill batch is settled, plus auto-retry support via an independent autoretry.List. The core barrier logic and the auto-retry drain interaction (Shift(ctx, false) blocking until exhausted(), so nacked backfill batches cannot be lost) look correct. Four issues below.

1. internal/component/input/async_reader_snapshot_test.go:197 — test can never pass. TestAsyncReaderSnapshotPhaseSkippedWhenEmpty asserts []string{"backfill-complete", "stream-ack-0"}, but the mock's BackfillComplete records the string "snapshot-complete" (line 81 of the same file). The preceding require.Eventually only checks len(Events()) == 2, so it succeeds and then the assert.Equal fails. Fix by matching the recorded string, or rename the recorded event to "backfill-complete" and also update TestAsyncReaderSnapshotPhaseBarrier, which asserts "snapshot-complete" at line 164.

2. internal/component/input/async_reader.go:178-185 — the ack barrier can stall the input indefinitely when a downstream component settles messages only after a flush that needs more input. Concrete case: an output with a count-based batching policy and no period (e.g. count: 100). The trailing backfill batches sit unflushed in the batcher, so their acks never arrive; runBackfillPhase returns after ErrBackfillComplete and then blocks in pendingBackfillAcks.Wait(), so ReadBatch is never reached and no further messages can arrive to trigger the flush. The stream only unwedges at shutdown. Same applies to a buffer, or a batch/group_by-style processor that holds messages back. Worth deciding explicitly how the barrier should interact with downstream batching (document the constraint, force a flush, or bound the wait) before CDC connectors depend on it.

3. internal/component/input/async_reader.go:228-230 — the backfill loop never records input_latency_ns. The steady-state loop captures startedAt := time.Now() before dispatch and calls mLatency.Timing(...) in the ack goroutine (lines 319 and 342). The backfill path increments input_received but omits the latency timing entirely, so end-to-end latency is unreported for the whole backfill phase — which for a CDC table snapshot may be the bulk of the messages an input ever emits.

4. public/service/errors.go:33-37 — godoc on new public API references a type that does not exist. The comment says ErrBackfillComplete is returned by "a SnapshotBatchInput's BackfillReadBatch method"; the interface added in public/service/input.go is BackfillBatchInput. The error string also says "snapshot phase complete" while the rest of the API is named around "backfill". Same wording mismatch in internal/component/errors.go for component.ErrBackfillComplete, and the leftover snapshot/Snapshot naming in the new tests (fnSnapshotBatchInput, snapshotAsyncBatchInput, async_reader_snapshot_test.go, TestAsyncReaderSnapshotPhase*) makes the new API harder to find. Aligning on Backfill* before this lands as public API would avoid a rename later.

[claude]

Backfill ack barrier can hang indefinitely and a graceful shutdown cannot break it.

The ack goroutine spawned below only unblocks on rChan or r.shutSig.HardStopChan() (async_reader.go#L259-L265). So if a trailing partial backfill batch is sitting in a downstream batching policy with no period — exactly the scenario the new warning describes — pendingAcks.Wait() never returns, and the <-done after the 60s warning blocks forever.

Crucially this differs from the pre-existing pendingAcks wait in loop(), which only runs at shutdown. Here the barrier runs mid-stream while the input is otherwise healthy, and because the backfill loop has already stopped producing, nothing will ever fill that downstream batch. TriggerStopConsuming() does not release it (the goroutine ignores SoftStopChan), so loop() never reaches r.shutSig.TriggerHasStopped() and WaitForClose blocks until its own context deadline. Only TriggerCloseNow() escapes.

Suggested fix: also select on r.shutSig.SoftStopChan() in the wait (returning ok = false, same as the hard-stop case) so a graceful shutdown can abort a stuck backfill barrier, and/or re-emit the warning periodically rather than once.

[claude]

A nacked backfill batch still lets BackfillComplete fire, contradicting the guarantee stated above.

The doc comment on runBackfillPhase says "ok must be false so the caller never calls BackfillComplete for unconfirmed rows", but the only thing that clears ok is a hard stop. If a backfill batch is delivered and then permanently nacked downstream, res != nil is passed to aFn, pendingAcks drains normally, ok stays true, and BackfillComplete is invoked at async_reader.go#L279-L287 — so a CDC connector persists its post-backfill resume position for rows that never landed. That is silent snapshot data loss.

Note this is only masked when the input is wrapped in AutoRetryNacksBatched; a BackfillAsync reader that does not use it (nothing in the API requires it) gets the unsafe behaviour. Either the nack result should be tracked and clear ok, or the guarantee in the comment and in BackfillCompleter's godoc (interface.go#L426-L434) needs to be reworded to make clear it means "settled", not "delivered".

[claude]

input_latency_ns is not recorded for the backfill phase.

The steady-state loop this block was adapted from captures startedAt := time.Now() before dispatch and reports mLatency.Timing(time.Since(startedAt).Nanoseconds()) in the ack goroutine (async_reader.go#L333-L357). The backfill loop increments mRcvd but never touches mLatency, so the entire snapshot phase — often the longest-running and slowest-acking part of a CDC input — is invisible in the input latency metric.

Suggested fix: capture startedAt here and record mLatency.Timing(...) in the backfill ack goroutine, matching the steady-state loop.

[claude]

The godoc for this new exported symbol references SnapshotBatchInput, which does not exist anywhere in the tree. The interface introduced by this PR is BackfillBatchInput (input.go#L478-L484). Since this is public API documentation, it should name the real type.

The same snapshot/backfill mix-up shows up in the error string "snapshot phase complete", and in interface.go's "distinct initial backfill (backfill) phase" — worth a sweep before this leaves POC state.

[claude]

Commits

1. api: backfillasyc poc — three problems: backfillasyc is a typo; api is not a system/area in this repo (the change lives in internal/component/input and public/service, so input: would be the right scope); and poc is not an imperative-mood description of the change. Something like input: add optional backfill read phase fits the policy.
2. add warning if partial flush is stuck waiting — missing the required system: prefix. The change is scoped to internal/component/input/async_reader.go, so it needs a scope (e.g. input: warn when backfill ack barrier stalls) rather than the sentence-case form reserved for repo-wide changes.
3. Granularity — commit 2 only refines code introduced by commit 1 in this same PR; it is effectively a fixup and should be squashed into it.

Review

Reviewed the new optional backfill API: component.ErrBackfillComplete, the BackfillAsync/BackfillCompleter internal interfaces, AsyncReader.runBackfillPhase, the airGapBackfillBatchReader shim, and the autoRetryInputBatchedBackfill auto-retry wrapper. The layering is sound — the type-assertion opt-in means non-backfill inputs are genuinely unaffected, newAirGapBatchReader feeds NewAsyncReader directly so the interface is not erased by an intermediate wrapper, and the independent autoretry.List gives the backfill phase a second barrier that composes correctly with the AsyncReader one. Test coverage of the barrier, the empty-backfill case, and the hard-stop abort is good for a POC.

The findings below are about the ack barrier's failure modes, which are the part of this design most worth settling before it stops being a POC.

1. Backfill ack barrier can hang indefinitely and a graceful shutdown cannot break it (internal/component/input/async_reader.go) — the ack goroutines only unblock on rChan or HardStopChan, so a trailing partial batch stuck in a downstream batching policy with no period blocks <-done forever. Unlike the pre-existing pendingAcks wait, this one runs mid-stream, and TriggerStopConsuming() will not release it — only TriggerCloseNow() does.
2. A nacked backfill batch still lets BackfillComplete fire (internal/component/input/async_reader.go) — only a hard stop clears ok, so a permanently nacked snapshot batch still results in the connector persisting a post-backfill resume position. This contradicts the "never calls BackfillComplete for unconfirmed rows" guarantee stated in the code, and is only masked when the input happens to be wrapped in AutoRetryNacksBatched.
3. input_latency_ns is not recorded during the backfill phase (internal/component/input/async_reader.go) — the loop was adapted from the steady-state loop but dropped startedAt/mLatency.Timing, so the slowest phase of a CDC input is invisible in the latency metric.
4. Public godoc references a non-existent SnapshotBatchInput (public/service/errors.go) — the type is BackfillBatchInput. The same snapshot/backfill mix-up appears in the error string and in interface.go.