diff --git a/internal/impl/mssqlserver/batcher.go b/internal/impl/mssqlserver/batcher.go
index 3907c58a01..d513a97892 100644
--- a/internal/impl/mssqlserver/batcher.go
+++ b/internal/impl/mssqlserver/batcher.go
@@ -12,8 +12,10 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -26,8 +28,45 @@ import (
 // batchPublisher is responsible processing individual events into a batch and flushing
 // them to the pipeline using service.Batcher.
 type batchPublisher struct {
-	batcher   *service.Batcher
+	batcher *service.Batcher
+	// batcherMu guards only the batcher's buffer (Add/Flush/UntilNext) and
+	// the pendingCheckpointLSN/buffered bookkeeping.
 	batcherMu sync.Mutex
+	// Flush tickets keep the checkpoint sequence exact without a lock held
+	// across Track: each flush (and CheckpointWindow marker) takes a ticket
+	// under batcherMu - atomically with the Flush, so the user's batching
+	// policy stays exact - and Track+send admission happens in ticket order.
+	// Track can block on checkpoint_limit under downstream backpressure;
+	// only the admitted ticket holder (and flushers queued behind it) waits,
+	// while Publish calls keep buffering and the timed-flush ticker keeps
+	// reading UntilNext. Admission is cancellable: an abandoned ticket is
+	// skipped when its turn comes, so a graceful stop unwinds queued
+	// flushers instead of wedging them behind a parked send.
+	ticketMu   sync.Mutex
+	nextTicket uint64                   // next ticket to hand out; guarded by batcherMu
+	admitted   uint64                   // next ticket allowed to Track+send; guarded by ticketMu
+	waiters    map[uint64]chan struct{} // parked admit calls; guarded by ticketMu
+	abandoned  map[uint64]struct{}      // cancelled tickets to skip; guarded by ticketMu
+	// sealed refuses all further admissions (guarded by ticketMu): set when
+	// an abandoned ticket owned a flushed batch. Admission is strictly
+	// ordered, so at that moment nothing after the dropped rows has been
+	// tracked - sealing guarantees nothing ever is, so no ack can persist an
+	// LSN past them before Connect rebuilds the poisoned publisher.
+	sealed bool
+	// stopping is set by the input's Close BEFORE any cancellation
+	// propagates, so sendTracked can distinguish the expected
+	// graceful-shutdown unwind (debug) from a send that fails while the
+	// pipeline is meant to be live (warn) - relying on the publisher's own
+	// shutSig alone is racy on the streaming path.
+	stopping atomic.Bool
+	// closed marks the batcher as torn down (guarded by batcherMu): the
+	// flush loop's deferred batcher.Close races in-flight Publish calls
+	// otherwise, and the batcher is not goroutine-safe.
+	closed bool
+	// poisoned is set when a tracked batch could not be handed to ReadBatch:
+	// its checkpoint slot can never resolve, so this publisher can never
+	// checkpoint past it. Connect rebuilds a poisoned publisher.
+	poisoned atomic.Bool
 
 	// tableSchemas caches the computed common schema for each table. No
 	// invalidation is needed because MSSQL CDC capture instances are immutable:
@@ -42,6 +81,27 @@ 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
+	// 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
+	// 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
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -54,16 +114,145 @@ func newBatchPublisher(batcher *service.Batcher, checkpoint *checkpoint.Capped[r
 		shutSig:      shutdown.NewSignaller(),
 		tableSchemas: make(map[string]any),
 	}
+	b.waiters = make(map[uint64]chan struct{})
+	b.abandoned = make(map[uint64]struct{})
 	go b.loop()
 	return b
 }
 
+// takeTicketLocked hands out the next flush ticket. MUST be called with
+// batcherMu held, atomically with the Flush that produced the batch, so
+// ticket order is exactly flush order.
+func (b *batchPublisher) takeTicketLocked() uint64 {
+	t := b.nextTicket
+	b.nextTicket++
+	return t
+}
+
+// errQueueSealed refuses admission after an abandoned ticket dropped a
+// flushed batch: nothing may be tracked past that gap until Connect rebuilds
+// the poisoned publisher.
+var errQueueSealed = errors.New("publisher flush queue sealed after an abandoned batch; reconnecting rebuilds the publisher")
+
+// admit blocks until it is ticket's turn to Track+send, or ctx is cancelled.
+// On success, pair with release. On cancellation the ticket is marked
+// abandoned - release skips it when its turn comes - and the caller must NOT
+// release it. ownsRows declares whether the ticket holds a non-empty flushed
+// batch: such an abandon seals and poisons IN THE SAME critical section that
+// records the abandonment, because the moment abandoned[ticket] is visible,
+// a release from the previous holder may skip it and admit the next ticket -
+// sealing any later would let that ticket track, deliver, and ack past the
+// dropped rows before the seal lands. Row-less abandons (barrier tickets,
+// window markers) skip benignly.
+func (b *batchPublisher) admit(ctx context.Context, ticket uint64, ownsRows bool) error {
+	b.ticketMu.Lock()
+	if b.sealed {
+		b.ticketMu.Unlock()
+		return errQueueSealed
+	}
+	if b.admitted == ticket {
+		b.ticketMu.Unlock()
+		return nil
+	}
+	ch := make(chan struct{})
+	b.waiters[ticket] = ch
+	b.ticketMu.Unlock()
+
+	wake := func() error {
+		b.ticketMu.Lock()
+		defer b.ticketMu.Unlock()
+		if b.sealed {
+			return errQueueSealed
+		}
+		return nil
+	}
+
+	select {
+	case <-ch:
+		return wake()
+	case <-ctx.Done():
+		b.ticketMu.Lock()
+		select {
+		case <-ch:
+			// Woken between cancellation and the lock: either admitted
+			// normally (caller owns the release) or the queue was sealed.
+			sealed := b.sealed
+			b.ticketMu.Unlock()
+			if sealed {
+				return errQueueSealed
+			}
+			return nil
+		default:
+		}
+		delete(b.waiters, ticket)
+		b.abandoned[ticket] = struct{}{}
+		if ownsRows {
+			b.sealLocked()
+		}
+		b.ticketMu.Unlock()
+		if ownsRows {
+			b.poisoned.Store(true)
+		}
+		return ctx.Err()
+	}
+}
+
+// sealLocked marks the queue sealed and wakes every waiter (they observe the
+// seal and refuse). Caller must hold ticketMu.
+func (b *batchPublisher) sealLocked() {
+	b.sealed = true
+	for t, ch := range b.waiters {
+		close(ch)
+		delete(b.waiters, t)
+	}
+}
+
+// sealQueue permanently refuses further admissions and poisons the publisher:
+// called when flushed-but-untracked rows were dropped (a failed Flush or
+// trackBatch), so no later batch can be tracked (and therefore no ack can
+// persist a position) past the dropped rows before Connect rebuilds. Safe to
+// call while holding batcherMu: the established order is batcherMu before
+// ticketMu, never the reverse.
+func (b *batchPublisher) sealQueue() {
+	b.ticketMu.Lock()
+	b.sealLocked()
+	b.ticketMu.Unlock()
+	b.poisoned.Store(true)
+}
+
+// release passes the sequence to the next live ticket, skipping abandoned
+// ones. Every ADMITTED ticket must be released exactly once, error paths
+// included, or the sequence wedges.
+func (b *batchPublisher) release() {
+	b.ticketMu.Lock()
+	b.admitted++
+	for {
+		if _, ok := b.abandoned[b.admitted]; !ok {
+			break
+		}
+		delete(b.abandoned, b.admitted)
+		b.admitted++
+	}
+	if ch, ok := b.waiters[b.admitted]; ok {
+		close(ch)
+		delete(b.waiters, b.admitted)
+	}
+	b.ticketMu.Unlock()
+}
+
 // loop creates a long-running process that periodically flushes batches by configured interval.
 // lifted from internal/impl/kafka/franz_reader_ordered.go.
 func (p *batchPublisher) loop() {
 	defer func() {
 		if p.batcher != nil {
-			p.batcher.Close(context.Background())
+			// The batcher is not goroutine-safe and in-flight Publish calls
+			// may still be mutating it under batcherMu when a shutdown stops
+			// this loop: close it under the same lock, and mark it closed so
+			// later flush paths refuse instead of touching a closed batcher.
+			p.batcherMu.Lock()
+			p.closed = true
+			_ = p.batcher.Close(context.Background())
+			p.batcherMu.Unlock()
 		}
 		p.shutSig.TriggerHasStopped()
 	}()
@@ -80,7 +269,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()
@@ -104,30 +297,61 @@ func (p *batchPublisher) loop() {
 		adjustTimedFlush()
 		select {
 		case <-flushBatch:
-			var sendBatch service.MessageBatch
-
-			// Wrap this in a closure to make locking/unlocking easier.
-			func() {
+			flushBatch = nil
+			if err := func() error {
 				p.batcherMu.Lock()
-				defer p.batcherMu.Unlock()
-
-				flushBatch = nil
 				if tNext, exists := p.batcher.UntilNext(); !exists || tNext > 1 {
 					// This can happen if a pushed message triggered a batch before
 					// the last known flush period. In this case we simply enter the
 					// loop again which readjusts our flush batch timer.
-					return
+					p.batcherMu.Unlock()
+					return nil
 				}
-
-				if sendBatch, _ = p.batcher.Flush(closeAtLeisureCtx); len(sendBatch) == 0 {
-					return
+				sendBatch, flushErr := p.batcher.Flush(closeAtLeisureCtx)
+				var (
+					checkpointLSN []byte
+					ticket        uint64
+				)
+				if flushErr == nil && len(sendBatch) > 0 {
+					p.buffered = 0
+					checkpointLSN = []byte(p.pendingCheckpointLSN)
+					ticket = p.takeTicketLocked()
+				}
+				if flushErr != nil {
+					// Defensive: the current benthos Batcher.Flush never
+					// assigns its error return (processor failures surface as
+					// errored messages), so this branch is unreachable today -
+					// but the signature declares the error, and if a future
+					// version does fail here the drained rows were never
+					// tracked. Seal BEFORE releasing batcherMu: in the gap
+					// after the unlock another flusher could take the next
+					// ticket and be admitted past the dropped rows.
+					p.sealQueue()
+				}
+				p.batcherMu.Unlock()
+				if flushErr != nil {
+					p.log.Errorf("Flushing timed batch failed; the publisher is marked for rebuild and its rows re-read from the last durable LSN on reconnect: %v", flushErr)
+					return flushErr
+				}
+				if len(sendBatch) == 0 {
+					return nil
 				}
-			}()
 
-			if len(sendBatch) > 0 {
-				if err := p.publishBatch(closeAtLeisureCtx, sendBatch); err != nil {
-					return
+				if err := p.admit(closeAtLeisureCtx, ticket, true); err != nil {
+					return err
+				}
+				defer p.release()
+				tracked, err := p.trackBatch(closeAtLeisureCtx, sendBatch, checkpointLSN)
+				if err != nil {
+					// The rows left the batcher but were never tracked, and
+					// the deferred release lets later tickets proceed: seal so
+					// nothing can be tracked (and persisted) past the gap.
+					p.sealQueue()
+					return err
 				}
+				return p.sendTracked(closeAtLeisureCtx, tracked)
+			}(); err != nil {
+				return
 			}
 		case <-p.shutSig.SoftStopChan():
 			return
@@ -176,60 +400,290 @@ func (b *batchPublisher) Publish(ctx context.Context, m replication.MessageEvent
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: s})
 	}
 
-	var flushedBatch []*service.Message
+	// Add and Flush are atomic under batcherMu so the user's batching policy
+	// stays exact, and the flush ticket taken in the same critical section
+	// pins this batch's position in the checkpoint sequence. Track+send then
+	// run outside batcherMu in ticket order: a Track blocked on
+	// checkpoint_limit stalls only the ticket queue, never concurrent
+	// buffering or the timed-flush ticker.
+	var (
+		flushedBatch  service.MessageBatch
+		checkpointLSN []byte
+		ticket        uint64
+	)
 	b.batcherMu.Lock()
+	if b.closed {
+		b.batcherMu.Unlock()
+		return context.Canceled
+	}
+	b.pendingCheckpointLSN = m.CheckpointLSN
 	if b.batcher.Add(msg) {
-		flushedBatch, err = b.batcher.Flush(ctx)
+		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			b.buffered = 0
+			checkpointLSN = []byte(b.pendingCheckpointLSN)
+			ticket = b.takeTicketLocked()
+		}
+	} else {
+		b.buffered++
+	}
+	if err != nil {
+		// The failed Flush drained rows that were never tracked. Seal BEFORE
+		// releasing batcherMu: in the gap after the unlock another flusher
+		// could flush, take the next ticket, and be admitted past the
+		// dropped rows.
+		b.sealQueue()
 	}
 	b.batcherMu.Unlock()
 	if err != nil {
 		return fmt.Errorf("flushing batch due to reaching count limit: %w", err)
 	}
-
-	// If a batch was flushed, publish it outside the lock
-	if len(flushedBatch) > 0 {
-		if err := b.publishBatch(ctx, flushedBatch); err != nil {
-			return fmt.Errorf("publishing flushed batch: %w", err)
-		}
+	if len(flushedBatch) == 0 {
+		return nil
 	}
 
+	if err := b.admit(ctx, ticket, true); err != nil {
+		return err
+	}
+	defer b.release()
+	tracked, err := b.trackBatch(ctx, flushedBatch, checkpointLSN)
+	if err != nil {
+		// The rows left the batcher but were never tracked, and the deferred
+		// release lets later tickets proceed: seal so nothing can be tracked
+		// (and persisted) past the gap.
+		b.sealQueue()
+		return err
+	}
+	if err := b.sendTracked(ctx, tracked); err != nil {
+		return fmt.Errorf("publishing flushed batch: %w", err)
+	}
 	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 {
+	msgs       asyncMessage
+	isSnapshot bool
+}
 
+// trackBatch registers the batch with the ordered checkpoint tracker and
+// builds its ack function. It MUST be called by the admitted ticket holder:
+// Track order defines the checkpoint sequence, so it has to match flush
+// (ticket) order exactly. Track may block on checkpoint_limit, which is why
+// batcherMu must NOT be held here. checkpointLSN is the pendingCheckpointLSN captured under
+// batcherMu at flush time: the last transaction whose rows are all published
+// (a 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).
+func (b *batchPublisher) trackBatch(ctx context.Context, batch service.MessageBatch, checkpointLSN []byte) (*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 {
-		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)
-	}
-	msg := asyncMessage{
-		msg: batch,
-		ackFn: func(ctx context.Context, _ error) error {
-			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,
+		msgs: asyncMessage{
+			msg: batch,
+			// Nacks resolve like acks: they 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. The drop is logged - it is the one
+			// place rows become unrecoverable by design.
+			ackFn: func(ctx context.Context, ackErr error) error {
+				if isSnapshotBatch {
+					defer b.snapshotAckWG.Done()
+				}
+				if ackErr != nil {
+					b.log.Warnf("Dropping batch of %d messages rejected downstream (snapshot=%v, checkpoint LSN %X): auto_replay_nacks is disabled, so the checkpoint advances past the dropped rows: %v", len(batch), isSnapshotBatch, checkpointLSN, ackErr)
+				}
+				b.persistMu.Lock()
+				defer b.persistMu.Unlock()
+				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 by the
+// admitted ticket holder, never under batcherMu: the send blocks until
+// consumed. A failed send releases the batch's snapshot-gate slot and poisons
+// the publisher.
+func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch) error {
+	select {
+	case b.msgChan <- tracked.msgs:
+		return nil
+	case <-ctx.Done():
+		if tracked.isSnapshot {
+			b.snapshotAckWG.Done()
+		}
+		// The batch's checkpoint slot is registered but its ackFn will never
+		// run, so the tracker is permanently pinned before this batch: mark
+		// the publisher poisoned so Connect rebuilds it with a fresh tracker.
+		// Resolving the slot here instead would be unsafe - another flusher
+		// may already have delivered a later-tracked batch, and its ack would
+		// then persist an LSN past these undelivered rows.
+		if b.stopping.Load() || b.shutSig.IsSoftStopSignalled() {
+			// Expected on a graceful stop: nothing drains msgChan once
+			// ReadBatch stops, and Close cancels this send. Not a fault.
+			b.log.Debugf("Batch of %d messages undelivered at shutdown; its rows re-read from the last durable LSN on the next run", len(tracked.msgs.msg))
+		} else {
+			b.log.Warnf("Batch of %d messages could not be handed to the pipeline; the publisher is marked for rebuild and its rows re-read from the last durable LSN on reconnect", len(tracked.msgs.msg))
+		}
+		b.poisoned.Store(true)
+		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 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() {
+		// May outlive this call if ctx fires first; bounded by process lifetime.
+		b.snapshotAckWG.Wait()
+		close(drained)
+	}()
 	select {
-	case b.msgChan <- msg:
+	case <-drained:
 		return nil
 	case <-ctx.Done():
 		return ctx.Err()
 	}
 }
 
+// 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 {
+	b.batcherMu.Lock()
+	if b.closed {
+		b.batcherMu.Unlock()
+		return context.Canceled
+	}
+	if b.buffered > 0 {
+		b.pendingCheckpointLSN = lsn
+		b.batcherMu.Unlock()
+		return nil
+	}
+	ticket := b.takeTicketLocked()
+	b.batcherMu.Unlock()
+
+	// The marker joins the checkpoint sequence like any flush: it takes a
+	// ticket so no later flush can Track ahead of it, and Track runs outside
+	// batcherMu (it may block on checkpoint_limit).
+	// The marker owns no rows: an abandoned marker drops nothing, so no
+	// seal is needed - the next drained window re-marks.
+	if err := b.admit(ctx, ticket, false); err != nil {
+		return err
+	}
+	defer b.release()
+	resolveFn, err := b.checkpoint.Track(ctx, lsn, 1)
+	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.
+	b.persistMu.Lock()
+	defer b.persistMu.Unlock()
+	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
+// 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()
+	if b.closed {
+		b.batcherMu.Unlock()
+		return context.Canceled
+	}
+	remaining, err := b.batcher.Flush(ctx)
+	var checkpointLSN []byte
+	if err == nil && len(remaining) > 0 {
+		b.buffered = 0
+		checkpointLSN = []byte(b.pendingCheckpointLSN)
+	}
+	// The ticket is taken unconditionally - even when the batcher is empty -
+	// so that admission below doubles as a sequence barrier: another flusher
+	// (the timed loop) may already hold the final snapshot rows while parked
+	// in checkpoint.Track, before it has counted them on the snapshot ack
+	// gate. Being admitted proves every earlier flush has finished
+	// trackBatch+send, so once flushCurrent returns the gate counts every
+	// published snapshot batch and waitSnapshotAcks cannot release early.
+	ticket := b.takeTicketLocked()
+	if err != nil {
+		// The failed Flush may have drained rows that were never tracked.
+		// Seal BEFORE releasing batcherMu: in the gap after the unlock
+		// another flusher could flush, take the next ticket, and be admitted
+		// past the dropped rows.
+		b.sealQueue()
+	}
+	b.batcherMu.Unlock()
+	if err != nil {
+		// The seal is already applied under batcherMu; return the real flush
+		// error rather than letting admit's sealed refusal mask it (the
+		// operator needs the batching.processors failure, not the seal).
+		return err
+	}
+	if admitErr := b.admit(ctx, ticket, len(remaining) > 0); admitErr != nil {
+		return admitErr
+	}
+	defer b.release()
+	if len(remaining) == 0 {
+		return nil
+	}
+	tracked, err := b.trackBatch(ctx, remaining, checkpointLSN)
+	if err != nil {
+		// Same gap as above: flushed but untracked.
+		b.sealQueue()
+		return err
+	}
+	return b.sendTracked(ctx, tracked)
+}
+
 func (b *batchPublisher) msgs() <-chan asyncMessage {
 	return b.msgChan
 }
+
+// close stops the publisher's flush-loop goroutine and waits for it to exit.
+// Used before a poisoned publisher is replaced; in-flight ack functions keep
+// working against the abandoned tracker.
+func (b *batchPublisher) close() {
+	b.shutSig.TriggerSoftStop()
+	<-b.shutSig.HasStoppedChan()
+}
diff --git a/internal/impl/mssqlserver/batcher_test.go b/internal/impl/mssqlserver/batcher_test.go
new file mode 100644
index 0000000000..1c275c625a
--- /dev/null
+++ b/internal/impl/mssqlserver/batcher_test.go
@@ -0,0 +1,876 @@
+// 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"
+	"fmt"
+	"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, snapshotEvent())
+
+		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, snapshotEvent())
+		// 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) {
+		ctx := t.Context()
+		publisher, _ := newTestBatchPublisher(t)
+
+		// Published but never acked: must not block the gate.
+		publishAndReceive(t, ctx, publisher, streamingEvent("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, snapshotEvent())
+
+		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()
+	// 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, snapshotEvent()))
+	}
+	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)
+		}
+	}
+
+	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 TestCheckpointSelection(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)
+
+		am := publishAndReceive(t, ctx, publisher, streamingEvent("00000042", "00000041"))
+		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 boundary yet (first transaction) persists nothing", 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(),
+			"a batch ending mid-transaction (no prior complete transaction) must not persist any LSN")
+	})
+
+	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"))
+
+		// 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))
+
+		lsns := cachedLSNs()
+		require.NotEmpty(t, lsns, "the checkpoint must continue advancing past a dropped batch")
+		require.Equal(t, "00000042", string(lsns[len(lsns)-1]))
+	})
+}
+
+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 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")))
+
+		// 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]))
+	})
+}
+
+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")
+}
+
+// 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 {
+		// %08d keeps lexicographic order == numeric order, like real LSNs.
+		lsn := fmt.Sprintf("%08d", i)
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(lsn, lsn)))
+	}
+	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)
+	}
+}
+
+// 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)
+	}
+}
+
+// TestPublishBuffersWhileTrackBlocked verifies that a flusher blocked in
+// checkpoint.Track (checkpoint_limit reached, nothing acked) waits only in
+// the ticket queue: other Publish calls must still be able to buffer rows
+// instead of freezing on batcherMu behind the blocked Track.
+func TestPublishBuffersWhileTrackBlocked(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	// Capacity for exactly one 2-row batch: the second flush blocks in Track.
+	cp := checkpoint.NewCapped[replication.LSN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	// Batch 1 fills the tracker to capacity; consume it but do not ack. The
+	// flushing Publish blocks on the unbuffered channel send until the batch
+	// is consumed, so it runs on its own goroutine.
+	require.NoError(t, publisher.Publish(ctx, streamingEvent("00000001", "00000001")))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, streamingEvent("00000002", "00000002")) }()
+	first := <-publisher.msgs()
+	require.NoError(t, <-firstPublished)
+
+	// Batch 2 flushes and blocks in Track (capacity exhausted).
+	blocked := make(chan error, 1)
+	go func() {
+		blocked <- func() error {
+			if err := publisher.Publish(ctx, streamingEvent("00000003", "00000003")); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent("00000004", "00000004"))
+		}()
+	}()
+
+	// Give the flusher time to reach Track and park there.
+	time.Sleep(100 * time.Millisecond)
+	select {
+	case err := <-blocked:
+		t.Fatalf("expected the second batch's flusher to block in Track, but it returned: %v", err)
+	default:
+	}
+
+	// The key assertion: a concurrent Publish that only buffers (no flush due)
+	// completes promptly even though a flusher is parked in Track.
+	buffered := make(chan error, 1)
+	go func() { buffered <- publisher.Publish(ctx, streamingEvent("00000005", "00000005")) }()
+	select {
+	case err := <-buffered:
+		require.NoError(t, err)
+	case <-time.After(2 * time.Second):
+		t.Fatal("a buffering Publish froze behind a Track blocked on checkpoint_limit")
+	}
+
+	// Ack batch 1 to release the blocked flusher and drain.
+	require.NoError(t, first.ackFn(ctx, nil))
+	go func() {
+		for m := range publisher.msgs() {
+			_ = m.ackFn(ctx, nil)
+		}
+	}()
+	select {
+	case err := <-blocked:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("blocked flusher never released after the ack freed tracker capacity")
+	}
+}
+
+// TestFlushCurrentBarriersParkedFlusher encodes the snapshot-handoff crash
+// window: the timed-flush loop can hold the final snapshot rows while parked
+// in checkpoint.Track (before counting them on the snapshot ack gate). The
+// handoff's flushCurrent must not return - and waitSnapshotAcks must not
+// release - until that parked flusher has registered and delivered its batch,
+// otherwise the post-snapshot LSN persists ahead of undelivered rows.
+func TestFlushCurrentBarriersParkedFlusher(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	// Capacity for exactly one 2-row batch: the second flush parks in Track.
+	cp := checkpoint.NewCapped[replication.LSN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	// Snapshot batch 1 fills the tracker; consume it but do not ack (gate=1).
+	require.NoError(t, publisher.Publish(ctx, snapshotEvent()))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, snapshotEvent()) }()
+	first := <-publisher.msgs()
+	require.NoError(t, <-firstPublished)
+
+	// Snapshot batch 2 flushes, takes its ticket, and parks in Track - the
+	// exact state the timed-flush loop can be in at the handoff.
+	parked := make(chan error, 1)
+	go func() {
+		parked <- func() error {
+			if err := publisher.Publish(ctx, snapshotEvent()); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, snapshotEvent())
+		}()
+	}()
+	time.Sleep(100 * time.Millisecond)
+
+	// The handoff: flushCurrent sees an empty batcher but must still barrier
+	// behind the parked flusher's ticket.
+	flushed := make(chan error, 1)
+	go func() { flushed <- publisher.flushCurrent(ctx) }()
+	select {
+	case err := <-flushed:
+		t.Fatalf("flushCurrent returned (%v) while a flusher holding snapshot rows was still parked in Track: the ack gate does not yet count those rows", err)
+	case <-time.After(300 * time.Millisecond):
+	}
+
+	// Ack batch 1: the parked flusher tracks, counts, and delivers batch 2.
+	require.NoError(t, first.ackFn(ctx, nil))
+	second := <-publisher.msgs()
+	require.NoError(t, <-parked)
+	require.NoError(t, <-flushed)
+
+	// The gate must now hold for batch 2: it is published but un-acked.
+	gate := make(chan error, 1)
+	go func() { gate <- publisher.waitSnapshotAcks(ctx) }()
+	select {
+	case err := <-gate:
+		t.Fatalf("waitSnapshotAcks returned (%v) with a published snapshot batch still un-acked", err)
+	case <-time.After(300 * time.Millisecond):
+	}
+	require.NoError(t, second.ackFn(ctx, nil))
+	select {
+	case err := <-gate:
+		require.NoError(t, err)
+	case <-time.After(5 * time.Second):
+		t.Fatal("gate never released after the final ack")
+	}
+}
+
+// TestShutdownUnwindsWedgedTicketChain encodes the shutdown wedge: the timed
+// flush loop parks in sendTracked holding its ticket (nothing drains msgChan
+// once ReadBatch stops), and another flusher waits in admit() with no escape
+// of its own. Triggering the publisher's soft stop - which the input's Close
+// now does - must release the loop's ticket and let the chain drain via each
+// caller's cancelled context, instead of leaking both goroutines past the
+// shutdown timeout.
+func TestShutdownUnwindsWedgedTicketChain(t *testing.T) {
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	// Count 2 with a tiny period keeps the timed loop flushing.
+	batcher, err := (service.BatchPolicy{Count: 2, Period: "1ms"}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+
+	producerCtx, cancelProducer := context.WithCancel(t.Context())
+
+	// One buffered event: the timed loop flushes it and parks in sendTracked
+	// (ticket 0) - nobody consumes msgs().
+	require.NoError(t, publisher.Publish(producerCtx, streamingEvent("00000001", "00000001")))
+	require.Eventually(t, func() bool {
+		publisher.batcherMu.Lock()
+		defer publisher.batcherMu.Unlock()
+		return publisher.nextTicket == 1
+	}, 5*time.Second, time.Millisecond, "the timed loop never flushed the first batch")
+
+	// A second flusher takes ticket 1 and wedges in admit behind the loop.
+	blocked := make(chan error, 1)
+	go func() {
+		blocked <- func() error {
+			if err := publisher.Publish(producerCtx, streamingEvent("00000002", "00000002")); err != nil {
+				return err
+			}
+			return publisher.Publish(producerCtx, streamingEvent("00000003", "00000003"))
+		}()
+	}()
+	time.Sleep(100 * time.Millisecond)
+	select {
+	case err := <-blocked:
+		t.Fatalf("expected the second flusher to wedge in admit, but it returned: %v", err)
+	default:
+	}
+
+	// Shutdown as the input's Close performs it: publisher soft stop plus
+	// cancellation of the producers' contexts.
+	publisher.shutSig.TriggerSoftStop()
+	cancelProducer()
+
+	select {
+	case <-blocked:
+	case <-time.After(5 * time.Second):
+		t.Fatal("the wedged flusher never unwound after shutdown")
+	}
+	select {
+	case <-publisher.shutSig.HasStoppedChan():
+	case <-time.After(5 * time.Second):
+		t.Fatal("the flush loop never stopped after shutdown")
+	}
+}
+
+// TestAdmitEscapesOnContextCancel verifies admission is cancellable: a
+// flusher queued in admit behind a ticket whose holder is parked under a
+// DIFFERENT, still-live context must unwind via its own context - the
+// abandoned ticket is skipped when its turn comes, and the sequence stays
+// intact for later flushers.
+func TestAdmitEscapesOnContextCancel(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	// Ticket 0's holder parks in sendTracked under a live context (nobody
+	// consumes msgs()) - the state the timed loop is in at a soft stop.
+	holderDone := make(chan error, 1)
+	go func() {
+		holderDone <- func() error {
+			if err := publisher.Publish(ctx, streamingEvent("00000001", "00000001")); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent("00000002", "00000002"))
+		}()
+	}()
+	require.Eventually(t, func() bool {
+		publisher.batcherMu.Lock()
+		defer publisher.batcherMu.Unlock()
+		return publisher.nextTicket == 1
+	}, 5*time.Second, time.Millisecond)
+
+	// A flusher with a cancellable context queues behind it (ticket 1).
+	flusherCtx, cancelFlusher := context.WithCancel(ctx)
+	queued := make(chan error, 1)
+	go func() { queued <- publisher.flushCurrent(flusherCtx) }()
+	time.Sleep(100 * time.Millisecond)
+	select {
+	case err := <-queued:
+		t.Fatalf("expected the flusher to queue in admit, but it returned: %v", err)
+	default:
+	}
+
+	// Cancelling ONLY the flusher's context must unwind it promptly, without
+	// touching the parked ticket holder.
+	cancelFlusher()
+	select {
+	case err := <-queued:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(2 * time.Second):
+		t.Fatal("admit did not escape on context cancellation")
+	}
+
+	// The sequence must remain intact: drain the parked holder and prove a
+	// later flusher still gets admitted (the abandoned ticket is skipped).
+	first := <-publisher.msgs()
+	require.NoError(t, first.ackFn(ctx, nil))
+	require.NoError(t, <-holderDone)
+
+	done := make(chan error, 1)
+	go func() {
+		done <- func() error {
+			if err := publisher.Publish(ctx, streamingEvent("00000003", "00000003")); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent("00000004", "00000004"))
+		}()
+	}()
+	select {
+	case m := <-publisher.msgs():
+		require.NoError(t, m.ackFn(ctx, nil))
+	case <-time.After(5 * time.Second):
+		t.Fatal("a later flusher was never admitted: the abandoned ticket wedged the sequence")
+	}
+	require.NoError(t, <-done)
+}
+
+// TestAbandonedBatchSealsQueue encodes the abandonment loss window: a flusher
+// whose rows are already out of the batcher abandons its ticket on
+// cancellation. Nothing pins the tracker for those rows, so if any LATER
+// batch could still be tracked and acked, its resolve would persist an LSN
+// past the dropped rows - silent loss on restart. The abandon must therefore
+// seal the queue (no later admission can ever track) and poison the
+// publisher so Connect rebuilds and re-reads from the last durable LSN.
+func TestAbandonedBatchSealsQueue(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.LSN](100)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	// Ticket 0's holder parks in sendTracked under a live context.
+	holderDone := make(chan error, 1)
+	go func() {
+		holderDone <- func() error {
+			if err := publisher.Publish(ctx, streamingEvent("00000001", "00000001")); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent("00000002", "00000002"))
+		}()
+	}()
+	require.Eventually(t, func() bool {
+		publisher.batcherMu.Lock()
+		defer publisher.batcherMu.Unlock()
+		return publisher.nextTicket == 1
+	}, 5*time.Second, time.Millisecond)
+
+	// A flusher with ROWS (ticket 1) queues behind it and is cancelled: its
+	// batch left the batcher but was never tracked.
+	abandonCtx, cancelAbandon := context.WithCancel(ctx)
+	abandoned := make(chan error, 1)
+	go func() {
+		abandoned <- func() error {
+			if err := publisher.Publish(abandonCtx, streamingEvent("00000003", "00000003")); err != nil {
+				return err
+			}
+			return publisher.Publish(abandonCtx, streamingEvent("00000004", "00000004"))
+		}()
+	}()
+	time.Sleep(100 * time.Millisecond)
+	cancelAbandon()
+	require.ErrorIs(t, <-abandoned, context.Canceled)
+
+	// The abandon dropped rows: the queue must be sealed and the publisher
+	// poisoned so nothing can ever be tracked (and persisted) past them from
+	// this generation.
+	require.True(t, publisher.poisoned.Load(),
+		"abandoning a flushed-but-untracked batch must poison the publisher")
+	laterErr := func() error {
+		if err := publisher.Publish(ctx, streamingEvent("00000005", "00000005")); err != nil {
+			return err
+		}
+		return publisher.Publish(ctx, streamingEvent("00000006", "00000006"))
+	}()
+	require.ErrorIs(t, laterErr, errQueueSealed,
+		"a later flusher must be refused: tracking past the dropped rows would let its ack persist an LSN that skips them")
+}
+
+// TestTrackFailureSealsQueue is TestAbandonedBatchSealsQueue's sibling: an
+// ADMITTED flusher parked in checkpoint.Track (capacity exhausted) whose
+// context cancels returns with its rows flushed but untracked, while the
+// deferred release advances the queue - the same unpinned gap. The failure
+// must seal and poison so no later ticket can persist past the dropped rows.
+func TestTrackFailureSealsQueue(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	// Capacity for exactly one 2-row batch: the next Track parks.
+	cp := checkpoint.NewCapped[replication.LSN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheLSN = func(context.Context, replication.LSN) error { return nil }
+	t.Cleanup(func() { publisher.shutSig.TriggerSoftStop() })
+
+	// Batch 1 fills the tracker; consume it but do not ack.
+	require.NoError(t, publisher.Publish(ctx, streamingEvent("00000001", "00000001")))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, streamingEvent("00000002", "00000002")) }()
+	<-publisher.msgs()
+	require.NoError(t, <-firstPublished)
+
+	// Batch 2 is admitted and parks in Track; cancelling its context fails
+	// the Track with the rows already out of the batcher.
+	trackCtx, cancelTrack := context.WithCancel(ctx)
+	parked := make(chan error, 1)
+	go func() {
+		parked <- func() error {
+			if err := publisher.Publish(trackCtx, streamingEvent("00000003", "00000003")); err != nil {
+				return err
+			}
+			return publisher.Publish(trackCtx, streamingEvent("00000004", "00000004"))
+		}()
+	}()
+	time.Sleep(100 * time.Millisecond)
+	cancelTrack()
+	require.Error(t, <-parked)
+
+	require.True(t, publisher.poisoned.Load(),
+		"a failed Track stranded flushed-but-untracked rows; the publisher must be marked for rebuild")
+	laterErr := func() error {
+		if err := publisher.Publish(ctx, streamingEvent("00000005", "00000005")); err != nil {
+			return err
+		}
+		return publisher.Publish(ctx, streamingEvent("00000006", "00000006"))
+	}()
+	require.ErrorIs(t, laterErr, errQueueSealed,
+		"a later flusher must be refused: tracking past the dropped rows would let its ack persist an LSN that skips them")
+}
+
+// TestFailedSendPoisonsPublisher verifies that a tracked batch that cannot be
+// handed to ReadBatch marks the publisher poisoned: its checkpoint slot can
+// never resolve, so Connect must rebuild the publisher rather than reuse a
+// permanently pinned tracker.
+func TestFailedSendPoisonsPublisher(t *testing.T) {
+	publisher, _ := newTestBatchPublisher(t)
+
+	sendCtx, cancel := context.WithCancel(t.Context())
+	cancel() // nobody consumes msgs(): the send can only fail
+
+	err := publisher.Publish(sendCtx, streamingEvent("00000001", "00000001"))
+	require.ErrorIs(t, err, context.Canceled)
+	require.True(t, publisher.poisoned.Load(),
+		"a failed send orphans its tracker slot; the publisher must be marked for rebuild")
+}
+
+// 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)
+
+	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 (
+		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 snapshotEvent() replication.MessageEvent {
+	return replication.MessageEvent{
+		Schema:    "dbo",
+		Table:     "t",
+		Operation: replication.MessageOperationRead.String(),
+		Data:      map[string]any{"a": 1},
+	}
+}
+
+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},
+	}
+}
+
+// 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.Publish(ctx, event)
+	}()
+	return <-publisher.msgs()
+}
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc.go b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
index 4b837c7f9b..8fd6b2315a 100644
--- a/internal/impl/mssqlserver/input_mssqlserver_cdc.go
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc.go
@@ -9,12 +9,14 @@
 package mssqlserver
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"errors"
 	"fmt"
 	"regexp"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -152,14 +154,29 @@ type sqlServerCDCInput struct {
 	cfg *config
 	db  *sql.DB
 
-	res       *service.Resources
-	publisher *batchPublisher
+	res *service.Resources
+	// publisher is rebuilt by Connect when poisoned, and read by ReadBatch
+	// and Close on other goroutines: atomic so those reads can never observe
+	// a torn or stale pointer and Close always stops the CURRENT publisher.
+	publisher atomic.Pointer[batchPublisher]
 	metrics   *service.Metrics
 
 	connMu  sync.Mutex
 	stopSig *shutdown.Signaller
 	log     *service.Logger
 	cpCache service.Cache
+
+	// batching and checkpointLimit are retained so Connect can rebuild a
+	// poisoned publisher (see batchPublisher.poisoned).
+	batching        service.BatchPolicy
+	checkpointLimit int
+
+	// lastPersistedMu serializes cacheLSN writes across publisher generations
+	// and lastPersistedLSN keeps them monotonic: after a rebuild a previous
+	// session's late acks may still arrive, and a stale write must never
+	// regress the durable resume position.
+	lastPersistedMu  sync.Mutex
+	lastPersistedLSN replication.LSN
 }
 
 func newMSSQLServerCDCInput(conf *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {
@@ -266,15 +283,18 @@ 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(),
+		cpCache:         cpCache,
+		batching:        policy,
+		checkpointLimit: checkpointLimit,
 	}
 
-	i.publisher.cacheLSN = i.cacheLSN
+	pub := newBatchPublisher(batcher, cp, logger)
+	pub.cacheLSN = i.cacheLSN
+	i.publisher.Store(pub)
 
 	// Has stopped is how we notify that we're not connected. This will get reset at connection time.
 	i.stopSig.TriggerHasStopped()
@@ -287,6 +307,29 @@ func newMSSQLServerCDCInput(conf *service.ParsedConfig, resources *service.Resou
 	return conf.WrapBatchInputExtractTracingSpanMapping("microsoft_sql_server_cdc", batchInput)
 }
 
+// rebuildPublisherIfPoisoned returns the current publisher, replacing it
+// first when a failed send or a sealed flush queue poisoned it: the old
+// generation is closed (its flush loop stops; in-flight ack functions keep
+// resolving into the abandoned tracker, where cacheLSN's monotonic guard
+// makes any stale persist a no-op) and a fresh batcher and tracker take its
+// place, so the new session resumes from the last durable LSN.
+func (i *sqlServerCDCInput) rebuildPublisherIfPoisoned() (*batchPublisher, error) {
+	publisher := i.publisher.Load()
+	if !publisher.poisoned.Load() {
+		return publisher, nil
+	}
+	i.log.Warn("Rebuilding publisher: a batch could not be handed to the pipeline, so the previous checkpoint tracker is pinned")
+	publisher.close()
+	batcher, err := i.batching.NewBatcher(i.res)
+	if err != nil {
+		return nil, fmt.Errorf("rebuilding batcher: %w", err)
+	}
+	publisher = newBatchPublisher(batcher, checkpoint.NewCapped[replication.LSN](int64(i.checkpointLimit)), i.log)
+	publisher.cacheLSN = i.cacheLSN
+	i.publisher.Store(publisher)
+	return publisher, nil
+}
+
 func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 	i.connMu.Lock()
 	defer i.connMu.Unlock()
@@ -300,8 +343,18 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 		return nil
 	}
 
+	// A failed batch send leaves an unresolvable slot in the ordered tracker
+	// (see sendTracked), so a poisoned publisher can never checkpoint again.
+	// Rebuild it with a fresh tracker: the new session resumes from the last
+	// durable LSN, which is necessarily before the orphaned rows, and the old
+	// session's late acks resolve into the abandoned tracker (cacheLSN's
+	// monotonic guard turns any stale write into a no-op).
+	publisher, err := i.rebuildPublisherIfPoisoned()
+	if err != nil {
+		return err
+	}
+
 	var (
-		err        error
 		userTables []replication.UserDefinedTable
 		cachedLSN  replication.LSN
 	)
@@ -336,14 +389,14 @@ func (i *sqlServerCDCInput) Connect(ctx context.Context) error {
 	)
 	// no cached LSN means we're not recovering from a restart
 	if i.cfg.streamSnapshot && len(cachedLSN) == 0 {
-		if snapshotter, err = replication.NewSnapshot(i.cfg.connectionString, userTables, i.publisher, i.log, i.metrics); err != nil {
+		if snapshotter, err = replication.NewSnapshot(i.cfg.connectionString, userTables, publisher, i.log, i.metrics); err != nil {
 			return fmt.Errorf("creating database snapshotter: %w", err)
 		}
 	} else {
 		i.log.Infof("Snapshotting disabled, skipping...")
 	}
 
-	streaming = replication.NewChangeTableStream(userTables, i.publisher, i.cfg.streamBackoffInterval, i.log)
+	streaming = replication.NewChangeTableStream(userTables, publisher, i.cfg.streamBackoffInterval, i.log)
 
 	// Reset our stop signal
 	i.stopSig = shutdown.NewSignaller()
@@ -366,6 +419,32 @@ 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 = publisher.flushCurrent(softCtx); err != nil {
+				// A graceful stop lands here whenever shutdown hits the
+				// handoff window (nothing drains msgChan any more, so the
+				// blocked send exits via softCtx): normal operation, Info.
+				// Genuine flush failures keep the error level.
+				if errors.Is(err, context.Canceled) && !i.stopSig.IsHardStopSignalled() {
+					i.log.Infof("Interrupted while flushing remaining snapshot batches. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				} else {
+					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 = 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)
@@ -429,6 +508,16 @@ func (i *sqlServerCDCInput) cacheLSN(ctx context.Context, lsn replication.LSN) e
 		return errors.New("LSN for caching is empty")
 	}
 
+	// Serialized and monotonic across publisher generations: a previous
+	// session's late acks must never land a stale LSN over a newer durable
+	// position. LSNs are fixed-width and byte-ordered, so skipping
+	// non-advancing writes is always safe.
+	i.lastPersistedMu.Lock()
+	defer i.lastPersistedMu.Unlock()
+	if len(i.lastPersistedLSN) != 0 && bytes.Compare(lsn, i.lastPersistedLSN) <= 0 {
+		return nil
+	}
+
 	var cErr error
 	if i.cpCache != nil {
 		cErr = i.cpCache.Set(ctx, i.cfg.lsnCacheKey, lsn, nil)
@@ -443,12 +532,13 @@ func (i *sqlServerCDCInput) cacheLSN(ctx context.Context, lsn replication.LSN) e
 	if cErr != nil {
 		return fmt.Errorf("unable persist checkpoint to cache: %w", cErr)
 	}
+	i.lastPersistedLSN = lsn
 	return nil
 }
 
 func (i *sqlServerCDCInput) ReadBatch(ctx context.Context) (service.MessageBatch, service.AckFunc, error) {
 	select {
-	case m := <-i.publisher.msgs():
+	case m := <-i.publisher.Load().msgs():
 		return m.msg, m.ackFn, nil
 	case <-i.stopSig.HasStoppedChan():
 		return nil, nil, service.ErrNotConnected
@@ -482,7 +572,24 @@ func (i *sqlServerCDCInput) Close(ctx context.Context) error {
 	if i.stopSig == nil {
 		return nil // Never connected
 	}
+	// Mark the publisher as stopping BEFORE any cancellation propagates: the
+	// session's contexts unwind off stopSig, and sendTracked needs the flag
+	// already visible to log the graceful unwind at debug rather than warn.
+	if pub := i.publisher.Load(); pub != nil {
+		pub.stopping.Store(true)
+	}
 	i.stopSig.TriggerSoftStop()
+	// Shut the publisher down alongside the session: its timed-flush loop
+	// runs under the publisher's OWN signaller, and a flush parked in
+	// sendTracked (nothing drains msgChan once ReadBatch stops) would
+	// otherwise hold its flush ticket forever - wedging every other flusher
+	// waiting in admit() and leaking the session goroutines past the
+	// timeout. Cancelling the loop's context releases its ticket, and the
+	// chain then drains: each later ticket holder's Track/send escapes via
+	// its stopSig-derived context.
+	if pub := i.publisher.Load(); pub != nil {
+		pub.shutSig.TriggerSoftStop()
+	}
 	select {
 	case <-ctx.Done():
 	case <-time.After(shutdownTimeout):
diff --git a/internal/impl/mssqlserver/input_mssqlserver_cdc_test.go b/internal/impl/mssqlserver/input_mssqlserver_cdc_test.go
new file mode 100644
index 0000000000..7c86f92bf7
--- /dev/null
+++ b/internal/impl/mssqlserver/input_mssqlserver_cdc_test.go
@@ -0,0 +1,138 @@
+// 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"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Jeffail/checkpoint"
+	"github.com/Jeffail/shutdown"
+	"github.com/stretchr/testify/require"
+
+	"github.com/redpanda-data/benthos/v4/public/service"
+
+	"github.com/redpanda-data/connect/v4/internal/impl/mssqlserver/replication"
+)
+
+// recordingCache is a minimal service.Cache capturing Set calls.
+type recordingCache struct {
+	service.Cache
+	mu   sync.Mutex
+	sets [][]byte
+}
+
+func (c *recordingCache) Set(_ context.Context, _ string, value []byte, _ *time.Duration) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.sets = append(c.sets, append([]byte(nil), value...))
+	return nil
+}
+
+func (c *recordingCache) recorded() [][]byte {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return append([][]byte(nil), c.sets...)
+}
+
+func newTestInput(t *testing.T) (*sqlServerCDCInput, *recordingCache) {
+	t.Helper()
+	cache := &recordingCache{}
+	i := &sqlServerCDCInput{
+		cfg:             &config{lsnCacheKey: "lsn"},
+		res:             service.MockResources(),
+		log:             service.NewLoggerFromSlog(slog.Default()),
+		stopSig:         shutdown.NewSignaller(),
+		cpCache:         cache,
+		batching:        service.BatchPolicy{Count: 1},
+		checkpointLimit: 8,
+	}
+	batcher, err := i.batching.NewBatcher(i.res)
+	require.NoError(t, err)
+	pub := newBatchPublisher(batcher, checkpoint.NewCapped[replication.LSN](8), i.log)
+	pub.cacheLSN = i.cacheLSN
+	i.publisher.Store(pub)
+	t.Cleanup(func() { i.publisher.Load().shutSig.TriggerSoftStop() })
+	return i, cache
+}
+
+// TestCacheLSNMonotonicGuard locks in the persist guard: advancing writes
+// land, equal and regressing writes are silently skipped - a stale ack from
+// an abandoned publisher generation must never move the durable resume
+// position backwards.
+func TestCacheLSNMonotonicGuard(t *testing.T) {
+	i, cache := newTestInput(t)
+	ctx := t.Context()
+
+	require.NoError(t, i.cacheLSN(ctx, replication.LSN("00000010")))
+	require.NoError(t, i.cacheLSN(ctx, replication.LSN("00000020")), "an advancing LSN must persist")
+	require.NoError(t, i.cacheLSN(ctx, replication.LSN("00000020")), "an equal LSN is a no-op, not an error")
+	require.NoError(t, i.cacheLSN(ctx, replication.LSN("00000015")), "a regressing LSN is a no-op, not an error")
+
+	got := cache.recorded()
+	require.Len(t, got, 2, "only the two advancing writes may reach the cache")
+	require.Equal(t, "00000010", string(got[0]))
+	require.Equal(t, "00000020", string(got[1]))
+
+	require.Error(t, i.cacheLSN(ctx, nil), "an empty LSN is rejected")
+}
+
+// TestRebuildPublisherIfPoisoned proves the rebuild actually swaps
+// generations: the old publisher is closed, the new one is a distinct
+// publisher with a fresh tracker wired to cacheLSN, and a late ack from the
+// OLD generation cannot regress the durable position past the guard.
+func TestRebuildPublisherIfPoisoned(t *testing.T) {
+	i, cache := newTestInput(t)
+	ctx := t.Context()
+
+	old := i.publisher.Load()
+
+	// Not poisoned: same generation back.
+	same, err := i.rebuildPublisherIfPoisoned()
+	require.NoError(t, err)
+	require.Same(t, old, same)
+
+	// Deliver a batch on the old generation but hold its ack (late ack). The
+	// flushing Publish blocks on the unbuffered channel until consumed.
+	oldPublished := make(chan error, 1)
+	go func() { oldPublished <- old.Publish(ctx, streamingEvent("00000010", "00000010")) }()
+	oldMsg := <-old.msgs()
+	require.NoError(t, <-oldPublished)
+
+	// Poison and rebuild.
+	old.poisoned.Store(true)
+	rebuilt, err := i.rebuildPublisherIfPoisoned()
+	require.NoError(t, err)
+	require.NotSame(t, old, rebuilt, "a poisoned publisher must be replaced")
+	require.Same(t, rebuilt, i.publisher.Load(), "the stored pointer must be the new generation")
+	select {
+	case <-old.shutSig.HasStoppedChan():
+	default:
+		t.Fatal("the old generation's flush loop must be stopped by the rebuild")
+	}
+
+	// The new generation persists progress normally.
+	newPublished := make(chan error, 1)
+	go func() { newPublished <- rebuilt.Publish(ctx, streamingEvent("00000030", "00000030")) }()
+	newMsg := <-rebuilt.msgs()
+	require.NoError(t, <-newPublished)
+	require.NoError(t, newMsg.ackFn(ctx, nil))
+
+	// The old generation's late ack resolves into its abandoned tracker and
+	// must be a no-op on the durable position (monotonic guard).
+	require.NoError(t, oldMsg.ackFn(ctx, nil))
+	got := cache.recorded()
+	require.Equal(t, "00000030", string(got[len(got)-1]), "a late ack from the abandoned generation must not regress the cache")
+	for _, v := range got {
+		require.NotEqual(t, "00000010", string(v), "the stale LSN must never have been persisted after the newer one")
+	}
+}
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/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 39145380e3..1d03720dba 100644
--- a/internal/impl/mssqlserver/replication/stream.go
+++ b/internal/impl/mssqlserver/replication/stream.go
@@ -332,9 +332,37 @@ 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
+	// 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
@@ -365,6 +393,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 +446,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 {
@@ -450,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...")
diff --git a/internal/impl/mssqlserver/replication/stream_message.go b/internal/impl/mssqlserver/replication/stream_message.go
index 27143a3c05..ffab898aed 100644
--- a/internal/impl/mssqlserver/replication/stream_message.go
+++ b/internal/impl/mssqlserver/replication/stream_message.go
@@ -83,11 +83,17 @@ 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"`
-	Schema    string `json:"schema"`
-	Table     string `json:"table"`
-	Data      any    `json:"data"`
+	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"`
 
 	// ColumnNames and ColumnTypes carry user-defined column metadata (excluding
 	// MSSQL system columns with __$ prefix). They are used to build schema
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)
+		}
+	}
+}