salesforce_cdc: apply backpressure instead of dropping events on full buffer (CON-504)

[squiidz]

Part of CON-504 (CDC at-least-once / ack-gated progress). The final connector fix of the audit.

The input's checkpoint architecture was already correct (ordered tracker, ack-gated replay persistence), but the Pub/Sub gRPC layer underneath it could lose events, and the ack functions mishandled nacks:

1. Buffer-full drop. When the internal event buffer was full, the receive loop dropped the event with a warning while the batch's replay ID advanced past it — triggered precisely under downstream backpressure, no crash needed. The loop now blocks on the buffer (escaping cleanly on close/reconnect). While blocked, no flow-control FetchRequest is issued, so Salesforce stops sending: real backpressure, and the replay cursor can never pass an undelivered event.

2. Schema-fetch / decode skip. A transient schema-fetch error (or Avro decode failure) logged and skipped the event while the replay cursor advanced. The stream now reconnects without advancing lastReplayID, so the batch is redelivered: transient failures heal on retry; a genuinely undecodable event stalls the topic loudly (reconnect loop with clear errors) instead of vanishing.

3. Nack handling. Both the streaming and snapshot ack functions ignored their error argument, so with auto_replay_nacks: false a nack resolved its checkpoint slot and later acks could persist replay state past undelivered batches. A nack now pins the checkpoint with an error log identifying the consequence — the semantics hardened across #4675/#4677/#4685.

Proof of Work

• New unit tests drive the real receive loop against a fake gRPC stream: a buffer of 1 against a 5-event batch delivers all 5 with zero drops once the consumer drains (the pre-fix code dropped 4 of them and advanced the cursor), and a loop parked on a backpressured send exits promptly on close. Run 10× under -race.
• Full salesforce packages green under -race; lint and docs clean.

Note for certification (tier C)

The package's integration tests require real Salesforce org credentials (SALESFORCE_ORG_URL etc.), so the adversarial proof is unit-level; a run against a real org with a deliberately slow consumer is recommended before merging. A pre-fix red-check was not mechanically possible because the fix changes receiveLoop's signature — the "must block, never drop" assertion is the new contract.

[claude]

Test coverage gap for the ack/nack semantics change (CONTRIBUTING.md §1.3.2 — "Tests should cover end-to-end functionality and prove that the connector works across supported configurations").

This PR changes ack semantics in two places — flushTopic's ackFn and emitSnapshot's ackFn — so that a nack pins the checkpoint instead of resolving it. The only tests added are for the gRPC backpressure path in salesforcegrpc/subscription_test.go; the checkpoint-pinning behaviour has no test at all, and there is currently no input_salesforce_cdc_test.go (only integration tests exist for this input).

This is the behaviour that matters most for correctness here: it only takes effect when auto_replay_nacks is disabled, which is exactly the configuration nobody exercises by accident. A unit test that tracks two batches against a checkpoint.Capped, nacks the first and acks the second, and asserts that e.state.Topics[topic] is not advanced (and that the same holds for e.state.RestCursor / SnapshotComplete on the snapshot path) would lock in the invariant the comments describe.

[squiidz]

Fixed in f8cf8bf — new unit tests drive flushTopic and emitSnapshot with real Capped trackers: nack the first batch, ack the second, and assert neither the topic replay ID nor RestCursor/SnapshotComplete is persisted past the rejected batch.

[squiidz]

Follow-up: the tests added for this were removed again in 6c4e103 along with the pinning behaviour they asserted — per the auto_replay_nacks contract (see the PR comment) a nack now resolves the checkpoint slot like an ack, so there is no pin-on-nack semantics left to cover.

[claude]

A permanently undecodable event does not "stall the topic loudly" — it produces an unbounded reconnect loop that keeps re-emitting duplicates downstream, and the input never sees an error.

Walking the path (subscription.go#L188-L206):

1. Events 0..k-1 of the batch are already pushed into eventBuffer; event k fails to decode, so failStream → return, and s.lastReplayID is never advanced (it is only updated after the events loop, #L243-L247).
2. reconnectWithBackoff reconnects with ReplayPreset_CUSTOM from that pre-batch replay ID (#L101-L106). reconnect_max_attempts defaults to 0 = unlimited (config.go#L179-L181), and the connection itself is healthy, so it always succeeds and returns nil.
3. The replayed batch decodes events 0..k-1 again (duplicates delivered downstream) and fails on k again — forever.

Because reconnection succeeds, s.streamErr stays nil, so subscribeAndPump's health tick on sub.StreamErr() (input_salesforce_cdc.go#L877-L883) never fires. eventsDecodeErrors / reconnectCount are only exposed via Health(), which has no caller in this repo, so the only signal is a log line repeating at the backoff interval while the topic silently makes no forward progress. That is the opposite of §1.2.2 ("Unexpected behavior should emit warning or error logs") being actionable, and it conflicts with the comment's stated intent.

Suggested fix: bound the retries for this specific failure — e.g. track consecutive decode failures at the same replay position and, past a small threshold, set s.streamErr / s.state = StreamStateDisconnected so subscribeAndPump surfaces the error and the pipeline fails loudly instead of spinning. Also worth noting that neither new test covers this reconnect-on-decode-failure path, only the buffer-backpressure and ack-nack paths.

[squiidz]

Fixed in 9cce455 — see the reply on the newer duplicate thread: consecutive decode failures at one replay position are bounded at 5, then the stream fails terminally via streamErr/Disconnected.

[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 (6c4e103): both ackFn nack pins (snapshot cursor and per-topic replay checkpoints), and the unit test file that asserted the pinning behavior is removed. Kept: the gRPC flow-control backpressure and schema/decode reconnect fixes with their tests.

[claude]

Turning a schema/decode failure into a reconnect does not produce the "stalls the topic loudly" behaviour the comment above describes — for a permanently undecodable payload it produces an unbounded redelivery loop.

s.lastReplayID is only advanced after the whole batch has been pushed (subscription.go#L242-L247), so failStream reconnects from a replay ID that points before the current batch. Salesforce then redelivers the batch: events preceding the bad one are decoded and pushed into eventBuffer again (and emitted downstream again), then the same event fails again, and the cycle repeats.

maxReconnect does not bound this: reconnectWithBackoff returns nil on a successful reconnect (subscription.go#L288-L332), so attempt restarts at 0 on every iteration and the "max reconnect attempts exceeded" guard never trips. streamErr is never set either, so subscribeAndPump's health tick (input_salesforce_cdc.go#L874-L880) never surfaces the failure — the input keeps running while re-emitting the same prefix of events forever. That is hard-to-diagnose error handling (CONTRIBUTING.md §3.2.2) rather than a loud stall.

Suggested fix: track consecutive decode failures at the same replay position and, once a small bound is exceeded, set s.streamErr/s.state = StreamStateDisconnected so the input fails visibly instead of looping — or give undecodable payloads an explicit terminal path. Either way the comment should describe the actual behaviour.

[squiidz]

Fixed in 9cce455. Consecutive schema/decode failures are now counted per replay position (recordDecodeFailure, cleared on any successful decode or when the failing position changes); past maxConsecutiveDecodeFailures=5 the stream fails terminally — streamErr set and state Disconnected, so subscribeAndPump's health tick surfaces it and the pipeline stops loudly instead of redelivering the batch prefix forever. Transient schema-fetch errors still heal by bounded redelivery. The comment now describes the actual behaviour.

[claude]

The new tests cover the buffer-backpressure fix well, but the PR's other behavioural change — schema-fetch / Avro-decode failure now reconnecting instead of skipping the event (subscription.go#L188-L206) — has no test. That path changes delivery semantics (skip → redeliver) and is the one most likely to regress.

fakeSubscribeStream already makes this cheap to cover: queue a FetchResponse whose event carries an unknown SchemaId or a corrupt payload, and assert the loop takes the failStream path rather than silently dropping the event (and that eventsDecodeErrors increments). Per CONTRIBUTING.md §1.3.2, tests should prove the connector works across the behaviours it claims.

[squiidz]

Addressed in 9cce455: TestReceiveLoopTerminalDecodeFailure drives an undecodable payload through the production receive loop and asserts the terminal outcome (streamErr surfaced, state Disconnected, no reconnect attempted), and TestRecordDecodeFailure locks in the counting semantics (same-position trip, new-position reset, clear on success). The intermediate reconnect-and-redeliver hop itself needs a real Pub/Sub connection, so it stays integration-level.

[claude]

The consecutive-failure bound is unreachable for any batch where the undecodable event is not the first one, so the infinite redelivery loop this commit is meant to bound still happens.

Walk a batch [e1 (decodable), e2 (undecodable)]:

1. e2 fails → failDecode → recordDecodeFailure(replay2) returns 1 >= 5 == false → failStream reconnects.
2. Reconnect resumes from s.lastReplayID, which was not advanced past this batch — the s.lastReplayID = resp.LatestReplayId update only runs after the per-event loop finishes (L301-L305). So the whole batch is redelivered, prefix included.
3. e1 decodes fine → clearDecodeFailures() on this line zeroes decodeFailures and nils decodeFailureReplayID.
4. e2 fails again → recordDecodeFailure sees !bytes.Equal(replay2, nil), resets the count to 0 and returns 1.

The count therefore oscillates 1 → 0 → 1 and never reaches maxConsecutiveDecodeFailures. streamErr is never set, the health tick never surfaces anything, and e1 is re-emitted downstream on every reconnect forever — exactly the failure mode described in the commit message. It only terminates in the special case where the undecodable event is the first event of the batch, which is also the only case the new tests cover (newBackpressureTestSubscription(..., 1, 1) / a single-event queue).

Suggested fix: stop resetting on any successful decode. Clear the tracked position only when the success is at decodeFailureReplayID (or track the failure per replay position and let a change of failing position, not an unrelated success, reset it). Please also add a test with a batch whose undecodable event is preceded by a decodable one — that is the realistic shape given BatchSize > 1.

Per CONTRIBUTING.md §3.1.4 ("The implementation is complete and correct, with no known bugs or missing core functionality") and §1.3.2 (tests prove the behaviour across supported configurations).

[squiidz]

Fixed in d90005a. clearDecodeFailures now takes the succeeding event's replay position and resets only when it matches the tracked failing position (a genuine heal); recordDecodeFailure already resets on a position change, so an unrelated prefix success leaves the count alone and it reaches the bound. New test TestReceiveLoopTerminalDecodeFailureAfterDecodablePrefix drives a [decodable, undecodable] batch through the receive loop and asserts the terminal outcome — proven red against the previous behavior (the loop attempted a reconnect instead of terminating).

[claude]

The "reconnect ⇒ redelivered" invariant does not hold when lastReplayID is still empty, so the first batch can still be silently lost.

The comment above states "lastReplayID still points before this batch, so it is redelivered", and failDecode → failStream → reconnectWithBackoff relies on that. But connectLocked only replays from a custom position when s.lastReplayID is non-empty:

if len(s.lastReplayID) > 0 {
    fetchReq.ReplayPreset = ReplayPreset_CUSTOM
    fetchReq.ReplayId = s.lastReplayID
} else {
    fetchReq.ReplayPreset = s.config.ReplayPreset
}

s.lastReplayID is only advanced after the whole resp.Events loop completes (or from an empty-events keepalive). So on a fresh subscription with no persisted replay ID and the default replay_preset: latest (input_salesforce_cdc.go:122), a schema-fetch or decode failure in the first batch reconnects with ReplayPreset_LATEST: the failing event, the rest of its batch, and everything published during the backoff are dropped without ever reaching the consumer. The consecutive-failure bound also never trips, because the position is never redelivered — so the loss is silent rather than loud, which is the exact failure mode this change is meant to eliminate.

Suggested fix: capture a resume position before the first response is fully processed (e.g. seed lastReplayID from resp.LatestReplayId/the first event's replay ID prior to the decode attempt, or refuse to reconnect via the configured preset when a decode failure is pending) so the redelivery guarantee holds for the first batch too.

Refs: CONTRIBUTING.md §5.4.2 (at-least-once delivery), and subscription.go#L110-L121.

[claude]

Schema-fetch failures and undecodable payloads share one counter, so a transient outage terminates the topic with a misleading "permanently undecodable" error.

Both call sites feed the same failDecode path: GetSchema errors (network/auth/5xx against the schema endpoint — subscription.go:257-262) and DecodeAvroPayload errors (a genuinely bad payload — subscription.go:264-269). Once five consecutive attempts at the same replay position fail for any of those reasons, the stream is failed terminally and the topic surfaces ... treating as permanently undecodable: get schema for event (schemaID=…): <connection refused>.

With reconnect_min_delay: 500ms / reconnect_max_delay: 30s, five consecutive attempts elapse in roughly a minute, so a schema-endpoint outage or an expired-token window longer than that permanently kills the topic — even though reconnect_max_attempts defaults to 0 (unlimited), i.e. the user explicitly asked for indefinite retry on transport failures. The error text then points a support engineer at the payload rather than at the actual cause.

Suggested fix: only count DecodeAvroPayload failures toward maxConsecutiveDecodeFailures (a decode failure against a successfully fetched schema really is deterministic), and route GetSchema failures through the normal reconnect/backoff path so they stay governed by reconnect_max_attempts. If they must share the bound, the terminal error should name the actual failing stage rather than asserting the payload is undecodable.

Refs: CONTRIBUTING.md §3.2.2 / §1.2.2 — poor error handling / difficult-to-diagnose failures, and subscription.go#L257-L269.