aws_dynamodb_cdc: ack-gate snapshot checkpoints and fix shard start_from handling (CON-504)

[squiidz]

Part of CON-504 (CDC at-least-once / ack-gated progress).

Closes the three gaps from the CON-504 audit that are still live on main (the audit's fourth finding — CDC-path nack pinning — is stale: RecordBatcher has since gained per-shard ordered tracking with documented pin-on-nack semantics, so only the retry wrapper is added on top).

1. Snapshot checkpoints and completion persisted at read time. The scanner wrote UpdateSnapshotProgress inline during the scan and MarkSnapshotComplete fired as soon as the scan finished, while the snapshot batches' ack function was a no-op. A crash (or terminal nack) after persistence but before delivery skipped the un-acked items on restart. The scanner no longer persists anything: each batch carries its scan resume position to a per-segment ordered tracker (mirroring the CDC path's RecordBatcher), segments are sealed behind their in-flight batches so Complete=true persists only after they all ack, and the snapshot is marked complete only once every emitted batch is acknowledged — with nack failure and per-attempt gate reset, per the pattern hardened on #4675/#4677.

2. start_from: latest applied to rotation children. Any checkpoint-less shard got LATEST, including children created by DynamoDB Streams' ~4h shard rotation (discovered by the periodic refresh) and children created while the pipeline was down — silently skipping their backlog in steady state. start_from is now honored only on the first discovery of a pipeline whose checkpoint store holds no prior state (new Checkpointer.HasAnyState probe); everything else starts at TRIM_HORIZON. Exact-checkpoint and global-table failover resume paths are unchanged.

3. No nack redelivery. The input lacked the standard auto_replay_nacks toggle: a rejected batch pinned its shard's frontier (correctly, no loss) but records were only redelivered by a restart. Wrapped with AutoRetryNacksBatchedToggled, consistent with the other CDC inputs.

Proof of Work

• New unit tests: the snapshot ack tracker (read-time persistence forbidden, out-of-order acks never skip, interval throttling, seal-behind-in-flight-batches, never-acked batch pins progress and completion) and HasAnyState (fresh/existing/namespaced).
• New adversarial integration test (TestIntegrationDynamoDBSnapshotAckGate): receive the whole snapshot without acking → assert zero checkpoint rows → restart re-delivers every item → completion persists only after a fully-acked run. Fails on pre-fix code: no snapshot checkpoint state may be persisted before the batch is acknowledged: Should be zero, but was 2 (segment progress + completion marker written at read time).
• All 4 existing integration tests pass (-tags integration: Streams, Snapshot incl. resume, MultiTable, TagDiscovery), full package green under -race, lint clean, docs regenerated for the new field.

[claude]

The persist bookkeeping (st.persistedBatches, st.persistedComplete) is committed before the store write is attempted, and there is no rollback when the write fails.

Failure scenario: a segment's last batch is acked, the frontier reaches the completion marker, so completeDue is true and st.persistedComplete = true is recorded — then UpdateSnapshotProgress returns a transient DynamoDB error (throttle/5xx). Nothing ever retries that write: persistedComplete now says the marker is durable, and no further Ack for this segment will ever arrive (the completion marker is the last tracked item). Meanwhile the ack function has already run snapshot.ackWG.Done() via defer, so the completion gate in waitAcks still drains cleanly and MarkSnapshotComplete proceeds. The segment's durable row is left at Complete=false with a stale LastKey, so if the snapshot ever has to resume (e.g. the CDC checkpoint goes stale and forces a re-snapshot) that segment restarts from the stale position. The same applies to the interval path: persistedBatches is advanced on a failed write, delaying the next checkpoint by another full interval.

Suggested fix: perform the store write first and only mutate persistedBatches/persistedComplete after it succeeds (re-acquiring t.mu to commit, or resetting the fields on error) so a failed write is retried by the next ack.

Reference: snapshot_ack.go L114-L127; CONTRIBUTING.md §3.1.4 (implementation complete and correct) and §3.2.2 (poor error handling).

[squiidz]

Fixed in 76385ac — the store write happens first and persistedBatches/persistedComplete commit (re-acquiring the mutex) only on success, so a failed write is retried by the next ack or seal. New unit test covers the failed-write-then-retry path.

[claude]

The gate treats context cancellation as a snapshot failure, unlike the Scan call directly above it which deliberately ignores context.Canceled (line 1449).

Failure scenario: a pipeline is stopped gracefully while a snapshot is still running. Close() calls TriggerSoftStop(), which cancels scanCtx; Scan returns context.Canceled and is correctly swallowed, but execution then falls through to waitAcks(scanCtx), which immediately returns ctx.Err(). The result is an ERROR-level log (snapshot completion gate for table ...: context canceled) on every graceful stop mid-snapshot, plus d.snapshot.state = snapshotStateFailed and d.snapshot.err set — which ReadBatch L3091-L3098 then returns to the framework as a pipeline error instead of ErrNotConnected/ErrEndOfInput. Nothing is actually wrong: the un-acked items simply resume next run.

Suggested fix: distinguish cancellation from a real gate failure — treat errors.Is(err, context.Canceled) as a normal shutdown (debug/info log, leave the snapshot state as in-progress) and reserve the failed state + error for a recorded nack.

This is a §1.2.2 issue as written in CONTRIBUTING.md: "Unexpected behavior should emit warning or error logs. Normal operation should emit no logs." A graceful stop is normal operation.

[squiidz]

Fixed in 76385ac — errors.Is(err, context.Canceled) is treated as normal shutdown: debug log, snapshot state left in progress, no error surfaced. The failed state is reserved for a recorded nack or a genuine gate failure.

[claude]

The new honorStartFrom guard covers shard discovery, but the iterator-expiry recovery path still applies start_from: latest unconditionally, so the exact failure mode this commit fixes remains reachable.

Failure scenario: a stream-rotation child shard is discovered on a later refresh cycle, so (correctly, post-fix) it starts at TRIM_HORIZON. Before anything is read from it, its iterator expires (DynamoDB Streams iterators expire after 15 minutes — easy on an idle or backpressured shard). refreshExpiredIterator is then called with lastSeq == "" and no checkpoint for that shard, and resolveResumeIterator L2091-L2102 hits case startFrom == "latest" and re-acquires LATEST — silently skipping that shard's backlog, which is precisely what the commit message says must not happen.

Suggested fix: thread the same "genuinely fresh pipeline" decision into resolveResumeIterator (e.g. pass the effective start position rather than d.conf.startFrom) so a shard that was positioned at TRIM_HORIZON cannot be re-positioned at LATEST on iterator refresh.

CONTRIBUTING.md §3.1.4 — "The implementation is complete and correct, with no known bugs or missing core functionality."

[squiidz]

Fixed in 76385ac — resolveResumeIterator no longer has a LATEST arm at all: an expired iterator means the shard was already positioned, so re-acquiring LATEST would skip everything published since (including a TRIM_HORIZON child's whole backlog). Recovery now falls back to lastSeq → checkpoint → TRIM_HORIZON, with the expired-iterator test table updated to lock that in.

[claude]

The start_from gating behavior change has no behavioral test on either path — neither this multi-table branch nor the single-table one at L1681-L1691. TestHasAnyState only covers the new probe; nothing asserts the actual contract, i.e. that with start_from: latest a checkpoint-less shard discovered on a later refresh cycle (or after a restart with existing state) is positioned at TRIM_HORIZON rather than LATEST. A regression here is silent data skipping with no test to catch it.

Suggested fix: extract the iterator-type decision into a pure helper (like the existing resolveResumeIterator, which is unit-tested in input_cdc_expired_iterator_test.go) taking (startFrom, honorStartFrom, resumeMode) and table-drive it, or add an integration case that runs a pipeline with existing checkpoint state and asserts a newly-discovered shard replays its backlog.

Per the project test patterns, changed behavior needs tests; this also backs CONTRIBUTING.md §1.3.2 ("Tests should cover end-to-end functionality and prove that the connector works across supported configurations").

[squiidz]

Fixed in 76385ac — the discovery decision is extracted into initialIteratorType(startFrom, honorStartFrom), used by both the single-table and multi-table arms, with a table-driven test asserting the contract: latest only on a fresh pipeline's first discovery; rotation children and restart-with-state always TRIM_HORIZON.

[claude]

The snapshot persist interval is a bare literal 10 here, duplicated as a second literal fallback inside newSnapshotAckTracker L74-L84 (if interval <= 0 { interval = 10 }). The project Go patterns require named constants: "Name all numeric constants. Every literal number in logic must have a clear meaning through a named constant or variable."

Suggested fix: declare one package-level constant (e.g. defaultSnapshotCheckpointBatchInterval = 10) and use it in both places, so the two defaults cannot drift.

[squiidz]

Fixed in 76385ac — one package constant, defaultSnapshotCheckpointBatchInterval, used at the call site and as the tracker's fallback.

[claude]

Race: the store write is issued outside t.mu, so concurrent acks on the same segment can land out of order.

Ack computes toPersist under the lock, then releases it before calling UpdateSnapshotProgress. Ack functions for a segment's batches are invoked concurrently by the framework (one per in-flight batch), so two Ack calls for the same segment can be in the write phase at the same time with different frontiers, and nothing orders the two PutItems.

Concrete failure (segment 0, interval 10, batches b1…b20):

1. Ack(b10) takes the lock, frontier k10, intervalDue fires, toPersist = {lastKey: k10}, unlocks, and is descheduled before the write.
2. The remaining acks plus SealSegment resolve; another Ack computes toPersist = {complete: true}, writes PutItem{Complete: true}, and sets st.persistedComplete = true.
3. Ack(b10)'s write now lands and, because UpdateSnapshotProgress does a whole-item PutItem (checkpoint.go:566-581), overwrites the row with LastKey: k10, Complete: false.

The durable segment row is left incomplete at an older position, and persistedComplete == true means no later ack will ever rewrite it. If the process dies before MarkSnapshotComplete runs, the restart re-scans the segment from k10 and re-delivers ~10 batches that were already acked. No loss (frontiers only move forward), but the window this PR is closing is partly reopened as duplicate delivery.

Suggested fix: serialize the write for a segment (hold a per-segment write lock across the store call), or make the commit monotonic — record the position actually written and drop a write whose frontier is behind what has already been persisted.

Ref: Review policy — bugs and race conditions, snapshot_ack.go#L124-L142

[squiidz]

Fixed in e7881f2 — see the reply on the newer duplicate thread: per-segment persistMu held across compute+write+commit, with a concurrency test proven red pre-fix.

[claude]

The new test was inserted between // TestIntegrationDynamoDBSnapshot tests snapshot functionality. and the function it documents, so that comment now reads as the first line of TestIntegrationDynamoDBSnapshotAckGate's doc comment, and TestIntegrationDynamoDBSnapshot (further down) is left undocumented.

Move the new test (with only its own comment) below TestIntegrationDynamoDBSnapshot, or move the stale line back down so it sits directly above func TestIntegrationDynamoDBSnapshot.

Ref: Go patterns — documentation

[squiidz]

Fixed in e7881f2 — see the newer duplicate thread.

[squiidz]

The nack-handling changes from the earlier review rounds have been unwound in the latest commit. The framework's documented contract for auto_replay_nacks reads: "If set to false these messages will instead be deleted." — disabling replay is an explicit opt-in to drop rejected messages (typically because failures are routed to a DLQ, which acks them). Pinning the checkpoint on nack contradicted that contract and produced permanent backpressure once the in-flight limit filled. Ack functions now resolve their checkpoint slot on nack exactly like on ack.

Unwound here (5765a61): the completion gate no longer fails on nack, and a nacked snapshot batch resolves its tracker slot so the segment checkpoint advances past it. Kept: ack-gated segment checkpoints (write-then-commit), the completion gate waiting for all in-flight batches to settle (the crash window), SealSegment, start_from scoping, never-LATEST resume, and the auto_replay_nacks toggle. Full integration suite re-verified green. The pre-existing RecordBatcher behavior is untouched.

[claude]

Completion gate can never drain after a reconnect (snapshot hangs)

snapshotState — and therefore the new ackWG — is allocated once, in the constructor (input_cdc.go#L876-L879), but the batches it counts are per-connection: Connect() unconditionally allocates a fresh d.msgChan (input_cdc.go#L1012-L1014) with no "already connected" guard, and the framework re-invokes Connect without Close after ReadBatch returns service.ErrNotConnected — which this input returns from several branches (input_cdc.go#L3078-L3084).

Failure scenario (snapshot_only, or snapshot_and_cdc before the snapshot finishes):

1. Snapshot emits batches; handleSnapshotBatch does snapshot.ackWG.Add(1) per batch and hands the ack fn to d.msgChan.
2. A reconnect occurs while batches are still buffered in d.msgChan. Connect() replaces the channel; those batches are orphaned and their ackFunc (the only caller of ackWG.Done()) is never invoked.
3. The snapshot checkpoint is still incomplete, so connectWithSnapshot runs a second scan (input_cdc.go#L1287 falls through).
4. That run's waitAcks blocks forever on the leaked counter → MarkSnapshotComplete is never written, and in snapshot_only mode TriggerSoftStop() is only reached after the gate (line 1464), so the input never returns ErrEndOfInput and stalls until the shutdown timeout.

Suggested fix: scope the ack gate to a connection — allocate the ackWG (and the ackTracker) alongside the new msgChan in Connect()/connectWithSnapshot rather than reusing the input-lifetime snapshotState, and/or settle the orphaned batches when the channel is replaced. A bounded wait instead of an unbounded one would also stop a single leaked batch from wedging the snapshot permanently.

[squiidz]

Fixed in e7881f2. The completion gate is no longer on the input-lifetime snapshotState: connectWithSnapshot allocates a fresh WaitGroup per connection attempt and the scanner batch callback captures it, so it only counts batches its own attempt emitted. Batches a previous attempt left buffered in a replaced msgChan settle (or leak) on their own attempt's gate and can never wedge the current run's waitAckGate. Full integration suite re-verified green.

[claude]

Race: the store write is issued outside the mutex, so segment positions can be persisted out of order

Ack releases t.mu before calling t.store.UpdateSnapshotProgress and re-acquires it afterwards. Acks for the same segment run concurrently (multiple in-flight batches, acked from the output/AutoRetryNacksBatched goroutines), so:

• Ack A: ackedBatches=10, persistedBatches=0 → intervalDue → unlocks with frontier k10, starts its PutItem.
• Ack B (concurrent): persistedBatches is still 0, so 11-0 >= 10 → also unlocks, with frontier k11, and starts its PutItem.

The two PutItems can land in either order, so the durable row can end up at the older k10, or — when one of them is the completion write (lastKey == nil ⇒ Complete=true, see checkpoint.go#L563-L576) — a stale in-flight write can overwrite Complete=true back to Complete=false with an older LastKey. On restart the segment then re-scans from an already-acked position or re-runs a completed segment, contradicting the "only ever called with the highest contiguous acknowledged position" invariant in the type doc.

Related, same window: st.persistedBatches = st.ackedBatches on re-lock credits acks that arrived after the write was computed, so those batches are counted as persisted when the completed write did not cover them, and the next interval persist is skipped.

The CDC path in this package holds the lock across the store write for exactly this reason — RecordBatcher.AckMessages does defer b.mu.Unlock() and calls cp.Set inside the critical section (batcher.go#L206-L237). Serializing per-segment writes the same way (or guarding with a per-segment write lock/sequence number) would restore the invariant and keep this consistent with the sibling implementation, per CONTRIBUTING.md §3.1.5.

[squiidz]

Fixed in e7881f2. Each segment now serializes the whole resolve+compute+write+commit sequence behind a per-segment persistMu (matching RecordBatcher's hold-the-lock-across-the-write discipline, without serializing segments against each other): every persist is computed after the previous write finished, so the durable row only moves forward, the interval check can't double-fire, and the commit records the acked count captured at compute time instead of crediting later acks. New concurrency test (200 concurrent acks + seal, interval 1) proven red against the unfixed code.

[claude]

Doc comment now documents the wrong function

The new test was inserted between the existing // TestIntegrationDynamoDBSnapshot tests snapshot functionality. comment (line 536) and the function it described, so that comment is now attached to TestIntegrationDynamoDBSnapshotAckGate and TestIntegrationDynamoDBSnapshot is left undocumented. Move the new function (with its own comment) below TestIntegrationDynamoDBSnapshot, or move the old comment back down to its function.

[squiidz]

Fixed in e7881f2 — the doc comment is reattached to TestIntegrationDynamoDBSnapshot and the new test carries only its own.

[claude]

The completion gate was correctly scoped per connection attempt (ackGate is a local captured by the scanner callbacks), but the tracker itself was left on the input-lifetime snapshotState and is dereferenced live at input_cdc.go#L1386-L1392 and input_cdc.go#L2820-L2822.

In the same "Connect re-runs while a previous attempt's scanner is still alive" scenario that motivated scoping the gate (per the commit message for e7881f2), the old scanner goroutine will:

1. race with this unsynchronized field write (-race sees write here vs. reads at :1387 / :2820, neither under d.mu), and
2. call TrackBatch/SealSegment on the new attempt's tracker, breaking the invariant documented on TrackBatch ("Must be called in scan order per segment (the segment's single scan goroutine)"). Two interleaved scan cursors on one segment's ordered tracker can resolve a contiguous frontier no single scan actually acked, and an old scanner's SealSegment can persist Complete=true for a segment whose new-attempt batches are still in flight — those items are then skipped on the next restart, which is the data-loss window this PR set out to close.

Suggested fix: allocate the tracker as a local next to ackGate and capture it in all three callbacks (batch, sealed) rather than reading d.snapshot.ackTracker at call time.

[squiidz]

Fixed in bea7052 — the tracker is allocated as a local next to ackGate and captured by the batch and seal callbacks; handleSnapshotBatch takes it as a parameter, and the now write-only snapshotState field is removed. Full integration suite re-verified green.

[claude]

HasAnyState is now the sole gate for honoring start_from: latest, but Query defaults to eventually consistent reads.

Failure scenario: a pipeline crashes and is restarted within a second or two (the common restart case). The checkpoint rows written moments before the crash are not yet visible to this probe, so HasAnyState returns false → honorStartFrom.Store(true) → shards whose per-shard checkpoint read (also eventually consistent, Get) is stale for the same reason fall into the default branch and are re-positioned at LATEST — silently skipping their backlog, which is exactly the hole this change closes.

Since the probe is Limit: 1, adding ConsistentRead: true makes the decision deterministic for a negligible extra RCU.

[squiidz]

Fixed in bea7052 — the probe now sets ConsistentRead: true. (The per-shard Get staleness alone is safe: with honorStartFrom false a missing checkpoint falls through to TRIM_HORIZON, so the probe was the only decision that needed the strong read.)

[claude]

require.NoError inside the wg.Go goroutine calls t.FailNow() from a non-test goroutine, which testify documents as unsupported — on failure this panics (or leaves the test wedged) instead of failing cleanly, hiding the very regression this test is meant to catch.

This is the same rule the project's test patterns state for polling helpers: "require calls FailNow() which panics when called from a non-test goroutine. Use assert or return bool."

Collect the errors from each goroutine (e.g. into a slice guarded by the existing mutex pattern, or a buffered channel) and assert with require after wg.Wait().

[squiidz]

Fixed in bea7052 — ack errors are collected through a buffered channel and asserted with require after wg.Wait().

[claude]

Project Go style rule: "Use new(X) instead of &X{} for zero-value struct pointers." This should be new(sync.WaitGroup).

[squiidz]

Fixed in bea7052 — new(sync.WaitGroup).

[claude]

start_from documentation is now stale for the new semantics.

This commit narrows start_from: latest so it applies only to the first shard discovery of a pipeline whose checkpoint store is empty. Every other checkpoint-less shard — rotation children found by the periodic refresh (~4h on active streams), and any shard after a restart where HasAnyState returns true — now starts at TRIM_HORIZON regardless of the configured value.

The user-facing description was not updated and still describes the old behaviour:

connect/internal/impl/aws/dynamodb/input_cdc.go

Lines 235 to 238 in bea7052

	service.NewStringEnumField(dciFieldStartFrom, "trim_horizon", "latest").
	Description("Where to start reading when no checkpoint exists. `trim_horizon` starts from the oldest available record, `latest` starts from new records.").
	ShortDescription("Where to start when no checkpoint exists: trim_horizon for the oldest record, or latest.").
	Default("trim_horizon"),

Where to start reading when no checkpoint exists. trim_horizon starts from the oldest available record, latest starts from new records.

That sentence is now wrong for the common case: a shard can have no checkpoint and still be forced to trim_horizon. The regenerated docs/modules/components/pages/inputs/aws_dynamodb_cdc.adoc inherits the same stale text (only auto_replay_nacks was added).

This is a silent behaviour change for anyone running start_from: latest — after a restart they will now replay the full 24h stream retention instead of only new records, which is exactly the kind of edge case CONTRIBUTING.md §1.2.3 requires to be documented ("Known limitations and edge cases are documented"), and §1.1.2 ("expected usage patterns, troubleshooting guidance, and known pitfalls").

Suggested fix: extend the Description()/ShortDescription() on dciFieldStartFrom to state that latest is honoured only on a genuinely fresh pipeline (no prior checkpoint state under this checkpoint_namespace + stream), and that shards discovered later or after a restart always start at trim_horizon so their backlog is not skipped — then re-run task docs so the adoc picks it up.