mssqlserver_cdc: ack-gate snapshot checkpoint and fix transaction-boundary resume (CON-504) by squiidz · Pull Request #4677 · redpanda-data/connect

squiidz · 2026-08-06T19:35:08Z

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

Fixes three at-least-once gaps found in the CON-504 audit, plus one data race the new tests surfaced:

1. Post-snapshot LSN persisted at read time. After the snapshot read loop finished, cacheLSN(maxLSN) ran while snapshot batches were still un-acked downstream (and a partial batch could still sit unflushed in the batcher). A crash in that window meant the restart saw a cached LSN, skipped the snapshot, and silently lost the un-delivered rows. The handoff now flushes the remaining partial batch, blocks until every snapshot batch is acknowledged (escapable by soft-stop), and only then persists the LSN — the same barrier as postgres (#4584) and oracledb (#4675).

2. Transaction tail skipped on resume (tie-group). All rows of a transaction share one __$start_lsn and resume is exclusive (> lsn), but the checkpoint used the last row's own LSN. Acking a batch that ended mid-transaction persisted that transaction's LSN; a crash then skipped every remaining row of the same transaction on restart. Rows now carry checkpoint_lsn — the start LSN of the most recent transaction whose rows are all published, computed in the globally LSN-ordered stream loop — and only that value is ever persisted. Partially-delivered transactions replay in full (duplicates, not loss). Known limitation: the final transaction of a burst is checkpointed once a later transaction is observed; until then a restart re-delivers it.

3. Out-of-order checkpoint tracking. checkpoint.Track ran outside the batcher mutex, so the count-triggered flush (Publish) and the timed-flush loop could register batches with the ordered tracker in the wrong order and persist a regressing LSN on ack. Track now happens under the same lock as the flush.

4. Data race on batcher state (found by the new stress test under -race). The timed-flush loop read batcher.UntilNext() unlocked while Publish mutated the batcher under the mutex. Now locked.

Proof of Work

Unit tests: snapshot ack gate (ack/nack/streaming/cancellation), flushCurrent, checkpoint selection (checkpoint_lsn preferred, row LSN never persisted, first-transaction batches persist nothing), transaction-boundary tracker, and a concurrent-flush ordering stress test (race-clean, checkpoint never regresses).
TestIntegration_MicrosoftSQLServerCDC_SnapshotAckBarrier: blocked consumer + simulated crash → no checkpoint persisted → restart re-runs the snapshot. Fails on pre-fix code: post-snapshot LSN must not be persisted before snapshot rows are acknowledged: Should be zero, but was 1.
TestIntegration_MicrosoftSQLServerCDC_TransactionSplitAcrossBatches (the ticket's commit-order ≠ batch-order criterion): a 4-row transaction split across count-2 batches, first batch acked, crash, restart. Fails on pre-fix code — every tail row reported never redelivered (checkpoint advanced past a partially-delivered transaction); passes with the fix.
Full package green under -race; existing resume/ordering integration tests unaffected.

Note: the Track-outside-mutex and unguarded-UntilNext races also exist in the oracledb batcher (same lifted pattern) — flagged as a follow-up for #4675.

Track order defines the ordered checkpoint sequence, but Track was called after releasing the batcher mutex, so the count-triggered flush (Publish) and the timed-flush loop could register batches out of order and persist a regressing LSN on ack. Track now happens under the same lock as the flush. Also guards the loop's UntilNext call, which read batcher state concurrently mutated by Publish (caught by the new stress test under -race).

All rows of a transaction share one __$start_lsn and resume is exclusive (> lsn), so persisting the last row's own LSN while its transaction was only partially delivered skipped the transaction's remaining rows on restart. Each row now carries checkpoint_lsn - the start LSN of the most recent transaction whose rows are all published - and only that value is persisted. Partially-delivered transactions replay in full on restart (duplicates, not loss). The final transaction of a burst is checkpointed once a later transaction is observed; until then a restart re-delivers it.

…it transactions

claude · 2026-08-07T13:18:34Z

+// batches release the gate too: redelivery is owned by auto_replay_nacks,
+// and the ctx escape prevents a permanently-failing downstream from
+// wedging shutdown.
+func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {


The gate treats a nack exactly like an ack, but the redelivery guarantee it relies on is user-toggleable, so the barrier can still lose snapshot rows.

ackFn discards the ack error (func(ctx context.Context, _ error) error) and unconditionally runs defer b.snapshotAckWG.Done(), so a nacked snapshot batch releases the gate. The doc comment justifies this with "redelivery is owned by auto_replay_nacks", but this input exposes that as an opt-out field — service.NewAutoRetryNacksToggleField() / AutoRetryNacksBatchedToggled.

Failure scenario: auto_replay_nacks: false, stream_snapshot: true, downstream rejects one snapshot batch. waitSnapshotAcks returns as if it had been delivered, cacheLSN(softCtx, maxLSN) persists the post-snapshot LSN, and on restart the snapshot is skipped — the rejected rows are gone. That is the exact data-loss window this PR set out to close, and it contradicts CONTRIBUTING.md §5.4.2 ("at-least-once delivery, with progress gated on downstream ack").

Suggested fix: track nacked snapshot batches separately (e.g. a flag set from ackFn when the error is non-nil) and have waitSnapshotAcks return an error in that case so the post-snapshot LSN is not persisted; alternatively gate the barrier on auto_replay_nacks actually being enabled.

Fixed in 53aef1f. A nacked batch no longer resolves its checkpoint slot (nothing can be persisted past its rows), and a nacked snapshot batch records the error so waitSnapshotAcks returns it — cacheLSN is skipped and the snapshot re-runs on restart. Covered by the reworked "a nack releases the gate but fails it" subtest plus a new "a nacked batch pins the checkpoint" streaming subtest. Note the same gap exists in postgres (#4584, merged) and oracledb (#4675) — follow-ups planned.

claude · 2026-08-07T13:18:46Z

+// that flush the batcher themselves must instead track under the same lock as
+// their flush (see Publish/loop/flushCurrent) to keep Track order == flush
+// order.
+func (b *batchPublisher) publishBatch(ctx context.Context, batch service.MessageBatch) error {


After this refactor publishBatch has no production callers left — the only reference in the repo is batcher_test.go:287. The three real flush paths (Publish, loop, flushCurrent) all call trackBatchLocked + sendTracked directly, and this function's own doc comment describes a caller shape ("a batch that was flushed elsewhere") that no longer exists.

Two consequences:

Dead code that golangci-lint's unused will not report, because a test file keeps it alive.

TestSnapshotAckGate and TestCheckpointSelection drive the gate/checkpoint logic exclusively through this orphaned entry point, so they never exercise the locking contract the PR is actually fixing (Track under the same batcherMu critical section as the flush). Per tester patterns, these should go through the production path.

Suggested fix: delete publishBatch and rewrite those tests to publish via Publish (count-triggered flush) and flushCurrent, which is what production does.

Fixed in 53aef1f. publishBatch is deleted and the batcher tests now drive the production paths — Publish with a count-1 batcher and flushCurrent — so the Track-under-batcherMu contract is what the tests actually exercise.

claude · 2026-08-07T13:18:55Z

 		msg.MetaSet("lsn", string(m.LSN))
 	}
+	if len(m.CheckpointLSN) != 0 {
+		msg.MetaSet("checkpoint_lsn", string(m.CheckpointLSN))


This adds a new user-visible metadata key checkpoint_lsn on every streaming message, but the connector's documented metadata contract is not updated. The == Metadata block in the config spec still lists only database_schema, schema, table, operation, lsn — see input_mssqlserver_cdc.go#L60-L67. That section is what renders on the component docs page, so users now receive an undocumented key (CONTRIBUTING.md §1.1.1 "concise, well-organized documentation", §1.2.3 "known limitations and edge cases are documented").

It is also worth reconsidering whether this should be message metadata at all: it is pure internal plumbing between ReadChangeTables and trackBatchLocked, and it is not one of the fleet metadata names in §5.2.2 (operation, schema, DB-native position key, table, database_schema, source_ts_ms, commit_ts_ms, transaction_id).

Suggested fix: either document checkpoint_lsn in the == Metadata section, or carry the value out-of-band (e.g. on the batch/tracking struct rather than as message metadata) so it stays an implementation detail.

Fixed in 53aef1f — went with the out-of-band option: checkpoint_lsn is no longer message metadata. The value is carried on the publisher as pendingCheckpointLSN, guarded by batcherMu, so at flush time it always mirrors the flushed batch's last message. The documented metadata contract is unchanged.

…p orphaned publishBatch, carry checkpoint LSN out-of-band - A nacked batch no longer resolves its checkpoint slot, and a nacked snapshot batch fails waitSnapshotAcks: auto_replay_nacks is user-toggleable, so a nack can be terminal and the post-snapshot LSN must not be persisted over undelivered rows (the snapshot re-runs on restart instead). - publishBatch had no production callers left after the flush/track refactor; deleted, and the batcher tests now drive the production Publish/flushCurrent paths (exercising the Track-under-mutex contract). - checkpoint_lsn is no longer message metadata: it is internal plumbing, now carried on the publisher (pendingCheckpointLSN, guarded by batcherMu) instead of an undocumented user-visible key.

claude · 2026-08-07T14:32:18Z

+func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
+	drained := make(chan struct{})
+	go func() {
+		// May outlive this call if ctx fires first; bounded by process lifetime.
+		b.snapshotAckWG.Wait()
+		close(drained)
+	}()
 	select {
-	case b.msgChan <- msg:
+	case <-drained:
+		b.snapshotNackMu.Lock()
+		defer b.snapshotNackMu.Unlock()
+		if b.snapshotNackErr != nil {
+			return fmt.Errorf("snapshot batch was rejected downstream: %w", b.snapshotNackErr)
+		}
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
 	}
 }


Bug: snapshotNackErr is sticky across reconnects, so one snapshot nack permanently wedges the input.

snapshotNackErr is only ever written (in recordSnapshotNack) and read here — it is never cleared. The batchPublisher is built once in the constructor (input_mssqlserver_cdc.go#L272-L277) and reused for every Connect, which is explicitly a re-entrant path (Connect resets i.stopSig at L348-L349, and ReadBatch returns service.ErrNotConnected at L471-L472 to trigger the framework's reconnect).

Failure scenario, with auto_replay_nacks: false:

Run 1 snapshot emits batches; one is nacked → snapshotNackErr set.

waitSnapshotAcks fails → input_mssqlserver_cdc.go#L381-L385 logs and calls TriggerHasStopped(); cacheLSN(maxLSN) is skipped (correct so far).

ReadBatch → ErrNotConnected → framework calls Connect again. Still no cached LSN, so a fresh snapshot runs and every row is re-delivered.

waitSnapshotAcks drains the new run's batches, then returns the stale run-1 error — even if every run-2 batch was acked successfully. The post-snapshot LSN is never persisted.

The result is an unbounded loop that re-runs and re-emits the full snapshot on every reconnect and can never make progress, rather than a one-shot retry. Reset the gate state (snapshotNackErr, and ideally assert the snapshotAckWG is drained) at the start of each snapshot run — e.g. an explicit resetSnapshotGate() called from Connect before processSnapshot — so the gate reflects only the current attempt.

This also violates CONTRIBUTING.md §3.1.4 ("The implementation is complete and correct, with no known bugs"). Worth a unit test covering two sequential snapshot runs on the same publisher where the first nacks and the second acks cleanly.

Fixed in 5c04730. resetSnapshotGate() clears the recorded nack at the start of each snapshot attempt (called from Connect before processSnapshot), so a nack fails only the attempt it belongs to and a clean re-run persists the LSN and proceeds to streaming. The WaitGroup is deliberately left untouched — a previous attempt's in-flight batches can still ack or nack and must keep counting. Covered by the suggested two-run unit test: run 1 nacks and fails the gate, run 2 acks cleanly and passes.

…lling window Transaction-boundary checkpointing left the final transaction of a burst un-checkpointed until a later transaction appeared, so a graceful stop on an idle stream re-delivered it on every restart (caught by CI: TestIntegration_MicrosoftSQLServerCDC_ResumesFromCheckpoint). When a polling window drains, every transaction <= lastLSN is fully published, so the stream now registers an empty marker slot carrying the window's end LSN with the ordered tracker. Once all of the window's batches are acked the exact position persists — no trailing-transaction lag on graceful stop or steady state, while a crash mid-window still replays from the last safe boundary (duplicates, never loss).

claude · 2026-08-07T15:56:15Z

+// waitSnapshotAcks blocks until every published snapshot batch has been
+// acknowledged or nacked downstream, or until ctx is cancelled (the escape
+// prevents a stalled downstream from wedging shutdown). Any nack fails the
+// gate: with auto_replay_nacks disabled a nack is terminal, so the
+// post-snapshot LSN must not be persisted and the snapshot must re-run.
+func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
+	drained := make(chan struct{})
+	go func() {
+		// May outlive this call if ctx fires first; bounded by process lifetime.
+		b.snapshotAckWG.Wait()
+		close(drained)
+	}()
 	select {
-	case b.msgChan <- msg:
+	case <-drained:
+		b.snapshotNackMu.Lock()
+		defer b.snapshotNackMu.Unlock()
+		if b.snapshotNackErr != nil {
+			return fmt.Errorf("snapshot batch was rejected downstream: %w", b.snapshotNackErr)
+		}
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
 	}
 }


snapshotNackErr is sticky for the lifetime of the batchPublisher, but the publisher is created once in the constructor (input_mssqlserver_cdc.go#L272-L277) and reused across every Connect — only stopSig is reset. That makes the failure permanent rather than per-connection:

auto_replay_nacks: false, a snapshot batch is nacked → recordSnapshotNack sets snapshotNackErr.

waitSnapshotAcks fails, the goroutine calls TriggerHasStopped, so ReadBatch returns service.ErrNotConnected (input_mssqlserver_cdc.go#L467-L475) and the framework calls Connect again.

No LSN was cached (by design), so len(cachedLSN) == 0 and the snapshot re-runs in full (input_mssqlserver_cdc.go#L338-L341).

On the new run waitSnapshotAcks returns the stale error immediately even if every batch was acked, so the LSN is never persisted and streaming is never reached.

The result is a livelock: the full snapshot is re-emitted downstream on every reconnect and the connector never makes progress, which conflicts with the completeness/diagnosability bar in CONTRIBUTING.md §3.1.4 and §3.2.2.

Suggested fix: reset the gate state (snapshotNackErr, and ideally assert/reset the snapshotAckWG bookkeeping) at the start of each snapshot run — e.g. a resetSnapshotGate() called from Connect before the snapshot goroutine starts — so a nack fails only the connection attempt it belongs to.

Fixed in 5c04730 — see the reply on the duplicate thread above: the gate error now resets per snapshot attempt via resetSnapshotGate(), with a two-run unit test.

snapshotNackErr was sticky for the publisher's lifetime, but the publisher is reused across Connect retries: after one nack, every re-run's waitSnapshotAcks returned the stale error even when the retried snapshot acked cleanly, livelocking the input into re-emitting the full snapshot on every reconnect. The gate error is now cleared at the start of each snapshot attempt; the WaitGroup is deliberately untouched since a previous attempt's in-flight batches can still ack or nack.

claude · 2026-08-10T14:19:07Z

+				if err != nil {
+					// auto_replay_nacks is user-toggleable, so a nack can be
+					// terminal. Never resolve: the checkpoint stays pinned
+					// before this batch so nothing can be persisted past its
+					// undelivered rows. Snapshot nacks additionally fail the
+					// handoff gate so the post-snapshot LSN is not persisted.
+					if isSnapshotBatch {
+						b.recordSnapshotNack(err)
+					}
+					return err
+				}


A nacked batch pins the ordered checkpointer permanently, not just past its own rows.

Returning early here means resolveFn() is never called for this batch, so its slot stays pending in the checkpoint.Capped tracker forever. Because Capped only resolves a contiguous prefix (see how postgres_cdc guards with if maxOffset == nil { return nil } — input_pg_stream.go#L608-L624), every subsequent ack also resolves to nil. Consequences after a single terminal nack (auto_replay_nacks: false, which is exactly the case this PR is designed around):

No LSN is ever persisted again for the lifetime of the process — not "pinned before the nacked batch" for a moment, but permanently. The publisher and its checkpoint.Capped are built once in the constructor (input_mssqlserver_cdc.go#L238-L273) and are reused across Connect retries, so the stale pending slot survives a reconnect too — including the snapshot re-run path that resetSnapshotGate was added for.

Once checkpoint_limit unresolved batches accumulate, checkpoint.Track (called under batcherMu in trackBatchLocked) blocks on capacity and the input stops emitting entirely, with no log line and no error surfaced.

Not advancing past undelivered rows is the right intent, but there is no terminal handling for it. Suggested fix: treat a nack as fatal for the stream — log at error level and trigger a stop so Connect restarts from the last durable LSN (which is what makes the "pin" recoverable) — rather than leaving the tracker wedged. Note this also diverges from the rest of the CDC fleet, which deliberately ignores the ack error and always resolves (oracledb batcher.go#L251-L265); CONTRIBUTING.md §3.1.5 and §3.2.2.

Fixed in 90d6ac3. A terminal nack now triggers a restart (via an onTerminalNack hook on the publisher), and Connect rebuilds the publisher — batcher and ordered tracker — per connection attempt, sealing the old one so late acks from the previous session can neither persist stale positions nor trigger spurious restarts. The restart therefore resumes from the last durable LSN and redelivers, instead of the tracker staying wedged for the process lifetime. Covered by new unit tests (terminal nack invokes the hook; sealed publishers neither persist nor restart) and the full 10-test integration suite. Note the fleet-consistency aside is somewhat stale — oracledb (#4675) gained the same nack guard in its review round; aligning it with this restart+rebuild approach is a reasonable follow-up.

claude · 2026-08-10T14:19:15Z

+			if err = i.publisher.waitSnapshotAcks(softCtx); err != nil {
+				i.log.Infof("Interrupted while waiting for snapshot acknowledgements. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				i.stopSig.TriggerHasStopped()
+				return
+			}


waitSnapshotAcks returns two very different errors through this one branch: ctx.Err() on soft-stop (expected, Info is right) and the wrapped snapshotNackErr when a snapshot batch was rejected downstream (unexpected — the whole snapshot is discarded and re-run on the next connect). The latter is logged at Info with the text "Interrupted while waiting for snapshot acknowledgements", which is both the wrong level and misleading about what happened.

CONTRIBUTING.md §1.2.2: "Unexpected behavior should emit warning or error logs." Suggested fix: distinguish the two cases (e.g. errors.Is(err, context.Canceled) → Info, otherwise Errorf/Warnf with a message that says the snapshot was rejected downstream).

Fixed in 90d6ac3 — cancellation keeps Info, a downstream rejection logs at error level with wording that names it.

Mirrors the mssqlserver review fixes (#4677): a nacked batch no longer resolves its checkpoint slot, and a nacked snapshot batch fails waitSnapshotAcks so the post-snapshot SCN is not persisted over undelivered rows (auto_replay_nacks is user-toggleable, so a nack can be terminal). publishBatch had no production callers left after the flush/track refactor; deleted, with the batcher tests rewritten to drive the production Publish/flushCurrent paths.

A terminal nack (auto_replay_nacks disabled) deliberately pins the checkpoint and eventually stalls the input behind checkpoint_limit, but that consequence was invisible: nothing was logged anywhere on the nack path. Emit an error identifying the batch's checkpoint LSN, whether it was a snapshot batch, and the pinned-checkpoint consequence so operators can connect a stalled input to the downstream rejection.

claude · 2026-08-10T15:08:48Z

+func (b *batchPublisher) CheckpointWindow(ctx context.Context, lsn replication.LSN) error {
+	// Flush buffered rows first: they belong to the window, so the marker
+	// must be tracked after them.
+	if err := b.flushCurrent(ctx); err != nil {


CheckpointWindow unconditionally calls flushCurrent, and ReadChangeTables calls it on every polling window that advanced lastLSN — i.e. on every window that produced any rows, not just on idle. The effect is that the user's batching policy is silently overridden during steady-state streaming: with batching: {count: 5000} (no period), each output batch is sized by however many rows the poll window happened to return rather than by the configured count, so a user configuring large batches for downstream efficiency gets many small ones instead.

The flush is load-bearing for correctness here (buffered rows must be tracked before the marker, otherwise the window-end LSN could resolve ahead of undelivered rows), so this isn't simply removable. Worth considering: only register the marker when the batcher is empty, or attach the window-end LSN as the checkpoint payload of the batch that is eventually flushed, so the configured batching policy still governs batch size.

Per CONTRIBUTING.md §1.1.3 ("UX should be intuitive and require minimal explanation"). At minimum, if this is the intended trade-off, the batching field description should say that batches are also flushed at each CDC polling-window boundary.

Fixed in 90d6ac3 — went with the checkpoint-payload option: CheckpointWindow no longer flushes. When rows from the window are still buffered, the window-end LSN is recorded as pendingCheckpointLSN so it rides as the checkpoint payload of whatever batch the user's batching policy eventually produces (safe: it resolves only after that batch and all predecessors ack, and later rows can only overwrite it with a boundary >= the window LSN). The marker slot is registered only when the batcher is empty. New unit test asserts no force-flush occurs and the deferred window LSN persists with the buffered batch; the batching field description needed no change since the policy is no longer overridden. ResumesFromCheckpoint and the rest of the integration suite stay green.

… tracker, batching policy preserved - A terminal nack (auto_replay_nacks disabled) pinned the ordered checkpoint tracker permanently: the publisher and tracker were built once and reused across Connect retries, so after one nack no LSN could ever be persisted again and the input eventually wedged behind checkpoint_limit. A nack now triggers a restart, and Connect rebuilds the publisher (batcher + tracker) per attempt - sealing the old one so late acks from the previous session cannot persist stale positions - letting the restart resume from the last durable LSN and redeliver. - CheckpointWindow force-flushed the partial batch at every drained polling window, silently overriding the user's batching policy during steady-state streaming. It now defers the window checkpoint onto the buffered batch (the window-end LSN rides as its checkpoint payload) and only registers a marker when the batcher is empty. - The snapshot gate's downstream-rejection failure is logged at error level with wording that names it; soft-stop cancellation keeps Info. Full integration suite (10 tests) green after the changes.

claude · 2026-08-10T20:30:40Z

+	i.publisher.onTerminalNack = func(error) {
+		// i.stopSig is only replaced while the input is stopped, and sealed
+		// publishers never invoke this, so the signaller here is current.
+		i.stopSig.TriggerSoftStop()
+	}


Data race on i.stopSig: this closure reads the i.stopSig field, and it is invoked from ackFn — which benthos calls on the downstream pipeline/output goroutine, not the reader goroutine that runs Connect. Connect reassigns that same field 20 lines below (i.stopSig = shutdown.NewSignaller(), line 379) while holding only connMu, which the ack path never takes.

The window is real: the guard at lines 306-310 only proves the previous session's goroutine exited — batches it already handed downstream can still be nacked at any time. Such a nack runs the previous publisher's onTerminalNack (that publisher is not sealed until line 348, after sql.Open, newCheckpointCache, VerifyUserDefinedTables and getCachedLSN — several DB round trips), so it reads i.stopSig concurrently with the write at line 379. Besides being a race -race can flag, if the read observes the new signaller it soft-stops the freshly-built session immediately, producing a spurious reconnect.

The comment "i.stopSig is only replaced while the input is stopped" holds for the reader goroutine but not for ack goroutines. Suggested fix: create the replacement signaller before installing the callback and capture it in the closure (sig := shutdown.NewSignaller(); i.stopSig = sig; ... func(error) { sig.TriggerSoftStop() }), so the callback never touches the shared field.

Ref: race conditions / concurrency patterns in .claude/agents/godev.md.

Obsolete as of the unwind (c3f2ff9): the ackFn no longer reads i.stopSig — the onTerminalNack hook and the per-Connect publisher rebuild were both removed.

claude · 2026-08-10T20:30:51Z

 		if snapshotter != nil {
+			// The publisher outlives reconnects: clear any nack recorded by a
+			// previous snapshot attempt so the gate judges only this run.
+			i.publisher.resetSnapshotGate()


Dead call and contradictory documentation. As of the publisher rebuild added in this same PR (lines 348-354), i.publisher is a brand-new batchPublisher on every Connect, so snapshotNackErr is always already nil here — resetSnapshotGate() is a no-op and this is its only production caller (the other reference is batcher_test.go:92).

The comment is also now false: "The publisher outlives reconnects" directly contradicts the rebuild 40 lines above. The same claim is repeated in batcher.go:98-103 ("the publisher outlives reconnects, and a stale error would fail every retry") and in the snapshotNackErr field comment. Leaving these in place misdocuments the checkpoint/nack lifetime for the next maintainer, and TestSnapshotAckGate/"a nack fails only the snapshot attempt it belongs to" now exercises a path that cannot occur in production.

Suggested fix: drop the resetSnapshotGate call and method (and the corresponding subtest), or, if it is kept as a defensive measure, reword all three comments to state that the publisher is rebuilt per Connect.

Ref: CONTRIBUTING.md §3.1.2 (self-documenting code) and §3.1.5 (consistency).

Obsolete as of the unwind (c3f2ff9): resetSnapshotGate, the rebuild, and the contradictory comments were all removed.

Aligns with the mssqlserver review outcome (#4677): a terminal nack (auto_replay_nacks disabled) pinned the ordered checkpoint tracker permanently — the publisher and tracker were built once and reused across Connect retries, so after one nack no SCN could ever be persisted again and the input eventually wedged behind checkpoint_limit. A nack now triggers a restart, and Connect rebuilds the publisher (batcher + tracker) per attempt, sealing the old one so late acks from the previous session can neither persist stale positions nor trigger spurious restarts. The restart resumes from the last durable SCN and redelivers.

…an opt-in drop) Unwinds the nack-pinning, gate-failure, and terminal-nack-restart changes from the review rounds. 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 - typically because failures are routed to a DLQ, which acks. Pinning the checkpoint (or restarting with a rebuilt publisher to force redelivery) contradicted that contract: pinning produced permanent backpressure once checkpoint_limit filled, and the restart variant turned a persistently-failing message into an infinite redelivery loop. The snapshot ack gate still guards the crash window; a nack now simply settles its slot and the stream continues. Snapshot barrier, split-transaction, and resume integration tests re-verified green.

squiidz · 2026-08-11T14:15:48Z

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 (c3f2ff9): the nack gate failure and per-connect publisher rebuild. Kept: the snapshot ack gate, transaction-boundary checkpointing (checkpoint_lsn carried out-of-band + CheckpointWindow markers), and the Flush→Track atomicity fix. SnapshotAckBarrier, TransactionSplitAcrossBatches, and ResumesFromCheckpoint integration tests re-verified green.

claude · 2026-08-11T14:23:26Z

+	resolveFn, err := b.checkpoint.Track(ctx, lsn, 1)
+	b.batcherMu.Unlock()
+	if err != nil {
+		return fmt.Errorf("tracking window checkpoint: %w", err)
+	}
+	// Resolve the marker immediately: if everything before it is already
+	// acked this persists lsn now; otherwise the last outstanding ack's
+	// resolve will surface it.
+	if resolved := resolveFn(); resolved != nil && len(*resolved) != 0 {
+		return b.cacheLSN(ctx, *resolved)
+	}


Checkpoint can regress: CheckpointWindow and ackFn write the cache from different goroutines with no ordering.

CheckpointWindow runs on the stream goroutine (ReadChangeTables), while the ack path runs on the pipeline's ack goroutine. Both call resolveFn() and then b.cacheLSN(...) outside any shared lock, so the ordered tracker's guarantee (resolve returns monotonically-increasing payloads) does not carry through to the cache write.

Failure scenario — batch B is tracked with payload L0 and is still in flight; the polling window drains with buffered == 0, so a marker M with payload L1 > L0 is registered:

Ack goroutine: B.resolveFn() → returns L0 (B is head, M not yet resolved).

Stream goroutine: M.resolveFn() → returns L1 (M is now head).

Stream goroutine wins the scheduling race and runs cacheLSN(L1) first.

Ack goroutine then runs cacheLSN(L0), overwriting the newer position.

The persisted checkpoint is now L0, so a restart replays everything in (L0, L1] — the exact non-regression invariant TestTrackOrderUnderConcurrentFlush asserts (that test only acks from a single goroutine, so it can't catch this). No data loss, but it defeats the "exact end position of each drained window" goal this path was added for.

Suggested fix: funnel all checkpoint persistence through one serialization point — e.g. hold a dedicated mutex across resolveFn() + cacheLSN(...) in both CheckpointWindow and the batch ackFn, or track the last persisted LSN and skip writes that would move it backwards.

Refs: batcher.go#L340-L359, ack path batcher.go#L281-L291. Relevant to CONTRIBUTING.md §5.4.1 / §5.4.2 (durable checkpoint + resume, correct progress gating).

Fixed in 16e7f8f. A persistMu now makes each resolveFn()+cacheLSN pair a single critical section in both the batch ackFn and CheckpointWindow, so cache writes land in tracker order. New test (TestPersistOrderUnderConcurrentAcksAndWindows) drives concurrent per-batch acks against interleaved window markers and asserts the persisted sequence never regresses — proven red against the unfixed code (regression reproduced under -race within one run). Full 10-test integration suite re-verified green.

…gression CheckpointWindow (stream goroutine) and batch ack functions (pipeline goroutines) each ran resolve+cacheLSN with no shared ordering, so two persists could land out of order and overwrite a newer resume position with an older one - bounded replay after restart, not loss. A persistMu critical section around each resolve+persist pair keeps the cache writes in tracker order. New concurrency test proven red against the unfixed code (regression reproduced under -race within one run).

claude · 2026-08-14T17:44:52Z

+		return nil
+	case <-ctx.Done():
+		if tracked.isSnapshot {
+			b.snapshotAckWG.Done()


The failed-send rollback is only partial: it releases the snapshot-gate slot but never resolves the checkpoint slot that trackBatchLocked just registered. When this branch is taken the trackedBatch (and with it resolveFn) is discarded, so that tracker entry stays pending forever.

Because checkpoint.Capped only advances contiguously (see the checkpoint_limit docs: "Any given Log Sequence Number (LSN) will not be acknowledged unless all messages under that offset are delivered"), and because both the publisher and the tracker are built once in the constructor (input_mssqlserver_cdc.go#L238-L273) and therefore outlive Connect retries, one orphaned slot permanently pins the resume position for the rest of the process.

Reachable path: Snapshot.Read uses errgroup.WithContext, so one worker's error cancels the shared ctx while sibling workers are blocked in Publish → sendTracked (snapshot.go#L203-L215) — likely whenever the pipeline is slower than the snapshot readers and max_parallel_snapshot_tables > 1. processSnapshot then fails, the input reconnects, the snapshot re-runs (still no cached LSN), and every batch of the new attempt tracks behind the orphan: no LSN is ever persisted again, and the orphan's len(batch) slots stay counted against checkpoint_limit.

Suggested fix: resolve (or otherwise discard) the tracker slot on the failed-send path alongside the snapshotAckWG.Done(), so the rollback is symmetric — the same way the gate slot is released. Relevant to CONTRIBUTING.md §5.4.1 (durable checkpoint + resume) and §3.2.2 (hard-to-diagnose failure modes), since the resulting stall is silent.

The orphaned-slot wedge is real, but resolving the slot on the failed-send path is unsafe — it opens the data-loss window this PR closes.

The failed send only proves this batch never reached the pipeline; it does not stop the other flusher. Publish sends under the caller's context (the snapshot errgroup's) while the timed-flush loop sends under the publisher's own shutdown context, so this interleaving is reachable: batch S is tracked, a timed flush tracks B (after S) and sends it successfully, then S's send fails on the cancelled errgroup ctx. B is downstream with rows after S's rows. If the rollback resolved S's slot, B's ack would persist an LSN past S's undelivered rows and a restart would skip them — silent loss, exactly what §5.4.2 forbids. Keeping the slot pinned is the conservative half of the current behavior; the bug is only that the pin outlives the session because the publisher and tracker are constructor-lifetime.

Proposed fix: treat a failed send as poisoning the publisher. sendTracked's failure path sets a poisoned flag alongside the existing gate release; Connect checks it and rebuilds the publisher (batcher + fresh tracker) before starting a session. The orphan then pins nothing beyond its own dead session, and the new session re-reads from the last durable LSN, which by construction is before the orphaned rows. Late acks from the old session resolve into the abandoned tracker; the only residual risk is a stale cacheLSN write racing the new session's, which the existing persistMu plus a last-persisted-LSN monotonic guard turns into bounded replay rather than regression.

This is deliberately narrower than the per-Connect rebuild that was unwound: it triggers only on the crash path (failed send), involves no nack semantics, and keeps the drop-contract behavior untouched. Happy to implement it that way if there are no objections.

squiidz added 5 commits August 6, 2026 15:02

mssqlserver_cdc: track in-flight snapshot batch acks in the publisher

9908f83

mssqlserver_cdc: gate post-snapshot checkpoint on downstream acks

0c9ee6f

mssqlserver_cdc: adversarial crash tests for snapshot barrier and spl…

c17934a

…it transactions

josephwoodward added the run-integration-tests label Aug 7, 2026

josephwoodward closed this Aug 7, 2026

josephwoodward reopened this Aug 7, 2026

claude Bot reviewed Aug 7, 2026

View reviewed changes

claude Bot reviewed Aug 10, 2026

View reviewed changes

squiidz mentioned this pull request Aug 10, 2026

oracledb_cdc: gate post-snapshot checkpoint on downstream acks (CON-504) #4675

Open

claude Bot reviewed Aug 10, 2026

View reviewed changes

claude Bot reviewed Aug 11, 2026

View reviewed changes

claude Bot reviewed Aug 14, 2026

View reviewed changes

Uh oh!

Conversation

squiidz commented Aug 6, 2026

Proof of Work

Uh oh!

claude Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 7, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 10, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

squiidz commented Aug 11, 2026

Uh oh!

claude Bot Aug 11, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

claude Bot Aug 14, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants