diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go
index baae1e9991..59089a6272 100644
--- a/internal/impl/mongodb/cdc/input.go
+++ b/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 {
diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go
index e6bdcfaeb0..e150299f0e 100644
--- a/internal/impl/postgresql/input_pg_stream.go
+++ b/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,24 @@ 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.
+					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 {
diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go
index bcb67946a1..5738888f0c 100644
--- a/internal/impl/postgresql/integration_test.go
+++ b/internal/impl/postgresql/integration_test.go
@@ -261,6 +261,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