diff --git a/internal/impl/oracledb/batcher.go b/internal/impl/oracledb/batcher.go
index bfad80612e..647366c43c 100644
--- a/internal/impl/oracledb/batcher.go
+++ b/internal/impl/oracledb/batcher.go
@@ -11,9 +11,11 @@ package oracledb
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Jeffail/checkpoint"
@@ -26,16 +28,56 @@ 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).
 	batcherMu sync.Mutex
+	// Flush tickets keep the checkpoint sequence exact without a lock held
+	// across Track: each flush 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 send parked under hardStopCtx.
+	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
+	// SCN past them before Connect rebuilds the poisoned publisher.
+	sealed bool
+	// closed marks the batcher as torn down (guarded by batcherMu): Close's
+	// batcher.Close races in-flight Publish calls otherwise, and the batcher
+	// is not goroutine-safe.
+	closed 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) - the publisher's own shutSig is
+	// triggered too late on the streaming path to make that call.
+	stopping atomic.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
 
 	checkpoint *checkpoint.Capped[replication.SCN]
 	msgChan    chan asyncMessage
 	cacheSCN   func(ctx context.Context, scn replication.SCN) error
 	schemas    *schemaCache
 
-	log     *service.Logger
-	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 SCN is never persisted while snapshot rows are in flight.
+	snapshotAckWG sync.WaitGroup
+	log           *service.Logger
+	shutSig       *shutdown.Signaller
 }
 
 // newBatchPublisher creates an instance of batchPublisher.
@@ -47,10 +89,132 @@ func newBatchPublisher(batcher *service.Batcher, checkpoint *checkpoint.Capped[r
 		log:        logger,
 		shutSig:    shutdown.NewSignaller(),
 	}
+	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() {
@@ -68,7 +232,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()
@@ -85,7 +253,7 @@ func (p *batchPublisher) loop() {
 		flushBatch = flushBatchTicker.C
 	}
 
-	// hardStopCtx survives a soft stop so that an in-flight publishBatch send can
+	// hardStopCtx survives a soft stop so that an in-flight sendTracked send can
 	// complete before the loop exits. Only a hard stop (triggered by Close)
 	// cancels it, which is the forced-shutdown last resort.
 	hardStopCtx, done := p.shutSig.HardStopCtx(context.Background())
@@ -95,30 +263,56 @@ 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(hardStopCtx); len(sendBatch) == 0 {
-					return
+				sendBatch, flushErr := p.batcher.Flush(hardStopCtx)
+				var ticket uint64
+				if flushErr == nil && len(sendBatch) > 0 {
+					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 SCN on reconnect: %v", flushErr)
+					return flushErr
+				}
+				if len(sendBatch) == 0 {
+					return nil
 				}
-			}()
 
-			if len(sendBatch) > 0 {
-				if err := p.publishBatch(hardStopCtx, sendBatch); err != nil {
-					return
+				if err := p.admit(hardStopCtx, ticket, true); err != nil {
+					return err
+				}
+				defer p.release()
+				tracked, err := p.trackBatch(hardStopCtx, sendBatch)
+				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(hardStopCtx, tracked)
+			}(); err != nil {
+				return
 			}
 		case <-p.shutSig.SoftStopChan():
 			return
@@ -193,31 +387,72 @@ func (b *batchPublisher) Publish(ctx context.Context, m *replication.MessageEven
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: schemaAny})
 	}
 
-	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
+		ticket       uint64
+	)
 	b.batcherMu.Lock()
+	if b.closed {
+		b.batcherMu.Unlock()
+		return context.Canceled
+	}
 	if b.batcher.Add(msg) {
-		flushedBatch, err = b.batcher.Flush(ctx)
+		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			ticket = b.takeTicketLocked()
+		}
+	}
+	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)
+	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.
+func (b *batchPublisher) trackBatch(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
 
 	// ensure we don't checkpoint snapshot batches
@@ -240,31 +475,92 @@ func (b *batchPublisher) publishBatch(ctx context.Context, batch service.Message
 		var parseErr error
 		checkpointSCN, parseErr = replication.ParseSCN(scn)
 		if parseErr != nil {
-			return fmt.Errorf("parsing checkpoint SCN: %w", parseErr)
+			return nil, fmt.Errorf("parsing checkpoint SCN: %w", parseErr)
 		}
 	}
 
 	resolveFn, err := b.checkpoint.Track(ctx, checkpointSCN, int64(len(batch)))
 	if err != nil {
-		return fmt.Errorf("tracking SCN checkpoint for batch: %w", err)
-	}
-	msg := asyncMessage{
-		msg: batch,
-		ackFn: func(ctx context.Context, _ error) error {
-			scn := resolveFn()
-			if scn == nil || !scn.IsValid() {
-				return nil
-			}
-			if isSnapshotBatch && *scn <= checkpointSCN {
-				// Resolved value is this snapshot batch's own shared SCN (or older) —
-				// nothing new to persist, and persisting it would be premature.
-				return nil
-			}
-			return b.cacheSCN(ctx, *scn)
+		return nil, fmt.Errorf("tracking SCN 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 SCN %s): auto_replay_nacks is disabled, so the checkpoint advances past the dropped rows: %v", len(batch), isSnapshotBatch, checkpointSCN, ackErr)
+				}
+				scn := resolveFn()
+				if scn == nil || !scn.IsValid() {
+					return nil
+				}
+				if isSnapshotBatch && *scn <= checkpointSCN {
+					// Resolved value is this snapshot batch's own shared SCN (or older) —
+					// nothing new to persist, and persisting it would be premature.
+					return nil
+				}
+				return b.cacheSCN(ctx, *scn)
+			},
 		},
+	}, 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 SCN 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 SCN 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 SCN 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()
@@ -288,32 +584,83 @@ func (b *batchPublisher) msgs() <-chan asyncMessage {
 	return b.msgChan
 }
 
-// FlushRemaining stops the loop goroutine and then flushes any partial batch
-// still held in the batcher, blocking until it is consumed by ReadBatch.
-func (b *batchPublisher) FlushRemaining(ctx context.Context) error {
+// 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 SCN is persisted.
+func (b *batchPublisher) flushCurrent(ctx context.Context) error {
 	if b.batcher == nil {
 		return nil
 	}
-	b.shutSig.TriggerSoftStop()
-	<-b.shutSig.HasStoppedChan()
-
 	b.batcherMu.Lock()
+	if b.closed {
+		b.batcherMu.Unlock()
+		return context.Canceled
+	}
 	remaining, err := b.batcher.Flush(ctx)
+	// 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 || len(remaining) == 0 {
+	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)
+	if err != nil {
+		// Same gap as above: flushed but untracked.
+		b.sealQueue()
 		return err
 	}
-	return b.publishBatch(ctx, remaining)
+	return b.sendTracked(ctx, tracked)
+}
+
+// FlushRemaining stops the loop goroutine and then flushes any partial batch
+// still held in the batcher, blocking until it is consumed by ReadBatch.
+func (b *batchPublisher) FlushRemaining(ctx context.Context) error {
+	if b.batcher == nil {
+		return nil
+	}
+	b.shutSig.TriggerSoftStop()
+	<-b.shutSig.HasStoppedChan()
+	return b.flushCurrent(ctx)
 }
 
 // Close signals the publisher's loop goroutine to stop and waits for it to exit.
-// TriggerHardStop cancels the HardStopCtx used by publishBatch, unblocking any
+// TriggerHardStop cancels the HardStopCtx used by the flush loop, unblocking any
 // send that is waiting on msgChan when no consumer is left.
 func (b *batchPublisher) Close() {
 	b.shutSig.TriggerSoftStop()
 	b.shutSig.TriggerHardStop()
 	<-b.shutSig.HasStoppedChan()
 	if b.batcher != nil {
+		// The batcher is not goroutine-safe and session goroutines may still
+		// be inside Publish: close it under batcherMu and mark it closed so
+		// later flush paths refuse instead of touching a closed batcher.
+		b.batcherMu.Lock()
+		b.closed = true
 		_ = b.batcher.Close(context.Background())
+		b.batcherMu.Unlock()
 	}
 }
diff --git a/internal/impl/oracledb/batcher_test.go b/internal/impl/oracledb/batcher_test.go
index 2352ef5593..8b0be107a1 100644
--- a/internal/impl/oracledb/batcher_test.go
+++ b/internal/impl/oracledb/batcher_test.go
@@ -10,9 +10,11 @@ package oracledb
 
 import (
 	"context"
+	"errors"
 	"log/slog"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/Jeffail/checkpoint"
 	"github.com/stretchr/testify/require"
@@ -26,8 +28,8 @@ func TestPublishBatch(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedSCNs := newTestBatchPublisher(t)
 
-		msg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage("100")})
-		msg1 := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage("100")})
+		msg := publishAndReceive(t, ctx, publisher, snapshotEvent(100))
+		msg1 := publishAndReceive(t, ctx, publisher, snapshotEvent(100))
 
 		require.NoError(t, msg.ackFn(ctx, nil))
 		require.Empty(t, cachedSCNs(), "cacheSCN must not be called after acking only the first snapshot batch")
@@ -40,8 +42,7 @@ func TestPublishBatch(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedSCNs := newTestBatchPublisher(t)
 
-		batch := service.MessageBatch{newStreamingMessage("200")}
-		msg := publishAndReceive(t, ctx, publisher, batch)
+		msg := publishAndReceive(t, ctx, publisher, streamingEvent(200))
 		require.NoError(t, msg.ackFn(ctx, nil))
 
 		scns := cachedSCNs()
@@ -51,13 +52,21 @@ func TestPublishBatch(t *testing.T) {
 
 	t.Run("mixed snapshot and streaming batch persists", func(t *testing.T) {
 		ctx := t.Context()
-		publisher, cachedSCNs := newTestBatchPublisher(t)
+		// Count=2 groups the snapshot and streaming events into one batch.
+		publisher, cachedSCNs := newTestBatchPublisherWithCount(t, 2)
+
+		got := make(chan asyncMessage, 1)
+		go func() { got <- <-publisher.msgs() }()
+		require.NoError(t, publisher.Publish(ctx, snapshotEvent(100)))
+		require.NoError(t, publisher.Publish(ctx, streamingEvent(300)))
 
-		batch := service.MessageBatch{
-			newSnapshotMessage("100"),
-			newStreamingMessage("300"),
+		var msg asyncMessage
+		select {
+		case msg = <-got:
+			require.Len(t, msg.msg, 2)
+		case <-time.After(5 * time.Second):
+			t.Fatal("mixed batch was never published")
 		}
-		msg := publishAndReceive(t, ctx, publisher, batch)
 		require.NoError(t, msg.ackFn(ctx, nil))
 
 		scns := cachedSCNs()
@@ -69,8 +78,8 @@ func TestPublishBatch(t *testing.T) {
 		ctx := t.Context()
 		publisher, cachedSCNs := newTestBatchPublisher(t)
 
-		snapshotMsg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newSnapshotMessage("100")})
-		streamingMsg := publishAndReceive(t, ctx, publisher, service.MessageBatch{newStreamingMessage("200")})
+		snapshotMsg := publishAndReceive(t, ctx, publisher, snapshotEvent(100))
+		streamingMsg := publishAndReceive(t, ctx, publisher, streamingEvent(200))
 
 		require.NoError(t, streamingMsg.ackFn(ctx, nil))
 		require.Empty(t, cachedSCNs(), "cacheSCN must not be called while the snapshot batch is still the unresolved head")
@@ -80,15 +89,568 @@ func TestPublishBatch(t *testing.T) {
 		require.Len(t, scns, 1, "expected cacheSCN to be called exactly once when the snapshot batch resolves the streaming SCN")
 		require.Equal(t, replication.SCN(200), scns[0], "expected the streaming batch's SCN to survive the out-of-order snapshot ack")
 	})
+
+	t.Run("a nack resolves too: auto_replay_nacks off is an opt-in drop", func(t *testing.T) {
+		ctx := t.Context()
+		publisher, cachedSCNs := newTestBatchPublisher(t)
+
+		b1 := publishAndReceive(t, ctx, publisher, streamingEvent(200))
+		b2 := publishAndReceive(t, ctx, publisher, streamingEvent(300))
+
+		// 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))
+
+		scns := cachedSCNs()
+		require.NotEmpty(t, scns, "the checkpoint must continue advancing past a dropped batch")
+		require.Equal(t, replication.SCN(300), scns[len(scns)-1])
+	})
+}
+
+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(100))
+
+		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(100))
+		// 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(200))
+
+		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(100))
+
+		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(v int) {
+		t.Helper()
+		e := snapshotEvent(100)
+		e.Data = map[string]any{"a": v}
+		require.NoError(t, publisher.Publish(ctx, e))
+	}
+	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(1)
+	receive("flushCurrent did not publish the buffered partial batch")
+
+	// The loop must still be alive after flushCurrent (unlike FlushRemaining):
+	// a second publish+flush must work identically.
+	publishEvent(2)
+	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 SCN. 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.SCN](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(publisher.Close)
+
+	var (
+		mu        sync.Mutex
+		persisted []replication.SCN
+	)
+	publisher.cacheSCN = func(_ context.Context, scn replication.SCN) error {
+		mu.Lock()
+		defer mu.Unlock()
+		persisted = append(persisted, scn)
+		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:        "S",
+			Table:         "T",
+			Operation:     replication.MessageOperationInsert,
+			CheckpointSCN: replication.SCN(i + 1),
+			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 && persisted[len(persisted)-1] == replication.SCN(events)
+	}, 10*time.Second, 10*time.Millisecond, "final SCN was never persisted")
+	stopConsumer()
+	<-consumerDone
+
+	mu.Lock()
+	defer mu.Unlock()
+	for i := 1; i < len(persisted); i++ {
+		require.GreaterOrEqual(t, persisted[i], 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.SCN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheSCN = func(context.Context, replication.SCN) error { return nil }
+	t.Cleanup(publisher.Close)
+
+	// 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(100)))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, streamingEvent(101)) }()
+	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(200)); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent(201))
+		}()
+	}()
+
+	// 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(300)) }()
+	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 SCN 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.SCN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheSCN = func(context.Context, replication.SCN) error { return nil }
+	t.Cleanup(publisher.Close)
+
+	// Snapshot batch 1 fills the tracker; consume it but do not ack (WG=1).
+	require.NoError(t, publisher.Publish(ctx, snapshotEvent(100)))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, snapshotEvent(100)) }()
+	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(100)); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, snapshotEvent(100))
+		}()
+	}()
+	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")
+	}
+}
+
+// TestAdmitEscapesOnContextCancel verifies admission is cancellable: a
+// flusher queued in admit behind a ticket whose holder is parked under a
+// DIFFERENT, still-live context (the timed loop uses hardStopCtx by design)
+// 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.SCN](100)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheSCN = func(context.Context, replication.SCN) error { return nil }
+	t.Cleanup(publisher.Close)
+
+	// Ticket 0's holder parks in sendTracked under a live context (nobody
+	// consumes msgs()).
+	holderDone := make(chan error, 1)
+	go func() {
+		holderDone <- func() error {
+			if err := publisher.Publish(ctx, streamingEvent(100)); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent(101))
+		}()
+	}()
+	require.Eventually(t, func() bool {
+		publisher.batcherMu.Lock()
+		defer publisher.batcherMu.Unlock()
+		return publisher.nextTicket == 1
+	}, 5*time.Second, time.Millisecond)
+
+	// The handoff's flushCurrent queues behind it with a cancellable context
+	// - the exact soft-stop-during-handoff scenario.
+	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 flushCurrent to queue in admit, but it returned: %v", err)
+	default:
+	}
+
+	// Cancelling ONLY the flusher's context must unwind it promptly - the
+	// parked holder's hardStopCtx is untouched.
+	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: a soft stop during the handoff would burn the shutdown timeouts")
+	}
+
+	// The sequence stays intact: drain the parked holder and prove a later
+	// flusher is still 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(200)); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent(201))
+		}()
+	}()
+	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 SCN
+// 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 SCN.
+func TestAbandonedBatchSealsQueue(t *testing.T) {
+	ctx := t.Context()
+	logger := service.NewLoggerFromSlog(slog.Default())
+	cp := checkpoint.NewCapped[replication.SCN](100)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheSCN = func(context.Context, replication.SCN) error { return nil }
+	t.Cleanup(publisher.Close)
+
+	// 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(100)); err != nil {
+				return err
+			}
+			return publisher.Publish(ctx, streamingEvent(101))
+		}()
+	}()
+	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(200)); err != nil {
+				return err
+			}
+			return publisher.Publish(abandonCtx, streamingEvent(201))
+		}()
+	}()
+	time.Sleep(100 * time.Millisecond)
+	cancelAbandon()
+	require.ErrorIs(t, <-abandoned, context.Canceled)
+
+	// The abandon dropped SCN 200-201 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(300)); err != nil {
+			return err
+		}
+		return publisher.Publish(ctx, streamingEvent(301))
+	}()
+	require.ErrorIs(t, laterErr, errQueueSealed,
+		"a later flusher must be refused: tracking past the dropped rows would let its ack persist an SCN 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.SCN](2)
+
+	batcher, err := (service.BatchPolicy{Count: 2}).NewBatcher(service.MockResources())
+	require.NoError(t, err)
+	publisher := newBatchPublisher(batcher, cp, logger)
+	publisher.cacheSCN = func(context.Context, replication.SCN) error { return nil }
+	t.Cleanup(publisher.Close)
+
+	// Batch 1 fills the tracker; consume it but do not ack.
+	require.NoError(t, publisher.Publish(ctx, streamingEvent(100)))
+	firstPublished := make(chan error, 1)
+	go func() { firstPublished <- publisher.Publish(ctx, streamingEvent(101)) }()
+	<-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(200)); err != nil {
+				return err
+			}
+			return publisher.Publish(trackCtx, streamingEvent(201))
+		}()
+	}()
+	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(300)); err != nil {
+			return err
+		}
+		return publisher.Publish(ctx, streamingEvent(301))
+	}()
+	require.ErrorIs(t, laterErr, errQueueSealed,
+		"a later flusher must be refused: tracking past the dropped rows would let its ack persist an SCN 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(100))
+	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->trackBatch->sendTracked path directly.
 func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.SCN) {
 	t.Helper()
+	return newTestBatchPublisherWithCount(t, 1)
+}
+
+func newTestBatchPublisherWithCount(t *testing.T, count int) (*batchPublisher, func() []replication.SCN) {
+	t.Helper()
 
 	logger := service.NewLoggerFromSlog(slog.Default())
 	cp := checkpoint.NewCapped[replication.SCN](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(publisher.Close)
 
 	var (
@@ -111,24 +673,33 @@ func newTestBatchPublisher(t *testing.T) (*batchPublisher, func() []replication.
 	return publisher, cachedSCNsFn
 }
 
-func newSnapshotMessage(scn string) *service.Message {
-	msg := service.NewMessage([]byte("{}"))
-	msg.MetaSet("operation", replication.MessageOperationRead.String())
-	msg.MetaSet("scn", scn)
-	return msg
+func snapshotEvent(scn replication.SCN) *replication.MessageEvent {
+	return &replication.MessageEvent{
+		Schema:    "S",
+		Table:     "T",
+		Operation: replication.MessageOperationRead,
+		SCN:       scn,
+		Data:      map[string]any{"a": 1},
+	}
 }
 
-func newStreamingMessage(checkpointSCN string) *service.Message {
-	msg := service.NewMessage([]byte("{}"))
-	msg.MetaSet("operation", replication.MessageOperationInsert.String())
-	msg.MetaSet("checkpoint_scn", checkpointSCN)
-	return msg
+func streamingEvent(checkpointSCN replication.SCN) *replication.MessageEvent {
+	return &replication.MessageEvent{
+		Schema:        "S",
+		Table:         "T",
+		Operation:     replication.MessageOperationInsert,
+		CheckpointSCN: checkpointSCN,
+		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()
 }
diff --git a/internal/impl/oracledb/input_oracledb_cdc.go b/internal/impl/oracledb/input_oracledb_cdc.go
index d77cdbffda..db5a3450ef 100644
--- a/internal/impl/oracledb/input_oracledb_cdc.go
+++ b/internal/impl/oracledb/input_oracledb_cdc.go
@@ -16,6 +16,7 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -257,14 +258,30 @@ type oracleDBCDCInput struct {
 	lmCfg *logminer.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
 
 	stopSig          *shutdown.Signaller
 	snapshotOnlyDone atomic.Bool
 	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
+
+	// persistMu serializes cacheSCN writes and lastPersistedSCN keeps them
+	// monotonic: ack functions run on concurrent pipeline goroutines, and
+	// after a publisher rebuild a previous session's late acks may still
+	// arrive - without ordering, a stale write could regress the durable
+	// resume position.
+	persistMu        sync.Mutex
+	lastPersistedSCN replication.SCN
 }
 
 func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resources) (s service.BatchInput, err error) {
@@ -403,23 +420,26 @@ func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resource
 				Exclude: tableExcludes,
 			},
 		},
-		lmCfg:     lmCfg,
-		res:       resources,
-		log:       logger,
-		metrics:   resources.Metrics(),
-		stopSig:   shutdown.NewSignaller(),
-		publisher: newBatchPublisher(batcher, cp, logger),
-		cpCache:   cpCache,
+		lmCfg:           lmCfg,
+		res:             resources,
+		log:             logger,
+		metrics:         resources.Metrics(),
+		stopSig:         shutdown.NewSignaller(),
+		cpCache:         cpCache,
+		batching:        policy,
+		checkpointLimit: checkpointLimit,
 	}
 
+	pub := newBatchPublisher(batcher, cp, logger)
+	pub.cacheSCN = o.cacheSCN
+	o.publisher.Store(pub)
+
 	defer func() {
 		if err != nil {
-			o.publisher.Close()
+			pub.Close()
 		}
 	}()
 
-	o.publisher.cacheSCN = o.cacheSCN
-
 	// Has stopped is how we notify that we're not connected. This will get reset at connection time.
 	o.stopSig.TriggerHasStopped()
 
@@ -431,13 +451,47 @@ func newOracleDBCDCInput(conf *service.ParsedConfig, resources *service.Resource
 	return conf.WrapBatchInputExtractTracingSpanMapping("oracledb_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 (in-flight ack functions keep resolving into the
+// abandoned tracker, where cacheSCN'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 SCN.
+func (o *oracleDBCDCInput) rebuildPublisherIfPoisoned() (*batchPublisher, error) {
+	publisher := o.publisher.Load()
+	if !publisher.poisoned.Load() {
+		return publisher, nil
+	}
+	o.log.Warn("Rebuilding publisher: a batch could not be handed to the pipeline, so the previous checkpoint tracker is pinned")
+	publisher.Close()
+	batcher, err := o.batching.NewBatcher(o.res)
+	if err != nil {
+		return nil, fmt.Errorf("rebuilding batcher: %w", err)
+	}
+	publisher = newBatchPublisher(batcher, checkpoint.NewCapped[replication.SCN](int64(o.checkpointLimit)), o.log)
+	publisher.cacheSCN = o.cacheSCN
+	o.publisher.Store(publisher)
+	return publisher, nil
+}
+
 func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 	var (
 		userTables []replication.UserTable
 		cachedSCN  replication.SCN
-		err        error
 		isCDB      bool
 	)
+
+	// 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 SCN, which is necessarily before the orphaned rows, and the old
+	// session's late acks resolve into the abandoned tracker (cacheSCN's
+	// monotonic guard turns any stale write into a no-op).
+	publisher, err := o.rebuildPublisherIfPoisoned()
+	if err != nil {
+		return err
+	}
+
 	if o.db != nil {
 		_ = o.db.Close()
 		o.db = nil
@@ -531,7 +585,8 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 			o.log.Warnf("Failed to pre-fetch schema for %s.%s: %v", t.Schema, t.Name, err)
 		}
 	}
-	o.publisher.schemas = schemas
+
+	publisher.schemas = schemas
 
 	if cachedSCN, err = o.getCachedSCN(ctx); err != nil {
 		if errors.Is(err, service.ErrKeyNotFound) {
@@ -564,7 +619,7 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 
 	// no cached SCN means we're not recovering from a restart
 	if !o.cfg.SnapshotMode.IsSnapshotNone() && cachedSCN == replication.InvalidSCN {
-		if snapshotter, err = replication.NewSnapshot(ctx, o.cfg.ConnectionString, userTables, o.cfg.SnapshotFilters, o.publisher, o.lmCfg.LOBEnabled, pdbNameForCache, o.log, o.metrics); err != nil {
+		if snapshotter, err = replication.NewSnapshot(ctx, o.cfg.ConnectionString, userTables, o.cfg.SnapshotFilters, publisher, o.lmCfg.LOBEnabled, pdbNameForCache, o.log, o.metrics); err != nil {
 			return fmt.Errorf("creating database snapshotter: %w", err)
 		}
 		defer func() {
@@ -581,7 +636,7 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 		if o.lmCfg.TransactionCacheConfig.CacheName != "" {
 			txnCache = logminer.NewConnectCacheResource(o.res, o.lmCfg.TransactionCacheConfig, o.metrics, o.log)
 		}
-		streaming = logminer.NewMiner(o.db, userTables, o.publisher, o.lmCfg, txnCache, o.metrics, o.log)
+		streaming = logminer.NewMiner(o.db, userTables, publisher, o.lmCfg, txnCache, o.metrics, o.log)
 	} else {
 		return errors.New("logminer configuration required for streaming")
 	}
@@ -617,6 +672,31 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 				return
 			}
 
+			// Flush the partial snapshot batch still held by the batcher, then
+			// block until every snapshot batch is acknowledged downstream.
+			// Persisting the SCN 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) && !o.stopSig.IsHardStopSignalled() {
+					o.log.Infof("Interrupted while flushing remaining snapshot batches. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				} else {
+					o.log.Errorf("Failed to flush remaining snapshot batches. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				}
+				o.stopSig.TriggerHasStopped()
+				return
+			}
+			if err = publisher.waitSnapshotAcks(softCtx); err != nil {
+				o.log.Infof("Interrupted while waiting for snapshot acknowledgements. Snapshot will re-run on restart (may cause duplicate data): %s", err)
+				o.stopSig.TriggerHasStopped()
+				return
+			}
+
 			if err = o.cacheSCN(softCtx, startSCN); err != nil {
 				o.log.Errorf("Failed to capture SCN after snapshot completion. Snapshot will re-run on restart (may cause duplicate data): %s", err)
 				o.stopSig.TriggerHasStopped()
@@ -627,7 +707,7 @@ func (o *oracleDBCDCInput) Connect(ctx context.Context) (resErr error) {
 		}
 
 		if o.cfg.SnapshotMode.IsSnapshotOnly() {
-			if err = o.publisher.FlushRemaining(softCtx); err != nil {
+			if err = publisher.FlushRemaining(softCtx); err != nil {
 				o.log.Errorf("Failed to flush remaining snapshot events: %s", err)
 			}
 			o.log.Infof("Snapshot-only mode complete, stopping at SCN %s", startSCN)
@@ -693,6 +773,16 @@ func (o *oracleDBCDCInput) cacheSCN(ctx context.Context, scn replication.SCN) er
 		return errors.New("SCN for caching is empty")
 	}
 
+	// Serialized and monotonic: concurrent acks (and, after a publisher
+	// rebuild, a previous session's late acks) must never land a stale SCN
+	// over a newer durable position. SCNs only grow, so skipping
+	// non-advancing writes is always safe.
+	o.persistMu.Lock()
+	defer o.persistMu.Unlock()
+	if o.lastPersistedSCN.IsValid() && scn <= o.lastPersistedSCN {
+		return nil
+	}
+
 	// Use internal Oracle-based cache if set (when no external cache configured),
 	// otherwise use external cache resource
 	var cErr error
@@ -709,12 +799,13 @@ func (o *oracleDBCDCInput) cacheSCN(ctx context.Context, scn replication.SCN) er
 	if cErr != nil {
 		return fmt.Errorf("persisting checkpoint to cache: %w", cErr)
 	}
+	o.lastPersistedSCN = scn
 	return nil
 }
 
 func (o *oracleDBCDCInput) ReadBatch(ctx context.Context) (service.MessageBatch, service.AckFunc, error) {
 	select {
-	case m := <-o.publisher.msgs():
+	case m := <-o.publisher.Load().msgs():
 		return m.msg, m.ackFn, nil
 	case <-o.stopSig.HasStoppedChan():
 		if o.snapshotOnlyDone.Load() {
@@ -751,6 +842,12 @@ func (o *oracleDBCDCInput) Close(ctx context.Context) error {
 	if o.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 := o.publisher.Load(); pub != nil {
+		pub.stopping.Store(true)
+	}
 	o.stopSig.TriggerSoftStop()
 	select {
 	case <-ctx.Done():
@@ -766,8 +863,8 @@ func (o *oracleDBCDCInput) Close(ctx context.Context) error {
 	case <-o.stopSig.HasStoppedChan():
 	}
 
-	if o.publisher != nil {
-		o.publisher.Close()
+	if pub := o.publisher.Load(); pub != nil {
+		pub.Close()
 	}
 
 	// Close both resources and combine errors to avoid resource leaks
diff --git a/internal/impl/oracledb/input_oracledb_cdc_unit_test.go b/internal/impl/oracledb/input_oracledb_cdc_unit_test.go
new file mode 100644
index 0000000000..adaf9e59f6
--- /dev/null
+++ b/internal/impl/oracledb/input_oracledb_cdc_unit_test.go
@@ -0,0 +1,134 @@
+// 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 oracledb
+
+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/oracledb/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) (*oracleDBCDCInput, *recordingCache) {
+	t.Helper()
+	cache := &recordingCache{}
+	o := &oracleDBCDCInput{
+		cfg:             Config{SCNCacheKey: "scn"},
+		res:             service.MockResources(),
+		log:             service.NewLoggerFromSlog(slog.Default()),
+		stopSig:         shutdown.NewSignaller(),
+		cpCache:         cache,
+		batching:        service.BatchPolicy{Count: 1},
+		checkpointLimit: 8,
+	}
+	batcher, err := o.batching.NewBatcher(o.res)
+	require.NoError(t, err)
+	pub := newBatchPublisher(batcher, checkpoint.NewCapped[replication.SCN](8), o.log)
+	pub.cacheSCN = o.cacheSCN
+	o.publisher.Store(pub)
+	t.Cleanup(func() { o.publisher.Load().Close() })
+	return o, cache
+}
+
+// TestCacheSCNMonotonicGuard 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 TestCacheSCNMonotonicGuard(t *testing.T) {
+	o, cache := newTestInput(t)
+	ctx := t.Context()
+
+	require.NoError(t, o.cacheSCN(ctx, replication.SCN(10)))
+	require.NoError(t, o.cacheSCN(ctx, replication.SCN(20)), "an advancing SCN must persist")
+	require.NoError(t, o.cacheSCN(ctx, replication.SCN(20)), "an equal SCN is a no-op, not an error")
+	require.NoError(t, o.cacheSCN(ctx, replication.SCN(15)), "a regressing SCN 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.Error(t, o.cacheSCN(ctx, replication.InvalidSCN), "an invalid SCN is rejected")
+}
+
+// TestRebuildPublisherIfPoisoned proves the rebuild actually swaps
+// generations: the old publisher is closed, the new one is distinct with a
+// fresh tracker wired to cacheSCN, and a late ack from the OLD generation
+// cannot regress the durable position past the guard.
+func TestRebuildPublisherIfPoisoned(t *testing.T) {
+	o, cache := newTestInput(t)
+	ctx := t.Context()
+
+	old := o.publisher.Load()
+
+	// Not poisoned: same generation back.
+	same, err := o.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(10)) }()
+	oldMsg := <-old.msgs()
+	require.NoError(t, <-oldPublished)
+
+	// Poison and rebuild.
+	old.poisoned.Store(true)
+	rebuilt, err := o.rebuildPublisherIfPoisoned()
+	require.NoError(t, err)
+	require.NotSame(t, old, rebuilt, "a poisoned publisher must be replaced")
+	require.Same(t, rebuilt, o.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(30)) }()
+	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.NotEmpty(t, got)
+	require.Equal(t, replication.SCN(30).Bytes(), got[len(got)-1], "a late ack from the abandoned generation must not regress the cache")
+}
diff --git a/internal/impl/oracledb/integration_test.go b/internal/impl/oracledb/integration_test.go
index a5eedd374e..bd2e1f196f 100644
--- a/internal/impl/oracledb/integration_test.go
+++ b/internal/impl/oracledb/integration_test.go
@@ -560,6 +560,127 @@ oracledb_cdc:
 	}
 }
 
+// TestIntegrationOracleDBCDCSnapshotAckBarrier 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 SCN is
+// only persisted once every snapshot batch is acked, the snapshot must re-run
+// on restart. See CON-504.
+func TestIntegrationOracleDBCDCSnapshotAckBarrier(t *testing.T) {
+	integration.CheckSkip(t)
+
+	connStr, db := oracledbtest.SetupTestWithOracleDBVersion(t)
+	require.NoError(t, db.CreateTableWithSupplementalLoggingIfNotExists(t.Context(), "testdb.ackbarrier", "CREATE TABLE testdb.ackbarrier (id NUMBER GENERATED ALWAYS AS IDENTITY PRIMARY KEY)"))
+
+	const rowCount = 5
+	for range rowCount {
+		db.MustExec("INSERT INTO testdb.ackbarrier (id) VALUES (DEFAULT)")
+	}
+	db.MustExec("COMMIT")
+
+	// 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(`
+oracledb_cdc:
+  connection_string: %s
+  snapshot_mode: snapshot_and_stream
+  logminer:
+    scn_window_size: 20000
+    min_scn_window_size: 0
+    backoff_interval: 1s
+  include: ["TESTDB.ACKBARRIER"]
+  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 SCN 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 SCN) 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 SCN from being
+	// persisted, since the snapshot rows were never acknowledged. This is the
+	// core guarantee: without it a cached SCN 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.CDC_CHECKPOINT_CACHE").Scan(&checkpoints))
+	require.Zero(t, checkpoints, "post-snapshot SCN 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 SCN 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))
+}
+
 func TestIntegrationOracleDBCDCStreaming(t *testing.T) {
 	integration.CheckSkip(t)
 	connStr, db := oracledbtest.SetupTestWithOracleDBVersion(t)