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.

[squiidz]

Fixed in 844c3af. The replay anchor now advances past each event as it is buffered, so any later failure reconnects from exactly the failing event — the first-batch case included, and re-emitted prefixes are gone as a side effect. When NO anchor exists (nothing delivered yet on a fresh stream) the reconnect path is never taken: schema fetches retry inline with backoff, and an undecodable first event fails the stream terminally and loudly (streamErr surfaced) instead of reconnecting via the preset and losing the batch silently. New tests cover the unanchored-terminal path and per-event anchor advancement.

[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.

[squiidz]

Fixed in 844c3af — only DecodeAvroPayload failures against a successfully fetched schema (deterministic) count toward the bound; GetSchema failures route through the normal reconnect path governed by reconnect_max_attempts, and the terminal error wording names the actual stage.

[claude]

Hardcoded retry delay — not YAML-configurable.

unanchoredSchemaRetryDelay is a new retry backoff baked into the package. The project Go patterns are explicit:

Configurable Time Parameters — Every time-related value (timeouts, backoffs, intervals, retry delays) must be exposed as a YAML-configurable field. Do not hardcode durations.

Every other timing knob on this connector (grpc.reconnect_*, timeouts) is user-facing; this one is not. It should be either derived from the existing gRPC backoff settings on s.client (baseBackoff/maxBackoff, already available here) or exposed as a config field.

Secondary point: the doc comment says the value is a var "so tests can shorten it", but no test in this PR sets it — the fresh-stream schema-fetch retry path added at subscription.go#L284-L301 is the one new branch with no coverage, while the decode branches all get tests. Per CONTRIBUTING.md §1.3.2, a mutable global that exists purely as a test seam should have the test that uses it.

[squiidz]

Fixed in 810750f — the hardcoded delay (and the untested test-seam var) are gone: the inline retry is now driven by grpcBackoffWithJitter over the connector's existing reconnect_min_delay/reconnect_max_delay settings, so no new knob was needed. The previously-uncovered fresh-stream retry branch now has a test (TestReceiveLoopUnanchoredSchemaRetryHonorsReconnectPolicy) driving a failing schema endpoint through the receive loop.

[claude]

The unanchored schema-fetch retry budget reuses an unrelated constant and bypasses the user's reconnect policy.

Two problems with this loop:

1. Wrong constant. maxConsecutiveDecodeFailures is used as the schema-fetch retry count, but failDecode's own doc comment states the opposite intent:

   Schema-fetch failures do NOT come through here - they are transport-class errors governed by the reconnect policy, not evidence of an undecodable payload.

   That split is the whole point of commit 844c3af. Here the decode bound leaks back into the schema-fetch path as a magic bound. The constant's own doc at subscription.go#L34-L40 is also stale for the same reason — it still says it bounds "an event whose schema fetch or Avro decode keeps failing" and talks about "redelivering the batch prefix forever", neither of which is true after the anchor now advances per event.

2. Fixed ~10s budget, ignoring reconnect_max_attempts/backoff. Attempts sleep 1s+2s+3s+4s, then failTerminal kills the topic permanently. Commit 844c3af's message identifies exactly this failure mode as the bug it fixes:

   a ~1 minute schema-endpoint outage terminally killed the topic with a misleading 'permanently undecodable' error even under reconnect_max_attempts: 0 (unlimited)

   The fix only covers the anchored path. On a fresh stream — which is every process start with replay_preset: latest, and every topic that has not yet delivered an event — a ~15s Salesforce schema-endpoint blip still fails the input terminally, with reconnect_max_attempts: 0 silently ignored. Reconnecting is genuinely unsafe here (LATEST would drop the batch), but the inline retry should be governed by the same configured backoff/attempt policy rather than a hardcoded ~10s, per CONTRIBUTING.md §3.2.3 (unfamiliar/confusing UX) — a documented reconnect setting that does not apply on the first batch is hard to reason about.

[squiidz]

Fixed in 810750f on both points. The retry budget is now reconnect_max_attempts itself — 0 retries indefinitely, matching what the user asked for on transport failures — with the configured jittered backoff between attempts, so a schema-endpoint blip on a fresh stream no longer terminally kills the topic under an unlimited policy. When a bounded policy is exhausted the terminal error names the schema stage and the attempt budget rather than claiming the payload is undecodable. The stale maxConsecutiveDecodeFailures doc is rewritten for what it bounds post-split (deterministic per-position decode failures only). Reconnecting stays deliberately off this path: with no anchor, the preset could silently drop the batch.

[claude]

Anchored schema-fetch failures are not actually bounded by the reconnect policy — a permanently unfetchable schema loops forever without ever surfacing an error.

The comment above claims schema-fetch failures "stay governed by the reconnect policy" (subscription.go#L273-L275), but reconnect_max_attempts cannot bound this path: reconnectWithBackoff restarts attempt at 0 on every call and returns nil as soon as connectLocked succeeds (subscription.go#L417-L461). Since the reconnect itself always succeeds here, the budget is never approached.

Failure scenario — a schema whose SchemaJson fails to compile (avro.Parse error in avro.go#L81-L84), or a schemaID that returns a permanent NotFound/PermissionDenied:

1. GetSchema fails deterministically; an anchor exists, so failStream is called and the stream reconnects from lastReplayID.
2. Salesforce redelivers starting at the same event; GetSchema fails again identically.
3. streamErr is never set (the reconnect succeeded), so the health tick in input_salesforce_cdc.go#L874-L880 never observes anything, and handleStreamErr is never reached.

Result: the topic reconnects roughly once per backoff interval indefinitely, never advances its replay position, and never fails — the pipeline hangs silently apart from a repeated Errorf. This is the same unbounded-redelivery livelock that maxConsecutiveDecodeFailures was introduced for in 9cce455, reintroduced for the schema-fetch class when 844c3af split schema fetches out of that bound. Note the unanchored path already terminates correctly (lines 288–310), so the two branches disagree on whether a permanently failing schema fetch is survivable.

Suggested fix: give this path its own bound rather than relying on the per-call reconnect budget — e.g. track consecutive schema-fetch failures at the same replay position (mirroring recordDecodeFailure/clearDecodeFailures), or carry a reconnect budget across failStream calls that are triggered by schema fetches, so an uncompilable/inaccessible schema eventually reaches failTerminal with an error naming the schema stage — the same terminal outcome the unanchored branch already produces.

Rules referenced: CONTRIBUTING.md §3.1.4 ("implementation is complete and correct, with no known bugs"), §3.2.2 ("poor error handling or difficult-to-diagnose bugs"), and §1.2.2 (unexpected behavior should surface, not silently stall).

[squiidz]

Confirmed and fixed in f9f6346. Consecutive schema-fetch failures are now counted per replay position (the decode counter's machinery, extracted as a shared positionFailures type) with reconnect_max_attempts as the budget — counted across failStream cycles precisely because each individual reconnect succeeds, so the per-call budget alone can never trip. 0 keeps the documented opt-in to indefinite retries; a bounded policy now genuinely bounds this path and reaches failTerminal with an error naming the schema stage and the exhausted budget, matching the unanchored branch. Success at the tracked position clears the count; unrelated successes never do. Covered by counting-semantics and exhausted-budget receive-loop tests, and the stale 'stay governed by the reconnect policy' comment now describes the actual mechanism.

[claude]

The new schema-fetch bound never engages under the shipped default, so the livelock this commit targets persists out of the box.

recordSchemaFailure returns budget > 0 && ... (subscription.go:241), and reconnect_max_attempts defaults to 0:

connect/internal/impl/salesforce/config.go

Lines 179 to 181 in f9f6346

	service.NewIntField(sfFieldGRPCReconnectMaxAttempts).
	Description("Maximum number of gRPC reconnection attempts. 0 means unlimited.").
	Default(0),

So with the default config a deterministically unfetchable schema (uncompilable SchemaJson, permanent NotFound) takes failStream → reconnectWithBackoff on every redelivery, forever: streamErr is never set, the health tick in subscribeAndPump stays blind, and the pipeline spins re-subscribing indefinitely — exactly the failure mode the commit message says this bound exists to break. The commit message describes 0 as "the user's explicit opt-in to indefinite retries", but nobody has to opt in — it is the default value, so the guard is inert for every user who does not set the field.

Suggested fix: give this path its own bound analogous to maxConsecutiveDecodeFailures (used only when reconnect_max_attempts is unlimited), or treat 0 as "unbounded reconnects, still bounded per replay position". Either way the default configuration should surface a terminal error rather than reconnect-redeliver forever, per CONTRIBUTING.md §3.1.4 (complete and correct, no known missing core functionality) and §1.2.2 (unexpected behavior emits warning/error logs and is diagnosable).

[squiidz]

Fixed in a66868a. Schema failures are now classified: deterministic ones — NotFound, InvalidArgument, PermissionDenied, Unimplemented, and compile failures (no gRPC status) — cannot heal by retrying, so they trip at maxConsecutiveDecodeFailures regardless of the reconnect policy; the shipped default no longer livelocks. Transport-class failures stay governed by the policy (0 = indefinite retries), preserving the earlier outage-must-not-kill-the-topic fix. Both the anchored and unanchored paths apply the same classification, and the counting test now covers the deterministic-under-default-policy case.

[claude]

Terminal schema-fetch errors now put a wrapped gRPC status into streamErr, where the stale-replay handler can misread it and delete the durable checkpoint.

Both new terminal schema paths (this one and subscription.go:365) wrap the error from SchemaCache.GetSchema, which itself wraps the raw gRPC error with %w (avro.go:78). That error becomes s.streamErr, and the input's health tick feeds it straight into handleStreamErr:

connect/internal/impl/salesforce/input_salesforce_cdc.go

Lines 972 to 987 in f9f6346

	func (e *salesforceCDCInputExecutor) handleStreamErr(ctx context.Context, topic string, err error) bool {
	if grpcErr, ok := status.FromError(err); ok && grpcErr.Code() == codes.InvalidArgument {
	e.logger.Warnf("topic %s replay_id rejected (%s); clearing and reconnecting via configured preset", topic, grpcErr.Message())
	e.stateMu.Lock()
	delete(e.state.Topics, topic)
	saveErr := e.saveStateLocked(ctx)
	e.stateMu.Unlock()
	if saveErr != nil {
	e.logger.Errorf("clear stale replay_id for %s: %v", topic, saveErr)
	}
	return true
	}
	return false
	}

status.FromError unwraps through %w (it uses errors.As on the GRPCStatus() interface), so if the Pub/Sub GetSchema RPC ever fails with codes.InvalidArgument (e.g. a malformed/rejected schema ID), this branch fires for a schema error: the persisted replay ID for the topic is deleted and the topic resubscribes via the configured preset — with the default LATEST that silently skips every event between the checkpoint and now. Before this change only Recv()/reconnect errors reached streamErr, so the InvalidArgument heuristic could only mean "replay ID rejected"; that assumption no longer holds.

Suggested fix: make the terminal decode/schema failures distinguishable (a sentinel or typed error checked with errors.Is/errors.As) and restrict the stale-replay reset in handleStreamErr to stream-level errors, so a schema-stage failure can never clear the checkpoint. Silent skip-ahead is the exact loss class this PR is closing elsewhere (CONTRIBUTING.md §5.4.1/§5.4.2 — durable checkpoint and at-least-once delivery).

[squiidz]

Fixed in a66868a — failTerminal now wraps its verdicts in a TerminalStreamError marker (Unwrap preserved for transparency), and handleStreamErr checks for it before the InvalidArgument stale-replay heuristic: a terminal schema/decode verdict can never clear the persisted replay ID. Terminal errors propagate as fatal topic errors with the checkpoint intact, so the stale-replay reset is once again reachable only by genuine stream-level errors.

[claude]

TerminalStreamError is never constructed, so the checkpoint shield this PR adds is dead code.

failTerminal stores the raw err into s.streamErr:

connect/internal/impl/salesforce/salesforcegrpc/subscription.go

Lines 160 to 168 in a66868a

	func (s *Subscription) failTerminal(err error) {
	s.lastError.Store(err)
	s.lastErrorTime.Store(time.Now().UnixNano())
	s.mu.Lock()
	s.streamErr = err
	s.state = StreamStateDisconnected
	s.mu.Unlock()
	s.client.log.Errorf("Pub/Sub stream failed permanently (topic=%s): %v", s.config.TopicName, err)
	}

TerminalStreamError is declared and documented here but has no construction site anywhere in the repo — the only other reference is the errors.As match in the consumer:

connect/internal/impl/salesforce/input_salesforce_cdc.go

Lines 978 to 983 in a66868a

	func (e *salesforceCDCInputExecutor) handleStreamErr(ctx context.Context, topic string, err error) bool {
	var terminal *salesforcegrpc.TerminalStreamError
	if errors.As(err, &terminal) {
	return false
	}
	if grpcErr, ok := status.FromError(err); ok && grpcErr.Code() == codes.InvalidArgument {

So errors.As(err, &terminal) can never be true, and the exact failure this PR's final commit describes ("a terminal schema error carrying InvalidArgument would have DELETED the topic's durable checkpoint and resubscribed via the preset") is still live:

1. deterministicSchemaFailure classifies codes.InvalidArgument as deterministic (L207), so a schema fetch rejected with InvalidArgument trips the bound and calls failTerminal(fmt.Errorf("fetching schema ...: %w", err)) (L407-L409).
2. StreamErr() hands that unwrapped error to handleStreamErr (L875-L876).
3. The terminal guard misses; status.FromError (grpc-go v1.82.1 resolves a GRPCStatus() through the error chain via errors.As) matches InvalidArgument, so delete(e.state.Topics, topic) runs and the durable checkpoint is destroyed — then the topic resubscribes via the preset, silently skipping everything since the checkpoint under the default LATEST.

Fix: have failTerminal wrap its argument in &TerminalStreamError{Err: err} before storing it in streamErr/lastError (or construct the wrapper at each failTerminal call site), so the guard actually engages.

Related gap: handleStreamErr has no unit test — internal/impl/salesforce has only integration tests for the CDC input, so nothing exercises the new terminal branch. A table-driven test over handleStreamErr (terminal-wrapped InvalidArgument → false and checkpoint intact; bare InvalidArgument → true) would have caught this and would lock the behavior in, per the test patterns referenced from CLAUDE.md.

[squiidz]

Fixed in 6eb239f — and the finding was doubly right: the previous commit's failTerminal edit had silently failed to apply, which is exactly the class of gap the missing consumer-side test would have caught. failTerminal now genuinely wraps (the receive-loop terminal tests assert the marker type reaches StreamErr), and the suggested table-driven handleStreamErr test is added in the input package: bare/stream-wrapped InvalidArgument resets the checkpoint, terminal verdicts never touch it, other errors are fatal with the checkpoint intact. The same commit also fixes a panic this surfaced: lastError is an atomic.Value and one stream lifetime stores differently-typed errors (wrapped fmt, gRPC status, now TerminalStreamError), so all stores are boxed through one wrapper type, with a mixed-type regression test.

[claude]

This doc comment states the opposite of what the code now does.

Schema-fetch failures never count toward this bound - they are transport-class errors governed by the reconnect policy.

recordSchemaFailure bounds deterministic schema-fetch failures by exactly this constant, explicitly regardless of the reconnect policy:

connect/internal/impl/salesforce/salesforcegrpc/subscription.go

Lines 275 to 281 in a66868a

	count := s.schemaFailures.record(replayID)
	if deterministic {
	return count >= maxConsecutiveDecodeFailures
	}
	budget := s.client.maxReconnect
	return budget > 0 && count >= budget
	}

and so does the unanchored inline-retry path:

connect/internal/impl/salesforce/salesforcegrpc/subscription.go

Lines 372 to 376 in a66868a

	for attempt := 0; err != nil; attempt++ {
	if deterministicSchemaFailure(err) && attempt >= maxConsecutiveDecodeFailures-1 {
	// A bad or inaccessible schema cannot heal by retrying.
	break
	}

failDecode's doc carries the same stale claim ("Schema-fetch failures do NOT come through here - they are transport-class errors governed by the reconnect policy", L172-L174): the first clause is still true, but the "transport-class / governed by the reconnect policy" characterisation no longer holds for the deterministic class.

A comment that contradicts the failure-handling policy it documents is a maintenance hazard on exactly the semantics this PR is iterating on — CONTRIBUTING.md §3.1.2 (self-documenting code) and §1.2.3 (known limitations and edge cases documented accurately). Suggest rewording both to say that only transient schema-fetch failures are governed by the reconnect policy, while deterministic ones share this bound, and renaming the constant to something that covers both paths it now bounds.

[squiidz]

Fixed in 6eb239f — the constant is renamed maxDeterministicPositionFailures (it bounds both decode failures and deterministic schema failures), and the failDecode, struct-field, and batch-loop comments now all state the deterministic/transient split: only transient schema-fetch failures are governed by the reconnect policy.

[claude]

Off-by-one in the transient schema-fetch budget: reconnect_max_attempts: 1 kills the topic with zero retries.

count here is the number of failures recorded at the position, while budget is reconnect_max_attempts — a number of attempts. Terminating on count >= budget therefore allows budget - 1 redelivery attempts:

• reconnect_max_attempts: 1 → the first transient schema-fetch failure returns true, so failTerminal fires with "failed repeatedly at replay position … with no sign of healing" before a single reconnect/redelivery has been tried. A one-second blip on the schema endpoint permanently fails the topic, and because the error now carries TerminalStreamError, handleStreamErr treats it as fatal for the topic.
• The message is also inaccurate in that case — nothing was retried.

This also contradicts the two paths this change is meant to align with, for every value of the knob:

• unanchored inline retry: attempt >= s.client.maxReconnect breaks at attempt == budget, i.e. budget retries (budget + 1 failures).
• reconnectWithBackoff makes budget connect attempts before giving up.

Suggested fix: trip on count > budget (or record the retry count rather than the failure count) so the anchored path grants the same budget redelivery attempts as the unanchored one, and extend TestRecordSchemaFailure with a maxReconnect: 1 case — the current table only covers 3 and 0, which is why the skew is invisible.

Refs: CONTRIBUTING §3.1.4 (complete and correct, no known bugs), §1.2.4 (config behaves as documented).

[squiidz]

Fixed in b3c2a32 — count > budget, so reconnect_max_attempts grants exactly that many redelivery attempts on the anchored path, matching the unanchored inline retry and reconnectWithBackoff. TestRecordSchemaFailure now includes the maxReconnect: 1 case (first failure earns its retry, second is terminal).

[claude]

Malformed RCL license URL in the new file's header.

The URL has an extra /v4 path segment (redpanda-data/connect/v4/blob/main/...), which is a 404 — the Go module path leaked into the link. Every other RCL header in this package uses the canonical form, including the other file added in this PR (subscription_test.go#L1-L7) and avro.go#L1-L7:

// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md

Drop the /v4 so the header matches the enterprise header format exactly (CONTRIBUTING §6: "A license header on every new .go file (including test and benchmark helpers), matching the component's distribution"; CLAUDE.md — "License headers matter: CI fails if headers don't match the component's distribution classification").

[squiidz]

Fixed in b3c2a32 — /v4 dropped, matching the package's canonical RCL header.

[claude]

This comment states the opposite of the invariant the same PR introduces.

"Because s.lastReplayID has not been advanced past the current batch, the reconnected stream redelivers it" was true before this PR, when the anchor only moved at batch boundaries. The buffer-send branch added below now advances it per delivered event (L461-L472), so on reconnect redelivery resumes from the last buffered event, not from the start of the batch — which is precisely the behaviour the per-event anchor exists to provide (no re-emitted prefix), and it is what the batch-loop comment at L359-L374 already says.

Since the whole no-silent-loss argument in this PR rests on exactly this invariant, the stale wording is actively misleading for the next reader. Suggest rewording to state that the anchor is not advanced past the failing event, so redelivery resumes at it.

Ref: CONTRIBUTING §3.1.2 (self-documenting code).

[squiidz]

Fixed in b3c2a32 — the comment now states the per-event invariant: the anchor never advances past the failing event, so redelivery resumes exactly there (duplicates at most, no re-emitted prefix).

[claude]

reconnect_max_attempts gains new user-visible semantics that its documentation does not mention.

This PR makes reconnect_max_attempts govern two things beyond reconnection: the inline schema-fetch retry budget on a fresh stream (here) and the per-position schema-failure budget on an anchored stream (L281-L290). Exhausting it now permanently fails the topic with a TerminalStreamError that the input deliberately refuses to recover from (handleStreamErr).

The field still documents only the old meaning (config.go#L179-L181):

Maximum number of gRPC reconnection attempts. 0 means unlimited.

A user who sets a small bound to cap reconnect churn has no way to know they are also choosing when a transient schema-endpoint failure becomes fatal for the topic. Please extend the field description (and re-run task docs) to cover the schema-fetch retry budget and the terminal outcome, and note that deterministic schema failures trip maxDeterministicPositionFailures regardless of this setting.

Refs: CONTRIBUTING §1.1.1 (documentation with configuration examples), §1.2.3 (known limitations and edge cases documented).

[squiidz]

Fixed in b3c2a32 — the field description (and regenerated docs) now state that reconnect_max_attempts also bounds transient schema-fetch retries with a permanent topic failure when exhausted, and that deterministic schema/decode failures give up after a fixed number of attempts regardless of this setting.

[claude]

The unanchored inline schema retry never refreshes credentials, so an expired token stalls the topic indefinitely.

This loop re-fetches the schema with the credentials already held by SchemaCache, and it neither stores the failure in lastError nor asks the client to refresh auth. refreshAuth is only ever invoked from reconnectWithBackoff (subscription.go#L546-L562), which this branch deliberately avoids, and nothing else in the connector refreshes the gRPC/schema-cache token (Client.UpdateAuth has refreshAuth as its only caller).

Failure scenario: no persisted replay ID for the topic (first run, or right after handleStreamErr cleared a stale checkpoint) and the first event arrives once the access token has expired — e.g. a low-traffic topic, or a long REST snapshot phase before streaming starts. GetSchema returns codes.Unauthenticated, which deterministicSchemaFailure classifies as transient (subscription.go#L210-L225), so under the shipped default reconnect_max_attempts: 0 the loop retries forever with the same dead token. streamErr/lastError are never set, so the input's health tick stays blind and the topic stalls silently until the process is restarted; under a bounded policy it instead dies with a misleading "retry budget is exhausted" error that a token refresh would have avoided. The anchored path recovers from exactly this case, so the two branches disagree.

Suggested fix: in this loop, record the error (s.lastError.Store(storedErr{err: err})) and call s.client.refreshAuth(ctx) when the failure carries codes.Unauthenticated before the next attempt, so the retry uses freshly-resolved credentials — per CONTRIBUTING.md §5.4.5 ("on connection loss, reconnect with freshly-resolved credentials") and §3.1.7.

[squiidz]

Fixed in 04c938f — the inline retry now mirrors everything reconnectWithBackoff does short of reconnecting: each failure is recorded in lastError (so Health() sees the stall), and an Unauthenticated failure triggers refreshAuth before the next attempt, propagating fresh credentials to the schema cache via UpdateAuth. Covered by a recovery test: GetSchema fails Unauthenticated until the refresh hook fires, then the event delivers inline with zero preset reconnects, with Health visibility asserted. Cross-checked the two branches for any remaining asymmetry — backoff/jitter, error recording, and auth refresh are the complete set reconnectWithBackoff contributes, and all three are now mirrored.