From c63981be768f02aa11abc2505a7a6197de12bd38 Mon Sep 17 00:00:00 2001 From: Indent Date: Sun, 8 Nov 2026 06:27:35 +0000 Subject: [PATCH 1/2] backfill: don't requeue history sync conversations that are still being created --- pkg/connector/backfill.go | 9 +++++++++ pkg/connector/client.go | 2 ++ 2 files changed, 11 insertions(+) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index cbc7eec..2d33535 100644 --- a/pkg/connector/backfill.go +++ b/pkg/connector/backfill.go @@ -442,6 +442,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 wa.historySyncQueued.Has(conv.ChatJID) { + wg.Done() + continue } // TODO can the chat info fetch be avoided entirely? select { @@ -482,6 +485,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } + wa.historySyncQueued.Add(conv.ChatJID) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -497,6 +501,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.Remove(conv.ChatJID) wg.Done() }, }, @@ -504,6 +509,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { + wa.historySyncQueued.Remove(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -511,6 +517,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.Size() > 0 { + return + } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false err = wa.UserLogin.Save(ctx) if err != nil { diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 678c8d2..3630972 100644 --- a/pkg/connector/client.go +++ b/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.NewSet[types.JID](), 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.Set[types.JID] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex From efdaad906581d8db6a24dd0a6c19a191711f5bbe Mon Sep 17 00:00:00 2001 From: Indent Date: Sun, 8 Nov 2026 07:23:11 +0000 Subject: [PATCH 2/2] backfill: let in-flight history sync entries expire so failed creations are retried --- pkg/connector/backfill.go | 12 +++++++----- pkg/connector/client.go | 4 ++-- 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/pkg/connector/backfill.go b/pkg/connector/backfill.go index 2d33535..e2a18b1 100644 --- a/pkg/connector/backfill.go +++ b/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,7 +444,7 @@ 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 wa.historySyncQueued.Has(conv.ChatJID) { + } else if queuedAt, ok := wa.historySyncQueued.Get(conv.ChatJID); ok && time.Since(queuedAt) < historySyncQueuedTimeout { wg.Done() continue } @@ -485,7 +487,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { wg.Done() continue } - wa.historySyncQueued.Add(conv.ChatJID) + wa.historySyncQueued.Set(conv.ChatJID, time.Now()) res := wa.UserLogin.QueueRemoteEvent(&simplevent.ChatResync{ EventMeta: simplevent.EventMeta{ Type: bridgev2.RemoteEventChatResync, @@ -501,7 +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.Remove(conv.ChatJID) + wa.historySyncQueued.Delete(conv.ChatJID) wg.Done() }, }, @@ -509,7 +511,7 @@ func (wa *WhatsAppClient) createPortalsFromHistorySync(ctx context.Context) { LatestMessageTS: conv.LastMessageTimestamp, }) if !res.Success { - wa.historySyncQueued.Remove(conv.ChatJID) + wa.historySyncQueued.Delete(conv.ChatJID) log.Debug().Msg("Cancelling history sync portal creation loop") return } @@ -517,7 +519,7 @@ 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.Size() > 0 { + if wa.historySyncQueued.Len() > 0 { return } wa.UserLogin.Metadata.(*waid.UserLoginMetadata).HistorySyncPortalsNeedCreating = false diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 3630972..9b3f190 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -50,7 +50,7 @@ func (wa *WhatsAppConnector) LoadUserLogin(ctx context.Context, login *bridgev2. MC: noopMCInstance, historySyncWakeup: make(chan struct{}, 1), - historySyncQueued: exsync.NewSet[types.JID](), + 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), @@ -115,7 +115,7 @@ type WhatsAppClient struct { MC mClient historySyncWakeup chan struct{} - historySyncQueued *exsync.Set[types.JID] + historySyncQueued *exsync.Map[types.JID, time.Time] stopLoops atomic.Pointer[context.CancelFunc] resyncQueue map[types.JID]resyncQueueItem resyncQueueLock sync.Mutex