From 8ddca3adc91175fb02ed55fa7ef769931427b209 Mon Sep 17 00:00:00 2001 From: Nick Mills-Barrett Date: Mon, 27 Jul 2026 15:01:40 +0100 Subject: [PATCH] connector: don't save reconnection state before events are handled The messagix sync cursor advances as soon as a table is parsed and queued, so persisting the dumped state while the derived events are still queued (or were dropped when the context was cancelled) checkpoints past work that never happened. Meta then never re-reports those thread ranges, so any messages in them are lost permanently. Count outstanding parsed tables and skip saving the state while any are unhandled. The counter is swapped for a fresh one when a new messagix client is created, so a late decrement from the old table loop can't make the new connection look up to date. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/connector/client.go | 12 ++++++++++ pkg/connector/handlemeta.go | 44 +++++++++++++++++++++++++------------ 2 files changed, 42 insertions(+), 14 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 8087b8e3..8032c6f9 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -40,6 +40,7 @@ type MetaClient struct { initialTable atomic.Pointer[table.LSTable] initialTableHandled atomic.Bool parsedTables chan *parsedTable + pendingTables atomic.Pointer[atomic.Int64] backfillCollectors map[int64]*BackfillCollector backfillLock sync.Mutex connectLock sync.Mutex @@ -95,6 +96,7 @@ func (m *MetaConnector) LoadUserLogin(ctx context.Context, login *bridgev2.UserL e2eeConnectWaiter: exsync.NewEvent(), } c.editChannels = exsync.NewMap[string, chan *FBEditEvent]() + c.pendingTables.Store(&atomic.Int64{}) login.Client = c return nil } @@ -143,6 +145,10 @@ func (m *MetaConnector) getProxy(reason string) (string, error) { func (m *MetaClient) ensureMessagixClient() { if m.LoginMeta.Cookies != nil && m.Client == nil { + // The new client will replay from the last saved state, so tables counted for the old one + // are no longer relevant. Swapping the counter rather than resetting it means a late + // decrement from the old table loop can't make the new one look up to date. + m.pendingTables.Store(&atomic.Int64{}) m.LoginMeta.Cookies.Platform = m.LoginMeta.Platform m.Client = messagix.NewClient( m.LoginMeta.Cookies, @@ -500,6 +506,12 @@ func (m *MetaClient) saveConnectionState(ctx context.Context, state json.RawMess if !m.Main.Config.CacheConnectionState { return } + if pending := m.pendingTables.Load().Load(); pending != 0 { + zerolog.Ctx(ctx).Warn(). + Int64("pending_tables", pending). + Msg("Not saving reconnection state, some events haven't been handled yet") + return + } m.lastStateSaveLock.Lock() ratelimited := time.Since(m.lastStateSave) < time.Minute if !ratelimited { diff --git a/pkg/connector/handlemeta.go b/pkg/connector/handlemeta.go index b01cf835..77d46e42 100644 --- a/pkg/connector/handlemeta.go +++ b/pkg/connector/handlemeta.go @@ -3,6 +3,7 @@ package connector import ( "context" "net/url" + "sync/atomic" "time" "github.com/coder/websocket" @@ -139,14 +140,20 @@ func (m *MetaClient) handleMetaEvent(ctx context.Context, rawEvt any) { func (m *MetaClient) parseAndQueueTable(ctx context.Context, tbl *table.LSTable, isInitial bool) { evts := m.parseTable(ctx, tbl) + pending := m.pendingTables.Load() wrapped := &parsedTable{ Table: tbl, Events: evts, IsInitial: isInitial, + Pending: pending, } if ctx.Err() != nil { zerolog.Ctx(ctx).Warn().Err(ctx.Err()).Msg("Not dispatching parsed table, context is canceled") } + // Count the table before returning: the sync manager advances the cursor past it as soon as + // this returns, so the reconnection state mustn't be saved until the events are handled. + // The dropped-table branch below deliberately doesn't undo this. + pending.Add(1) select { case m.parsedTables <- wrapped: default: @@ -163,6 +170,7 @@ type parsedTable struct { Table *table.LSTable Events []bridgev2.RemoteEvent IsInitial bool + Pending *atomic.Int64 } func (m *MetaClient) handleTableLoop(ctx context.Context) { @@ -177,7 +185,9 @@ func (m *MetaClient) handleTableLoop(ctx context.Context) { select { case evt := <-m.parsedTables: m.notifyBackgroundConnAboutEvent(true) - m.handleParsedTable(ctx, evt.IsInitial, evt.Table, evt.Events) + if m.handleParsedTable(ctx, evt.IsInitial, evt.Table, evt.Events) { + evt.Pending.Add(-1) + } m.saveConnectionState(ctx, nil) m.notifyBackgroundConnAboutEvent(false) case <-ctx.Done(): @@ -185,27 +195,31 @@ func (m *MetaClient) handleTableLoop(ctx context.Context) { } } } -func (m *MetaClient) handleParsedTable(ctx context.Context, isInitial bool, tbl *table.LSTable, innerQueue []bridgev2.RemoteEvent) { + +// handleParsedTable returns whether every event in the table was handled. The caller must not +// persist the reconnection state unless it did, as the sync cursor has already moved past this +// table and anything dropped here would never be delivered again. +func (m *MetaClient) handleParsedTable(ctx context.Context, isInitial bool, tbl *table.LSTable, innerQueue []bridgev2.RemoteEvent) bool { for _, contact := range tbl.LSDeleteThenInsertContact { if ctx.Err() != nil { - return + return false } m.syncGhost(ctx, contact) } for _, contact := range tbl.LSVerifyContactRowExists { if ctx.Err() != nil { - return + return false } m.syncGhost(ctx, contact) } if ctx.Err() != nil { - return + return false } if m.Client.GetPlatform() == types.Instagram { contactsWithoutIGID := []int64{} for _, contact := range tbl.LSVerifyContactRowExists { if ctx.Err() != nil { - return + return false } igid, err := m.Main.DB.GetIGUserForFBID(ctx, contact.GetFBID()) if err != nil { @@ -257,23 +271,25 @@ func (m *MetaClient) handleParsedTable(ctx context.Context, isInitial bool, tbl } for _, evt := range innerQueue { if ctx.Err() != nil { - return + return false } res := m.UserLogin.QueueRemoteEvent(evt) if !res.Success { zerolog.Ctx(ctx).Warn(). Any("queue_result", res). Msg("Queue remote event returned non-success status, cancelling table handling") - return + return false } } - if ctx.Err() == nil { - if isInitial { - m.initialTableHandled.Store(true) - } else { - m.Client.PostHandlePublishResponse(tbl) - } + if ctx.Err() != nil { + return false + } + if isInitial { + m.initialTableHandled.Store(true) + } else { + m.Client.PostHandlePublishResponse(tbl) } + return true } func (m *MetaClient) syncGhost(ctx context.Context, info types.UserInfo) {