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

[squiidz]

Part of CON-504 (CDC at-least-once / ack-gated progress).

Two small fixes closing silent-loss paths found in the CON-504 audit:

mongodb_cdc — the snapshot batch ackFn ignored its error argument and resolved the checkpoint slot unconditionally. With auto_replay_nacks: false, a nacked snapshot batch freed its slot in the shared capped tracker, so later streaming batches advanced the resume token past the undelivered rows and a restart never re-read them. The ackFn (extracted as snapshotAckFn) now mirrors the streaming ackFn: a nack returns the error without resolving, keeping the tracker pinned so the resume token can never persist past undelivered snapshot rows.

postgres_cdc — a json.Marshal failure in the stream loop logged and break-ed, silently dropping that row plus the rest of its WAL batch while the stream kept running and checkpointed past them. This is reachable with real data: the decoder passes float8 NaN through as a float64, which encoding/json rejects (numeric NaN is already special-cased as a string). The stream now soft-stops instead — the LSN was never acked, so the restart resumes before the poison row. Note the behavior change: an unmarshalable row now stalls the stream loudly (restart loop with a clear error) rather than vanishing. Actually supporting NaN floats (e.g. encoding as a string like numeric does) is a possible follow-up, out of scope here.

Proof of Work

• TestSnapshotAckFn unit tests: nack returns the error without resolving; ack resolves; unexpected resume token rejected.
• TestIntegrationPostgresMarshalFailureStopsStream: live stream delivers a sentinel row, then a 'NaN'::double precision row followed by a normal row — asserts nothing is delivered past the poison row. Verified it fails against the pre-fix code, demonstrating the loss:

"[{"id":1,"value":1.5} {"id":3,"value":2.5}]" should have 1 item(s), but has 2
(id 2, the NaN row, was silently dropped while id 3 was delivered)

[squiidz]

The nack-handling changes from the earlier review rounds have been unwound in the latest commit. The framework's documented contract for auto_replay_nacks reads: "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 them). Pinning the checkpoint on nack contradicted that contract and produced permanent backpressure once the in-flight limit filled. Ack functions now resolve their checkpoint slot on nack exactly like on ack.

Unwound here (0d016a2): the snapshot ackFn nack pin (back to resolve-always). The pre-existing streaming ackFn guard is untouched, and the postgres marshal-failure fix stays.

[claude]

msg.LSN is a *string (stream_message.go#L47-L49), so %v prints the pointer address (e.g. 0xc0000b4010) rather than the LSN. This warning is the only operator-facing signal that identifies which WAL row was published unmarshalable, so the position it is meant to convey is lost — see how the same value is dereferenced two lines below at input_pg_stream.go#L634-L636.

Suggested fix: dereference the pointer when non-nil (falling back to an empty/unknown rendering when msg.LSN == nil, which is the snapshot case) before formatting it into the log line.

Ref: CONTRIBUTING.md §1.2.2 — "Provides relevant logging to support troubleshooting."

[claude]

Fixed sleep as a readiness gate makes this test flaky. With stream_snapshot: false the input only sees rows inserted after the replication slot exists, so if slot creation takes longer than 5s on a loaded CI runner the sentinel INSERT at line 328 is never captured, and the test fails 30s later at require.Eventually(len(received) == 1) — a hard failure with no retry path, since the sentinel is inserted exactly once.

Suggested fix: replace the sleep with a readiness poll — e.g. require.Eventually on SELECT 1 FROM pg_replication_slots WHERE slot_name = 'test_slot_marshal_failure' before inserting the sentinel, or retry the sentinel insert inside the Eventually until it is received.

Ref: .claude/agents/tester.md — "Readiness retries beyond wait strategy: When you need to retry application-level checks after the container is up ... use require.Eventually."