mongodb_cdc, postgres_cdc: respect nacks and marshal failures in ack paths (CON-504) - #4676
mongodb_cdc, postgres_cdc: respect nacks and marshal failures in ack paths (CON-504)#4676squiidz wants to merge 6 commits into
Conversation
squiidz
changed the title
| // 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) |
There was a problem hiding this comment.
Wouldn't restarting just cause the connector to fail again (getting stuck in a restart loop)? I agree we shouldn't be dropping this type of error, I just wonder whether we should encode the correct course of action in the error message so it's actionable.
| } | ||
| resumeToken := resolve() | ||
| if resumeToken != nil && *resumeToken != nil { | ||
| return fmt.Errorf("unexpected resume token for snapshot batch: %s", resumeToken.String()) |
There was a problem hiding this comment.
I presume if we see this then we know there's been a regression somewhere and the snapshot has suddenly got a resume token?
There was a problem hiding this comment.
Correct — it's an invariant guard, not an operational error. Snapshot slots are tracked with a nil token (only streaming slots carry one), so a token surfacing here means snapshot and streaming acks were misrouted in the checkpoint tracker, i.e. a regression. The error message now states that explicitly (also covers the wording ask on the adjacent thread).
| func snapshotAckFn(resolve func() *bson.Raw) service.AckFunc { | ||
| return func(_ context.Context, err error) error { | ||
| if err != nil { | ||
| return err |
There was a problem hiding this comment.
minor (non-blocking): would it be worth adding some context to this error?
There was a problem hiding this comment.
Done — the error now names the invariant and what a violation implies: "invariant violation: snapshot batch resolved with resume token %s, which only streaming batches carry; snapshot and streaming acks were misrouted in the checkpoint tracker".
…f 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.
|
The nack-handling changes from the earlier review rounds have been unwound in the latest commit. The framework's documented contract for 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[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| 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 | ||
| } |
There was a problem hiding this comment.
Unrecoverable rows now stall the connector permanently, with no operator escape hatch.
Preferring a restart over silent data loss is the right instinct, but the failure this guards against is never transient. json.Marshal on decoded WAL data can only fail deterministically — in practice non-finite float64 (NaN/±Inf), exactly what the new integration test exercises. So the recovery path can never make progress:
TriggerSoftStop()unwindsprocessStream, which firesTriggerHasStopped()(input_pg_stream.go#L461-L472).ReadBatchthen returnsservice.ErrNotConnected(input_pg_stream.go#L643-L652), so the framework callsConnectagain — which succeeds, since nothing about the connection is broken.- Replication resumes from the last acked LSN, immediately re-reads the same row, and fails identically. The connector loops forever, emitting an error log per cycle and re-establishing the replication stream each time, and the pipeline never advances again.
internal/impl/postgresql/integration_test.goasserts exactly this terminal state.
This is the difference between this and the existing pgStream.Errors() soft stop above (L576-L579): that one restarts on a condition that can clear, this one restarts on a condition that cannot. Per CONTRIBUTING.md §3.1.4 (complete and correct, no missing core functionality) and §3.2.2 (poor error handling / difficult-to-diagnose bugs), the row needs an eventual exit path rather than an unbounded retry — e.g. serialize non-finite floats deterministically (consistent with the canonicalisation requirement in §5.4.6), or gate skip-vs-stop behind a config field so operators can route the row onward and let a DLQ handle it.
Two smaller points on the same lines:
- The error log identifies neither the table nor the LSN of the offending row, so an operator hitting the permanent stall has no way to find or fix the row from the logs alone (§1.2.2 — "provides relevant logging to support troubleshooting").
TriggerSoftStop()cancels thectxderived at L459 before theif flushblock below runs, so the successfully-marshalled rows that preceded the poison row in the batch are flushed with an already-cancelled context and dropped. That is safe for delivery (they were never acked), but it makes the surroundingflushBatchwork dead code on this path — worthbreaking out of theselectcase directly instead.
…uard 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).
| // 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()) |
There was a problem hiding this comment.
The stated invariant ("a token here means snapshot and streaming acks were misrouted in the checkpoint tracker - a regression, not an operational error") looks reachable in normal operation, so this error message will misdirect whoever hits it.
Snapshot and streaming share one checkpoint.Capped (input.go#L497-L512): the snapshot phase tracks with a nil token (L726-L730) and streaming then tracks resume tokens on the same tracker (L936). g.Wait() only guarantees snapshot batches were enqueued, not acked, so snapshot batches can still be pending when streaming batches are tracked. resolve() returns the furthest contiguously-resolved value, not the caller's own value — so if a streaming batch is acked before an earlier snapshot batch (out-of-order acks, e.g. an output with max_in_flight > 1), resolving the snapshot batch legitimately yields a streaming resume token. That is exactly the case the sibling connector treats as normal: pg_stream's ack func obtains maxLSN from a snapshot batch's resolveFn() and acks it rather than erroring (input_pg_stream.go#L686-L707).
If that reading is right, the ack fails and the resolved token is dropped instead of being persisted via the same path as the streaming ack fn (m.resumeToken / m.checkpoint.Store), and the new unit case ack rejects an unexpected non-nil resume token locks the behavior in. Suggested fix: treat a non-nil token here as a resolved streaming checkpoint and persist it (share the token-persisting logic with the streaming ack fn), or — if the guard is meant to stay — drop the "regression, not an operational error" wording since out-of-order acks can produce it.
There was a problem hiding this comment.
Reading confirmed — the tracker ordering makes the token legitimate, and because every snapshot slot precedes every streaming slot, a streaming token surfacing from a snapshot resolve proves the whole snapshot has settled, so persisting it (and skipping the snapshot on a subsequent restart) is correct. Fixed in 2e245cf: the guard is gone, snapshotAckFn persists a non-nil token through the same path as the streaming ack (shared persistResumeToken), and the unit case now asserts the persist instead of locking in the drop.
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.
| return m.checkpoint.Store(ctx, m.resumeToken) | ||
| } | ||
| return nil | ||
| return m.persistResumeToken(ctx, *resumeToken) |
There was a problem hiding this comment.
The snapshot ack path now resolves the checkpoint slot on nack (snapshotAckFn ignores the error argument), but this streaming ack path still bails at the top with if err != nil { return err } and never calls resolve().
The rationale in this PR's own commit message applies equally here: with auto_replay_nacks: false a nack is a documented opt-in drop, so pinning the slot is not correct. Because a nacked streaming batch never resolves, the shared checkpoint.Capped tracker keeps that slot pending forever; once checkpoint_limit further batches accumulate, cp.Track blocks and the input stalls permanently with no way to recover short of a restart — exactly the permanent-backpressure failure the snapshot change was made to avoid.
Suggest routing the streaming ack through the same resolve-then-persist shape as snapshotAckFn so the two halves of the one tracker agree on nack semantics.
| // 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() |
There was a problem hiding this comment.
This changes user-visible behaviour materially: an unmarshalable row used to be skipped and the pipeline kept running; it now halts the connector, and since the LSN is never acked every reconnect re-reads the same row and soft-stops again. That is the right trade-off for data safety, but it means a single poison row permanently stalls a postgres_cdc/pg_stream pipeline, with no config knob to skip it and only an Errorf line to diagnose it.
CONTRIBUTING.md §1.2.3 requires that "known limitations and edge cases are documented". Please document this edge case in the connector's Description() (it feeds the generated docs): what a permanently stalled stream looks like, and what an operator can do about it.
2 participants
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 assnapshotAckFn) 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.Marshalfailure in the stream loop logged andbreak-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 passesfloat8NaN through as afloat64, whichencoding/jsonrejects (numericNaN 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 likenumericdoes) is a possible follow-up, out of scope here.Proof of Work
TestSnapshotAckFnunit tests: nack returns the error without resolving; ack resolves; unexpected resume token rejected.TestIntegrationPostgresMarshalFailureStopsStream: live stream delivers a sentinel row, then a'NaN'::double precisionrow followed by a normal row — asserts nothing is delivered past the poison row. Verified it fails against the pre-fix code, demonstrating the loss: