connector: don't save reconnection state before events are handled

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 {

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,35 +185,41 @@ 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():
 			return
 		}
 	}
 }
-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) {