mongodb_cdc, postgres_cdc: respect nacks and marshal failures in ack paths (CON-504)

internal/impl/mongodb/cdc/input.go

@@ -666,6 +666,45 @@ func (m *mongoCDC) readParallelSnapshot(
 	return g.Wait()
 }
 
+// snapshotAckFn builds the ack function for a snapshot batch. Nacks resolve
+// the checkpoint slot just like acks: auto_replay_nacks defaults to replaying
+// rejections in-process, and disabling it is a documented opt-in to DROP
+// messages that fail ("If set to false these messages will instead be
+// deleted"), so the stream must continue past them rather than pin the
+// tracker and back-pressure forever.
+//
+// A non-nil resolved token is a legitimate outcome, not a misroute: snapshot
+// and streaming batches share one ordered tracker, and streaming tracking
+// starts once snapshot batches are enqueued (not acked). Under out-of-order
+// acks a snapshot slot's resolve can therefore surface a streaming batch's
+// resume token as the new contiguous frontier - and because every snapshot
+// slot precedes every streaming slot in the tracker, that frontier proves the
+// whole snapshot has settled. It must be persisted exactly like the streaming
+// ack path would, or the checkpoint is silently dropped.
+func snapshotAckFn(resolve func() *bson.Raw, persist func(context.Context, bson.Raw) error) service.AckFunc {
+	return func(ctx context.Context, _ error) error {
+		resumeToken := resolve()
+		if resumeToken == nil || *resumeToken == nil {
+			return nil
+		}
+		return persist(ctx, *resumeToken)
+	}
+}
+
+// persistResumeToken records token as the in-memory resume position and, when
+// no interval flusher owns persistence, stores it in the checkpoint cache.
+// Shared by the streaming ack path and snapshot acks that surface a streaming
+// token via the shared tracker.
+func (m *mongoCDC) persistResumeToken(ctx context.Context, token bson.Raw) error {
+	m.resumeTokenMu.Lock()
+	defer m.resumeTokenMu.Unlock()
+	m.resumeToken = token
+	if m.checkpointFlusher == nil {
+		return m.checkpoint.Store(ctx, m.resumeToken)
+	}
+	return nil
+}
+
 func (m *mongoCDC) readSnapshotRange(
 	ctx context.Context,
 	coll *mongo.Collection,
@@ -707,13 +746,7 @@ func (m *mongoCDC) readSnapshotRange(
 			if err != nil {
 				return fmt.Errorf("unable to create batch: %w", err)
 			}
-			b := mongoBatch{mb, func(context.Context, error) error {
-				resumeToken := resolve()
-				if resumeToken != nil && *resumeToken != nil {
-					return fmt.Errorf("unexpected resume token for snapshot batch: %s", resumeToken.String())
-				}
-				return nil
-			}}
+			b := mongoBatch{mb, snapshotAckFn(resolve, m.persistResumeToken)}
 			select {
 			case m.readChan <- b:
 			case <-ctx.Done():
@@ -931,13 +964,7 @@ func (m *mongoCDC) readFromStream(ctx context.Context, cp *checkpoint.Capped[bso
 				if resumeToken == nil || *resumeToken == nil {
 					return nil
 				}
-				m.resumeTokenMu.Lock()
-				defer m.resumeTokenMu.Unlock()
-				m.resumeToken = *resumeToken
-				if m.checkpointFlusher == nil {
-					return m.checkpoint.Store(ctx, m.resumeToken)
-				}
-				return nil
+				return m.persistResumeToken(ctx, *resumeToken)
 			}
 			select {
 			case m.readChan <- mongoBatch{mb, ackFn}:

internal/impl/mongodb/cdc/input_test.go

@@ -0,0 +1,59 @@
+// 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/v4/blob/main/licenses/rcl.md
+
+package cdc
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/bson"
+)
+
+func TestSnapshotAckFn(t *testing.T) {
+	noPersist := func(context.Context, bson.Raw) error {
+		return errors.New("persist must not be called for a nil token")
+	}
+
+	t.Run("a nack resolves too: auto_replay_nacks off is an opt-in drop", func(t *testing.T) {
+		resolved := false
+		ackFn := snapshotAckFn(func() *bson.Raw {
+			resolved = true
+			return nil
+		}, noPersist)
+		require.NoError(t, ackFn(t.Context(), errors.New("downstream failure")))
+		require.True(t, resolved, "a nacked batch is deleted per the auto_replay_nacks contract; the stream must continue past it")
+	})
+
+	t.Run("ack resolves and accepts a nil resume token", func(t *testing.T) {
+		resolved := false
+		ackFn := snapshotAckFn(func() *bson.Raw {
+			resolved = true
+			return nil
+		}, noPersist)
+		require.NoError(t, ackFn(t.Context(), nil))
+		require.True(t, resolved)
+	})
+
+	t.Run("a streaming token surfaced by out-of-order acks is persisted", func(t *testing.T) {
+		// Snapshot and streaming share one ordered tracker: when a streaming
+		// batch acks before an earlier snapshot batch, the snapshot slot's
+		// resolve legitimately returns the streaming token as the contiguous
+		// frontier. It must be persisted, not dropped.
+		token := bson.Raw("streaming-token")
+		var persisted bson.Raw
+		ackFn := snapshotAckFn(func() *bson.Raw { return &token }, func(_ context.Context, tok bson.Raw) error {
+			persisted = tok
+			return nil
+		})
+		require.NoError(t, ackFn(t.Context(), nil))
+		require.Equal(t, token, persisted, "the resolved streaming checkpoint must persist through the same path as a streaming ack")
+	})
+}

internal/impl/postgresql/input_pg_stream.go

@@ -530,7 +530,11 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 			)
 			for _, msg := range batch {
 				if mb, err = json.Marshal(msg.Data); err != nil {
-					p.logger.Errorf("failure to marshal message: %s", err)
+					// Skipping the row would silently lose it while later rows
+					// advance the checkpoint past it. Restart instead: the LSN
+					// was never acked, so the stream resumes before this row.
+					p.logger.Errorf("failure to marshal message, restarting stream to avoid data loss: %s", err)
+					p.stopSig.TriggerSoftStop()
 					break
 				}
 				batchMsg := service.NewMessage(mb)

internal/impl/postgresql/integration_test.go

@@ -275,6 +275,83 @@ pg_stream:
 	require.NoError(t, streamOut.StopWithin(time.Second*10))
 }
 
+// TestIntegrationPostgresMarshalFailureStopsStream verifies that a row whose
+// value cannot be marshalled to JSON (float8 NaN: the decoder passes it
+// through as a float64, which encoding/json rejects) stops the stream instead
+// of being silently skipped. Before the fix the row and the remainder of its
+// WAL batch were dropped while the stream kept running and checkpointed past
+// them. See CON-504.
+func TestIntegrationPostgresMarshalFailureStopsStream(t *testing.T) {
+	integration.CheckSkip(t)
+	databaseURL, db, err := ResourceWithPostgreSQLVersion(t, "16")
+	require.NoError(t, err)
+
+	_, err = db.Exec("CREATE TABLE IF NOT EXISTS nan_floats (id serial PRIMARY KEY, value DOUBLE PRECISION);")
+	require.NoError(t, err)
+
+	template := fmt.Sprintf(`
+pg_stream:
+    dsn: %s
+    slot_name: test_slot_marshal_failure
+    stream_snapshot: false
+    schema: public
+    tables:
+       - nan_floats
+`, databaseURL)
+
+	var (
+		receivedMu sync.Mutex
+		received   []string
+	)
+	builder := service.NewStreamBuilder()
+	require.NoError(t, builder.SetLoggerYAML(`level: OFF`))
+	require.NoError(t, builder.AddInputYAML(template))
+	require.NoError(t, builder.AddConsumerFunc(func(_ context.Context, m *service.Message) error {
+		b, err := m.AsBytes()
+		if err != nil {
+			return err
+		}
+		receivedMu.Lock()
+		received = append(received, string(b))
+		receivedMu.Unlock()
+		return nil
+	}))
+	stream, err := builder.Build()
+	require.NoError(t, err)
+	license.InjectTestService(stream.Resources())
+	go func() { _ = stream.Run(t.Context()) }()
+
+	// Give the input time to create the replication slot: streaming-only mode
+	// only sees rows inserted after the slot exists.
+	time.Sleep(5 * time.Second)
+
+	// Sentinel row proves the stream is live before the poison row arrives.
+	_, err = db.Exec("INSERT INTO nan_floats (value) VALUES (1.5);")
+	require.NoError(t, err)
+	require.Eventually(t, func() bool {
+		receivedMu.Lock()
+		defer receivedMu.Unlock()
+		return len(received) == 1
+	}, 30*time.Second, 100*time.Millisecond, "sentinel row was never streamed - stream not live")
+
+	// Poison row (unmarshalable), then a normal row behind it.
+	_, err = db.Exec("INSERT INTO nan_floats (value) VALUES ('NaN'::double precision);")
+	require.NoError(t, err)
+	_, err = db.Exec("INSERT INTO nan_floats (value) VALUES (2.5);")
+	require.NoError(t, err)
+
+	// The core guarantee: nothing may be delivered past the poison row. In the
+	// buggy version the NaN row was silently dropped and 2.5 arrived here.
+	time.Sleep(10 * time.Second)
+	receivedMu.Lock()
+	got := append([]string(nil), received...)
+	receivedMu.Unlock()
+	require.Len(t, got, 1, "no row may be delivered past an unmarshalable row; got: %v", got)
+	require.Contains(t, got[0], "1.5")
+
+	require.NoError(t, stream.StopWithin(30*time.Second))
+}
+
 // TestIntegrationPostgresSnapshotAckBarrier verifies that a crash during the
 // snapshot->stream handoff (after snapshot rows are emitted but before they are
 // acknowledged) does not lose data: because the replication slot is only