Skip bad events in sync and backfill instead of aborting

handleSync aborted the whole batch on any room or event error and
runSync retried it forever, so one deterministic failure stopped
bridging for the entire account. Log and skip failures per room and per
event, only holding the cursor when the sync is cancelled.

FetchMessages likewise failed the whole page on one unconvertible
message. Messages that fail to parse or convert now backfill as a
placeholder notice mapped to their ID, edits are skipped, and a message
whose reactions can't be fetched is backfilled without them. Cursor and
identity problems still fail the page.

Reaction images that fail to download fall back to the matching
unicode emoji or shortcode text.

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

Author: Fizzadar

pkg/connector/backfill.go

@@ -7,6 +7,7 @@ import (
 	"sort"
 	"time"
 
+	"github.com/rs/zerolog"
 	"maunium.net/go/mautrix"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/database"
@@ -111,10 +112,11 @@ func (r *RedditClient) FetchMessages(ctx context.Context, params bridgev2.FetchM
 				return nil, errors.New("reddit history event has no stable identity")
 			}
 			if err = evt.Content.ParseRaw(event.EventMessage); err != nil && !errors.Is(err, event.ErrContentAlreadyParsed) {
-				return nil, err
+				result.Messages = append(result.Messages, r.historyPlaceholder(ctx, evt, err))
+				continue
 			}
 			if evt.Content.AsMessage().RelatesTo.GetReplaceID() != "" {
-				return nil, errors.New("reddit history edit reconciliation has not been implemented")
+				continue
 			}
 			if params.ThreadRoot != "" && evt.Unsigned.RedactedBecause == nil && makeMessageID(evt.Content.AsMessage().RelatesTo.GetThreadParent()) != params.ThreadRoot {
 				return nil, errors.New("reddit thread history contains a reply to another root")
@@ -125,13 +127,14 @@ func (r *RedditClient) FetchMessages(ctx context.Context, params bridgev2.FetchM
 			}
 			converted, err := r.convertMessage(ctx, params.Portal, intent, evt)
 			if err != nil {
-				return nil, err
+				result.Messages = append(result.Messages, r.historyPlaceholder(ctx, evt, err))
+				continue
 			}
 			var reactions []*bridgev2.BackfillReaction
 			if evt.Unsigned.RedactedBecause == nil {
 				reactions, err = r.backfillReactions(ctx, params.Portal.PortalKey, makeMessageID(evt.ID))
 				if err != nil {
-					return nil, err
+					zerolog.Ctx(ctx).Warn().Err(err).Stringer("event_id", evt.ID).Msg("Failed to fetch Reddit history reactions, backfilling message without them")
 				}
 			}
 			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})
@@ -146,6 +149,9 @@ 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) {
@@ -155,3 +161,16 @@ func (r *RedditClient) FetchMessages(ctx context.Context, params bridgev2.FetchM
 	})
 	return result, nil
 }
+
+// Map the placeholder to the native ID so replies and reactions to it resolve.
+func (r *RedditClient) historyPlaceholder(ctx context.Context, evt *event.Event, err error) *bridgev2.BackfillMessage {
+	zerolog.Ctx(ctx).Warn().Err(err).Stringer("event_id", evt.ID).Msg("Failed to convert Reddit history message, using placeholder")
+	return &bridgev2.BackfillMessage{
+		ConvertedMessage: &bridgev2.ConvertedMessage{Parts: []*bridgev2.ConvertedMessagePart{{
+			Type:    event.EventMessage,
+			Content: &event.MessageEventContent{MsgType: event.MsgNotice, Body: "An error occurred while processing an incoming message", Mentions: &event.Mentions{}},
+			Extra:   map[string]any{"fi.mau.bridge.internal_error": err.Error()},
+		}}},
+		ID: makeMessageID(evt.ID), Sender: r.makeSender(evt.Sender), Timestamp: time.UnixMilli(evt.Timestamp),
+	}
+}

pkg/connector/chatsync.go

@@ -86,16 +86,9 @@ func (r *RedditClient) runSync(ctx context.Context, done chan struct{}) {
 			continue
 		}
 
-		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
+		r.handleSync(ctx, resp)
+		if ctx.Err() != nil {
+			return
 		}
 		// All framework event calls have returned. Advance the native token;
 		// event delivery and crash safety belong to bridgev2 and its runtime.
@@ -118,90 +111,79 @@ func (r *RedditClient) runSync(ctx context.Context, done chan struct{}) {
 	}
 }
 
-func (r *RedditClient) handleSync(ctx context.Context, resp *redditchat.SyncResponse) error {
-	unhidden, err := r.applyHiddenChatData(ctx, resp)
-	if err != nil {
-		return err
-	}
+// 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)
 	for roomID, joined := range resp.Rooms.Join {
 		if r.meta.HiddenRooms[roomID] {
 			continue
 		}
-		if err := r.handleRoomUpdate(ctx, roomID, joined); err != nil {
-			return err
-		}
+		logRoomError(ctx, roomID, r.handleRoomUpdate(ctx, roomID, joined))
 	}
 	for roomID, invited := range resp.Rooms.Invite {
 		if r.meta.HiddenRooms[roomID] {
 			continue
 		}
-		if err := r.handleInvitedRoom(ctx, roomID, invited); err != nil {
-			return err
-		}
+		logRoomError(ctx, roomID, r.handleInvitedRoom(ctx, roomID, invited))
 	}
 	// Pending requests receive subsequent messages in rooms.peek. Process
 	// invitation state first, then use the same queue/backfill path as joined
 	// timelines. Membership comes from state, never from the response bucket.
 	for roomID, peek := range resp.Rooms.Peek {
-		if r.meta.HiddenRooms[roomID] || resp.Rooms.Join[roomID] != nil || resp.Rooms.Leave[roomID] != nil {
+		if peek == nil || r.meta.HiddenRooms[roomID] || resp.Rooms.Join[roomID] != nil || resp.Rooms.Leave[roomID] != nil {
 			continue
 		}
 		if len(peek.State.Events)+len(peek.Timeline.Events)+len(peek.Ephemeral.Events) == 0 {
 			continue
 		}
-		if err := r.handleRoomUpdate(ctx, roomID, peek); err != nil {
-			return err
-		}
+		logRoomError(ctx, roomID, r.handleRoomUpdate(ctx, roomID, peek))
 	}
 	for roomID := range resp.Rooms.Leave {
-		if err := r.handleLeftRoom(ctx, roomID); err != nil {
-			return err
-		}
+		logRoomError(ctx, roomID, r.handleLeftRoom(ctx, roomID))
 	}
 	if len(unhidden) > 0 {
 		state, err := r.fetchRoomSnapshots(ctx, unhidden)
 		if err != nil {
-			return err
+			if ctx.Err() == nil {
+				zerolog.Ctx(ctx).Err(err).Msg("Failed to fetch unhidden Reddit chats, skipping")
+			}
+			return
 		}
-		return r.handleSync(ctx, state)
+		r.handleSync(ctx, state)
+	}
+}
+
+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")
 	}
-	return nil
 }
 
-func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat.SyncResponse) ([]id.RoomID, error) {
+func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat.SyncResponse) []id.RoomID {
 	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.
 	for _, rooms := range []map[id.RoomID]*mautrix.SyncJoinedRoom{resp.Rooms.Join, resp.Rooms.Peek} {
 		for roomID, room := range rooms {
 			if room == nil {
-				return nil, errors.New("reddit sync contains a null room")
+				continue
 			}
 			for _, evt := range room.AccountData.Events {
 				if evt == nil || evt.Type.Type != "com.reddit.hidden_chat" {
 					continue
 				}
-				var content struct {
-					Hidden *bool `json:"hidden"`
-				}
-				raw, err := json.Marshal(&evt.Content)
+				hidden, err := parseHiddenChat(evt)
 				if err != nil {
-					return nil, err
-				}
-				if err = json.Unmarshal(raw, &content); err != nil {
-					return nil, err
-				}
-				if content.Hidden == nil {
-					return nil, errors.New("reddit hidden chat state has no hidden flag")
+					logRoomError(ctx, roomID, err)
+					continue
 				}
-				if *content.Hidden {
+				if hidden {
 					if r.meta.HiddenRooms == nil {
 						r.meta.HiddenRooms = make(map[id.RoomID]bool)
 					}
 					r.meta.HiddenRooms[roomID] = true
-					if err = r.handleLeftRoom(ctx, roomID); err != nil {
-						return nil, err
-					}
+					logRoomError(ctx, roomID, r.handleLeftRoom(ctx, roomID))
 				} else {
 					delete(r.meta.HiddenRooms, roomID)
 					if resp.Rooms.Join[roomID] == nil && resp.Rooms.Invite[roomID] == nil {
@@ -211,10 +193,30 @@ func (r *RedditClient) applyHiddenChatData(ctx context.Context, resp *redditchat
 			}
 		}
 	}
-	return unhidden, nil
+	return unhidden
+}
+
+func parseHiddenChat(evt *event.Event) (bool, error) {
+	var content struct {
+		Hidden *bool `json:"hidden"`
+	}
+	raw, err := json.Marshal(&evt.Content)
+	if err != nil {
+		return false, err
+	}
+	if err = json.Unmarshal(raw, &content); err != nil {
+		return false, err
+	}
+	if content.Hidden == nil {
+		return false, errors.New("reddit hidden chat state has no hidden flag")
+	}
+	return *content.Hidden, nil
 }
 
 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")
+	}
 	state, portalKey, err := r.resolveRoomState(ctx, roomID, joined)
 	if err != nil {
 		return err
@@ -224,11 +226,12 @@ 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 {
-			return err
+		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")
 		}
 	}
-	return r.handleEphemeralEvents(ctx, portalKey, joined.Ephemeral.Events)
+	r.handleEphemeralEvents(ctx, portalKey, joined.Ephemeral.Events)
+	return nil
 }
 
 func (r *RedditClient) handleInvitedRoom(ctx context.Context, roomID id.RoomID, invited *mautrix.SyncInvitedRoom) error {

pkg/connector/emoji.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/rs/zerolog"
 	"go.mau.fi/util/variationselector"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/database"
@@ -95,6 +96,22 @@ 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.
+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 {
+		return emoji, shortcode, err
+	}
+	asset, _ := redditchat.ReactionAssetByKey(key)
+	zerolog.Ctx(ctx).Warn().Err(err).Str("reaction", asset.Name).Msg("Failed to fetch Reddit reaction image, using text")
+	for unicode, name := range reactionAliases {
+		if name == asset.Name {
+			return unicode, "", nil
+		}
+	}
+	return ":" + asset.Name + ":", "", nil
+}
+
 func (r *RedditClient) resolveOutgoingReaction(ctx context.Context, value string) (string, error) {
 	if strings.HasPrefix(value, "mxc://") {
 		key := r.main.Bridge.DB.KV.Get(ctx, database.Key("reddit_emoji_v1/mxc/"+value))

pkg/connector/reactionhistory.go

@@ -81,7 +81,7 @@ func (r *RedditClient) backfillReactions(ctx context.Context, key networkid.Port
 			continue
 		}
 		seen[identity] = evt.ID
-		emoji, shortcode, err := r.reactionImage(ctx, key)
+		emoji, shortcode, err := r.reactionEmoji(ctx, key)
 		if err != nil {
 			return nil, err
 		}

pkg/connector/reactions.go

@@ -16,7 +16,7 @@ func (r *RedditClient) convertReaction(ctx context.Context, key networkid.Portal
 	if !ok || content.RelatesTo.EventID == "" || content.RelatesTo.Type != event.RelAnnotation || content.RelatesTo.Key == "" {
 		return nil, errors.New("reaction is missing its target")
 	}
-	emoji, shortcode, err := r.reactionImage(ctx, content.RelatesTo.Key)
+	emoji, shortcode, err := r.reactionEmoji(ctx, content.RelatesTo.Key)
 	if err != nil {
 		return nil, err
 	}

pkg/connector/receipts.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/rs/zerolog"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/bridgev2/networkid"
 	"maunium.net/go/mautrix/bridgev2/simplevent"
@@ -70,15 +71,16 @@ func (r *RedditClient) convertReadReceipts(key networkid.PortalKey, evt *event.E
 	return out, nil
 }
 
-func (r *RedditClient) handleEphemeralEvents(ctx context.Context, key networkid.PortalKey, events []*event.Event) error {
+func (r *RedditClient) handleEphemeralEvents(ctx context.Context, key networkid.PortalKey, events []*event.Event) {
 	for _, evt := range events {
 		if evt == nil {
 			continue
 		}
 		if evt.Type == event.EphemeralEventTyping {
 			typings, err := r.convertTyping(key, evt)
 			if err != nil {
-				return err
+				zerolog.Ctx(ctx).Err(err).Stringer("portal_key", key).Msg("Failed to convert Reddit typing event, skipping")
+				continue
 			}
 			for _, typing := range typings {
 				r.userLogin.QueueRemoteEvent(typing)
@@ -90,7 +92,8 @@ func (r *RedditClient) handleEphemeralEvents(ctx context.Context, key networkid.
 		}
 		receipts, err := r.convertReadReceipts(key, evt)
 		if err != nil {
-			return err
+			zerolog.Ctx(ctx).Err(err).Stringer("portal_key", key).Msg("Failed to convert Reddit read receipt, skipping")
+			continue
 		}
 		for _, receipt := range receipts {
 			if receipt.Timestamp.IsZero() {
@@ -99,5 +102,4 @@ func (r *RedditClient) handleEphemeralEvents(ctx context.Context, key networkid.
 			r.userLogin.QueueRemoteEvent(receipt)
 		}
 	}
-	return nil
 }