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

pkg/connector/backfill.go

@@ -399,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").
@@ -442,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 {
@@ -482,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,
@@ -497,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()
 				},
 			},
@@ -504,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
 		}
@@ -511,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