backfill: let in-flight history sync entries expire so failed creations are retried

pkg/connector/backfill.go

@@ -293,17 +293,11 @@ func (wa *WhatsAppClient) handleWAHistorySync(
 		for i, rawMsg := range rawMessages {
 			// Don't store messages that will just be skipped.
 			msgEvt, err := wa.Client.ParseWebMessage(jid, rawMsg.GetMessage())
-			if err != nil && getMessageType(rawMsg.GetMessage().GetMessage()) == "ignore" {
+			if err != nil {
-				ignoredTypes++
-				continue
-			} else if err != nil {
 				log.Warn().Err(err).
 					Int("msg_index", i).
 					Str("msg_id", rawMsg.GetMessage().GetKey().GetID()).
 					Uint64("msg_time_seconds", rawMsg.GetMessage().GetMessageTimestamp()).
-					Stringer("stub_type", rawMsg.GetMessage().GetMessageStubType()).
-					Str("message_type", getMessageType(rawMsg.GetMessage().GetMessage())).
-					Bool("from_me", rawMsg.GetMessage().GetKey().GetFromMe()).
 					Msg("Dropping historical message due to parse error")
 				log.Trace().
 					Any("web_message_info", rawMsg.GetMessage()).
@@ -405,6 +399,8 @@ func (wa *WhatsAppClient) handleWAHistorySync(
 	return nil
 }
 
+const historySyncQueuedTimeout = 10 * time.Minute
+
 func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 	log := wa.UserLogin.Log.With().
 		Str("action", "create portals from history sync").
@@ -448,6 +444,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 			// We don't currently support new PSAs, so don't bother backfilling them either
 			wg.Done()
 			continue
+		} else if queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout {
+			wg.Done()
+			continue
 		}
 		// TODO can the chat info fetch be avoided entirely?
 		select {
@@ -488,6 +487,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 			wg.Done()
 			continue
 		}
+		wa.historySyncQueued.Set(conv.ChatJID, time.Now())
 		res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{
 			EventMeta: simplevent.EventMeta{
 				Type: bridgev2.RemoteEventChatResync,
@@ -503,6 +503,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 					if err != nil {
 						zerolog.Ctx(ctx).Err(err).Msg("Failed to mark conversation as bridged")
 					}
+					wa.historySyncQueued.Delete(conv.ChatJID)
 					wg.Done()
 				},
 			},
@@ -510,6 +511,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 			LatestMessageTS: conv.LastMessageTimestamp,
 		})
 		if !res.Success {
+			wa.historySyncQueued.Delete(conv.ChatJID)
 			log.Debug().Msg("Cancelling history sync portal creation loop")
 			return
 		}
@@ -517,6 +519,9 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) {
 	log.Info().Int("conversation_count", len(conversations)).Msg("Finished creating portals from history sync")
 	go func() {
 		wg.Wait()
+		if wa.historySyncQueued.Len() > 0 {
+			return
+		}
 		wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false
 		err = wa.UserLogin.Save(ctx)
 		if err != nil {

pkg/connector/client.go

@@ -50,6 +50,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2.
 		MC:        noopMCInstance,
 
 		historySyncWakeup:         make(chan struct{}, 1),
+		historySyncQueued:         exsync.NewMap[types.JID, time.Time](),
 		resyncQueue:               make(map[types.JID]resyncQueueItem),
 		directMediaRetries:        make(map[networkid.MessageID]*directMediaRetry),
 		mediaRetryLock:            semaphore.NewWeighted(wa.Config.HistorySync.MediaRequests.MaxAsyncHandle),
@@ -114,6 +115,7 @@ type WhatsAppClient struct {
 	MC        mClient
 
 	historySyncWakeup  chan struct{}
+	historySyncQueued  *exsync.Map[types.JID, time.Time]
 	stopLoops          atomic.Pointer[context.CancelFunc]
 	resyncQueue        map[types.JID]resyncQueueItem
 	resyncQueueLock    sync.Mutex

pkg/connector/handlewhatsapp.go

@@ -374,14 +374,6 @@ func (wa *WhatsAppClient) handleWAMessage(ctx context.Context, evt *events.Messa
 	if assocType := messageAssoc.GetAssociationType(); assocType == waE2E.MessageAssociation_HD_IMAGE_DUAL_UPLOAD || assocType == waE2E.MessageAssociation_HD_VIDEO_DUAL_UPLOAD {
 		parentKey := messageAssoc.GetParentMessageKey()
 		protocolMsg, shouldHideEdit := makeHDMediaReplacementEdit(evt.Message, parentKey)
-		if protocolMsg.GetEditedMessage() == nil {
-			wa.UserLogin.Log.Debug().
-				Str("message_id", evt.Info.ID).
-				Str("parent_id", parentKey.GetID()).
-				Stringer("assoc_type", assocType).
-				Msg("Ignoring HD replacement message without replacement content")
-			return
-		}
 		dontRenderEdited = shouldHideEdit
 		wa.UserLogin.Log.Debug().
 			Str("message_id", evt.Info.ID).

pkg/msgconv/from-whatsapp.go

@@ -322,7 +322,6 @@ func (mc *MessageConverter) ToMatrix(
 		cm.ThreadRoot = ptr.Ptr(waid.MakeMessageID(chat, pcp, commentTarget.GetID()))
 	}
 	mc.cutOffLongMessage(part.Content)
-	logOversizedPart(ctx, part)
 
 	return cm
 }

pkg/msgconv/wa-text.go

@@ -18,7 +18,6 @@ package msgconv
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"html"
 	"regexp"
@@ -26,7 +25,6 @@ import (
 	"strconv"
 	"strings"
 
-	"github.com/rs/zerolog"
 	"go.mau.fi/whatsmeow/proto/waE2E"
 	"maunium.net/go/mautrix/bridgev2"
 	"maunium.net/go/mautrix/event"
@@ -74,36 +72,6 @@ func (mc *MessageConverter) cutOffLongMessage(content *event.MessageEventContent
 	}
 }
 
-func logOversizedPart(ctx context.Context, part *bridgev2.ConvertedMessagePart) {
-	if part == nil || part.Content == nil {
-		return
-	}
-	contentJSON, _ := json.Marshal(part.Content)
-	extraJSON, _ := json.Marshal(part.Extra)
-	if len(contentJSON)+len(extraJSON) <= 40000 {
-		return
-	}
-	extraKeys := make([]string, 0, len(part.Extra))
-	for key := range part.Extra {
-		extraKeys = append(extraKeys, key)
-	}
-	var mentionCount int
-	if part.Content.Mentions != nil {
-		mentionCount = len(part.Content.Mentions.UserIDs)
-	}
-	linkPreviewJSON, _ := json.Marshal(part.Content.BeeperLinkPreviews)
-	zerolog.Ctx(ctx).Warn().
-		Str("msg_type", string(part.Content.MsgType)).
-		Int("content_json_size", len(contentJSON)).
-		Int("extra_json_size", len(extraJSON)).
-		Int("body_size", len(part.Content.Body)).
-		Int("formatted_body_size", len(part.Content.FormattedBody)).
-		Int("mention_count", mentionCount).
-		Int("link_preview_json_size", len(linkPreviewJSON)).
-		Strs("extra_keys", extraKeys).
-		Msg("Converted message part is larger than the cut-off limit")
-}
-
 func (mc *MessageConverter) parseFormatting(content *event.MessageEventContent, allowInlineURL, forceHTML bool) {
 	parsedHTML := parseWAFormattingToHTML(content.Body, allowInlineURL)
 	if forceHTML || parsedHTML != event.TextToHTML(content.Body) {