From 9908f832758a16acf6f7106ac8b3680f7849ad4d Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Thu, 6 Aug 2026 15:02:55 -0400
Subject: [PATCH 01/12] mssqlserver_cdc: track in-flight snapshot batch acks in
 the publisher

---
 internal/impl/mssqlserver/batcher.go      |  58 +++++++
 internal/impl/mssqlserver/batcher_test.go | 182 ++++++++++++++++++++++
 2 files changed, 240 insertions(+)
 create mode 100644 internal/impl/mssqlserver/batcher_test.go

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 3907c58a01..15716ee75d 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -42,6 +42,11 @@ type batchPublisher struct {
 	log        *service.Logger
 	cacheLSN   func(ctx context.Context, lsn replication.LSN) error
 	shutSig    *shutdown.Signaller
+
+	// snapshotAckWG counts published snapshot batches that have not yet been
+	// acknowledged downstream. The snapshot->streaming handoff blocks on it so
+	// the post-snapshot LSN is never persisted while snapshot rows are in flight.
+	snapshotAckWG sync.WaitGroup
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -208,6 +213,13 @@ func (b *batchPublisher) publishBatch(ctx context.Context, batch service.Message
 		checkpointLSN = replication.LSN(lsn)
 	}
 
+	// Snapshot batches are tracked so the snapshot->streaming handoff can block
+	// until they are acknowledged downstream (see waitSnapshotAcks).
+	isSnapshotBatch := false
+	if op, ok := lastMsg.MetaGet("operation"); ok && op == replication.MessageOperationRead.String() {
+		isSnapshotBatch = true
+	}
+
 	resolveFn, err := b.checkpoint.Track(ctx, checkpointLSN, int64(len(batch)))
 	if err != nil {
 		return fmt.Errorf("tracking LSN checkpoint for batch: %w", err)
@@ -215,6 +227,9 @@ func (b *batchPublisher) publishBatch(ctx context.Context, batch service.Message
 	msg := asyncMessage{
 		msg: batch,
 		ackFn: func(ctx context.Context, _ error) error {
+			if isSnapshotBatch {
+				defer b.snapshotAckWG.Done()
+			}
 			lsn := resolveFn()
 			if lsn != nil && len(*lsn) != 0 {
 				return b.cacheLSN(ctx, *lsn)
@@ -222,14 +237,57 @@ func (b *batchPublisher) publishBatch(ctx context.Context, batch service.Message
 			return nil
 		},
 	}
+	if isSnapshotBatch {
+		b.snapshotAckWG.Add(1)
+	}
 	select {
 	case b.msgChan <- msg:
 		return nil
 	case <-ctx.Done():
+		if isSnapshotBatch {
+			b.snapshotAckWG.Done()
+		}
 		return ctx.Err()
 	}
 }
 
+// waitSnapshotAcks blocks until every published snapshot batch has been
+// acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
+// 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 {
+	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 <-drained:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// flushCurrent flushes any partial batch still held by the batcher and
+// publishes it, leaving the publisher loop running. Used at the
+// snapshot->streaming handoff so every snapshot row is published (and can be
+// awaited via waitSnapshotAcks) before the post-snapshot LSN is persisted.
+func (b *batchPublisher) flushCurrent(ctx context.Context) error {
+	if b.batcher == nil {
+		return nil
+	}
+	b.batcherMu.Lock()
+	remaining, err := b.batcher.Flush(ctx)
+	b.batcherMu.Unlock()
+	if err != nil || len(remaining) == 0 {
+		return err
+	}
+	return b.publishBatch(ctx, remaining)
+}
+
 func (b *batchPublisher) msgs() <-chan asyncMessage {
 	return b.msgChan
 }
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
new file mode 100644
index 0000000000..a15ed851ce
--- /dev/null
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -0,0 +1,182 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package mssqlserver
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Jeffail/checkpoint"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+	"github.com/redpanda-data/connect/v4/internal/impl/mssqlserver/replication"
+)
+
+func TestSnapshotAckGate(t *testing.T) {
+	t.Run("blocks until the snapshot batch is acked", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		msg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
+
+		done := make(chan error, 1)
+		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
+
+		select {
+		case err := <-done:
+			t.Fatalf("waitSnapshotAcks returned before the snapshot batch was acked: %v", err)
+		case <-time.After(100 * time.Millisecond):
+		}
+
+		require.NoError(t, msg.ackFn(ctx, nil))
+		select {
+		case err := <-done:
+			require.NoError(t, err)
+		case <-time.After(5 * time.Second):
+			t.Fatal("waitSnapshotAcks did not return after the snapshot batch was acked")
+		}
+	})
+
+	t.Run("a nack also releases the gate", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		msg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
+		require.NoError(t, msg.ackFn(ctx, errors.New("downstream failure")))
+
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+	})
+
+	t.Run("streaming batches do not hold the gate", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		// Published but never acked: must not block the gate.
+		publishAndReceive(t, ctx, publisher, service.MessageBatch{newStreamingMessage("00000030")})
+
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+	})
+
+	t.Run("context cancellation escapes the gate", func(t *testing.T) {
+		publisher, _ := newTestBatchPublisher(t)
+
+		ctx, cancel := context.WithCancel(t.Context())
+		publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
+
+		done := make(chan error, 1)
+		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
+		cancel()
+
+		select {
+		case err := <-done:
+			require.ErrorIs(t, err, context.Canceled)
+		case <-time.After(5 * time.Second):
+			t.Fatal("waitSnapshotAcks did not return after context cancellation")
+		}
+	})
+}
+
+func TestFlushCurrent(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	batcher, err := (service.BatchPolicy{Count: 100}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+
+	publishEvent := func() {
+		t.Helper()
+		require.NoError(t, publisher.Publish(ctx, replication.MessageEvent{
+			Schema:    "dbo",
+			Table:     "t",
+			Operation: replication.MessageOperationRead.String(),
+			Data:      map[string]any{"a": 1},
+		}))
+	}
+	receive := func(failMsg string) {
+		t.Helper()
+		got := make(chan asyncMessage, 1)
+		go func() { got <- <-publisher.msgs() }()
+		require.NoError(t, publisher.flushCurrent(ctx))
+		select {
+		case m := <-got:
+			require.Len(t, m.msg, 1)
+		case <-time.After(5 * time.Second):
+			t.Fatal(failMsg)
+		}
+	}
+
+	// Count=100 keeps a single event buffered in the batcher until flushed.
+	publishEvent()
+	receive("flushCurrent did not publish the buffered partial batch")
+
+	// The loop must still be alive after flushCurrent: a second
+	// publish+flush must work identically.
+	publishEvent()
+	receive("publisher loop no longer functional after flushCurrent")
+}
+
+func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.LSN) {
+	t.Helper()
+
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	publisher := newBatchPublisher(nil, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu         sync.Mutex
+		cachedLSNs []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		cachedLSNs = append(cachedLSNs, lsn)
+		return nil
+	}
+
+	cachedLSNsFn := func() []replication.LSN {
+		mu.Lock()
+		defer mu.Unlock()
+		return append([]replication.LSN(nil), cachedLSNs...)
+	}
+
+	return publisher, cachedLSNsFn
+}
+
+func newSnapshotMessage() *service.Message {
+	msg := service.NewMessage([]byte("{}"))
+	msg.MetaSet("operation", replication.MessageOperationRead.String())
+	return msg
+}
+
+func newStreamingMessage(lsn string) *service.Message {
+	msg := service.NewMessage([]byte("{}"))
+	msg.MetaSet("operation", replication.MessageOperationInsert.String())
+	msg.MetaSet("lsn", lsn)
+	return msg
+}
+
+func publishAndReceive(t *testing.T, ctx context.Context, publisher *batchPublisher, batch service.MessageBatch) asyncMessage {
+	t.Helper()
+	go func() {
+		_ = publisher.publishBatch(ctx, batch)
+	}()
+	return <-publisher.msgs()
+}

From c5fdc32d90350ea21e099157b2bf7af72592d0df Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Thu, 6 Aug 2026 15:06:14 -0400
Subject: [PATCH 02/12] mssqlserver_cdc: make batch tracking atomic with batch
 flushing

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).
---
 internal/impl/mssqlserver/batcher.go      | 119 ++++++++++++++++------
 internal/impl/mssqlserver/batcher_test.go |  75 ++++++++++++++
 2 files changed, 163 insertions(+), 31 deletions(-)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 15716ee75d..573b0ffcac 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -85,7 +85,11 @@ func (p *batchPublisher) loop() {
 			return
 		}
 
+		// UntilNext reads the batcher's internal state, which concurrent
+		// Publish calls mutate under batcherMu — take the same lock.
+		p.batcherMu.Lock()
 		tNext, exists := p.batcher.UntilNext()
+		p.batcherMu.Unlock()
 		if !exists {
 			if flushBatchTicker != nil {
 				flushBatchTicker.Stop()
@@ -109,9 +113,14 @@ func (p *batchPublisher) loop() {
 		adjustTimedFlush()
 		select {
 		case <-flushBatch:
-			var sendBatch service.MessageBatch
-
-			// Wrap this in a closure to make locking/unlocking easier.
+			var (
+				tracked  *trackedBatch
+				trackErr error
+			)
+
+			// Wrap this in a closure to make locking/unlocking easier. Track
+			// happens under the same lock as the flush so the checkpoint
+			// sequence matches flush order.
 			func() {
 				p.batcherMu.Lock()
 				defer p.batcherMu.Unlock()
@@ -124,13 +133,18 @@ func (p *batchPublisher) loop() {
 					return
 				}
 
+				var sendBatch service.MessageBatch
 				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
 					return
 				}
+				tracked, trackErr = p.trackBatchLocked(closeAtLeisureCtx, sendBatch)
 			}()
+			if trackErr != nil {
+				return
+			}
 
-			if len(sendBatch) > 0 {
-				if err := p.publishBatch(closeAtLeisureCtx, sendBatch); err != nil {
+			if tracked != nil {
+				if err := p.sendTracked(closeAtLeisureCtx, tracked); err != nil {
 					return
 				}
 			}
@@ -181,10 +195,17 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: s})
 	}
 
-	var flushedBatch []*service.Message
+	// Flush and Track must be atomic: Track order defines the checkpoint
+	// sequence, so another flusher (the timed-flush loop) must not interleave
+	// between our flush and our Track. Only the channel send happens outside
+	// the lock.
+	var tracked *trackedBatch
 	b.batcherMu.Lock()
 	if b.batcher.Add(msg) {
-		flushedBatch, err = b.batcher.Flush(ctx)
+		var flushedBatch []*service.Message
+		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			tracked, err = b.trackBatchLocked(ctx, flushedBatch)
+		}
 	}
 	b.batcherMu.Unlock()
 	if err != nil {
@@ -192,8 +213,8 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	}
 
 	// If a batch was flushed, publish it outside the lock
-	if len(flushedBatch) > 0 {
-		if err := b.publishBatch(ctx, flushedBatch); err != nil {
+	if tracked != nil {
+		if err := b.sendTracked(ctx, tracked); err != nil {
 			return fmt.Errorf("publishing flushed batch: %w", err)
 		}
 	}
@@ -201,11 +222,17 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	return nil
 }
 
-func (b *batchPublisher) publishBatch(ctx context.Context, batch service.MessageBatch) error {
-	if len(batch) == 0 {
-		return nil
-	}
+// trackedBatch pairs a ready-to-send asyncMessage with the bookkeeping needed
+// to roll back its snapshot-gate slot if the send fails.
+type trackedBatch struct {
+	msg        asyncMessage
+	isSnapshot bool
+}
 
+// trackBatchLocked registers the batch with the ordered checkpoint tracker and
+// builds its ack function. It MUST be called with batcherMu held: Track order
+// defines the checkpoint sequence, so it has to match flush order exactly.
+func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
 	var checkpointLSN []byte
 	// snapshot records don't have a lsn as we don't track those
@@ -222,35 +249,61 @@ func (b *batchPublisher) publishBatch(ctx context.Context, batch service.Message
 
 	resolveFn, err := b.checkpoint.Track(ctx, checkpointLSN, int64(len(batch)))
 	if err != nil {
-		return fmt.Errorf("tracking LSN checkpoint for batch: %w", err)
-	}
-	msg := asyncMessage{
-		msg: batch,
-		ackFn: func(ctx context.Context, _ error) error {
-			if isSnapshotBatch {
-				defer b.snapshotAckWG.Done()
-			}
-			lsn := resolveFn()
-			if lsn != nil && len(*lsn) != 0 {
-				return b.cacheLSN(ctx, *lsn)
-			}
-			return nil
-		},
+		return nil, fmt.Errorf("tracking LSN checkpoint for batch: %w", err)
 	}
 	if isSnapshotBatch {
 		b.snapshotAckWG.Add(1)
 	}
+	return &trackedBatch{
+		isSnapshot: isSnapshotBatch,
+		msg: asyncMessage{
+			msg: batch,
+			ackFn: func(ctx context.Context, _ error) error {
+				if isSnapshotBatch {
+					defer b.snapshotAckWG.Done()
+				}
+				lsn := resolveFn()
+				if lsn != nil && len(*lsn) != 0 {
+					return b.cacheLSN(ctx, *lsn)
+				}
+				return nil
+			},
+		},
+	}, nil
+}
+
+// sendTracked hands a tracked batch to ReadBatch. Must be called WITHOUT
+// batcherMu held (the send blocks until consumed). A failed send releases the
+// batch's snapshot-gate slot.
+func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch) error {
 	select {
-	case b.msgChan <- msg:
+	case b.msgChan <- tracked.msg:
 		return nil
 	case <-ctx.Done():
-		if isSnapshotBatch {
+		if tracked.isSnapshot {
 			b.snapshotAckWG.Done()
 		}
 		return ctx.Err()
 	}
 }
 
+// publishBatch tracks and sends a batch that was flushed elsewhere. Callers
+// 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 {
+	if len(batch) == 0 {
+		return nil
+	}
+	b.batcherMu.Lock()
+	tracked, err := b.trackBatchLocked(ctx, batch)
+	b.batcherMu.Unlock()
+	if err != nil {
+		return err
+	}
+	return b.sendTracked(ctx, tracked)
+}
+
 // waitSnapshotAcks blocks until every published snapshot batch has been
 // acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
 // batches release the gate too: redelivery is owned by auto_replay_nacks,
@@ -279,13 +332,17 @@ func (b *batchPublisher) flushCurrent(ctx context.Context) error {
 	if b.batcher == nil {
 		return nil
 	}
+	var tracked *trackedBatch
 	b.batcherMu.Lock()
 	remaining, err := b.batcher.Flush(ctx)
+	if err == nil && len(remaining) > 0 {
+		tracked, err = b.trackBatchLocked(ctx, remaining)
+	}
 	b.batcherMu.Unlock()
-	if err != nil || len(remaining) == 0 {
+	if err != nil || tracked == nil {
 		return err
 	}
-	return b.publishBatch(ctx, remaining)
+	return b.sendTracked(ctx, tracked)
 }
 
 func (b *batchPublisher) msgs() <-chan asyncMessage {
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index a15ed851ce..2fdc611b26 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -11,6 +11,7 @@ package mssqlserver
 import (
 	"context"
 	"errors"
+	"fmt"
 	"log/slog"
 	"sync"
 	"testing"
@@ -131,6 +132,80 @@ func TestFlushCurrent(t *testing.T) {
 	receive("publisher loop no longer functional after flushCurrent")
 }
 
+// TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
+// count-triggered flush in Publish and the timed-flush loop) and asserts the
+// persisted checkpoint never regresses when batches are acked in delivery
+// order. Before Track was moved under the batcher mutex, the two flushers
+// could interleave between flush and Track, registering batches with the
+// ordered tracker in the wrong order and persisting a regressing LSN. Run with
+// -race to also catch the underlying data race structurally.
+func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](1000)
+
+	// Count 2 + a tiny period keeps both flush paths active concurrently.
+	batcher, err := (service.BatchPolicy{Count: 2, Period: "1ms"}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu        sync.Mutex
+		persisted []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		persisted = append(persisted, lsn)
+		return nil
+	}
+
+	// Consumer: ack every batch immediately, in delivery order.
+	consumerDone := make(chan struct{})
+	consumerCtx, stopConsumer := context.WithCancel(ctx)
+	go func() {
+		defer close(consumerDone)
+		for {
+			select {
+			case m := <-publisher.msgs():
+				_ = m.ackFn(ctx, nil)
+			case <-consumerCtx.Done():
+				return
+			}
+		}
+	}()
+
+	const events = 500
+	for i := range events {
+		require.NoError(t, publisher.Publish(ctx, replication.MessageEvent{
+			Schema:    "dbo",
+			Table:     "t",
+			Operation: replication.MessageOperationInsert.String(),
+			// %08d keeps lexicographic order == numeric order, like real LSNs.
+			LSN:  replication.LSN(fmt.Sprintf("%08d", i)),
+			Data: map[string]any{"i": i},
+		}))
+	}
+	require.NoError(t, publisher.flushCurrent(ctx))
+
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(persisted) > 0 && string(persisted[len(persisted)-1]) == fmt.Sprintf("%08d", events-1)
+	}, 10*time.Second, 10*time.Millisecond, "final LSN was never persisted")
+	stopConsumer()
+	<-consumerDone
+
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 1; i < len(persisted); i++ {
+		require.GreaterOrEqual(t, string(persisted[i]), string(persisted[i-1]),
+			"persisted checkpoint regressed at index %d: %v", i, persisted)
+	}
+}
+
 func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.LSN) {
 	t.Helper()
 

From 2539c00170385c5b5226b70d03a60daa38a5af36 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Thu, 6 Aug 2026 15:08:21 -0400
Subject: [PATCH 03/12] mssqlserver_cdc: checkpoint only fully-published
 transaction boundaries

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.
---
 internal/impl/mssqlserver/batcher.go          | 11 ++++-
 internal/impl/mssqlserver/batcher_test.go     | 45 ++++++++++++++++---
 .../impl/mssqlserver/replication/stream.go    | 39 +++++++++++++---
 .../mssqlserver/replication/stream_message.go | 10 ++++-
 .../mssqlserver/replication/stream_test.go    | 41 +++++++++++++++++
 5 files changed, 129 insertions(+), 17 deletions(-)
 create mode 100644 internal/impl/mssqlserver/replication/stream_test.go

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 573b0ffcac..13b12369de 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -191,6 +191,9 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	if len(m.LSN) != 0 {
 		msg.MetaSet("lsn", string(m.LSN))
 	}
+	if len(m.CheckpointLSN) != 0 {
+		msg.MetaSet("checkpoint_lsn", string(m.CheckpointLSN))
+	}
 	if s := b.getOrComputeTableSchema(m.Table, m.ColumnNames, m.ColumnTypes); s != nil {
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: s})
 	}
@@ -235,8 +238,12 @@ type trackedBatch struct {
 func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
 	var checkpointLSN []byte
-	// snapshot records don't have a lsn as we don't track those
-	if lsn, ok := lastMsg.MetaGet("lsn"); ok {
+	// Checkpoint only checkpoint_lsn: the last transaction whose rows are all
+	// published. The row's own lsn must never be persisted — all rows of a
+	// transaction share a start LSN and resume is exclusive (> lsn), so
+	// persisting it mid-transaction would skip the transaction's remaining
+	// rows on restart. Snapshot rows carry neither meta; we don't track those.
+	if lsn, ok := lastMsg.MetaGet("checkpoint_lsn"); ok {
 		checkpointLSN = replication.LSN(lsn)
 	}
 
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 2fdc611b26..7e8548a8ec 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -132,6 +132,37 @@ func TestFlushCurrent(t *testing.T) {
 	receive("publisher loop no longer functional after flushCurrent")
 }
 
+func TestCheckpointSelection(t *testing.T) {
+	t.Run("persists checkpoint_lsn, never the row's own lsn", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		msg := service.NewMessage([]byte("{}"))
+		msg.MetaSet("operation", replication.MessageOperationInsert.String())
+		msg.MetaSet("lsn", "00000042")
+		msg.MetaSet("checkpoint_lsn", "00000041")
+
+		am := publishAndReceive(t, ctx, publisher, service.MessageBatch{msg})
+		require.NoError(t, am.ackFn(ctx, nil))
+
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000041", string(lsns[0]),
+			"the checkpoint must be the last fully-published transaction boundary, not the row's own LSN")
+	})
+
+	t.Run("no checkpoint_lsn (first transaction) persists nothing", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, service.MessageBatch{newStreamingMessage("00000042")})
+		require.NoError(t, am.ackFn(ctx, nil))
+
+		require.Empty(t, cachedLSNs(),
+			"a batch ending mid-transaction (no prior complete transaction) must not persist any LSN")
+	})
+}
+
 // TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
 // count-triggered flush in Publish and the timed-flush loop) and asserts the
 // persisted checkpoint never regresses when batches are acked in delivery
@@ -179,13 +210,15 @@ func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
 
 	const events = 500
 	for i := range events {
+		// %08d keeps lexicographic order == numeric order, like real LSNs.
+		lsn := replication.LSN(fmt.Sprintf("%08d", i))
 		require.NoError(t, publisher.Publish(ctx, replication.MessageEvent{
-			Schema:    "dbo",
-			Table:     "t",
-			Operation: replication.MessageOperationInsert.String(),
-			// %08d keeps lexicographic order == numeric order, like real LSNs.
-			LSN:  replication.LSN(fmt.Sprintf("%08d", i)),
-			Data: map[string]any{"i": i},
+			Schema:        "dbo",
+			Table:         "t",
+			Operation:     replication.MessageOperationInsert.String(),
+			LSN:           lsn,
+			CheckpointLSN: lsn,
+			Data:          map[string]any{"i": i},
 		}))
 	}
 	require.NoError(t, publisher.flushCurrent(ctx))
diff --git a/internal/impl/mssqlserver/replication/stream.go b/internal/impl/mssqlserver/replication/stream.go
index 39145380e3..df7de5e1d2 100644
--- a/internal/impl/mssqlserver/replication/stream.go
+++ b/internal/impl/mssqlserver/replication/stream.go
@@ -332,6 +332,27 @@ func mapScannedValue(val any, colType *sql.ColumnType) any {
 	return val
 }
 
+// txnBoundary tracks transaction boundaries in the globally LSN-ordered row
+// stream. All rows of one transaction share a __$start_lsn, so an LSN change
+// between consecutive rows proves the previous transaction is fully read (and,
+// because rows are published synchronously in read order, fully published).
+type txnBoundary struct {
+	prev         LSN
+	lastComplete LSN
+}
+
+// Observe records the current row's LSN and returns the start LSN of the most
+// recent transaction whose rows have all been observed — empty until the first
+// boundary is crossed.
+func (t *txnBoundary) Observe(lsn LSN) LSN {
+	if len(t.prev) != 0 && !bytes.Equal(lsn, t.prev) {
+		t.lastComplete = t.prev
+	}
+	// Copy: the iterator may reuse the underlying array on the next scan.
+	t.prev = append(t.prev[:0:0], lsn...)
+	return t.lastComplete
+}
+
 // ChangePublisher is responsible for handling and processing of a replication.MessageEvent.
 type ChangePublisher interface {
 	Publish(ctx context.Context, msg MessageEvent) error
@@ -365,6 +386,9 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 		startLSN LSN // load last checkpoint; nil means start from beginning in tables
 		endLSN   LSN // often set to fn_cdc_get_max_lsn(); nil means no upper bound
 		lastLSN  LSN
+		// boundary computes each row's CheckpointLSN: the last transaction
+		// whose rows are all published, the only safe resume position.
+		boundary txnBoundary
 	)
 
 	if len(startPos) != 0 {
@@ -415,13 +439,14 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 			cur := item.iter.current
 
 			msg := MessageEvent{
-				Table:       item.iter.table.Name,
-				Schema:      item.iter.table.Schema,
-				Data:        cur.columns,
-				LSN:         cur.startLSN,
-				Operation:   cur.operation.String(),
-				ColumnNames: item.iter.userColNames,
-				ColumnTypes: item.iter.userColTypes,
+				Table:         item.iter.table.Name,
+				Schema:        item.iter.table.Schema,
+				Data:          cur.columns,
+				LSN:           cur.startLSN,
+				CheckpointLSN: boundary.Observe(cur.startLSN),
+				Operation:     cur.operation.String(),
+				ColumnNames:   item.iter.userColNames,
+				ColumnTypes:   item.iter.userColTypes,
 			}
 
 			if err := r.publisher.Publish(ctx, msg); err != nil {
diff --git a/internal/impl/mssqlserver/replication/stream_message.go b/internal/impl/mssqlserver/replication/stream_message.go
index 27143a3c05..c860d3969b 100644
--- a/internal/impl/mssqlserver/replication/stream_message.go
+++ b/internal/impl/mssqlserver/replication/stream_message.go
@@ -83,8 +83,14 @@ func (op OpType) String() string {
 
 // MessageEvent represents a single change from Table's change table in the database.
 type MessageEvent struct {
-	LSN       LSN    `json:"start_lsn"`
-	Operation string `json:"operation"`
+	LSN LSN `json:"start_lsn"`
+	// CheckpointLSN is the start LSN of the most recent transaction whose rows
+	// have all been published — the only value safe to persist as a resume
+	// position (resume is exclusive and all rows of a transaction share a
+	// start LSN). Empty for snapshot rows and until the first transaction
+	// boundary is observed.
+	CheckpointLSN LSN    `json:"-"`
+	Operation     string `json:"operation"`
 	Schema    string `json:"schema"`
 	Table     string `json:"table"`
 	Data      any    `json:"data"`
diff --git a/internal/impl/mssqlserver/replication/stream_test.go b/internal/impl/mssqlserver/replication/stream_test.go
new file mode 100644
index 0000000000..ce41aae709
--- /dev/null
+++ b/internal/impl/mssqlserver/replication/stream_test.go
@@ -0,0 +1,41 @@
+// Copyright 2026 Redpanda Data, Inc.
+//
+// Licensed as a Redpanda Enterprise file under the Redpanda Community
+// License (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md
+
+package replication
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestTxnBoundaryObserve(t *testing.T) {
+	var b txnBoundary
+
+	// Sequence AAABBC: the last complete transaction only advances when the
+	// LSN changes, and always lags one transaction behind the current row.
+	observations := []struct {
+		lsn          string
+		lastComplete string // "" = no complete transaction yet
+	}{
+		{"A", ""},
+		{"A", ""},
+		{"A", ""},
+		{"B", "A"},
+		{"B", "A"},
+		{"C", "B"},
+	}
+	for i, o := range observations {
+		got := b.Observe(LSN(o.lsn))
+		if o.lastComplete == "" {
+			require.Emptyf(t, got, "observation %d (lsn %s)", i, o.lsn)
+		} else {
+			require.Equalf(t, o.lastComplete, string(got), "observation %d (lsn %s)", i, o.lsn)
+		}
+	}
+}

From 0c9ee6fc45e6388b3f816afa7dee68003ff6e731 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Thu, 6 Aug 2026 15:08:43 -0400
Subject: [PATCH 04/12] mssqlserver_cdc: gate post-snapshot checkpoint on
 downstream acks

---
 .../impl/mssqlserver/input_mssqlserver_cdc.go  | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index 4b837c7f9b..4034ccd745 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -366,6 +366,24 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 				i.stopSig.TriggerHasStopped()
 				return
 			}
+
+			// Flush the partial snapshot batch still held by the batcher, then
+			// block until every snapshot batch is acknowledged downstream.
+			// Persisting the LSN any earlier would let a crash in this window
+			// skip un-acked snapshot rows on restart. Blocks until acks drain
+			// or soft-stop (no timeout, by design; see postgres_cdc's
+			// equivalent barrier).
+			if err = i.publisher.flushCurrent(softCtx); err != nil {
+				i.log.Errorf("Failed to flush remaining snapshot batches. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				i.stopSig.TriggerHasStopped()
+				return
+			}
+			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
+			}
+
 			if err = i.cacheLSN(softCtx, maxLSN); err != nil {
 				if i.stopSig.IsHardStopSignalled() {
 					i.log.Errorf("Shutting down snapshotting process: %s", err)

From c17934af56de21fa9c5ce2a83a36d6ff0ee4b210 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Thu, 6 Aug 2026 15:27:11 -0400
Subject: [PATCH 05/12] mssqlserver_cdc: adversarial crash tests for snapshot
 barrier and split transactions

---
 internal/impl/mssqlserver/integration_test.go | 237 ++++++++++++++++++
 .../mssqlserver/replication/stream_message.go |   6 +-
 2 files changed, 240 insertions(+), 3 deletions(-)

diff --git a/internal/impl/mssqlserver/integration_test.go b/internal/impl/mssqlserver/integration_test.go
index 359193f5f8..7f8c7efb9d 100644
--- a/internal/impl/mssqlserver/integration_test.go
+++ b/internal/impl/mssqlserver/integration_test.go
@@ -11,9 +11,11 @@ package mssqlserver_test
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -429,6 +431,241 @@ microsoft_sql_server_cdc:
 	require.NoError(t, stream.StopWithin(time.Second*10))
 }
 
+// TestIntegration_MicrosoftSQLServerCDC_SnapshotAckBarrier verifies that a
+// crash during the snapshot->streaming handoff (after snapshot rows are
+// emitted but before they are acknowledged) does not lose data: because the
+// post-snapshot LSN is only persisted once every snapshot batch is acked, the
+// snapshot must re-run on restart. See CON-504.
+func TestIntegration_MicrosoftSQLServerCDC_SnapshotAckBarrier(t *testing.T) {
+	integration.CheckSkip(t)
+
+	connStr, db := mssqlservertest.SetupTestWithMicrosoftSQLServerVersion(t)
+	require.NoError(t, db.CreateTableWithCDCEnabledIfNotExists(t.Context(), "dbo.barrier", "CREATE TABLE dbo.barrier (id INT IDENTITY(1,1) PRIMARY KEY);"))
+
+	const rowCount = 5
+	for range rowCount {
+		db.MustExec("INSERT INTO dbo.barrier DEFAULT VALUES")
+	}
+	db.WaitForCDCChanges(t.Context(), rowCount, "dbo.barrier")
+
+	// batching.count == rowCount forces all snapshot rows into a single output
+	// batch, so the run-1 consumer receives them all at once and can then block
+	// without acking - reproducing the "emitted but not yet acked" handoff state.
+	cfg := fmt.Sprintf(`
+microsoft_sql_server_cdc:
+  connection_string: %s
+  stream_snapshot: true
+  checkpoint_cache: ""
+  include: ["dbo.barrier"]
+  batching:
+    count: %d
+    period: 1h`, connStr, rowCount)
+
+	// Run 1: receive the snapshot rows but never acknowledge them, then
+	// simulate a crash by cancelling the run before the LSN can be persisted.
+	t.Log("Launching run 1 (blocked consumer, simulated crash)...")
+	received := make(chan struct{}, 1)
+	run1Builder := service.NewStreamBuilder()
+	require.NoError(t, run1Builder.AddInputYAML(cfg))
+	require.NoError(t, run1Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run1Builder.AddBatchConsumerFunc(func(ctx context.Context, _ service.MessageBatch) error {
+		select {
+		case received <- struct{}{}:
+		default:
+		}
+		// Block without acking until the simulated crash cancels our context.
+		<-ctx.Done()
+		return ctx.Err()
+	}))
+	run1, err := run1Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run1.Resources())
+
+	run1Ctx, crash := context.WithCancel(t.Context())
+	run1Done := make(chan struct{})
+	go func() {
+		defer close(run1Done)
+		_ = run1.Run(run1Ctx)
+	}()
+
+	select {
+	case <-received:
+	case <-time.After(5 * time.Minute):
+		t.Fatal("snapshot rows were never delivered to the run-1 output")
+	}
+	// Give the input time to reach the ack barrier (and, in the buggy version,
+	// to persist the post-snapshot LSN) before we crash.
+	time.Sleep(5 * time.Second)
+	crash()
+	select {
+	case <-run1Done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("run 1 did not stop after the simulated crash")
+	}
+
+	// The barrier must have prevented the post-snapshot LSN from being
+	// persisted, since the snapshot rows were never acknowledged. Without it a
+	// cached LSN would exist here and the snapshot would be skipped on
+	// restart, silently losing the un-acked rows.
+	var checkpoints int
+	require.NoError(t, db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM rpcn.CdcCheckpointCache").Scan(&checkpoints))
+	require.Zero(t, checkpoints, "post-snapshot LSN must not be persisted before snapshot rows are acknowledged")
+
+	// Run 2: restart against the same checkpoint cache. Since run 1 never
+	// acked the snapshot, no LSN was cached, so the snapshot re-runs and every
+	// row is delivered again.
+	t.Log("Launching run 2 (verifying the snapshot re-runs)...")
+	var (
+		readsMu sync.Mutex
+		reads   int
+	)
+	run2Builder := service.NewStreamBuilder()
+	require.NoError(t, run2Builder.AddInputYAML(cfg))
+	require.NoError(t, run2Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run2Builder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error {
+		readsMu.Lock()
+		defer readsMu.Unlock()
+		for _, msg := range mb {
+			if op, _ := msg.MetaGet("operation"); op == "read" {
+				reads++
+			}
+		}
+		return nil
+	}))
+	run2, err := run2Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run2.Resources())
+	go func() {
+		if err := run2.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		readsMu.Lock()
+		defer readsMu.Unlock()
+		assert.Equal(c, rowCount, reads, "snapshot should have re-run and re-delivered every row after the crash")
+	}, 5*time.Minute, 500*time.Millisecond)
+	require.NoError(t, run2.StopWithin(time.Second*30))
+}
+
+// TestIntegration_MicrosoftSQLServerCDC_TransactionSplitAcrossBatches verifies
+// that acking a batch which ends mid-transaction never persists that
+// transaction's own start LSN: all rows of a transaction share a start LSN and
+// resume is exclusive (> lsn), so doing so would skip the transaction's
+// remaining rows after a crash. The checkpoint may only advance to the last
+// fully-published transaction boundary. See CON-504.
+func TestIntegration_MicrosoftSQLServerCDC_TransactionSplitAcrossBatches(t *testing.T) {
+	integration.CheckSkip(t)
+
+	connStr, db := mssqlservertest.SetupTestWithMicrosoftSQLServerVersion(t)
+	require.NoError(t, db.CreateTableWithCDCEnabledIfNotExists(t.Context(), "dbo.splittx", "CREATE TABLE dbo.splittx (id INT IDENTITY(1,1) PRIMARY KEY, val INT NOT NULL);"))
+
+	// T1: a single-row transaction, establishing a prior transaction boundary.
+	// T2: four rows committed in ONE transaction - they all share a start LSN.
+	db.MustExec("INSERT INTO dbo.splittx (val) VALUES (101)")
+	db.MustExec("BEGIN TRAN; INSERT INTO dbo.splittx (val) VALUES (102); INSERT INTO dbo.splittx (val) VALUES (103); INSERT INTO dbo.splittx (val) VALUES (104); INSERT INTO dbo.splittx (val) VALUES (105); COMMIT")
+	db.WaitForCDCChanges(t.Context(), 5, "dbo.splittx")
+
+	// batching.count = 2 splits T2 across batches: [T1r1, T2r1], [T2r2, T2r3], ...
+	cfg := fmt.Sprintf(`
+microsoft_sql_server_cdc:
+  connection_string: %s
+  stream_snapshot: false
+  checkpoint_cache: ""
+  include: ["dbo.splittx"]
+  batching:
+    count: 2
+    period: 1h`, connStr)
+
+	// Run 1: ack ONLY the first batch (which ends on T2's first row), block on
+	// everything after it, then crash once the ack's checkpoint write lands.
+	t.Log("Launching run 1 (ack first batch only, simulated crash)...")
+	var firstBatch atomic.Bool
+	firstBatch.Store(true)
+	run1Builder := service.NewStreamBuilder()
+	require.NoError(t, run1Builder.AddInputYAML(cfg))
+	require.NoError(t, run1Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run1Builder.AddBatchConsumerFunc(func(ctx context.Context, _ service.MessageBatch) error {
+		if firstBatch.CompareAndSwap(true, false) {
+			return nil // ack the first batch
+		}
+		<-ctx.Done()
+		return ctx.Err()
+	}))
+	run1, err := run1Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run1.Resources())
+
+	run1Ctx, crash := context.WithCancel(t.Context())
+	run1Done := make(chan struct{})
+	go func() {
+		defer close(run1Done)
+		_ = run1.Run(run1Ctx)
+	}()
+
+	// Wait for the first batch's ack to persist a checkpoint, then crash.
+	require.Eventually(t, func() bool {
+		var checkpoints int
+		if err := db.QueryRowContext(t.Context(), "SELECT COUNT(*) FROM rpcn.CdcCheckpointCache").Scan(&checkpoints); err != nil {
+			return false
+		}
+		return checkpoints == 1
+	}, 5*time.Minute, 500*time.Millisecond, "the first batch's ack never persisted a checkpoint")
+	crash()
+	select {
+	case <-run1Done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("run 1 did not stop after the simulated crash")
+	}
+
+	// Run 2: restart. The checkpoint must point at the T1/T2 boundary, so all
+	// of T2 is redelivered - especially rows 102-105's tail (103, 104, 105),
+	// which the pre-fix code skipped by persisting T2's own start LSN.
+	t.Log("Launching run 2 (verifying the split transaction replays in full)...")
+	var (
+		seenMu sync.Mutex
+		seen   = map[int]bool{}
+	)
+	run2Builder := service.NewStreamBuilder()
+	require.NoError(t, run2Builder.AddInputYAML(cfg))
+	require.NoError(t, run2Builder.SetLoggerYAML(`level: INFO`))
+	require.NoError(t, run2Builder.AddBatchConsumerFunc(func(_ context.Context, mb service.MessageBatch) error {
+		seenMu.Lock()
+		defer seenMu.Unlock()
+		for _, msg := range mb {
+			var row struct {
+				Val int `json:"val"`
+			}
+			b, err := msg.AsBytes()
+			if err != nil {
+				return err
+			}
+			if err := json.Unmarshal(b, &row); err == nil && row.Val != 0 {
+				seen[row.Val] = true
+			}
+		}
+		return nil
+	}))
+	run2, err := run2Builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(run2.Resources())
+	go func() {
+		if err := run2.Run(t.Context()); err != nil && !errors.Is(err, context.Canceled) {
+			t.Error(err)
+		}
+	}()
+
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		seenMu.Lock()
+		defer seenMu.Unlock()
+		for _, val := range []int{102, 103, 104, 105} {
+			assert.Truef(c, seen[val], "row val=%d from the split transaction was never redelivered (checkpoint advanced past a partially-delivered transaction)", val)
+		}
+	}, 5*time.Minute, 500*time.Millisecond)
+	require.NoError(t, run2.StopWithin(time.Second*30))
+}
+
 func TestIntegration_MicrosoftSQLServerCDC_ResumesFromCheckpoint(t *testing.T) {
 	integration.CheckSkip(t)
 
diff --git a/internal/impl/mssqlserver/replication/stream_message.go b/internal/impl/mssqlserver/replication/stream_message.go
index c860d3969b..ffab898aed 100644
--- a/internal/impl/mssqlserver/replication/stream_message.go
+++ b/internal/impl/mssqlserver/replication/stream_message.go
@@ -91,9 +91,9 @@ type MessageEvent struct {
 	// boundary is observed.
 	CheckpointLSN LSN    `json:"-"`
 	Operation     string `json:"operation"`
-	Schema    string `json:"schema"`
-	Table     string `json:"table"`
-	Data      any    `json:"data"`
+	Schema        string `json:"schema"`
+	Table         string `json:"table"`
+	Data          any    `json:"data"`
 
 	// ColumnNames and ColumnTypes carry user-defined column metadata (excluding
 	// MSSQL system columns with __$ prefix). They are used to build schema

From 53aef1ffc5ca99ea65ee8e28d3ae696ad5a2ac65 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Fri, 7 Aug 2026 10:03:56 -0400
Subject: [PATCH 06/12] mssqlserver_cdc: address review - fail the snapshot
 gate on nack, drop 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.
---
 internal/impl/mssqlserver/batcher.go      |  82 ++++++++------
 internal/impl/mssqlserver/batcher_test.go | 127 +++++++++++++---------
 2 files changed, 121 insertions(+), 88 deletions(-)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 13b12369de..69844fe89b 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -47,6 +47,26 @@ type batchPublisher struct {
 	// acknowledged downstream. The snapshot->streaming handoff blocks on it so
 	// the post-snapshot LSN is never persisted while snapshot rows are in flight.
 	snapshotAckWG sync.WaitGroup
+	// snapshotNackErr records the first snapshot batch nack. auto_replay_nacks
+	// is user-toggleable, so a nack can be terminal: the gate must fail rather
+	// than let the post-snapshot LSN persist over undelivered rows.
+	snapshotNackMu  sync.Mutex
+	snapshotNackErr error
+
+	// pendingCheckpointLSN mirrors the CheckpointLSN of the most recently
+	// added message: the start LSN of the last transaction whose rows are all
+	// published, the only value safe to persist as a resume position. Guarded
+	// by batcherMu, so at flush time it always belongs to the flushed batch's
+	// last message.
+	pendingCheckpointLSN replication.LSN
+}
+
+func (b *batchPublisher) recordSnapshotNack(err error) {
+	b.snapshotNackMu.Lock()
+	defer b.snapshotNackMu.Unlock()
+	if b.snapshotNackErr == nil {
+		b.snapshotNackErr = err
+	}
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -191,9 +211,6 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	if len(m.LSN) != 0 {
 		msg.MetaSet("lsn", string(m.LSN))
 	}
-	if len(m.CheckpointLSN) != 0 {
-		msg.MetaSet("checkpoint_lsn", string(m.CheckpointLSN))
-	}
 	if s := b.getOrComputeTableSchema(m.Table, m.ColumnNames, m.ColumnTypes); s != nil {
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: s})
 	}
@@ -204,6 +221,7 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	// the lock.
 	var tracked *trackedBatch
 	b.batcherMu.Lock()
+	b.pendingCheckpointLSN = m.CheckpointLSN
 	if b.batcher.Add(msg) {
 		var flushedBatch []*service.Message
 		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
@@ -237,15 +255,12 @@ type trackedBatch struct {
 // defines the checkpoint sequence, so it has to match flush order exactly.
 func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
-	var checkpointLSN []byte
-	// Checkpoint only checkpoint_lsn: the last transaction whose rows are all
-	// published. The row's own lsn must never be persisted — all rows of a
-	// transaction share a start LSN and resume is exclusive (> lsn), so
-	// persisting it mid-transaction would skip the transaction's remaining
-	// rows on restart. Snapshot rows carry neither meta; we don't track those.
-	if lsn, ok := lastMsg.MetaGet("checkpoint_lsn"); ok {
-		checkpointLSN = replication.LSN(lsn)
-	}
+	// Checkpoint only the pending checkpoint LSN: the last transaction whose
+	// rows are all published. The row's own lsn must never be persisted — all
+	// rows of a transaction share a start LSN and resume is exclusive (> lsn),
+	// so persisting it mid-transaction would skip the transaction's remaining
+	// rows on restart. Snapshot rows never carry one; we don't track those.
+	checkpointLSN := []byte(b.pendingCheckpointLSN)
 
 	// Snapshot batches are tracked so the snapshot->streaming handoff can block
 	// until they are acknowledged downstream (see waitSnapshotAcks).
@@ -265,10 +280,21 @@ func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.Mes
 		isSnapshot: isSnapshotBatch,
 		msg: asyncMessage{
 			msg: batch,
-			ackFn: func(ctx context.Context, _ error) error {
+			ackFn: func(ctx context.Context, err error) error {
 				if isSnapshotBatch {
 					defer b.snapshotAckWG.Done()
 				}
+				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
+				}
 				lsn := resolveFn()
 				if lsn != nil && len(*lsn) != 0 {
 					return b.cacheLSN(ctx, *lsn)
@@ -294,28 +320,11 @@ func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch)
 	}
 }
 
-// publishBatch tracks and sends a batch that was flushed elsewhere. Callers
-// 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 {
-	if len(batch) == 0 {
-		return nil
-	}
-	b.batcherMu.Lock()
-	tracked, err := b.trackBatchLocked(ctx, batch)
-	b.batcherMu.Unlock()
-	if err != nil {
-		return err
-	}
-	return b.sendTracked(ctx, tracked)
-}
-
 // waitSnapshotAcks blocks until every published snapshot batch has been
-// acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
-// batches release the gate too: redelivery is owned by auto_replay_nacks,
-// and the ctx escape prevents a permanently-failing downstream from
-// wedging shutdown.
+// 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() {
@@ -325,6 +334,11 @@ func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
 	}()
 	select {
 	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()
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 7e8548a8ec..73d0338325 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -29,7 +29,7 @@ func TestSnapshotAckGate(t *testing.T) {
 		ctx := t.Context()
 		publisher, _ := newTestBatchPublisher(t)
 
-		msg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
 
 		done := make(chan error, 1)
 		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
@@ -49,14 +49,20 @@ func TestSnapshotAckGate(t *testing.T) {
 		}
 	})
 
-	t.Run("a nack also releases the gate", func(t *testing.T) {
+	t.Run("a nack releases the gate but fails it", func(t *testing.T) {
 		ctx := t.Context()
-		publisher, _ := newTestBatchPublisher(t)
+		publisher, cachedLSNs := newTestBatchPublisher(t)
 
-		msg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
-		require.NoError(t, msg.ackFn(ctx, errors.New("downstream failure")))
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
+		nackErr := errors.New("downstream failure")
+		require.ErrorIs(t, msg.ackFn(ctx, nackErr), nackErr)
 
-		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+		// auto_replay_nacks is user-toggleable, so a nack can be terminal:
+		// the gate must report it so the post-snapshot LSN is not persisted
+		// and the snapshot re-runs on restart.
+		err := publisher.waitSnapshotAcks(ctx)
+		require.ErrorIs(t, err, nackErr)
+		require.Empty(t, cachedLSNs())
 	})
 
 	t.Run("streaming batches do not hold the gate", func(t *testing.T) {
@@ -64,7 +70,7 @@ func TestSnapshotAckGate(t *testing.T) {
 		publisher, _ := newTestBatchPublisher(t)
 
 		// Published but never acked: must not block the gate.
-		publishAndReceive(t, ctx, publisher, service.MessageBatch{newStreamingMessage("00000030")})
+		publishAndReceive(t, ctx, publisher, streamingEvent("00000030", ""))
 
 		require.NoError(t, publisher.waitSnapshotAcks(ctx))
 	})
@@ -73,7 +79,7 @@ func TestSnapshotAckGate(t *testing.T) {
 		publisher, _ := newTestBatchPublisher(t)
 
 		ctx, cancel := context.WithCancel(t.Context())
-		publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage()})
+		publishAndReceive(t, ctx, publisher, snapshotEvent())
 
 		done := make(chan error, 1)
 		go func() { done <- publisher.waitSnapshotAcks(ctx) }()
@@ -90,24 +96,12 @@ func TestSnapshotAckGate(t *testing.T) {
 
 func TestFlushCurrent(t *testing.T) {
 	ctx := t.Context()
-	logger := service.NewLoggerFromSlog(slog.Default())
-	cp := checkpoint.NewCapped[replication.LSN](100)
-
-	batcher, err := (service.BatchPolicy{Count: 100}).NewBatcher(service.MockResources())
-	require.NoError(t, err)
-
-	publisher := newBatchPublisher(batcher, cp, logger)
-	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
-	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	// Count=100 keeps published events buffered in the batcher until flushed.
+	publisher, _ := newTestBatchPublisherWithCount(t, 100)
 
 	publishEvent := func() {
 		t.Helper()
-		require.NoError(t, publisher.Publish(ctx, replication.MessageEvent{
-			Schema:    "dbo",
-			Table:     "t",
-			Operation: replication.MessageOperationRead.String(),
-			Data:      map[string]any{"a": 1},
-		}))
+		require.NoError(t, publisher.Publish(ctx, snapshotEvent()))
 	}
 	receive := func(failMsg string) {
 		t.Helper()
@@ -122,7 +116,6 @@ func TestFlushCurrent(t *testing.T) {
 		}
 	}
 
-	// Count=100 keeps a single event buffered in the batcher until flushed.
 	publishEvent()
 	receive("flushCurrent did not publish the buffered partial batch")
 
@@ -133,16 +126,11 @@ func TestFlushCurrent(t *testing.T) {
 }
 
 func TestCheckpointSelection(t *testing.T) {
-	t.Run("persists checkpoint_lsn, never the row's own lsn", func(t *testing.T) {
+	t.Run("persists the transaction boundary, never the row's own lsn", func(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedLSNs := newTestBatchPublisher(t)
 
-		msg := service.NewMessage([]byte("{}"))
-		msg.MetaSet("operation", replication.MessageOperationInsert.String())
-		msg.MetaSet("lsn", "00000042")
-		msg.MetaSet("checkpoint_lsn", "00000041")
-
-		am := publishAndReceive(t, ctx, publisher, service.MessageBatch{msg})
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
 		require.NoError(t, am.ackFn(ctx, nil))
 
 		lsns := cachedLSNs()
@@ -151,16 +139,33 @@ func TestCheckpointSelection(t *testing.T) {
 			"the checkpoint must be the last fully-published transaction boundary, not the row's own LSN")
 	})
 
-	t.Run("no checkpoint_lsn (first transaction) persists nothing", func(t *testing.T) {
+	t.Run("no boundary yet (first transaction) persists nothing", func(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedLSNs := newTestBatchPublisher(t)
 
-		am := publishAndReceive(t, ctx, publisher, service.MessageBatch{newStreamingMessage("00000042")})
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
 		require.NoError(t, am.ackFn(ctx, nil))
 
 		require.Empty(t, cachedLSNs(),
 			"a batch ending mid-transaction (no prior complete transaction) must not persist any LSN")
 	})
+
+	t.Run("a nacked batch pins the checkpoint", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		b1 := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		b2 := publishAndReceive(t, ctx, publisher, streamingEvent("00000043", "00000042"))
+
+		// Nack b1: with auto_replay_nacks disabled this is terminal, so b2's
+		// ack must not persist anything past the undelivered b1.
+		nackErr := errors.New("downstream failure")
+		require.ErrorIs(t, b1.ackFn(ctx, nackErr), nackErr)
+		require.NoError(t, b2.ackFn(ctx, nil))
+
+		require.Empty(t, cachedLSNs(),
+			"a checkpoint must never be persisted past a nacked batch")
+	})
 }
 
 // TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
@@ -211,15 +216,8 @@ func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
 	const events = 500
 	for i := range events {
 		// %08d keeps lexicographic order == numeric order, like real LSNs.
-		lsn := replication.LSN(fmt.Sprintf("%08d", i))
-		require.NoError(t, publisher.Publish(ctx, replication.MessageEvent{
-			Schema:        "dbo",
-			Table:         "t",
-			Operation:     replication.MessageOperationInsert.String(),
-			LSN:           lsn,
-			CheckpointLSN: lsn,
-			Data:          map[string]any{"i": i},
-		}))
+		lsn := fmt.Sprintf("%08d", i)
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(lsn, lsn)))
 	}
 	require.NoError(t, publisher.flushCurrent(ctx))
 
@@ -239,13 +237,24 @@ func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
 	}
 }
 
+// newTestBatchPublisher builds a publisher whose batcher flushes on every
+// published event (count=1), so tests drive the production
+// Publish->trackBatchLocked->sendTracked path directly.
 func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.LSN) {
 	t.Helper()
+	return newTestBatchPublisherWithCount(t, 1)
+}
+
+func newTestBatchPublisherWithCount(t *testing.T, count int) (*batchPublisher, func() []replication.LSN) {
+	t.Helper()
 
 	logger := service.NewLoggerFromSlog(slog.Default())
 	cp := checkpoint.NewCapped[replication.LSN](100)
 
-	publisher := newBatchPublisher(nil, cp, logger)
+	batcher, err := (service.BatchPolicy{Count: count}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
 	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
 
 	var (
@@ -268,23 +277,33 @@ func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.
 	return publisher, cachedLSNsFn
 }
 
-func newSnapshotMessage() *service.Message {
-	msg := service.NewMessage([]byte("{}"))
-	msg.MetaSet("operation", replication.MessageOperationRead.String())
-	return msg
+func snapshotEvent() replication.MessageEvent {
+	return replication.MessageEvent{
+		Schema:    "dbo",
+		Table:     "t",
+		Operation: replication.MessageOperationRead.String(),
+		Data:      map[string]any{"a": 1},
+	}
 }
 
-func newStreamingMessage(lsn string) *service.Message {
-	msg := service.NewMessage([]byte("{}"))
-	msg.MetaSet("operation", replication.MessageOperationInsert.String())
-	msg.MetaSet("lsn", lsn)
-	return msg
+func streamingEvent(lsn, checkpointLSN string) replication.MessageEvent {
+	return replication.MessageEvent{
+		Schema:        "dbo",
+		Table:         "t",
+		Operation:     replication.MessageOperationInsert.String(),
+		LSN:           replication.LSN(lsn),
+		CheckpointLSN: replication.LSN(checkpointLSN),
+		Data:          map[string]any{"a": 1},
+	}
 }
 
-func publishAndReceive(t *testing.T, ctx context.Context, publisher *batchPublisher, batch service.MessageBatch) asyncMessage {
+// publishAndReceive publishes a single event through the production Publish
+// path (count=1 batcher: every event flushes, tracks, and sends immediately)
+// and returns the delivered asyncMessage.
+func publishAndReceive(t *testing.T, ctx context.Context, publisher *batchPublisher, event replication.MessageEvent) asyncMessage {
 	t.Helper()
 	go func() {
-		_ = publisher.publishBatch(ctx, batch)
+		_ = publisher.Publish(ctx, event)
 	}()
 	return <-publisher.msgs()
 }

From 2fdb39b09823c211f63dcebcfa6260bb6b48cfb6 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Fri, 7 Aug 2026 11:16:43 -0400
Subject: [PATCH 07/12] mssqlserver_cdc: checkpoint the exact end position of
 each drained polling window
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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).
---
 internal/impl/mssqlserver/batcher.go          | 29 ++++++++++++
 internal/impl/mssqlserver/batcher_test.go     | 44 +++++++++++++++++++
 .../mssqlserver/replication/snapshot_test.go  |  4 ++
 .../impl/mssqlserver/replication/stream.go    | 13 ++++++
 4 files changed, 90 insertions(+)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 69844fe89b..4acde3bebc 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -345,6 +345,35 @@ func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
 	}
 }
 
+// CheckpointWindow registers an empty marker slot carrying lsn with the
+// ordered tracker, after flushing any partial batch belonging to the window.
+// The marker resolves immediately, so lsn is persisted as soon as every batch
+// published before it has been acked — giving the stream an exact resume
+// position at each drained polling window instead of lagging one transaction
+// behind (which would re-deliver the final transaction of a burst on every
+// restart).
+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 {
+		return fmt.Errorf("flushing window remainder: %w", err)
+	}
+
+	b.batcherMu.Lock()
+	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)
+	}
+	return nil
+}
+
 // flushCurrent flushes any partial batch still held by the batcher and
 // publishes it, leaving the publisher loop running. Used at the
 // snapshot->streaming handoff so every snapshot row is published (and can be
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 73d0338325..1ff28065d6 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -168,6 +168,50 @@ func TestCheckpointSelection(t *testing.T) {
 	})
 }
 
+func TestCheckpointWindow(t *testing.T) {
+	t.Run("persists immediately when all prior batches are acked", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, am.ackFn(ctx, nil))
+		require.Empty(t, cachedLSNs(), "mid-transaction batch must not persist anything on its own")
+
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000042", string(lsns[0]),
+			"a drained window must checkpoint its exact end position")
+	})
+
+	t.Run("waits for outstanding acks before surfacing", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+		require.Empty(t, cachedLSNs(), "the window end must not persist while its batches are unacked")
+
+		require.NoError(t, am.ackFn(ctx, nil))
+		lsns := cachedLSNs()
+		require.Len(t, lsns, 1)
+		require.Equal(t, "00000042", string(lsns[0]))
+	})
+
+	t.Run("a nacked batch pins the window checkpoint", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
+		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+
+		nackErr := errors.New("downstream failure")
+		require.ErrorIs(t, am.ackFn(ctx, nackErr), nackErr)
+		require.Empty(t, cachedLSNs(), "the window end must never persist past a nacked batch")
+	})
+}
+
 // TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
 // count-triggered flush in Publish and the timed-flush loop) and asserts the
 // persisted checkpoint never regresses when batches are acked in delivery
diff --git a/internal/impl/mssqlserver/replication/snapshot_test.go b/internal/impl/mssqlserver/replication/snapshot_test.go
index a89ed0d361..2b8be4f569 100644
--- a/internal/impl/mssqlserver/replication/snapshot_test.go
+++ b/internal/impl/mssqlserver/replication/snapshot_test.go
@@ -158,6 +158,10 @@ func (m *publisherStub) Publish(_ context.Context, msg replication.MessageEvent)
 	return nil
 }
 
+func (*publisherStub) CheckpointWindow(context.Context, replication.LSN) error {
+	return nil
+}
+
 func (m *publisherStub) count() int {
 	m.mu.Lock()
 	defer m.mu.Unlock()
diff --git a/internal/impl/mssqlserver/replication/stream.go b/internal/impl/mssqlserver/replication/stream.go
index df7de5e1d2..1d03720dba 100644
--- a/internal/impl/mssqlserver/replication/stream.go
+++ b/internal/impl/mssqlserver/replication/stream.go
@@ -356,6 +356,13 @@ func (t *txnBoundary) Observe(lsn LSN) LSN {
 // ChangePublisher is responsible for handling and processing of a replication.MessageEvent.
 type ChangePublisher interface {
 	Publish(ctx context.Context, msg MessageEvent) error
+	// CheckpointWindow records that every transaction up to and including lsn
+	// has been fully published (a polling window drained). Once all batches
+	// published before this call are acknowledged, lsn may be persisted as the
+	// resume position — without it the final transaction of a burst would only
+	// be checkpointed when a later transaction appears, re-delivering it on
+	// every restart of an idle stream.
+	CheckpointWindow(ctx context.Context, lsn LSN) error
 }
 
 // ChangeTableStream tracks and streams all change events from the configured change
@@ -475,6 +482,12 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st
 
 		if len(lastLSN) != 0 {
 			if !bytes.Equal(startLSN, lastLSN) {
+				// The window is drained: every transaction <= lastLSN is fully
+				// published, so the exact end position may be checkpointed once
+				// the window's batches are acked.
+				if err := r.publisher.CheckpointWindow(ctx, lastLSN); err != nil {
+					return fmt.Errorf("checkpointing window end: %w", err)
+				}
 				startLSN = lastLSN
 			} else {
 				r.log.Debug("No more changes across all change tables, backing off...")

From 5c047302d267de6453ef059627be13d25f4aa48f Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Mon, 10 Aug 2026 10:09:35 -0400
Subject: [PATCH 08/12] mssqlserver_cdc: reset the snapshot gate per attempt

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.
---
 internal/impl/mssqlserver/batcher.go          | 12 +++++++++++
 internal/impl/mssqlserver/batcher_test.go     | 20 +++++++++++++++++++
 .../impl/mssqlserver/input_mssqlserver_cdc.go |  3 +++
 3 files changed, 35 insertions(+)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 4acde3bebc..485cbc4bb3 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -69,6 +69,18 @@ func (b *batchPublisher) recordSnapshotNack(err error) {
 	}
 }
 
+// resetSnapshotGate clears any nack recorded by a previous snapshot attempt so
+// the gate reflects only the current run: the publisher outlives reconnects,
+// and a stale error would fail every retry even after a clean re-run. The
+// WaitGroup is deliberately left untouched — batches from a previous attempt
+// that are still in flight can yet be acked or nacked, and both must keep
+// counting.
+func (b *batchPublisher) resetSnapshotGate() {
+	b.snapshotNackMu.Lock()
+	defer b.snapshotNackMu.Unlock()
+	b.snapshotNackErr = nil
+}
+
 // newBatchPublisher creates an instance of batchPublisher.
 func newBatchPublisher(batcher *service.Batcher, checkpoint *checkpoint.Capped[replication.LSN], logger *service.Logger) *batchPublisher {
 	b := &batchPublisher{
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 1ff28065d6..9f46f0814e 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -75,6 +75,26 @@ func TestSnapshotAckGate(t *testing.T) {
 		require.NoError(t, publisher.waitSnapshotAcks(ctx))
 	})
 
+	t.Run("a nack fails only the snapshot attempt it belongs to", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		// Run 1: a snapshot batch is nacked; the gate fails.
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
+		nackErr := errors.New("downstream failure")
+		require.ErrorIs(t, msg.ackFn(ctx, nackErr), nackErr)
+		require.ErrorIs(t, publisher.waitSnapshotAcks(ctx), nackErr)
+
+		// Run 2 (reconnect reuses the publisher): the gate is reset, the
+		// re-run snapshot acks cleanly, and the gate must pass — a stale
+		// run-1 error here would livelock the input re-snapshotting forever.
+		publisher.resetSnapshotGate()
+		msg2 := publishAndReceive(t, ctx, publisher, snapshotEvent())
+		require.NoError(t, msg2.ackFn(ctx, nil))
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
+		require.Empty(t, cachedLSNs())
+	})
+
 	t.Run("context cancellation escapes the gate", func(t *testing.T) {
 		publisher, _ := newTestBatchPublisher(t)
 
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index 4034ccd745..9635e6d878 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -357,6 +357,9 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 
 		// snapshot if no LSN exists then store checkpoint once complete
 		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()
 			if maxLSN, err = i.processSnapshot(softCtx, snapshotter); err != nil {
 				if i.stopSig.IsHardStopSignalled() {
 					i.log.Errorf("Shutting down snapshotting process: %s", err)

From 4e7ea3d345f6d310e690e4569cbe8282b4c6780d Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Mon, 10 Aug 2026 10:45:59 -0400
Subject: [PATCH 09/12] mssqlserver_cdc: log downstream batch rejections

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.
---
 internal/impl/mssqlserver/batcher.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 485cbc4bb3..ad8443a815 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -305,6 +305,7 @@ func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.Mes
 					if isSnapshotBatch {
 						b.recordSnapshotNack(err)
 					}
+					b.log.Errorf("Batch rejected downstream (snapshot=%v, checkpoint LSN '%s'): the checkpoint is now pinned before this batch and the input will stall once checkpoint_limit is reached, unless the batch is redelivered (auto_replay_nacks) or the pipeline restarts: %v", isSnapshotBatch, replication.LSN(checkpointLSN), err)
 					return err
 				}
 				lsn := resolveFn()

From 90d6ac34eb8ab63e0075cc27e82da3f40b040bf6 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Mon, 10 Aug 2026 16:19:51 -0400
Subject: [PATCH 10/12] mssqlserver_cdc: address review - terminal nacks
 restart with a fresh 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.
---
 internal/impl/mssqlserver/batcher.go          | 86 +++++++++++++++----
 internal/impl/mssqlserver/batcher_test.go     | 74 ++++++++++++++++
 .../impl/mssqlserver/input_mssqlserver_cdc.go | 48 +++++++++--
 3 files changed, 182 insertions(+), 26 deletions(-)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index ad8443a815..7108c8238f 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -14,6 +14,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -54,11 +55,36 @@ type batchPublisher struct {
 	snapshotNackErr error
 
 	// pendingCheckpointLSN mirrors the CheckpointLSN of the most recently
-	// added message: the start LSN of the last transaction whose rows are all
-	// published, the only value safe to persist as a resume position. Guarded
-	// by batcherMu, so at flush time it always belongs to the flushed batch's
-	// last message.
+	// added message (or a stronger drained-window LSN, see CheckpointWindow):
+	// the start LSN of the last transaction whose rows are all published, the
+	// only value safe to persist as a resume position. Guarded by batcherMu,
+	// so at flush time it always belongs to the flushed batch's last message.
 	pendingCheckpointLSN replication.LSN
+	// buffered counts messages currently held by the batcher (guarded by
+	// batcherMu). CheckpointWindow uses it to decide between deferring the
+	// window checkpoint to the buffered batch and registering a marker.
+	buffered int
+
+	// onTerminalNack, when set, is invoked once a batch is rejected
+	// downstream: a nack pins the ordered tracker, so the input must restart
+	// (with a fresh publisher) to resume from the last durable LSN.
+	onTerminalNack func(error)
+	// sealed marks a publisher that has been replaced by a reconnect. Late
+	// acks from its session must not persist checkpoints (they could regress
+	// the new session's positions) nor trigger restarts.
+	sealed atomic.Bool
+}
+
+// seal marks the publisher as replaced; see the sealed field.
+func (b *batchPublisher) seal() {
+	b.sealed.Store(true)
+}
+
+// Close stops the publisher's flush loop and waits for it to exit; the
+// batcher is closed by the loop's defer.
+func (b *batchPublisher) Close() {
+	b.shutSig.TriggerSoftStop()
+	<-b.shutSig.HasStoppedChan()
 }
 
 func (b *batchPublisher) recordSnapshotNack(err error) {
@@ -169,6 +195,7 @@ func (p *batchPublisher) loop() {
 				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
 					return
 				}
+				p.buffered = 0
 				tracked, trackErr = p.trackBatchLocked(closeAtLeisureCtx, sendBatch)
 			}()
 			if trackErr != nil {
@@ -237,8 +264,11 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 	if b.batcher.Add(msg) {
 		var flushedBatch []*service.Message
 		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			b.buffered = 0
 			tracked, err = b.trackBatchLocked(ctx, flushedBatch)
 		}
+	} else {
+		b.buffered++
 	}
 	b.batcherMu.Unlock()
 	if err != nil {
@@ -301,13 +331,28 @@ func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.Mes
 					// 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.
+					// handoff gate so the post-snapshot LSN is not persisted,
+					// and the input restarts with a fresh tracker to resume
+					// from the last durable LSN (the pinned slot would
+					// otherwise wedge checkpointing for the process lifetime).
 					if isSnapshotBatch {
 						b.recordSnapshotNack(err)
 					}
-					b.log.Errorf("Batch rejected downstream (snapshot=%v, checkpoint LSN '%s'): the checkpoint is now pinned before this batch and the input will stall once checkpoint_limit is reached, unless the batch is redelivered (auto_replay_nacks) or the pipeline restarts: %v", isSnapshotBatch, replication.LSN(checkpointLSN), err)
+					if b.sealed.Load() {
+						return err
+					}
+					b.log.Errorf("Batch rejected downstream (snapshot=%v, checkpoint LSN '%s'): restarting to redeliver from the last durable checkpoint: %v", isSnapshotBatch, replication.LSN(checkpointLSN), err)
+					if b.onTerminalNack != nil {
+						b.onTerminalNack(err)
+					}
 					return err
 				}
+				if b.sealed.Load() {
+					// A late ack from a replaced session: resolving its own
+					// tracker is harmless, but persisting could regress the
+					// new session's checkpoints.
+					return nil
+				}
 				lsn := resolveFn()
 				if lsn != nil && len(*lsn) != 0 {
 					return b.cacheLSN(ctx, *lsn)
@@ -358,21 +403,23 @@ func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
 	}
 }
 
-// CheckpointWindow registers an empty marker slot carrying lsn with the
-// ordered tracker, after flushing any partial batch belonging to the window.
-// The marker resolves immediately, so lsn is persisted as soon as every batch
-// published before it has been acked — giving the stream an exact resume
-// position at each drained polling window instead of lagging one transaction
-// behind (which would re-deliver the final transaction of a burst on every
-// restart).
+// CheckpointWindow records that every transaction up to and including lsn is
+// fully published (a polling window drained), giving the stream an exact
+// resume position instead of lagging one transaction behind (which would
+// re-deliver the final transaction of a burst on every restart).
+//
+// The user's batching policy stays in charge of batch sizes: if rows from the
+// window are still buffered, the window-end LSN simply becomes their batch's
+// checkpoint payload (safe, and stronger than the last row's transaction
+// boundary). Only when the batcher is empty is an immediately-resolved marker
+// slot registered, so lsn persists once every published batch is acked.
 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 {
-		return fmt.Errorf("flushing window remainder: %w", err)
-	}
-
 	b.batcherMu.Lock()
+	if b.buffered > 0 {
+		b.pendingCheckpointLSN = lsn
+		b.batcherMu.Unlock()
+		return nil
+	}
 	resolveFn, err := b.checkpoint.Track(ctx, lsn, 1)
 	b.batcherMu.Unlock()
 	if err != nil {
@@ -399,6 +446,7 @@ func (b *batchPublisher) flushCurrent(ctx context.Context) error {
 	b.batcherMu.Lock()
 	remaining, err := b.batcher.Flush(ctx)
 	if err == nil && len(remaining) > 0 {
+		b.buffered = 0
 		tracked, err = b.trackBatchLocked(ctx, remaining)
 	}
 	b.batcherMu.Unlock()
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 9f46f0814e..29ba173be2 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"log/slog"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -232,6 +233,79 @@ func TestCheckpointWindow(t *testing.T) {
 	})
 }
 
+func TestCheckpointWindowDefersToBufferedBatch(t *testing.T) {
+	ctx := t.Context()
+	// Count=100 keeps published events buffered: the window checkpoint must
+	// ride on the eventual batch instead of forcing a flush (which would
+	// override the user's batching policy).
+	publisher, cachedLSNs := newTestBatchPublisherWithCount(t, 100)
+
+	require.NoError(t, publisher.Publish(ctx, streamingEvent("00000042", "")))
+	require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
+	require.Empty(t, cachedLSNs(), "a deferred window checkpoint must not persist before its batch is acked")
+
+	// No batch may have been force-flushed by CheckpointWindow.
+	select {
+	case m := <-publisher.msgs():
+		t.Fatalf("CheckpointWindow force-flushed a batch of %d messages, overriding the batching policy", len(m.msg))
+	case <-time.After(100 * time.Millisecond):
+	}
+
+	// When the batch eventually flushes and acks, it carries the window LSN.
+	got := make(chan asyncMessage, 1)
+	go func() { got <- <-publisher.msgs() }()
+	require.NoError(t, publisher.flushCurrent(ctx))
+	var am asyncMessage
+	select {
+	case am = <-got:
+	case <-time.After(5 * time.Second):
+		t.Fatal("buffered batch was never flushed")
+	}
+	require.NoError(t, am.ackFn(ctx, nil))
+
+	lsns := cachedLSNs()
+	require.Len(t, lsns, 1)
+	require.Equal(t, "00000042", string(lsns[0]), "the drained-window LSN must ride on the buffered batch's checkpoint")
+}
+
+func TestTerminalNack(t *testing.T) {
+	t.Run("invokes onTerminalNack so the input can restart", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		var got atomic.Value
+		publisher.onTerminalNack = func(err error) { got.Store(err) }
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		nackErr := errors.New("downstream failure")
+		require.ErrorIs(t, am.ackFn(ctx, nackErr), nackErr)
+
+		stored, _ := got.Load().(error)
+		require.ErrorIs(t, stored, nackErr)
+		require.Empty(t, cachedLSNs())
+	})
+
+	t.Run("a sealed publisher neither persists nor restarts", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedLSNs := newTestBatchPublisher(t)
+
+		restarted := false
+		publisher.onTerminalNack = func(error) { restarted = true }
+
+		am1 := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		am2 := publishAndReceive(t, ctx, publisher, streamingEvent("00000043", "00000042"))
+		publisher.seal()
+
+		// Late ack from a replaced session: must not persist.
+		require.NoError(t, am1.ackFn(ctx, nil))
+		require.Empty(t, cachedLSNs(), "a sealed publisher must not persist checkpoints")
+
+		// Late nack: must not trigger a restart of the new session.
+		require.Error(t, am2.ackFn(ctx, errors.New("late failure")))
+		require.False(t, restarted, "a sealed publisher must not trigger restarts")
+	})
+}
+
 // TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
 // count-triggered flush in Publish and the timed-flush loop) and asserts the
 // persisted checkpoint never regresses when batches are acked in delivery
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index 9635e6d878..d4b2cb5cad 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -156,6 +156,13 @@ type sqlServerCDCInput struct {
 	publisher *batchPublisher
 	metrics   *service.Metrics
 
+	// batching and checkpointLimit rebuild the publisher (batcher + ordered
+	// checkpoint tracker) on every Connect: a terminal nack pins a tracker
+	// slot by design, and only a fresh tracker lets the restart resume from
+	// the last durable LSN instead of staying wedged behind the stale slot.
+	batching        service.BatchPolicy
+	checkpointLimit int
+
 	connMu  sync.Mutex
 	stopSig *shutdown.Signaller
 	log     *service.Logger
@@ -266,12 +273,14 @@ func newMSSQLServerCDCInput(conf *service.ParsedConfig, resources *service.Resou
 				Exclude: tableExcludes,
 			},
 		},
-		res:       resources,
-		log:       logger,
-		metrics:   resources.Metrics(),
-		stopSig:   shutdown.NewSignaller(),
-		publisher: newBatchPublisher(batcher, cp, logger),
-		cpCache:   cpCache,
+		res:             resources,
+		log:             logger,
+		metrics:         resources.Metrics(),
+		stopSig:         shutdown.NewSignaller(),
+		publisher:       newBatchPublisher(batcher, cp, logger),
+		batching:        policy,
+		checkpointLimit: checkpointLimit,
+		cpCache:         cpCache,
 	}
 
 	i.publisher.cacheLSN = i.cacheLSN
@@ -329,6 +338,27 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 		return fmt.Errorf("unable to get cached LSN: %s", err)
 	}
 
+	// Rebuild the publisher (batcher + ordered checkpoint tracker) for this
+	// connection attempt. A terminal nack pins a tracker slot by design;
+	// reusing the old tracker would leave every future checkpoint stuck
+	// behind the stale slot, wedging the input for the process lifetime
+	// instead of letting this restart resume from the last durable LSN. The
+	// old publisher is sealed so late acks from the previous session cannot
+	// persist stale positions.
+	i.publisher.seal()
+	i.publisher.Close()
+	newBatcher, err := i.batching.NewBatcher(i.res)
+	if err != nil {
+		return fmt.Errorf("creating batcher: %w", err)
+	}
+	i.publisher = newBatchPublisher(newBatcher, checkpoint.NewCapped[replication.LSN](int64(i.checkpointLimit)), i.log)
+	i.publisher.cacheLSN = i.cacheLSN
+	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()
+	}
+
 	// setup snapshotting and streaming
 	var (
 		snapshotter *replication.Snapshot
@@ -382,7 +412,11 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 				return
 			}
 			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)
+				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+					i.log.Infof("Interrupted while waiting for snapshot acknowledgements. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				} else {
+					i.log.Errorf("Snapshot batch was rejected downstream. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				}
 				i.stopSig.TriggerHasStopped()
 				return
 			}

From c3f2ff9ae4b615138b3735d38e2e8a1b44505fe8 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Tue, 11 Aug 2026 10:01:15 -0400
Subject: [PATCH 11/12] mssqlserver_cdc: nacks resolve checkpoints
 (auto_replay_nacks off is 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.
---
 internal/impl/mssqlserver/batcher.go          |  94 ++--------------
 internal/impl/mssqlserver/batcher_test.go     | 102 ++++--------------
 .../impl/mssqlserver/input_mssqlserver_cdc.go |  51 ++-------
 3 files changed, 37 insertions(+), 210 deletions(-)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 7108c8238f..779e2052c5 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -14,7 +14,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -48,12 +47,6 @@ type batchPublisher struct {
 	// acknowledged downstream. The snapshot->streaming handoff blocks on it so
 	// the post-snapshot LSN is never persisted while snapshot rows are in flight.
 	snapshotAckWG sync.WaitGroup
-	// snapshotNackErr records the first snapshot batch nack. auto_replay_nacks
-	// is user-toggleable, so a nack can be terminal: the gate must fail rather
-	// than let the post-snapshot LSN persist over undelivered rows.
-	snapshotNackMu  sync.Mutex
-	snapshotNackErr error
-
 	// pendingCheckpointLSN mirrors the CheckpointLSN of the most recently
 	// added message (or a stronger drained-window LSN, see CheckpointWindow):
 	// the start LSN of the last transaction whose rows are all published, the
@@ -64,47 +57,6 @@ type batchPublisher struct {
 	// batcherMu). CheckpointWindow uses it to decide between deferring the
 	// window checkpoint to the buffered batch and registering a marker.
 	buffered int
-
-	// onTerminalNack, when set, is invoked once a batch is rejected
-	// downstream: a nack pins the ordered tracker, so the input must restart
-	// (with a fresh publisher) to resume from the last durable LSN.
-	onTerminalNack func(error)
-	// sealed marks a publisher that has been replaced by a reconnect. Late
-	// acks from its session must not persist checkpoints (they could regress
-	// the new session's positions) nor trigger restarts.
-	sealed atomic.Bool
-}
-
-// seal marks the publisher as replaced; see the sealed field.
-func (b *batchPublisher) seal() {
-	b.sealed.Store(true)
-}
-
-// Close stops the publisher's flush loop and waits for it to exit; the
-// batcher is closed by the loop's defer.
-func (b *batchPublisher) Close() {
-	b.shutSig.TriggerSoftStop()
-	<-b.shutSig.HasStoppedChan()
-}
-
-func (b *batchPublisher) recordSnapshotNack(err error) {
-	b.snapshotNackMu.Lock()
-	defer b.snapshotNackMu.Unlock()
-	if b.snapshotNackErr == nil {
-		b.snapshotNackErr = err
-	}
-}
-
-// resetSnapshotGate clears any nack recorded by a previous snapshot attempt so
-// the gate reflects only the current run: the publisher outlives reconnects,
-// and a stale error would fail every retry even after a clean re-run. The
-// WaitGroup is deliberately left untouched — batches from a previous attempt
-// that are still in flight can yet be acked or nacked, and both must keep
-// counting.
-func (b *batchPublisher) resetSnapshotGate() {
-	b.snapshotNackMu.Lock()
-	defer b.snapshotNackMu.Unlock()
-	b.snapshotNackErr = nil
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -322,37 +274,14 @@ func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.Mes
 		isSnapshot: isSnapshotBatch,
 		msg: asyncMessage{
 			msg: batch,
-			ackFn: func(ctx context.Context, err error) error {
+			// The ack error is deliberately ignored: nacks are replayed by
+			// auto_replay_nacks (the default), and disabling that is a
+			// documented opt-in to DROP rejected messages, so the checkpoint
+			// must advance past them rather than pin the tracker.
+			ackFn: func(ctx context.Context, _ error) error {
 				if isSnapshotBatch {
 					defer b.snapshotAckWG.Done()
 				}
-				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,
-					// and the input restarts with a fresh tracker to resume
-					// from the last durable LSN (the pinned slot would
-					// otherwise wedge checkpointing for the process lifetime).
-					if isSnapshotBatch {
-						b.recordSnapshotNack(err)
-					}
-					if b.sealed.Load() {
-						return err
-					}
-					b.log.Errorf("Batch rejected downstream (snapshot=%v, checkpoint LSN '%s'): restarting to redeliver from the last durable checkpoint: %v", isSnapshotBatch, replication.LSN(checkpointLSN), err)
-					if b.onTerminalNack != nil {
-						b.onTerminalNack(err)
-					}
-					return err
-				}
-				if b.sealed.Load() {
-					// A late ack from a replaced session: resolving its own
-					// tracker is harmless, but persisting could regress the
-					// new session's checkpoints.
-					return nil
-				}
 				lsn := resolveFn()
 				if lsn != nil && len(*lsn) != 0 {
 					return b.cacheLSN(ctx, *lsn)
@@ -379,10 +308,10 @@ func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch)
 }
 
 // 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.
+// acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
+// batches release the gate too: redelivery is owned by auto_replay_nacks,
+// and disabling that is a documented opt-in to drop rejections. The ctx
+// escape prevents a permanently-failing downstream from wedging shutdown.
 func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
 	drained := make(chan struct{})
 	go func() {
@@ -392,11 +321,6 @@ func (b *batchPublisher) waitSnapshotAcks(ctx context.Context) error {
 	}()
 	select {
 	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()
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 29ba173be2..6833782f0a 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -14,7 +14,6 @@ import (
 	"fmt"
 	"log/slog"
 	"sync"
-	"sync/atomic"
 	"testing"
 	"time"
 
@@ -50,20 +49,15 @@ func TestSnapshotAckGate(t *testing.T) {
 		}
 	})
 
-	t.Run("a nack releases the gate but fails it", func(t *testing.T) {
+	t.Run("a nack also releases the gate", func(t *testing.T) {
 		ctx := t.Context()
-		publisher, cachedLSNs := newTestBatchPublisher(t)
+		publisher, _ := newTestBatchPublisher(t)
 
 		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
-		nackErr := errors.New("downstream failure")
-		require.ErrorIs(t, msg.ackFn(ctx, nackErr), nackErr)
-
-		// auto_replay_nacks is user-toggleable, so a nack can be terminal:
-		// the gate must report it so the post-snapshot LSN is not persisted
-		// and the snapshot re-runs on restart.
-		err := publisher.waitSnapshotAcks(ctx)
-		require.ErrorIs(t, err, nackErr)
-		require.Empty(t, cachedLSNs())
+		// Nacks count as settled: replay is owned by auto_replay_nacks, and
+		// disabling it is a documented opt-in to drop rejections.
+		require.NoError(t, msg.ackFn(ctx, errors.New("downstream failure")))
+		require.NoError(t, publisher.waitSnapshotAcks(ctx))
 	})
 
 	t.Run("streaming batches do not hold the gate", func(t *testing.T) {
@@ -76,26 +70,6 @@ func TestSnapshotAckGate(t *testing.T) {
 		require.NoError(t, publisher.waitSnapshotAcks(ctx))
 	})
 
-	t.Run("a nack fails only the snapshot attempt it belongs to", func(t *testing.T) {
-		ctx := t.Context()
-		publisher, cachedLSNs := newTestBatchPublisher(t)
-
-		// Run 1: a snapshot batch is nacked; the gate fails.
-		msg := publishAndReceive(t, ctx, publisher, snapshotEvent())
-		nackErr := errors.New("downstream failure")
-		require.ErrorIs(t, msg.ackFn(ctx, nackErr), nackErr)
-		require.ErrorIs(t, publisher.waitSnapshotAcks(ctx), nackErr)
-
-		// Run 2 (reconnect reuses the publisher): the gate is reset, the
-		// re-run snapshot acks cleanly, and the gate must pass — a stale
-		// run-1 error here would livelock the input re-snapshotting forever.
-		publisher.resetSnapshotGate()
-		msg2 := publishAndReceive(t, ctx, publisher, snapshotEvent())
-		require.NoError(t, msg2.ackFn(ctx, nil))
-		require.NoError(t, publisher.waitSnapshotAcks(ctx))
-		require.Empty(t, cachedLSNs())
-	})
-
 	t.Run("context cancellation escapes the gate", func(t *testing.T) {
 		publisher, _ := newTestBatchPublisher(t)
 
@@ -171,21 +145,22 @@ func TestCheckpointSelection(t *testing.T) {
 			"a batch ending mid-transaction (no prior complete transaction) must not persist any LSN")
 	})
 
-	t.Run("a nacked batch pins the checkpoint", func(t *testing.T) {
+	t.Run("a nack resolves too: auto_replay_nacks off is an opt-in drop", func(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedLSNs := newTestBatchPublisher(t)
 
 		b1 := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
 		b2 := publishAndReceive(t, ctx, publisher, streamingEvent("00000043", "00000042"))
 
-		// Nack b1: with auto_replay_nacks disabled this is terminal, so b2's
-		// ack must not persist anything past the undelivered b1.
-		nackErr := errors.New("downstream failure")
-		require.ErrorIs(t, b1.ackFn(ctx, nackErr), nackErr)
+		// A nacked batch is deleted per the auto_replay_nacks contract: its
+		// slot resolves so the stream continues past it instead of pinning
+		// the tracker and back-pressuring forever.
+		require.NoError(t, b1.ackFn(ctx, errors.New("downstream failure")))
 		require.NoError(t, b2.ackFn(ctx, nil))
 
-		require.Empty(t, cachedLSNs(),
-			"a checkpoint must never be persisted past a nacked batch")
+		lsns := cachedLSNs()
+		require.NotEmpty(t, lsns, "the checkpoint must continue advancing past a dropped batch")
+		require.Equal(t, "00000042", string(lsns[len(lsns)-1]))
 	})
 }
 
@@ -220,16 +195,19 @@ func TestCheckpointWindow(t *testing.T) {
 		require.Equal(t, "00000042", string(lsns[0]))
 	})
 
-	t.Run("a nacked batch pins the window checkpoint", func(t *testing.T) {
+	t.Run("a nacked batch settles the window checkpoint too", func(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedLSNs := newTestBatchPublisher(t)
 
 		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", ""))
 		require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN("00000042")))
 
-		nackErr := errors.New("downstream failure")
-		require.ErrorIs(t, am.ackFn(ctx, nackErr), nackErr)
-		require.Empty(t, cachedLSNs(), "the window end must never persist past a nacked batch")
+		// A nacked batch is deleted per the auto_replay_nacks contract, so
+		// the window checkpoint behind it still persists.
+		require.NoError(t, am.ackFn(ctx, errors.New("downstream failure")))
+		lsns := cachedLSNs()
+		require.NotEmpty(t, lsns)
+		require.Equal(t, "00000042", string(lsns[len(lsns)-1]))
 	})
 }
 
@@ -268,44 +246,6 @@ func TestCheckpointWindowDefersToBufferedBatch(t *testing.T) {
 	require.Equal(t, "00000042", string(lsns[0]), "the drained-window LSN must ride on the buffered batch's checkpoint")
 }
 
-func TestTerminalNack(t *testing.T) {
-	t.Run("invokes onTerminalNack so the input can restart", func(t *testing.T) {
-		ctx := t.Context()
-		publisher, cachedLSNs := newTestBatchPublisher(t)
-
-		var got atomic.Value
-		publisher.onTerminalNack = func(err error) { got.Store(err) }
-
-		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
-		nackErr := errors.New("downstream failure")
-		require.ErrorIs(t, am.ackFn(ctx, nackErr), nackErr)
-
-		stored, _ := got.Load().(error)
-		require.ErrorIs(t, stored, nackErr)
-		require.Empty(t, cachedLSNs())
-	})
-
-	t.Run("a sealed publisher neither persists nor restarts", func(t *testing.T) {
-		ctx := t.Context()
-		publisher, cachedLSNs := newTestBatchPublisher(t)
-
-		restarted := false
-		publisher.onTerminalNack = func(error) { restarted = true }
-
-		am1 := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
-		am2 := publishAndReceive(t, ctx, publisher, streamingEvent("00000043", "00000042"))
-		publisher.seal()
-
-		// Late ack from a replaced session: must not persist.
-		require.NoError(t, am1.ackFn(ctx, nil))
-		require.Empty(t, cachedLSNs(), "a sealed publisher must not persist checkpoints")
-
-		// Late nack: must not trigger a restart of the new session.
-		require.Error(t, am2.ackFn(ctx, errors.New("late failure")))
-		require.False(t, restarted, "a sealed publisher must not trigger restarts")
-	})
-}
-
 // TestTrackOrderUnderConcurrentFlush stresses the two concurrent flushers (the
 // count-triggered flush in Publish and the timed-flush loop) and asserts the
 // persisted checkpoint never regresses when batches are acked in delivery
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index d4b2cb5cad..4034ccd745 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -156,13 +156,6 @@ type sqlServerCDCInput struct {
 	publisher *batchPublisher
 	metrics   *service.Metrics
 
-	// batching and checkpointLimit rebuild the publisher (batcher + ordered
-	// checkpoint tracker) on every Connect: a terminal nack pins a tracker
-	// slot by design, and only a fresh tracker lets the restart resume from
-	// the last durable LSN instead of staying wedged behind the stale slot.
-	batching        service.BatchPolicy
-	checkpointLimit int
-
 	connMu  sync.Mutex
 	stopSig *shutdown.Signaller
 	log     *service.Logger
@@ -273,14 +266,12 @@ func newMSSQLServerCDCInput(conf *service.ParsedConfig, resources *service.Resou
 				Exclude: tableExcludes,
 			},
 		},
-		res:             resources,
-		log:             logger,
-		metrics:         resources.Metrics(),
-		stopSig:         shutdown.NewSignaller(),
-		publisher:       newBatchPublisher(batcher, cp, logger),
-		batching:        policy,
-		checkpointLimit: checkpointLimit,
-		cpCache:         cpCache,
+		res:       resources,
+		log:       logger,
+		metrics:   resources.Metrics(),
+		stopSig:   shutdown.NewSignaller(),
+		publisher: newBatchPublisher(batcher, cp, logger),
+		cpCache:   cpCache,
 	}
 
 	i.publisher.cacheLSN = i.cacheLSN
@@ -338,27 +329,6 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 		return fmt.Errorf("unable to get cached LSN: %s", err)
 	}
 
-	// Rebuild the publisher (batcher + ordered checkpoint tracker) for this
-	// connection attempt. A terminal nack pins a tracker slot by design;
-	// reusing the old tracker would leave every future checkpoint stuck
-	// behind the stale slot, wedging the input for the process lifetime
-	// instead of letting this restart resume from the last durable LSN. The
-	// old publisher is sealed so late acks from the previous session cannot
-	// persist stale positions.
-	i.publisher.seal()
-	i.publisher.Close()
-	newBatcher, err := i.batching.NewBatcher(i.res)
-	if err != nil {
-		return fmt.Errorf("creating batcher: %w", err)
-	}
-	i.publisher = newBatchPublisher(newBatcher, checkpoint.NewCapped[replication.LSN](int64(i.checkpointLimit)), i.log)
-	i.publisher.cacheLSN = i.cacheLSN
-	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()
-	}
-
 	// setup snapshotting and streaming
 	var (
 		snapshotter *replication.Snapshot
@@ -387,9 +357,6 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 
 		// snapshot if no LSN exists then store checkpoint once complete
 		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()
 			if maxLSN, err = i.processSnapshot(softCtx, snapshotter); err != nil {
 				if i.stopSig.IsHardStopSignalled() {
 					i.log.Errorf("Shutting down snapshotting process: %s", err)
@@ -412,11 +379,7 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 				return
 			}
 			if err = i.publisher.waitSnapshotAcks(softCtx); err != nil {
-				if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
-					i.log.Infof("Interrupted while waiting for snapshot acknowledgements. Snapshot will re-run on restart (may cause duplicate data): %s", err)
-				} else {
-					i.log.Errorf("Snapshot batch was rejected downstream. Snapshot will re-run on restart (may cause duplicate data): %s", err)
-				}
+				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
 			}

From 16e7f8fab538f0ebf992c5330380e1e9d746b803 Mon Sep 17 00:00:00 2001
From: Jonathan Chaput <jon.chaput@redpanda.com>
Date: Fri, 14 Aug 2026 11:09:51 -0400
Subject: [PATCH 12/12] mssqlserver_cdc: serialize checkpoint persistence to
 prevent cache regression

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).
---
 internal/impl/mssqlserver/batcher.go      | 10 +++
 internal/impl/mssqlserver/batcher_test.go | 75 +++++++++++++++++++++++
 2 files changed, 85 insertions(+)

diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 779e2052c5..2b83ada77d 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -47,6 +47,12 @@ type batchPublisher struct {
 	// acknowledged downstream. The snapshot->streaming handoff blocks on it so
 	// the post-snapshot LSN is never persisted while snapshot rows are in flight.
 	snapshotAckWG sync.WaitGroup
+	// persistMu serializes resolve+persist pairs. The ordered tracker hands
+	// out monotonically increasing frontiers, but ack functions and
+	// CheckpointWindow run on different goroutines: without a shared critical
+	// section around resolveFn()+cacheLSN, two persists can land out of order
+	// and regress the cached resume position.
+	persistMu sync.Mutex
 	// pendingCheckpointLSN mirrors the CheckpointLSN of the most recently
 	// added message (or a stronger drained-window LSN, see CheckpointWindow):
 	// the start LSN of the last transaction whose rows are all published, the
@@ -282,6 +288,8 @@ func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.Mes
 				if isSnapshotBatch {
 					defer b.snapshotAckWG.Done()
 				}
+				b.persistMu.Lock()
+				defer b.persistMu.Unlock()
 				lsn := resolveFn()
 				if lsn != nil && len(*lsn) != 0 {
 					return b.cacheLSN(ctx, *lsn)
@@ -352,6 +360,8 @@ func (b *batchPublisher) CheckpointWindow(ctx context.Context, lsn replication.L
 	// 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.
+	b.persistMu.Lock()
+	defer b.persistMu.Unlock()
 	if resolved := resolveFn(); resolved != nil && len(*resolved) != 0 {
 		return b.cacheLSN(ctx, *resolved)
 	}
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
index 6833782f0a..f6d6931453 100644
--- a/internal/impl/mssqlserver/batcher_test.go
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -315,6 +315,81 @@ func TestTrackOrderUnderConcurrentFlush(t *testing.T) {
 	}
 }
 
+// TestPersistOrderUnderConcurrentAcksAndWindows locks in that the cached
+// resume position never regresses when batch acks (pipeline goroutines) and
+// CheckpointWindow markers (stream goroutine) persist concurrently: the
+// resolve+persist pair must be a single critical section, otherwise two
+// persists can land out of order.
+func TestPersistOrderUnderConcurrentAcksAndWindows(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](1000)
+
+	batcher, err := (service.BatchPolicy{Count: 1}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+
+	publisher := newBatchPublisher(batcher, cp, logger)
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	var (
+		mu        sync.Mutex
+		persisted []replication.LSN
+	)
+	publisher.cacheLSN = func(_ context.Context, lsn replication.LSN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		persisted = append(persisted, lsn)
+		return nil
+	}
+
+	// Consumer: ack every batch on its own goroutine so acks complete out of
+	// order relative to each other and to the window markers.
+	var ackWG sync.WaitGroup
+	consumerDone := make(chan struct{})
+	consumerCtx, stopConsumer := context.WithCancel(ctx)
+	go func() {
+		defer close(consumerDone)
+		for {
+			select {
+			case m := <-publisher.msgs():
+				ackWG.Go(func() {
+					_ = m.ackFn(ctx, nil)
+				})
+			case <-consumerCtx.Done():
+				return
+			}
+		}
+	}()
+
+	const events = 400
+	for i := range events {
+		lsn := fmt.Sprintf("%08d", i)
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(lsn, lsn)))
+		// A drained polling window ends every 10 rows; its end LSN persists
+		// via an immediately-resolved marker racing the in-flight acks.
+		if i%10 == 9 {
+			require.NoError(t, publisher.CheckpointWindow(ctx, replication.LSN(lsn)))
+		}
+	}
+
+	finalLSN := fmt.Sprintf("%08d", events-1)
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return len(persisted) > 0 && string(persisted[len(persisted)-1]) == finalLSN
+	}, 10*time.Second, 10*time.Millisecond, "final LSN was never persisted")
+	stopConsumer()
+	<-consumerDone
+	ackWG.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 1; i < len(persisted); i++ {
+		require.GreaterOrEqual(t, string(persisted[i]), string(persisted[i-1]),
+			"persisted checkpoint regressed at index %d: %v", i, persisted)
+	}
+}
+
 // newTestBatchPublisher builds a publisher whose batcher flushes on every
 // published event (count=1), so tests drive the production
 // Publish->trackBatchLocked->sendTracked path directly.