Only skip Reddit data that can never be bridged

Network, database and other transient failures retry the sync batch with
the cursor retained, or fail the backfill page. Errors explicitly marked
unbridgeable (invalid native data, 404/410, unsupported or oversized media)
are skipped, or become a placeholder in history.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: Fizzadar

pkg/connector/backfill.go

@@ -126,15 +126,19 @@ func (r *RedditClient) FetchMessages(ctx context.Context, params bridgev2.FetchM
 				intent, _ = params.Portal.GetIntentFor(ctx, r.makeSender(evt.Sender), r.userLogin, bridgev2.RemoteEventMessage)
 			}
 			converted, err := r.convertMessage(ctx, params.Portal, intent, evt)
-			if err != nil {
+			if isUnbridgeable(err) {
 				result.Messages = append(result.Messages, r.historyPlaceholder(ctx, evt, err))
 				continue
+			} else if err != nil {
+				return nil, err
 			}
 			var reactions []*bridgev2.BackfillReaction
 			if evt.Unsigned.RedactedBecause == nil {
 				reactions, err = r.backfillReactions(ctx, params.Portal.PortalKey, makeMessageID(evt.ID))
-				if err != nil {
-					zerolog.Ctx(ctx).Warn().Err(err).Stringer("event_id", evt.ID).Msg("Failed to fetch Reddit history reactions, backfilling message without them")
+				if isUnbridgeable(err) {
+					zerolog.Ctx(ctx).Warn().Err(err).Stringer("event_id", evt.ID).Msg("Skipping unbridgeable Reddit history reactions")
+				} else if err != nil {
+					return nil, err
 				}
 			}
 			result.Messages = append(result.Messages, &bridgev2.BackfillMessage{ConvertedMessage: converted, ID: makeMessageID(evt.ID), Sender: r.makeSender(evt.Sender), Timestamp: time.UnixMilli(evt.Timestamp), Reactions: reactions})
@@ -149,9 +153,6 @@ func (r *RedditClient) FetchMessages(ctx context.Context, params bridgev2.FetchM
 	if (params.Forward || seekBackwardAnchor) && anchor != nil && !anchorFound {
 		return nil, errors.New("reddit history ended before the known message; cannot locate the history anchor")
 	}
-	if err := ctx.Err(); err != nil {
-		return nil, err
-	}
 	sort.SliceStable(result.Messages, func(i, j int) bool {
 		left, right := result.Messages[i], result.Messages[j]
 		if left.Timestamp.Equal(right.Timestamp) {

pkg/connector/chatsync.go

@@ -86,9 +86,16 @@ func (r *RedditClient) runSync(ctx context.Context, done chan struct{}) {
 			continue
 		}
 
-		r.handleSync(ctx, resp)
-		if ctx.Err() != nil {
-			return
+		err = r.handleSync(ctx, resp)
+		if err != nil {
+			log.Error().Err(err).Msg("Failed to process Reddit sync; retaining cursor")
+			r.userLogin.BridgeState.Send(status.BridgeState{StateEvent: status.StateTransientDisconnect, Error: "reddit-sync-apply-failed", Message: "Unable to synchronize Reddit chat state. Retrying."})
+			select {
+			case <-time.After(backoff):
+			case <-ctx.Done():
+				return
+			}
+			continue
 		}
 		// All framework event calls have returned. Advance the native token;
 		// event delivery and crash safety belong to bridgev2 and its runtime.
@@ -111,21 +118,28 @@ func (r *RedditClient) runSync(ctx context.Context, done chan struct{}) {
 	}
 }
 
-// Failures are isolated per room and event: a deterministic failure must not
-// retain the cursor and stop every other chat on the account.
-func (r *RedditClient) handleSync(ctx context.Context, resp *redditchat.SyncResponse) {
-	unhidden := r.applyHiddenChatData(ctx, resp)
+// Unbridgeable native data is skipped so it cannot hold the cursor forever.
+// Any other failure retries the whole batch.
+func (r *RedditClient) handleSync(ctx context.Context, resp *redditchat.SyncResponse) error {
+	unhidden, err := r.applyHiddenChatData(ctx, resp)
+	if err != nil {
+		return err
+	}
 	for roomID, joined := range resp.Rooms.Join {
 		if r.meta.HiddenRooms[roomID] {
 			continue
 		}
-		logRoomError(ctx, roomID, r.handleRoomUpdate(ctx, roomID, joined))
+		if err = skipUnbridgeable(ctx, r.handleRoomUpdate(ctx, roomID, joined), roomID, ""); err != nil {
+			return err
+		}
 	}
 	for roomID, invited := range resp.Rooms.Invite {
 		if r.meta.HiddenRooms[roomID] {
 			continue
 		}
-		logRoomError(ctx, roomID, r.handleInvitedRoom(ctx, roomID, invited))
+		if err = skipUnbridgeable(ctx, r.handleInvitedRoom(ctx, roomID, invited), roomID, ""); err != nil {
+			return err
+		}
 	}
 	// Pending requests receive subsequent messages in rooms.peek. Process
 	// invitation state first, then use the same queue/backfill path as joined
@@ -137,30 +151,34 @@ func (r *RedditClient) handleSync(ctx context.Context, resp *redditchat.SyncResp
 		if len(peek.State.Events)+len(peek.Timeline.Events)+len(peek.Ephemeral.Events) == 0 {
 			continue
 		}
-		logRoomError(ctx, roomID, r.handleRoomUpdate(ctx, roomID, peek))
+		if err = skipUnbridgeable(ctx, r.handleRoomUpdate(ctx, roomID, peek), roomID, ""); err != nil {
+			return err
+		}
 	}
 	for roomID := range resp.Rooms.Leave {
-		logRoomError(ctx, roomID, r.handleLeftRoom(ctx, roomID))
+		if err = r.handleLeftRoom(ctx, roomID); err != nil {
+			return err
+		}
 	}
 	if len(unhidden) > 0 {
 		state, err := r.fetchRoomSnapshots(ctx, unhidden)
 		if err != nil {
-			if ctx.Err() == nil {
-				zerolog.Ctx(ctx).Err(err).Msg("Failed to fetch unhidden Reddit chats, skipping")
-			}
-			return
+			return err
 		}
-		r.handleSync(ctx, state)
+		return r.handleSync(ctx, state)
 	}
+	return nil
 }
 
-func logRoomError(ctx context.Context, roomID id.RoomID, err error) {
-	if err != nil && ctx.Err() == nil {
-		zerolog.Ctx(ctx).Err(err).Stringer("room_id", roomID).Msg("Failed to handle Reddit room update, skipping")
+func skipUnbridgeable(ctx context.Context, err error, roomID id.RoomID, eventID id.EventID) error {
+	if !isUnbridgeable(err) {
+		return err
 	}
+	zerolog.Ctx(ctx).Warn().Err(err).Stringer("room_id", roomID).Stringer("event_id", eventID).Msg("Skipping unbridgeable Reddit data")
+	return nil
 }
 
-func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat.SyncResponse) []id.RoomID {
+func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat.SyncResponse) ([]id.RoomID, error) {
 	var unhidden []id.RoomID
 	// Apply account data before invitations: full native sync includes hidden
 	// DMs in both peek and invite. Never mistake a preview for membership.
@@ -175,15 +193,17 @@ func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat
 				}
 				hidden, err := parseHiddenChat(evt)
 				if err != nil {
-					logRoomError(ctx, roomID, err)
+					zerolog.Ctx(ctx).Warn().Err(err).Stringer("room_id", roomID).Stringer("event_id", evt.ID).Msg("Skipping unbridgeable Reddit hidden chat state")
 					continue
 				}
 				if hidden {
 					if r.meta.HiddenRooms == nil {
 						r.meta.HiddenRooms = make(map[id.RoomID]bool)
 					}
 					r.meta.HiddenRooms[roomID] = true
-					logRoomError(ctx, roomID, r.handleLeftRoom(ctx, roomID))
+					if err = r.handleLeftRoom(ctx, roomID); err != nil {
+						return nil, err
+					}
 				} else {
 					delete(r.meta.HiddenRooms, roomID)
 					if resp.Rooms.Join[roomID] == nil && resp.Rooms.Invite[roomID] == nil {
@@ -193,7 +213,7 @@ func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat
 			}
 		}
 	}
-	return unhidden
+	return unhidden, nil
 }
 
 func parseHiddenChat(evt *event.Event) (bool, error) {
@@ -215,7 +235,7 @@ func parseHiddenChat(evt *event.Event) (bool, error) {
 
 func (r *RedditClient) handleRoomUpdate(ctx context.Context, roomID id.RoomID, joined *mautrix.SyncJoinedRoom) error {
 	if joined == nil {
-		return errors.New("reddit sync contains a null room")
+		return unbridgeable(errors.New("reddit sync contains a null room"))
 	}
 	state, portalKey, err := r.resolveRoomState(ctx, roomID, joined)
 	if err != nil {
@@ -226,8 +246,11 @@ func (r *RedditClient) handleRoomUpdate(ctx context.Context, roomID id.RoomID, j
 	r.userLogin.QueueRemoteEvent(resync)
 
 	for _, evt := range joined.Timeline.Events {
-		if err = r.handleTimelineEvent(ctx, portalKey, evt); err != nil && ctx.Err() == nil {
-			zerolog.Ctx(ctx).Err(err).Stringer("room_id", roomID).Stringer("event_id", evt.ID).Msg("Failed to handle Reddit event, skipping")
+		if evt == nil {
+			continue
+		}
+		if err = skipUnbridgeable(ctx, r.handleTimelineEvent(ctx, portalKey, evt), roomID, evt.ID); err != nil {
+			return err
 		}
 	}
 	r.handleEphemeralEvents(ctx, portalKey, joined.Ephemeral.Events)
@@ -237,7 +260,7 @@ func (r *RedditClient) handleRoomUpdate(ctx context.Context, roomID id.RoomID, j
 func (r *RedditClient) handleInvitedRoom(ctx context.Context, roomID id.RoomID, invited *mautrix.SyncInvitedRoom) error {
 	joined, err := r.inviteSnapshot(invited)
 	if err != nil {
-		return err
+		return unbridgeable(err)
 	}
 	state, key, err := r.roomState(ctx, roomID, joined)
 	if err != nil {
@@ -267,15 +290,12 @@ func (r *RedditClient) handleLeftRoom(ctx context.Context, roomID id.RoomID) err
 }
 
 func (r *RedditClient) handleTimelineEvent(ctx context.Context, portalKey networkid.PortalKey, evt *event.Event) error {
-	if evt == nil {
-		return nil
-	}
 	if evt.Type == event.EventMessage || evt.Type == event.EventReaction || evt.Type == event.EventRedaction {
 		if evt.ID == "" || evt.Sender == "" || (evt.RoomID != "" && makePortalID(evt.RoomID) != portalKey.ID) {
-			return errors.New("reddit timeline event has invalid room or event identity")
+			return unbridgeable(errors.New("reddit timeline event has invalid room or event identity"))
 		}
 		if err := evt.Content.ParseRaw(evt.Type); err != nil && !errors.Is(err, event.ErrContentAlreadyParsed) {
-			return err
+			return unbridgeable(err)
 		}
 	}
 	switch evt.Type {
@@ -308,7 +328,7 @@ func (r *RedditClient) queueMessage(ctx context.Context, key networkid.PortalKey
 	if evt.Unsigned.RedactedBecause != nil {
 		removal, err := r.convertDeletedEvent(key, evt)
 		if err != nil {
-			return err
+			return unbridgeable(err)
 		}
 		r.userLogin.QueueRemoteEvent(removal)
 		return nil
@@ -339,7 +359,7 @@ func (r *RedditClient) queueReaction(ctx context.Context, key networkid.PortalKe
 		// Old sync responses may retain the original reaction content.
 		removal, err := r.convertDeletedEvent(key, evt)
 		if err != nil {
-			return err
+			return unbridgeable(err)
 		}
 		r.userLogin.QueueRemoteEvent(removal)
 		return nil

pkg/connector/emoji.go

@@ -3,6 +3,7 @@ package connector
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/http"
 	"strings"
 	"time"
@@ -69,8 +70,10 @@ func (r *RedditClient) reactionImage(ctx context.Context, key string) (string, s
 			return "", "", mediaError(ctx, "Reddit reaction download", err)
 		}
 		defer resp.Body.Close()
-		if resp.StatusCode != http.StatusOK {
-			return "", "", errors.New("reddit reaction image is unavailable")
+		if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone {
+			return "", "", unbridgeable(errors.New("reddit reaction image no longer exists"))
+		} else if resp.StatusCode != http.StatusOK {
+			return "", "", fmt.Errorf("reddit reaction image is unavailable (HTTP %d)", resp.StatusCode)
 		}
 		data, err := readMediaBounded(resp.Body, 2*1024*1024)
 		if err != nil {
@@ -96,10 +99,10 @@ func (r *RedditClient) reactionImage(ctx context.Context, key string) (string, s
 	return mxc, ":" + asset.Name + ":", nil
 }
 
-// Bridged reactions must not depend on the image CDN: fall back to text.
+// A reaction image that can never be fetched falls back to text.
 func (r *RedditClient) reactionEmoji(ctx context.Context, key string) (string, string, error) {
 	emoji, shortcode, err := r.reactionImage(ctx, key)
-	if err == nil || ctx.Err() != nil {
+	if !isUnbridgeable(err) {
 		return emoji, shortcode, err
 	}
 	asset, _ := redditchat.ReactionAssetByKey(key)

pkg/connector/media.go

@@ -71,13 +71,16 @@ func (r *RedditClient) downloadRedditMedia(ctx context.Context, uri id.ContentUR
 	}
 	parsed, err := uri.Parse()
 	if err != nil || parsed.IsEmpty() || parsed.Homeserver != "reddit.com" {
-		return nil, errors.New("attachment is not a Reddit media URI")
+		return nil, unbridgeable(errors.New("attachment is not a Reddit media URI"))
 	}
 	resp, err := r.remote().DownloadMedia(ctx, parsed)
 	if err != nil {
 		if resp != nil && resp.Body != nil {
 			resp.Body.Close()
 		}
+		if isNotFound(err) {
+			return nil, unbridgeable(mediaError(ctx, "Reddit media download", err))
+		}
 		return nil, mediaError(ctx, "Reddit media download", err)
 	}
 	defer resp.Body.Close()
@@ -90,7 +93,7 @@ func (r *RedditClient) downloadRedditMedia(ctx context.Context, uri id.ContentUR
 	}
 	if file != nil {
 		if err = file.DecryptInPlace(data); err != nil {
-			return nil, errors.New("reddit attachment failed decryption or integrity verification")
+			return nil, unbridgeable(errors.New("reddit attachment failed decryption or integrity verification"))
 		}
 	}
 	return data, nil

pkg/connector/reactionhistory.go

@@ -30,22 +30,22 @@ func (r *RedditClient) fetchReactionEvents(ctx context.Context, key networkid.Po
 		}
 		for _, evt := range page.Chunk {
 			if evt == nil || evt.Type != event.EventReaction || evt.ID == "" || (evt.RoomID != "" && makePortalID(evt.RoomID) != key.ID) {
-				return nil, errors.New("reaction history has an invalid event identity")
+				return nil, unbridgeable(errors.New("reaction history has an invalid event identity"))
 			}
 			local, server, err := evt.Sender.Parse()
 			if err != nil || server != "reddit.com" || !strings.HasPrefix(local, "t2_") {
-				return nil, errors.New("reaction history has an invalid sender")
+				return nil, unbridgeable(errors.New("reaction history has an invalid sender"))
 			}
 			if err = evt.Content.ParseRaw(event.EventReaction); err != nil && !errors.Is(err, event.ErrContentAlreadyParsed) {
-				return nil, err
+				return nil, unbridgeable(err)
 			}
 			relation := evt.Content.AsReaction().RelatesTo
 			if relation.Type != event.RelAnnotation || makeMessageID(relation.EventID) != target || relation.Key == "" {
-				return nil, errors.New("reaction history belongs to another message")
+				return nil, unbridgeable(errors.New("reaction history belongs to another message"))
 			}
 			if redaction := evt.Unsigned.RedactedBecause; redaction != nil {
 				if redaction.Type != event.EventRedaction || redaction.Redacts != evt.ID || (redaction.RoomID != "" && makePortalID(redaction.RoomID) != key.ID) {
-					return nil, errors.New("reaction history has an invalid removal identity")
+					return nil, unbridgeable(errors.New("reaction history has an invalid removal identity"))
 				}
 			}
 			events = append(events, evt)
@@ -54,7 +54,7 @@ func (r *RedditClient) fetchReactionEvents(ctx context.Context, key networkid.Po
 			return events, nil
 		}
 		if page.NextBatch == cursor || seen[page.NextBatch] {
-			return nil, errors.New("reddit reaction history cursor did not advance")
+			return nil, unbridgeable(errors.New("reddit reaction history cursor did not advance"))
 		}
 		seen[page.NextBatch] = true
 		cursor = page.NextBatch
@@ -76,7 +76,7 @@ func (r *RedditClient) backfillReactions(ctx context.Context, key networkid.Port
 		identity := string(evt.Sender) + "\x00" + key
 		if previous := seen[identity]; previous != "" {
 			if previous != evt.ID {
-				return nil, errors.New("reaction history has conflicting active identities")
+				return nil, unbridgeable(errors.New("reaction history has conflicting active identities"))
 			}
 			continue
 		}

pkg/connector/reactions.go

@@ -14,7 +14,7 @@ import (
 func (r *RedditClient) convertReaction(ctx context.Context, key networkid.PortalKey, evt *event.Event) (*simplevent.Reaction, error) {
 	content, ok := evt.Content.Parsed.(*event.ReactionEventContent)
 	if !ok || content.RelatesTo.EventID == "" || content.RelatesTo.Type != event.RelAnnotation || content.RelatesTo.Key == "" {
-		return nil, errors.New("reaction is missing its target")
+		return nil, unbridgeable(errors.New("reaction is missing its target"))
 	}
 	emoji, shortcode, err := r.reactionEmoji(ctx, content.RelatesTo.Key)
 	if err != nil {

pkg/connector/redactions.go

@@ -68,26 +68,29 @@ func validateTombstone(key networkid.PortalKey, target *event.Event) error {
 // event endpoint; the framework owns the corresponding local lookup/removal.
 func (r *RedditClient) convertRedaction(ctx context.Context, key networkid.PortalKey, removal *event.Event) (bridgev2.RemoteEvent, error) {
 	if err := validateNativeEvent(key, removal); err != nil {
-		return nil, err
+		return nil, unbridgeable(err)
 	}
 	targetID, err := redactionTarget(removal)
 	if err != nil {
-		return nil, err
+		return nil, unbridgeable(err)
 	}
 	target, err := r.remote().GetEvent(ctx, portalIDToRoomID(key.ID), targetID)
-	if err != nil {
+	if isNotFound(err) {
+		return nil, unbridgeable(fmt.Errorf("resolve deleted Reddit event: %w", err))
+	} else if err != nil {
 		return nil, fmt.Errorf("resolve deleted Reddit event: %w", err)
 	}
 	if target == nil || target.ID != targetID {
-		return nil, errors.New("reddit returned another deletion target")
+		return nil, unbridgeable(errors.New("reddit returned another deletion target"))
 	}
 	if err = validateTombstone(key, target); err != nil {
-		return nil, err
+		return nil, unbridgeable(err)
 	}
 	if proof := target.Unsigned.RedactedBecause; proof.ID != removal.ID || proof.Sender != removal.Sender || proof.Timestamp != removal.Timestamp {
-		return nil, errors.New("reddit returned a different deletion proof")
+		return nil, unbridgeable(errors.New("reddit returned a different deletion proof"))
 	}
-	return r.convertDeletedEvent(key, target)
+	removed, err := r.convertDeletedEvent(key, target)
+	return removed, unbridgeable(err)
 }
 
 func (r *RedditClient) convertDeletedEvent(key networkid.PortalKey, target *event.Event) (bridgev2.RemoteEvent, error) {

pkg/connector/requests.go

@@ -35,8 +35,6 @@ func (r *RedditClient) fetchRoomSnapshots(ctx context.Context, rooms []id.RoomID
 			state.Rooms.Join[room] = joined
 		} else if invited := full.Rooms.Invite[room]; invited != nil {
 			state.Rooms.Invite[room] = invited
-		} else {
-			return nil, errors.New("reddit did not return the requested conversation state")
 		}
 		if peek := full.Rooms.Peek[room]; peek != nil {
 			state.Rooms.Peek[room] = peek