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():
@@ -923,21 +956,21 @@ func (m *mongoCDC) readFromStream(ctx context.Context, cp *checkpoint.Capped[bso
 			if err != nil {
 				return err
 			}
-			ackFn := func(ctx context.Context, err error) error {
-				if err != nil {
-					return err
+			// 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 shared tracker (which would block cp.Track
+			// at checkpoint_limit and stall the input permanently) - the same
+			// contract the snapshot ack path follows.
+			ackFn := func(ctx context.Context, ackErr error) error {
+				if ackErr != nil {
+					m.logger.Warnf("Advancing past a batch rejected downstream: auto_replay_nacks is disabled, so the rejected messages are dropped by contract: %v", ackErr)
 				}
 				resumeToken := resolve()
 				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

@@ -88,6 +88,10 @@ This input adds the following metadata fields to each message:
 - schema: The table schema in benthos common schema format, compatible with processors like parquet_encode
 - commit_ts_ms: The commit timestamp of the transaction as a Unix millisecond timestamp. Not set for snapshot reads.
 - before: The pre-change state of the row for update and delete operations, in benthos common schema format. For updates, availability depends on the table's REPLICA IDENTITY setting - with the default identity only key columns are present, with REPLICA IDENTITY FULL all columns are present.
+
+== Unserializable rows
+
+A row whose decoded WAL data cannot be marshalled to JSON (in practice non-finite floating point values such as NaN or Infinity) is published with its error set and a plain-text rendering of the row as the payload, rather than stalling the stream or silently dropping the row. Such messages can be inspected with the ` + "`errored()`" + ` Bloblang function and routed with error-handling components (for example a ` + "`switch`" + ` output with ` + "`reject_errored`" + `, or a dead-letter queue); if not handled they flow through the pipeline like any other message. The replication checkpoint advances past them normally once acknowledged.
 		`).
 		Field(service.NewStringField(fieldDSN).
 			Description("The Data Source Name for the PostgreSQL database in the form of `postgres://[user[:password]@][netloc][:port][/dbname][?param1=value1&...]`. Please note that Postgres enforces SSL by default, you can override this with the parameter `sslmode=disable` if required.").
@@ -526,14 +530,26 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 			var (
 				flush bool
 				mb    []byte
-				err   error
 			)
 			for _, msg := range batch {
-				if mb, err = json.Marshal(msg.Data); err != nil {
-					p.logger.Errorf("failure to marshal message: %s", err)
-					break
+				var marshalErr error
+				if mb, marshalErr = json.Marshal(msg.Data); marshalErr != nil {
+					// A marshal failure is deterministic (in practice
+					// non-finite floats), so neither skipping the row (silent
+					// loss) nor restarting (the same row fails on every
+					// reconnect, and the stalled slot blocks WAL retention on
+					// the server) can make progress. Publish the row with its
+					// error set instead: the stream keeps moving,
+					// at-least-once holds (the row IS delivered, flagged),
+					// and operators can inspect or route it with
+					// error-handling components.
+					p.logger.Warnf("Publishing unmarshalable row from table %s (LSN %v) with its error set for error-routing: %v", msg.Table, msg.LSN, marshalErr)
+					mb = fmt.Appendf(nil, "%+v", msg.Data)
 				}
 				batchMsg := service.NewMessage(mb)
+				if marshalErr != nil {
+					batchMsg.SetError(fmt.Errorf("marshalling WAL row from table %s: %w", msg.Table, marshalErr))
+				}
 				batchMsg.MetaSet("table", msg.Table)
 				batchMsg.MetaSet("operation", string(msg.Operation))
 				if msg.LSN != nil {

internal/impl/postgresql/integration_test.go

@@ -275,6 +275,105 @@ pg_stream:
 	require.NoError(t, streamOut.StopWithin(time.Second*10))
 }
 
+// TestIntegrationPostgresUnmarshalableRowRoutedWithError 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) is published with its
+// error set - inspectable and routable by error-handling components - while
+// the stream keeps moving. The original bug silently dropped the row and
+// checkpointed past it; the interim fix stalled the stream (restart loop,
+// with the stalled slot blocking WAL retention). See CON-504.
+func TestIntegrationPostgresUnmarshalableRowRoutedWithError(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)
+
+	type receivedMsg struct {
+		body    string
+		errored bool
+		errText string
+	}
+	var (
+		receivedMu sync.Mutex
+		received   []receivedMsg
+	)
+	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
+		}
+		rm := receivedMsg{body: string(b)}
+		if mErr := m.GetError(); mErr != nil {
+			rm.errored = true
+			rm.errText = mErr.Error()
+		}
+		receivedMu.Lock()
+		received = append(received, rm)
+		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: every row is delivered in order - the unmarshalable
+	// one flagged with its error, the rows after it unaffected. (The original
+	// bug dropped the NaN row silently; the interim fix stalled the stream.)
+	require.Eventually(t, func() bool {
+		receivedMu.Lock()
+		defer receivedMu.Unlock()
+		return len(received) == 3
+	}, 30*time.Second, 100*time.Millisecond, "all three rows must be delivered, the unmarshalable one included")
+
+	receivedMu.Lock()
+	got := append([]receivedMsg(nil), received...)
+	receivedMu.Unlock()
+	require.Contains(t, got[0].body, "1.5")
+	require.False(t, got[0].errored, "a normal row must not carry an error")
+	require.True(t, got[1].errored, "the unmarshalable row must be published with its error set")
+	require.Contains(t, got[1].errText, "nan_floats", "the error must name the table")
+	require.Contains(t, got[1].body, "NaN", "the fallback payload must render the row for inspection")
+	require.Contains(t, got[2].body, "2.5")
+	require.False(t, got[2].errored)
+
+	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