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

internal/impl/mongodb/cdc/input.go

@@ -1520,9 +1520,15 @@ func (m *mongoCDC) readFromStream(ctx context.Context, epoch uint64, cp *checkpo
 			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 {

internal/impl/mongodb/cdc/integration_test.go

@@ -1989,3 +1989,79 @@ mongodb_cdc:
 	assert.Equal(t, "age", schemas[1].Children[1].Name)
 	assert.Equal(t, "name", schemas[1].Children[2].Name)
 }
+
+// TestIntegrationMongoCDCNackedStreamBatchDropsByContract locks in the
+// streaming nack semantics: with auto_replay_nacks: false a rejected batch is
+// dropped by documented contract, so its checkpoint slot must resolve - the
+// input keeps reading past checkpoint_limit instead of wedging on a pinned
+// slot - and the committed resume token must advance past the dropped events
+// so a restart does not replay them.
+func TestIntegrationMongoCDCNackedStreamBatchDropsByContract(t *testing.T) {
+	stream, db, output := setup(t, `
+mongodb_cdc:
+  url: '$URI'
+  database: '$DATABASE'
+  stream_snapshot: false
+  checkpoint_cache: '$CACHE'
+  checkpoint_interval: 0s
+  checkpoint_limit: 2
+  auto_replay_nacks: false
+  collections:
+    - 'foo'
+`)
+	db.CreateCollection(t, "foo")
+
+	wait := stream.RunAsync(t)
+	time.Sleep(time.Second)
+
+	// Sentinel proves the stream is live.
+	db.InsertOne(t, "foo", bson.M{"_id": 1, "data": "keep-1"})
+	require.Eventually(t, func() bool {
+		msgs, err := output.messages()
+		return err == nil && len(msgs) == 1
+	}, 10*time.Second, 10*time.Millisecond, "sentinel was never streamed")
+
+	// Reject everything and push well past checkpoint_limit: with the old
+	// pin-on-nack behavior the first rejected slot wedged cp.Track after
+	// checkpoint_limit more events and nothing could ever be delivered again.
+	output.NackAll()
+	for i := 2; i <= 7; i++ {
+		db.InsertOne(t, "foo", bson.M{"_id": i, "data": "dropped"})
+	}
+	// Let every rejected delivery settle before accepting again, so none of
+	// the dropped documents race into the accepted window.
+	time.Sleep(3 * time.Second)
+
+	output.AckAll()
+	db.InsertOne(t, "foo", bson.M{"_id": 8, "data": "keep-8"})
+	require.Eventually(t, func() bool {
+		msgs, err := output.messages()
+		return err == nil && len(msgs) == 2
+	}, 10*time.Second, 10*time.Millisecond, "the input wedged: an event after the nacked batches was never delivered, so a rejected slot pinned the tracker past checkpoint_limit")
+
+	stream.StopWithin(t, 5*time.Second)
+	wait()
+
+	// Restart: the committed resume token must be past the dropped events, so
+	// only new activity is delivered - never a replay of documents 2..7.
+	wait = stream.RunAsync(t)
+	time.Sleep(time.Second)
+	db.InsertOne(t, "foo", bson.M{"_id": 9, "data": "keep-9"})
+	require.Eventually(t, func() bool {
+		msgs, err := output.messages()
+		return err == nil && len(msgs) == 3
+	}, 10*time.Second, 10*time.Millisecond, "the post-restart event was never delivered")
+	stream.StopWithin(t, 5*time.Second)
+	wait()
+
+	var data []string
+	for _, m := range output.Messages(t) {
+		doc, ok := m.(map[string]any)
+		require.True(t, ok)
+		val, ok := doc["data"].(string)
+		require.True(t, ok, "unexpected data field shape: %v", doc)
+		data = append(data, val)
+	}
+	require.ElementsMatch(t, []string{"keep-1", "keep-8", "keep-9"}, data,
+		"exactly the accepted documents may be delivered: the rejected ones are dropped by the auto_replay_nacks contract and must not replay after restart")
+}

internal/impl/postgresql/input_pg_stream.go

@@ -90,6 +90,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.").
@@ -599,7 +603,6 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 			var (
 				flush bool
 				mb    []byte
-				err   error
 			)
 			for _, msg := range batch {
 				// noop if not configured
@@ -608,11 +611,28 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher
 					p.logger.Errorf("failed to detect control signal in change event: %s", err)
 				}
 
-				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.
+					rowLSN := "unknown"
+					if msg.LSN != nil {
+						rowLSN = *msg.LSN
+					}
+					p.logger.Warnf("Publishing unmarshalable row from table %s (LSN %s) with its error set for error-routing: %v", msg.Table, rowLSN, 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

@@ -261,6 +261,109 @@ 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()) }()
+
+	// Streaming-only mode only sees rows inserted after the replication slot
+	// exists: poll for the slot instead of sleeping, which is flaky on loaded
+	// runners.
+	require.Eventually(t, func() bool {
+		var one int
+		return db.QueryRow("SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_slot_marshal_failure'").Scan(&one) == nil
+	}, 30*time.Second, 250*time.Millisecond, "replication slot was never created")
+
+	// 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