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

internal/impl/oracledb/batcher.go

@@ -34,8 +34,12 @@ type batchPublisher struct {
 	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.
@@ -68,7 +72,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 +93,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,9 +103,14 @@ func (p *batchPublisher) loop() {
 		adjustTimedFlush()
 		select {
 		case <-flushBatch:
-			var sendBatch service.MessageBatch
-
-			// Wrap this in a closure to make locking/unlocking easier.
+			var (
+				tracked  *trackedBatch
+				trackErr error
+			)
+
+			// Wrap this in a closure to make locking/unlocking easier. Track
+			// happens under the same lock as the flush so the checkpoint
+			// sequence matches flush order.
 			func() {
 				p.batcherMu.Lock()
 				defer p.batcherMu.Unlock()
@@ -110,13 +123,18 @@ func (p *batchPublisher) loop() {
 					return
 				}
 
+				var sendBatch service.MessageBatch
 				if sendBatch, _ = p.batcher.Flush(hardStopCtx); len(sendBatch) == 0 {
 					return
 				}
+				tracked, trackErr = p.trackBatchLocked(hardStopCtx, sendBatch)
 			}()
+			if trackErr != nil {
+				return
+			}
 
-			if len(sendBatch) > 0 {
-				if err := p.publishBatch(hardStopCtx, sendBatch); err != nil {
+			if tracked != nil {
+				if err := p.sendTracked(hardStopCtx, tracked); err != nil {
 					return
 				}
 			}
@@ -193,31 +211,44 @@ func (b *batchPublisher) Publish(ctx context.Context, m *replication.MessageEven
 		msg.MetaSetImmut("schema", service.ImmutableAny{V: schemaAny})
 	}
 
-	var flushedBatch []*service.Message
+	// Flush and Track must be atomic: Track order defines the checkpoint
+	// sequence, so another flusher (the timed-flush loop) must not interleave
+	// between our flush and our Track. Only the channel send happens outside
+	// the lock.
+	var tracked *trackedBatch
 	b.batcherMu.Lock()
 	if b.batcher.Add(msg) {
-		flushedBatch, err = b.batcher.Flush(ctx)
+		var flushedBatch []*service.Message
+		if flushedBatch, err = b.batcher.Flush(ctx); err == nil && len(flushedBatch) > 0 {
+			tracked, err = b.trackBatchLocked(ctx, flushedBatch)
+		}
 	}
 	b.batcherMu.Unlock()
 	if err != nil {
 		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 {
+	if tracked != nil {
+		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 {
+	msg        asyncMessage
+	isSnapshot bool
+}
 
+// trackBatchLocked registers the batch with the ordered checkpoint tracker and
+// builds its ack function. It MUST be called with batcherMu held: Track order
+// defines the checkpoint sequence, so it has to match flush order exactly.
+func (b *batchPublisher) trackBatchLocked(ctx context.Context, batch service.MessageBatch) (*trackedBatch, error) {
 	lastMsg := batch[len(batch)-1]
 
 	// ensure we don't checkpoint snapshot batches
@@ -240,31 +271,73 @@ 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)
+		return nil, 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)
+	if isSnapshotBatch {
+		b.snapshotAckWG.Add(1)
+	}
+	return &trackedBatch{
+		isSnapshot: isSnapshotBatch,
+		msg: asyncMessage{
+			msg: batch,
+			// The ack error is deliberately ignored: nacks are replayed by
+			// auto_replay_nacks (the default), and disabling that is a
+			// documented opt-in to DROP rejected messages, so the checkpoint
+			// must advance past them rather than pin the tracker.
+			ackFn: func(ctx context.Context, _ error) error {
+				if isSnapshotBatch {
+					defer b.snapshotAckWG.Done()
+				}
+				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 WITHOUT
+// batcherMu held (the send blocks until consumed). A failed send releases the
+// batch's snapshot-gate slot.
+func (b *batchPublisher) sendTracked(ctx context.Context, tracked *trackedBatch) error {
+	select {
+	case b.msgChan <- tracked.msg:
+		return nil
+	case <-ctx.Done():
+		if tracked.isSnapshot {
+			b.snapshotAckWG.Done()
+		}
+		return ctx.Err()
 	}
+}
+
+// waitSnapshotAcks blocks until every published snapshot batch has been
+// acknowledged (or nacked) downstream, or until ctx is cancelled. Nacked
+// batches release the gate too: redelivery is owned by auto_replay_nacks,
+// and 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,26 +361,40 @@ 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()
-
+	var tracked *trackedBatch
 	b.batcherMu.Lock()
 	remaining, err := b.batcher.Flush(ctx)
+	if err == nil && len(remaining) > 0 {
+		tracked, err = b.trackBatchLocked(ctx, remaining)
+	}
 	b.batcherMu.Unlock()
-	if err != nil || len(remaining) == 0 {
+	if err != nil || tracked == nil {
 		return err
 	}
-	return b.publishBatch(ctx, remaining)
+	return b.sendTracked(ctx, tracked)
+}
+
+// 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()