From d01137af236ddac057b276814a934a06f33b6f7e Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 6 Aug 2026 14:31:55 -0400 Subject: [PATCH 1/9] mongodb_cdc: don't resolve snapshot checkpoints on nack --- internal/impl/mongodb/cdc/input.go | 26 ++++++++++---- internal/impl/mongodb/cdc/input_test.go | 47 +++++++++++++++++++++++++ 2 files changed, 66 insertions(+), 7 deletions(-) create mode 100644 internal/impl/mongodb/cdc/input_test.go diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go index 0b95eb6d6d..6763771946 100644 --- a/internal/impl/mongodb/cdc/input.go +++ b/internal/impl/mongodb/cdc/input.go @@ -666,6 +666,24 @@ func (m *mongoCDC) readParallelSnapshot( return g.Wait() } +// snapshotAckFn builds the ack function for a snapshot batch. Mirroring the +// streaming ackFn, a nack returns the error without resolving the checkpoint +// slot: the shared capped tracker stays pinned, so the resume token can never +// be persisted past un-delivered snapshot rows (redelivery is owned by +// auto_replay_nacks). +func snapshotAckFn(resolve func() *bson.Raw) service.AckFunc { + return func(_ context.Context, err error) error { + if err != nil { + return err + } + resumeToken := resolve() + if resumeToken != nil && *resumeToken != nil { + return fmt.Errorf("unexpected resume token for snapshot batch: %s", resumeToken.String()) + } + return nil + } +} + func (m *mongoCDC) readSnapshotRange( ctx context.Context, coll *mongo.Collection, @@ -707,13 +725,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)} select { case m.readChan <- b: case <-ctx.Done(): diff --git a/internal/impl/mongodb/cdc/input_test.go b/internal/impl/mongodb/cdc/input_test.go new file mode 100644 index 0000000000..3cf378d75e --- /dev/null +++ b/internal/impl/mongodb/cdc/input_test.go @@ -0,0 +1,47 @@ +// 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 ( + "errors" + "testing" + + "github.com/stretchr/testify/require" + "go.mongodb.org/mongo-driver/v2/bson" +) + +func TestSnapshotAckFn(t *testing.T) { + t.Run("nack returns the error without resolving", func(t *testing.T) { + resolved := false + ackFn := snapshotAckFn(func() *bson.Raw { + resolved = true + return nil + }) + nackErr := errors.New("downstream failure") + err := ackFn(t.Context(), nackErr) + require.ErrorIs(t, err, nackErr) + require.False(t, resolved, "a nacked snapshot batch must not resolve its checkpoint slot") + }) + + 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 + }) + require.NoError(t, ackFn(t.Context(), nil)) + require.True(t, resolved) + }) + + t.Run("ack rejects an unexpected non-nil resume token", func(t *testing.T) { + token := bson.Raw("unexpected") + ackFn := snapshotAckFn(func() *bson.Raw { return &token }) + require.Error(t, ackFn(t.Context(), nil)) + }) +} From 7ed6bc5a20ad9f63cc9c079e29147c1ebb618fd2 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 6 Aug 2026 14:32:11 -0400 Subject: [PATCH 2/9] postgres_cdc: restart the stream on marshal failure instead of skipping rows --- internal/impl/postgresql/input_pg_stream.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index 2e1cd15b6b..b15f3ebb48 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/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) From b6fed3948608c1594476e093949ad5689caee32b Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 6 Aug 2026 14:34:44 -0400 Subject: [PATCH 3/9] postgres_cdc: integration test that marshal failures stop the stream --- internal/impl/postgresql/integration_test.go | 77 ++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 52d3b1e71c..a6bff37339 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/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 From 0d016a2bc46b14c7fbd894628aa29c050a938cf2 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Tue, 11 Aug 2026 09:50:33 -0400 Subject: [PATCH 4/9] mongodb_cdc: nacks resolve snapshot checkpoints (auto_replay_nacks off is an opt-in drop) Reverts the nack guard added earlier on this branch. Per the framework's documented contract for auto_replay_nacks ("If set to false these messages will instead be deleted"), disabling replay is an explicit opt-in to drop rejected messages - typically because failures are routed to a DLQ, which acks. Pinning the checkpoint on nack contradicted that contract and produced permanent backpressure once the checkpoint limit filled. --- internal/impl/mongodb/cdc/input.go | 16 +++++++--------- internal/impl/mongodb/cdc/input_test.go | 8 +++----- 2 files changed, 10 insertions(+), 14 deletions(-) diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go index 6763771946..abc0e45183 100644 --- a/internal/impl/mongodb/cdc/input.go +++ b/internal/impl/mongodb/cdc/input.go @@ -666,16 +666,14 @@ func (m *mongoCDC) readParallelSnapshot( return g.Wait() } -// snapshotAckFn builds the ack function for a snapshot batch. Mirroring the -// streaming ackFn, a nack returns the error without resolving the checkpoint -// slot: the shared capped tracker stays pinned, so the resume token can never -// be persisted past un-delivered snapshot rows (redelivery is owned by -// auto_replay_nacks). +// 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. func snapshotAckFn(resolve func() *bson.Raw) service.AckFunc { - return func(_ context.Context, err error) error { - if err != nil { - return err - } + return func(_ context.Context, _ error) error { resumeToken := resolve() if resumeToken != nil && *resumeToken != nil { return fmt.Errorf("unexpected resume token for snapshot batch: %s", resumeToken.String()) diff --git a/internal/impl/mongodb/cdc/input_test.go b/internal/impl/mongodb/cdc/input_test.go index 3cf378d75e..1e620bd645 100644 --- a/internal/impl/mongodb/cdc/input_test.go +++ b/internal/impl/mongodb/cdc/input_test.go @@ -17,16 +17,14 @@ import ( ) func TestSnapshotAckFn(t *testing.T) { - t.Run("nack returns the error without resolving", func(t *testing.T) { + 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 }) - nackErr := errors.New("downstream failure") - err := ackFn(t.Context(), nackErr) - require.ErrorIs(t, err, nackErr) - require.False(t, resolved, "a nacked snapshot batch must not resolve its checkpoint slot") + 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) { From 00c7cdd9cacc985a05d8edbb57798436014c2c92 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Fri, 14 Aug 2026 13:29:47 -0400 Subject: [PATCH 5/9] mongodb_cdc: name the misrouted-ack invariant in the snapshot token guard Review ask: the bare 'unexpected resume token' error gave no hint what it meant. The message now states the invariant (snapshot slots carry no token) and what a violation implies (snapshot and streaming acks misrouted in the checkpoint tracker - a regression, not an operational error). --- internal/impl/mongodb/cdc/input.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go index abc0e45183..f96bc54971 100644 --- a/internal/impl/mongodb/cdc/input.go +++ b/internal/impl/mongodb/cdc/input.go @@ -676,7 +676,11 @@ func snapshotAckFn(resolve func() *bson.Raw) service.AckFunc { return func(_ context.Context, _ error) error { resumeToken := resolve() if resumeToken != nil && *resumeToken != nil { - return fmt.Errorf("unexpected resume token for snapshot batch: %s", resumeToken.String()) + // Snapshot slots are tracked with a nil token (only streaming + // slots carry one), so a token here means snapshot and streaming + // acks were misrouted in the checkpoint tracker - a regression, + // not an operational error. + return fmt.Errorf("invariant violation: snapshot batch resolved with resume token %s, which only streaming batches carry; snapshot and streaming acks were misrouted in the checkpoint tracker", resumeToken.String()) } return nil } From 2e245cf03c66ad0bba7e8014a4e70b5c2ba18a3d Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Sat, 15 Aug 2026 20:37:27 -0400 Subject: [PATCH 6/9] mongodb_cdc: persist streaming tokens surfaced by snapshot acks The snapshot ack guard treated a non-nil resolved token as an internal misroute, but it is a legitimate outcome: snapshot and streaming batches share one ordered tracker and streaming tracking starts once snapshot batches are enqueued, not acked. Under out-of-order acks (any output with max_in_flight > 1) a snapshot slot's resolve can surface a streaming batch's resume token as the new contiguous frontier - and since every snapshot slot precedes every streaming slot, that frontier proves the whole snapshot has settled. Erroring dropped that checkpoint. The token now persists through the same path as a streaming ack (shared persistResumeToken), matching how pg_stream handles the equivalent case. --- internal/impl/mongodb/cdc/input.go | 47 ++++++++++++++++--------- internal/impl/mongodb/cdc/input_test.go | 26 ++++++++++---- 2 files changed, 50 insertions(+), 23 deletions(-) diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go index f96bc54971..cd8393235c 100644 --- a/internal/impl/mongodb/cdc/input.go +++ b/internal/impl/mongodb/cdc/input.go @@ -672,20 +672,39 @@ func (m *mongoCDC) readParallelSnapshot( // 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. -func snapshotAckFn(resolve func() *bson.Raw) service.AckFunc { - return func(_ context.Context, _ error) error { +// +// 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 { - // Snapshot slots are tracked with a nil token (only streaming - // slots carry one), so a token here means snapshot and streaming - // acks were misrouted in the checkpoint tracker - a regression, - // not an operational error. - return fmt.Errorf("invariant violation: snapshot batch resolved with resume token %s, which only streaming batches carry; snapshot and streaming acks were misrouted in the checkpoint tracker", resumeToken.String()) + if resumeToken == nil || *resumeToken == nil { + return 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, @@ -727,7 +746,7 @@ func (m *mongoCDC) readSnapshotRange( if err != nil { return fmt.Errorf("unable to create batch: %w", err) } - b := mongoBatch{mb, snapshotAckFn(resolve)} + b := mongoBatch{mb, snapshotAckFn(resolve, m.persistResumeToken)} select { case m.readChan <- b: case <-ctx.Done(): @@ -945,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}: diff --git a/internal/impl/mongodb/cdc/input_test.go b/internal/impl/mongodb/cdc/input_test.go index 1e620bd645..785b3be3c4 100644 --- a/internal/impl/mongodb/cdc/input_test.go +++ b/internal/impl/mongodb/cdc/input_test.go @@ -9,6 +9,7 @@ package cdc import ( + "context" "errors" "testing" @@ -17,12 +18,16 @@ import ( ) 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") }) @@ -32,14 +37,23 @@ func TestSnapshotAckFn(t *testing.T) { ackFn := snapshotAckFn(func() *bson.Raw { resolved = true return nil - }) + }, noPersist) require.NoError(t, ackFn(t.Context(), nil)) require.True(t, resolved) }) - t.Run("ack rejects an unexpected non-nil resume token", func(t *testing.T) { - token := bson.Raw("unexpected") - ackFn := snapshotAckFn(func() *bson.Raw { return &token }) - require.Error(t, ackFn(t.Context(), nil)) + 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") }) } From dd2bb0d1fdcb6187440672a3538caff3cb713523 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 20 Aug 2026 20:03:41 -0400 Subject: [PATCH 7/9] postgres_cdc: route unmarshalable rows with SetError; mongodb_cdc: align streaming nacks Resolves the two open design threads on this PR: - An unmarshalable WAL row (non-finite floats) previously soft-stopped the connector, which reconnected onto the same row forever - a restart loop that also pins the replication slot and risks WAL disk exhaustion on the server (reviewer finding). Per the reviewer's proposal the row is now published with its error set and a plain-text rendering as the payload: inspectable via errored(), routable with error-handling components, at-least-once preserved (the row IS delivered, flagged), and the checkpoint advances normally. The warn log names the table and LSN, the connector description documents the behavior, and the integration test asserts in-order delivery with the error flag instead of the stall. - The streaming ack path still pinned its checkpoint slot on nack, which under auto_replay_nacks: false wedges the shared tracker permanently - the exact backpressure failure the ruling on this PR resolved for the snapshot path. Nacks now resolve and persist like acks, with the contract-drop logged at warn. --- internal/impl/mongodb/cdc/input.go | 12 +++-- internal/impl/postgresql/input_pg_stream.go | 28 ++++++++--- internal/impl/postgresql/integration_test.go | 52 ++++++++++++++------ 3 files changed, 66 insertions(+), 26 deletions(-) diff --git a/internal/impl/mongodb/cdc/input.go b/internal/impl/mongodb/cdc/input.go index cd8393235c..785af0e384 100644 --- a/internal/impl/mongodb/cdc/input.go +++ b/internal/impl/mongodb/cdc/input.go @@ -956,9 +956,15 @@ 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 { diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index b15f3ebb48..c87013feb7 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/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,18 +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 { - // 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 + 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 a6bff37339..2c3108a70e 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -275,13 +275,14 @@ 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) { +// 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) @@ -299,9 +300,14 @@ pg_stream: - nan_floats `, databaseURL) + type receivedMsg struct { + body string + errored bool + errText string + } var ( receivedMu sync.Mutex - received []string + received []receivedMsg ) builder := service.NewStreamBuilder() require.NoError(t, builder.SetLoggerYAML(`level: OFF`)) @@ -311,8 +317,13 @@ pg_stream: 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, string(b)) + received = append(received, rm) receivedMu.Unlock() return nil })) @@ -340,14 +351,25 @@ pg_stream: _, 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) + // 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([]string(nil), received...) + got := append([]receivedMsg(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.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)) } From a45a1b4cda434e29b855554a76ff652b9f7e3812 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Sun, 23 Aug 2026 21:02:24 -0400 Subject: [PATCH 8/9] postgres_cdc: render the LSN in the unmarshalable-row warning, poll for slot readiness Two review findings: the warn log formatted the *string LSN with %v (printing a pointer address, losing the one operator-facing position signal) - now dereferenced with an 'unknown' fallback for the nil snapshot case; and the integration test's fixed 5s sleep readiness gate is replaced with a poll on pg_replication_slots, which is both reliable on loaded runners and faster. --- internal/impl/postgresql/input_pg_stream.go | 6 +++++- internal/impl/postgresql/integration_test.go | 10 +++++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/internal/impl/postgresql/input_pg_stream.go b/internal/impl/postgresql/input_pg_stream.go index e150299f0e..8cd5e457ef 100644 --- a/internal/impl/postgresql/input_pg_stream.go +++ b/internal/impl/postgresql/input_pg_stream.go @@ -622,7 +622,11 @@ func (p *pgStreamInput) processStream(pgStream *pglogicalstream.Stream, batcher // 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) + 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) diff --git a/internal/impl/postgresql/integration_test.go b/internal/impl/postgresql/integration_test.go index 5738888f0c..a03f7809b0 100644 --- a/internal/impl/postgresql/integration_test.go +++ b/internal/impl/postgresql/integration_test.go @@ -318,9 +318,13 @@ pg_stream: 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) + // 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);") From 1dca56916aa797c74866a5034ef80f6e8af0d3b1 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Mon, 24 Aug 2026 11:02:15 -0400 Subject: [PATCH 9/9] mongodb_cdc: integration-test the streaming nack drop contract Review ask: the streaming nack change (resolve and commit instead of pinning) had no coverage. The new test runs with auto_replay_nacks: false and checkpoint_limit: 2, rejects six consecutive streaming events, and asserts the two properties the change claims: an event delivered after the rejections still arrives (no wedge at checkpoint_limit - the old pin behavior fails exactly here, verified red), and a restart delivers only new activity, never a replay of the dropped events (the committed resume token advanced past them). --- internal/impl/mongodb/cdc/integration_test.go | 76 +++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/internal/impl/mongodb/cdc/integration_test.go b/internal/impl/mongodb/cdc/integration_test.go index 8bae0f7366..9eb22d5403 100644 --- a/internal/impl/mongodb/cdc/integration_test.go +++ b/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") +}