salesforce_cdc: apply backpressure instead of dropping events on full buffer (CON-504) #4689
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -9,6 +9,7 @@ | |
| package salesforcegrpc | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
|
|
@@ -30,6 +31,14 @@ import ( | |
| // provides safety margin without meaningfully slowing startup. | ||
| const subscribeSettleDelay = 5 * time.Second | ||
|
|
||
| // maxConsecutiveDecodeFailures bounds redelivery attempts for an event whose | ||
| // schema fetch or Avro decode keeps failing at the same replay position. Each | ||
| // failure reconnects and redelivers the batch (transient schema-fetch errors | ||
| // heal that way); once the same position has failed this many times in a row | ||
| // the payload is treated as permanently undecodable and the stream fails | ||
| // loudly instead of redelivering the batch prefix forever. | ||
| const maxConsecutiveDecodeFailures = 5 | ||
|
|
||
| // Subscription owns one subscribe stream for a single Pub/Sub topic. It reuses | ||
| // the parent Client's connection, auth, and schema cache. | ||
| type Subscription struct { | ||
|
|
@@ -46,6 +55,11 @@ type Subscription struct { | |
| ready chan struct{} | ||
| streamErr error | ||
| state StreamState | ||
| // decodeFailures counts consecutive schema/decode failures at | ||
| // decodeFailureReplayID; guarded by mu. Reset on any successful decode or | ||
| // when the failing position changes. | ||
| decodeFailures int | ||
| decodeFailureReplayID []byte | ||
|
|
||
| // Atomic counters for health reporting. | ||
| eventsReceived atomic.Int64 | ||
|
|
@@ -121,22 +135,87 @@ func (s *Subscription) connectLocked(ctx context.Context) error { | |
| // so waiting for the first Recv would block indefinitely on idle topics. | ||
| s.markReadyLocked() | ||
|
|
||
| go s.receiveLoop(ctx, streamCtx) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // failDecode routes a schema/decode failure: reconnect-and-redeliver while | ||
| // the position is under the consecutive-failure bound, terminal stream | ||
| // failure once it is exceeded. | ||
| func (s *Subscription) failDecode(replayID []byte, failStream func(error), err error) { | ||
| if !s.recordDecodeFailure(replayID) { | ||
| failStream(err) | ||
| return | ||
| } | ||
| terminalErr := fmt.Errorf("decoding event at replay position %x: %d consecutive failures, treating as permanently undecodable: %w", replayID, maxConsecutiveDecodeFailures, err) | ||
| s.lastError.Store(terminalErr) | ||
| s.lastErrorTime.Store(time.Now().UnixNano()) | ||
| s.mu.Lock() | ||
| s.streamErr = terminalErr | ||
| s.state = StreamStateDisconnected | ||
| s.mu.Unlock() | ||
| s.client.log.Errorf("Pub/Sub stream failed permanently (topic=%s): %v", s.config.TopicName, terminalErr) | ||
| } | ||
|
|
||
| // recordDecodeFailure counts a schema/decode failure at the given replay | ||
| // position and reports whether the position has now failed | ||
| // maxConsecutiveDecodeFailures times in a row. | ||
| func (s *Subscription) recordDecodeFailure(replayID []byte) bool { | ||
| s.mu.Lock() | ||
| defer s.mu.Unlock() | ||
| if !bytes.Equal(replayID, s.decodeFailureReplayID) { | ||
| s.decodeFailureReplayID = append([]byte(nil), replayID...) | ||
| s.decodeFailures = 0 | ||
| } | ||
| s.decodeFailures++ | ||
| return s.decodeFailures >= maxConsecutiveDecodeFailures | ||
| } | ||
|
|
||
| // clearDecodeFailures resets the consecutive-failure count when the event at | ||
| // the tracked failing position decodes successfully. Successes at OTHER | ||
| // positions must not reset it: a reconnect redelivers the whole batch, so the | ||
| // decodable events preceding a permanently undecodable one succeed on every | ||
| // redelivery cycle - clearing on any success would keep the count oscillating | ||
| // below the bound forever. | ||
| func (s *Subscription) clearDecodeFailures(replayID []byte) { | ||
| s.mu.Lock() | ||
| if s.decodeFailureReplayID != nil && bytes.Equal(replayID, s.decodeFailureReplayID) { | ||
| s.decodeFailures = 0 | ||
| s.decodeFailureReplayID = nil | ||
| } | ||
| s.mu.Unlock() | ||
| } | ||
|
|
||
| // receiveLoop reads from the gRPC stream and pushes decoded events into the | ||
| // buffer. On stream errors it attempts reconnection with backoff instead of | ||
| // exiting. streamCtx is this stream's cancellation context: it unblocks a | ||
| // backpressured buffer send when the subscription closes or reconnects. | ||
| func (s *Subscription) receiveLoop(ctx, streamCtx context.Context) { | ||
| // Capture done at goroutine start. reconnectWithBackoff → connectLocked | ||
| // replaces s.done with a fresh channel for the new goroutine; closing the | ||
| // old reference here prevents a double-close panic when both goroutines | ||
| // eventually return. | ||
| done := s.done | ||
| defer close(done) | ||
|
|
||
| // failStream logs err and hands control to the reconnect path. Because | ||
| // s.lastReplayID has not been advanced past the current batch, the | ||
| // reconnected stream redelivers it: duplicates, never loss. | ||
|
There was a problem hiding this comment. This comment states the opposite of the invariant the same PR introduces. "Because 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).
Contributor
Author
There was a problem hiding this comment. 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). |
||
| failStream := func(err error) { | ||
| s.client.log.Errorf("Pub/Sub stream error (topic=%s), reconnecting: %v", s.config.TopicName, err) | ||
| s.lastError.Store(err) | ||
| s.lastErrorTime.Store(time.Now().UnixNano()) | ||
|
|
||
| if reconnErr := s.reconnectWithBackoff(ctx); reconnErr != nil { | ||
| s.mu.Lock() | ||
| s.streamErr = reconnErr | ||
| s.state = StreamStateDisconnected | ||
| s.mu.Unlock() | ||
| s.client.log.Errorf("Reconnection failed permanently (topic=%s): %v", s.config.TopicName, reconnErr) | ||
| } | ||
| } | ||
|
|
||
| for { | ||
| resp, err := s.stream.Recv() | ||
| if err != nil { | ||
|
|
@@ -148,17 +227,7 @@ func (s *Subscription) receiveLoop(ctx context.Context) { | |
| } | ||
| s.mu.Unlock() | ||
|
|
||
| failStream(err) | ||
| return | ||
| } | ||
|
|
||
|
|
@@ -177,19 +246,28 @@ func (s *Subscription) receiveLoop(ctx context.Context) { | |
| continue | ||
| } | ||
|
|
||
| // A schema fetch or decode failure must not skip the event: the | ||
| // batch's replay ID would advance past it and the event would be | ||
| // silently lost. Reconnect instead — lastReplayID still points | ||
| // before this batch, so it is redelivered and transient failures | ||
| // (schema fetch) heal on retry. Redelivery is bounded: once the | ||
| // same position fails maxConsecutiveDecodeFailures times in a row | ||
| // the stream fails terminally (streamErr is surfaced through the | ||
| // health tick) instead of re-emitting the batch prefix forever. | ||
| schema, err := s.client.schemaCache.GetSchema(ctx, event.SchemaId) | ||
| if err != nil { | ||
| s.eventsDecodeErrors.Add(1) | ||
| s.failDecode(consumerEvent.ReplayId, failStream, fmt.Errorf("get schema for event (schemaID=%s): %w", event.SchemaId, err)) | ||
|
There was a problem hiding this comment. The "reconnect ⇒ redelivered" invariant does not hold when The comment above states "lastReplayID still points before this batch, so it is redelivered", and if len(s.lastReplayID) > 0 {
fetchReq.ReplayPreset = ReplayPreset_CUSTOM
fetchReq.ReplayId = s.lastReplayID
} else {
fetchReq.ReplayPreset = s.config.ReplayPreset
}
Suggested fix: capture a resume position before the first response is fully processed (e.g. seed Refs:
Contributor
Author
There was a problem hiding this comment. 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. |
||
| return | ||
| } | ||
|
Comment on lines
430
to
443
There was a problem hiding this comment. 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" ( Failure scenario — a schema whose
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 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 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).
Contributor
Author
There was a problem hiding this comment. 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. |
||
|
|
||
| decoded, err := DecodeAvroPayload(schema, event.Payload) | ||
| if err != nil { | ||
| s.eventsDecodeErrors.Add(1) | ||
| s.failDecode(consumerEvent.ReplayId, failStream, fmt.Errorf("decode Avro payload (schemaID=%s): %w", event.SchemaId, err)) | ||
| return | ||
| } | ||
| s.clearDecodeFailures(consumerEvent.ReplayId) | ||
|
|
||
| pubsubEvent := &PubSubEvent{ | ||
| ReplayID: consumerEvent.ReplayId, | ||
|
|
@@ -209,14 +287,20 @@ func (s *Subscription) receiveLoop(ctx context.Context) { | |
| } | ||
| } | ||
|
|
||
| // A full buffer applies backpressure instead of dropping: while | ||
| // this send blocks, no flow-control FetchRequest is issued, so | ||
| // Salesforce stops sending and the replay cursor cannot advance | ||
| // past an undelivered event. The stream context unblocks the send | ||
| // on close or reconnect. | ||
| select { | ||
| case s.eventBuffer <- pubsubEvent: | ||
| s.eventsReceived.Add(1) | ||
| s.lastEventTime.Store(time.Now().UnixNano()) | ||
| s.client.log.Debugf("Pub/Sub event received (topic=%s, schemaID=%s, replayID=%x)", pubsubEvent.TopicName, pubsubEvent.SchemaID, pubsubEvent.ReplayID) | ||
| case <-streamCtx.Done(): | ||
| return | ||
| case <-ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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
failDecodepath:GetSchemaerrors (network/auth/5xx against the schema endpoint —subscription.go:257-262) andDecodeAvroPayloaderrors (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 thoughreconnect_max_attemptsdefaults to0(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
DecodeAvroPayloadfailures towardmaxConsecutiveDecodeFailures(a decode failure against a successfully fetched schema really is deterministic), and routeGetSchemafailures through the normal reconnect/backoff path so they stay governed byreconnect_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, andsubscription.go#L257-L269.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
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.