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

internal/impl/salesforce/input_salesforce_cdc.go

@@ -727,6 +727,10 @@ func (e *salesforceCDCInputExecutor) emitSnapshot(
 		return fmt.Errorf("track snapshot checkpoint: %w", err)
 	}
 
+	// The ack error is deliberately ignored: nacks are replayed by
+	// auto_replay_nacks (the default), and disabling that is a documented
+	// opt-in to DROP rejected messages, so the checkpoint must advance past
+	// them rather than pin the tracker.
 	ackFn := func(ackCtx context.Context, _ error) error {
 		resolved := resolveFn()
 		if resolved == nil {
@@ -740,11 +744,11 @@ func (e *salesforceCDCInputExecutor) emitSnapshot(
 		e.stateMu.Lock()
 		e.state.SnapshotComplete = acked.SnapshotComplete
 		e.state.RestCursor = acked.RestCursor
-		err := e.saveStateLocked(ackCtx)
+		persistErr := e.saveStateLocked(ackCtx)
 		e.stateMu.Unlock()
 
-		if err != nil {
-			return fmt.Errorf("persist snapshot checkpoint: %w", err)
+		if persistErr != nil {
+			return fmt.Errorf("persist snapshot checkpoint: %w", persistErr)
 		}
 		return nil
 	}
@@ -900,6 +904,10 @@ func (e *salesforceCDCInputExecutor) flushTopic(
 		return fmt.Errorf("track checkpoint for %s: %w", topic, err)
 	}
 
+	// The ack error is deliberately ignored: nacks are replayed by
+	// auto_replay_nacks (the default), and disabling that is a documented
+	// opt-in to DROP rejected messages, so the checkpoint must advance past
+	// them rather than pin the tracker.
 	ackFn := func(ackCtx context.Context, _ error) error {
 		resolved := resolveFn()
 		if resolved == nil {
@@ -912,11 +920,11 @@ func (e *salesforceCDCInputExecutor) flushTopic(
 
 		e.stateMu.Lock()
 		e.state.Topics[topic] = acked
-		err := e.saveStateLocked(ackCtx)
+		persistErr := e.saveStateLocked(ackCtx)
 		e.stateMu.Unlock()
 
-		if err != nil {
-			return fmt.Errorf("persist checkpoint: %w", err)
+		if persistErr != nil {
+			return fmt.Errorf("persist checkpoint: %w", persistErr)
 		}
 		return nil
 	}

internal/impl/salesforce/salesforcegrpc/subscription.go

@@ -9,6 +9,7 @@
 package salesforcegrpc
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
@@ -30,6 +31,16 @@ import (
 // provides safety margin without meaningfully slowing startup.
 const subscribeSettleDelay = 5 * time.Second
 
+// maxConsecutiveDecodeFailures bounds redelivery attempts for an event whose
+// Avro decode keeps failing at the same replay position. A decode failure
+// against a successfully fetched schema is deterministic, and the replay
+// anchor advances per delivered event, so each reconnect redelivers from
+// exactly the failing event; once the same position has failed this many
+// times in a row the payload is treated as permanently undecodable and the
+// stream fails loudly. Schema-fetch failures never count toward this bound -
+// they are transport-class errors governed by the reconnect policy.
+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 +57,13 @@ type Subscription struct {
 	ready        chan struct{}
 	streamErr    error
 	state        StreamState
+	// decodeFailures and schemaFailures count consecutive failures pinned to
+	// one replay position each (guarded by mu): deterministic Avro-decode
+	// failures bounded by maxConsecutiveDecodeFailures, and schema-fetch
+	// failures bounded by the reconnect policy (reconnect_max_attempts).
+	// Position-scoped so an unrelated event's success never resets them.
+	decodeFailures positionFailures
+	schemaFailures positionFailures
 
 	// Atomic counters for health reporting.
 	eventsReceived     atomic.Int64
@@ -121,22 +139,145 @@ 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)
+	go s.receiveLoop(ctx, streamCtx)
 
 	return nil
 }
 
+// anchored reports whether a replay resume position exists: only then can a
+// reconnect redeliver the current batch (CUSTOM replay is exclusive-after).
+// On a fresh stream with no anchor, a reconnect falls back to the configured
+// preset - with LATEST that silently drops the batch, so unanchored failures
+// must never take the reconnect path.
+func (s *Subscription) anchored() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.lastReplayID) > 0
+}
+
+// failTerminal fails the stream permanently: streamErr surfaces through the
+// health tick and no reconnect is attempted.
+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)
+}
+
+// failDecode routes an Avro decode failure: reconnect-and-redeliver while the
+// position is under the consecutive-failure bound, terminal stream failure
+// once it is exceeded. Schema-fetch failures do NOT come through here - they
+// are transport-class errors governed by the reconnect policy, not evidence
+// of an undecodable payload.
+func (s *Subscription) failDecode(replayID []byte, failStream func(error), err error) {
+	if !s.recordDecodeFailure(replayID) {
+		failStream(err)
+		return
+	}
+	s.failTerminal(fmt.Errorf("decoding event at replay position %x: %d consecutive failures, treating as permanently undecodable: %w", replayID, maxConsecutiveDecodeFailures, err))
+}
+
+// positionFailures counts consecutive failures pinned to one replay
+// position. A failure at a different position resets the count (the stream
+// has moved on); a success clears it only when it lands at the tracked
+// position - successes elsewhere must not reset it, since a redelivery cycle
+// decodes the events around a permanently failing one successfully every
+// time, which would keep the count oscillating below any bound forever.
+// Callers must hold Subscription.mu.
+type positionFailures struct {
+	count    int
+	replayID []byte
+}
+
+func (p *positionFailures) record(replayID []byte) int {
+	if !bytes.Equal(replayID, p.replayID) {
+		p.replayID = append([]byte(nil), replayID...)
+		p.count = 0
+	}
+	p.count++
+	return p.count
+}
+
+func (p *positionFailures) clear(replayID []byte) {
+	if p.replayID != nil && bytes.Equal(replayID, p.replayID) {
+		p.count = 0
+		p.replayID = nil
+	}
+}
+
+// recordDecodeFailure counts an Avro-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()
+	return s.decodeFailures.record(replayID) >= maxConsecutiveDecodeFailures
+}
+
+// clearDecodeFailures resets the decode-failure count when the event at the
+// tracked failing position decodes successfully.
+func (s *Subscription) clearDecodeFailures(replayID []byte) {
+	s.mu.Lock()
+	s.decodeFailures.clear(replayID)
+	s.mu.Unlock()
+}
+
+// recordSchemaFailure counts a schema-fetch failure at the given replay
+// position against the reconnect policy: with reconnect_max_attempts 0
+// (unlimited) it never trips - the user asked for indefinite retries on
+// transport-class failures - but a bounded policy must also bound this path.
+// Each reconnect here succeeds (the transport is healthy, the schema is
+// not), so reconnectWithBackoff's own per-call budget never accumulates: a
+// deterministically unfetchable schema (uncompilable SchemaJson, permanent
+// NotFound) would otherwise reconnect-redeliver forever without ever
+// surfacing an error, the same livelock the decode bound exists for.
+func (s *Subscription) recordSchemaFailure(replayID []byte) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	budget := s.client.maxReconnect
+	return budget > 0 && s.schemaFailures.record(replayID) >= budget
+}
+
+// clearSchemaFailures resets the schema-failure count when a fetch succeeds
+// at the tracked failing position.
+func (s *Subscription) clearSchemaFailures(replayID []byte) {
+	s.mu.Lock()
+	s.schemaFailures.clear(replayID)
+	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.
-func (s *Subscription) receiveLoop(ctx context.Context) {
+// 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.
+	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 +289,7 @@ func (s *Subscription) receiveLoop(ctx context.Context) {
 			}
 			s.mu.Unlock()
 
-			s.client.log.Errorf("Pub/Sub stream error (topic=%s): %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)
-			}
+			failStream(err)
 			return
 		}
 
@@ -177,19 +308,79 @@ 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. When a replay anchor exists, reconnect instead —
+			// lastReplayID advances per delivered event, so redelivery resumes
+			// exactly at the failing event. Both failure classes are bounded
+			// per replay position: Avro decode failures against a fetched
+			// schema (deterministic) by maxConsecutiveDecodeFailures, and
+			// schema-fetch failures by the reconnect policy itself
+			// (reconnect_max_attempts; 0 = retry forever) — counted here
+			// because each individual reconnect succeeds, so the per-call
+			// reconnect budget alone can never trip. On a fresh stream with
+			// NO anchor a reconnect would fall back to the configured preset
+			// and could silently drop the batch (LATEST), so schema fetches
+			// retry inline under the same policy and decode failures fail the
+			// stream terminally instead.
 			schema, err := s.client.schemaCache.GetSchema(ctx, event.SchemaId)
+			if err != nil && !s.anchored() {
+				// A reconnect here would resume via the configured preset and
+				// could silently drop this batch (LATEST), so the fetch retries
+				// inline instead - governed by the same reconnect policy a
+				// reconnect would use (reconnect_min_delay/_max_delay and
+				// reconnect_max_attempts; 0 = retry indefinitely).
+				for attempt := 0; err != nil; attempt++ {
+					if s.client.maxReconnect > 0 && attempt >= s.client.maxReconnect {
+						break
+					}
+					delay := grpcBackoffWithJitter(s.client.baseBackoff, s.client.maxBackoff, attempt)
+					s.client.log.Warnf("Schema fetch failed on a fresh stream with no replay anchor to redeliver from (topic=%s, schemaID=%s), retrying inline in %v (attempt %d): %v", s.config.TopicName, event.SchemaId, delay, attempt+1, err)
+					t := time.NewTimer(delay)
+					select {
+					case <-t.C:
+					case <-streamCtx.Done():
+						t.Stop()
+						return
+					case <-ctx.Done():
+						t.Stop()
+						return
+					}
+					schema, err = s.client.schemaCache.GetSchema(ctx, event.SchemaId)
+				}
+				if err != nil {
+					s.eventsDecodeErrors.Add(1)
+					s.failTerminal(fmt.Errorf("fetching schema for the first event of a fresh stream (schemaID=%s): no replay anchor to redeliver from and reconnect_max_attempts (%d) exhausted: %w", event.SchemaId, s.client.maxReconnect, err))
+					return
+				}
+			}
 			if err != nil {
-				s.client.log.Errorf("get schema for event (schemaID=%s): %v", event.SchemaId, err)
 				s.eventsDecodeErrors.Add(1)
-				continue
+				// Bounded by the reconnect policy: each reconnect below
+				// succeeds (the transport is healthy), so the per-call
+				// reconnect budget never accumulates - the per-position count
+				// is what makes reconnect_max_attempts meaningful for a
+				// deterministically unfetchable schema. 0 = retry forever.
+				if s.recordSchemaFailure(consumerEvent.ReplayId) {
+					s.failTerminal(fmt.Errorf("fetching schema (schemaID=%s) failed %d consecutive times at replay position %x (reconnect_max_attempts exhausted): %w", event.SchemaId, s.client.maxReconnect, consumerEvent.ReplayId, err))
+					return
+				}
+				failStream(fmt.Errorf("get schema for event (schemaID=%s): %w", event.SchemaId, err))
+				return
 			}
+			s.clearSchemaFailures(consumerEvent.ReplayId)
 
 			decoded, err := DecodeAvroPayload(schema, event.Payload)
 			if err != nil {
-				s.client.log.Errorf("decode Avro payload (schemaID=%s): %v", event.SchemaId, err)
 				s.eventsDecodeErrors.Add(1)
-				continue
+				if !s.anchored() {
+					s.failTerminal(fmt.Errorf("decoding the first event of a fresh stream (schemaID=%s, replay position %x): payload is undecodable and no replay anchor exists to redeliver from: %w", event.SchemaId, consumerEvent.ReplayId, err))
+					return
+				}
+				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 +400,27 @@ 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())
+				// Advance the replay anchor past this event: CUSTOM replay is
+				// exclusive-after, so a reconnect now redelivers from exactly
+				// the next (possibly failing) event - no lost first batch, no
+				// re-emitted prefix.
+				s.mu.Lock()
+				s.lastReplayID = pubsubEvent.ReplayID
+				s.mu.Unlock()
 				s.client.log.Debugf("Pub/Sub event received (topic=%s, schemaID=%s, replayID=%x)", pubsubEvent.TopicName, pubsubEvent.SchemaID, pubsubEvent.ReplayID)
-			default:
-				s.eventsDropped.Add(1)
-				s.client.log.Warnf("Pub/Sub event buffer full (topic=%s), dropping event", s.config.TopicName)
+			case <-streamCtx.Done():
+				return
+			case <-ctx.Done():
+				return
 			}
 		}