From 620a20a56ab3e4ea0d25299d2920534dfef971a5 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 17 Sep 2026 15:55:43 -0400 Subject: [PATCH 1/2] chatsync: fixed delayed messages during reconnect catch-up --- pkg/connector/client.go | 5 +- pkg/connector/conversationdata.go | 1 + pkg/connector/xchat_reconnect.go | 202 +++++++++++++++++++++++--- pkg/twittermeow/client.go | 6 +- pkg/twittermeow/websocket.go | 219 ++++++++++++++++++++++------- pkg/twittermeow/xchat_processor.go | 24 +++- 6 files changed, 382 insertions(+), 75 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index e3a43f5..109ce46 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -406,13 +406,14 @@ func (tc *TwitterClient) connect(ctx context.Context) { copy := *value msgPullVersion = © } - tc.client.SetXChatConnectHandler(func(connectCtx context.Context) error { + tc.client.SetXChatConnectHandler(func(connectCtx context.Context, drain twittermeow.XChatLiveDrain) error { return tc.syncXChatInboxAfterConnect( connectCtx, getMaxSeqID, setMaxSeqID, getMessagePullVersion, setMessagePullVersion, + drain, ) }) @@ -722,6 +723,8 @@ func (tc *TwitterClient) cacheUsersFromItem(item *response.XChatInboxItem) []str // HandleConversationDataRefresh is called when conversation data is fetched on-demand. // It syncs the room data (members, name, avatar, etc.) from the fetched conversation data. func (tc *TwitterClient) HandleConversationDataRefresh(ctx context.Context, conversationID string, item *response.XChatInboxItem) { + tc.xchatInboxSyncLock.Lock() + defer tc.xchatInboxSyncLock.Unlock() if item == nil { return } diff --git a/pkg/connector/conversationdata.go b/pkg/connector/conversationdata.go index e832450..a3133f2 100644 --- a/pkg/connector/conversationdata.go +++ b/pkg/connector/conversationdata.go @@ -70,6 +70,7 @@ func (tc *TwitterClient) fetchConversationData(ctx context.Context, conversation collect(data.ConversationDetail.GroupMembersResults) collect(data.ConversationDetail.GroupAdminsResults) + tc.cacheUsersFromItem(&data) if err := tc.ensureUsersInCacheByID(ctx, missingIDs); err != nil { // Profile metadata is useful for room naming, but it must not prevent a // newly received message from creating/syncing its portal. The member-list diff --git a/pkg/connector/xchat_reconnect.go b/pkg/connector/xchat_reconnect.go index 90c3c3a..3a6a101 100644 --- a/pkg/connector/xchat_reconnect.go +++ b/pkg/connector/xchat_reconnect.go @@ -2,9 +2,9 @@ package connector import ( "context" + "encoding/json" "errors" "fmt" - "sync" "sync/atomic" "github.com/rs/zerolog" @@ -284,6 +284,7 @@ func (tc *TwitterClient) syncXChatInboxAfterConnect( setMaxSequenceID func(string), getMessagePullVersion func() *int, setMessagePullVersion func(*int), + drain twittermeow.XChatLiveDrain, ) error { tc.xchatInboxSyncLock.Lock() defer tc.xchatInboxSyncLock.Unlock() @@ -300,17 +301,15 @@ func (tc *TwitterClient) syncXChatInboxAfterConnect( } processor := tc.client.GetXChatProcessor() - var stagedMaxSequenceID string - var stagedMaxLock sync.Mutex - processor.SetSequenceIDCallback(func(sequenceID string) { - stagedMaxLock.Lock() - stagedMaxSequenceID = maxXChatSequenceID(stagedMaxSequenceID, sequenceID) - stagedMaxLock.Unlock() - }) - defer processor.SetSequenceIDCallback(setMaxSequenceID) + publishedSequence := maxSequenceID + processor.SetSequenceIDCallback(nil) + defer func() { + processor.CapHandledSequenceID(publishedSequence) + processor.SetSequenceIDCallback(setMaxSequenceID) + }() var totalItems atomic.Int32 - result, err := runXChatInboxCatchup(ctx, xchatInboxCatchupState{ + result, err := runXChatPriorityCatchup(ctx, xchatInboxCatchupState{ MaxSequenceID: maxSequenceID, MessagePullVersion: messagePullVersion, Cursor: xchatInboxCursorFromMetadata(meta.XChatInboxCursor), @@ -332,23 +331,26 @@ func (tc *TwitterClient) syncXChatInboxAfterConnect( }, ProcessPage: func(ctx context.Context, page response.XChatInboxPage) (xchatInboxPageProcessResult, error) { _, err := tc.processXChatInboxPage(ctx, page, &totalItems, true) - stagedMaxLock.Lock() - observedMax := stagedMaxSequenceID - stagedMaxLock.Unlock() - observedMax = maxXChatSequenceID(observedMax, processor.MaxHandledSequenceID()) return xchatInboxPageProcessResult{ - MaxSequenceID: observedMax, CheckpointBlocked: processor.SequenceCheckpointBlocked(), }, err }, - Checkpoint: tc.saveXChatInboxCheckpoint, - }) + }, tc.prepareXChatSnapshotProfiles, processor.ConversationRecoveryPending, drain) if err != nil { return err } - setMaxSequenceID(result.MaxSequenceID) - setMessagePullVersion(result.MessagePullVersion) + if !processor.SequenceCheckpointBlocked() { + if err = tc.saveXChatInboxCheckpoint(ctx, nil, result.MaxSequenceID, result.MessagePullVersion); err != nil { + return err + } + publishedSequence = result.MaxSequenceID + } + setMaxSequenceID(publishedSequence) + if !processor.SequenceCheckpointBlocked() { + setMessagePullVersion(result.MessagePullVersion) + } + result.CheckpointBlocked = processor.SequenceCheckpointBlocked() completionLog := log.Info() if result.CheckpointBlocked { completionLog = log.Warn() @@ -356,8 +358,168 @@ func (tc *TwitterClient) syncXChatInboxAfterConnect( completionLog. Int("pages", result.Pages). Int("items", result.Items). - Str("max_sequence_id", result.MaxSequenceID). + Str("max_sequence_id", publishedSequence). Bool("checkpoint_blocked", result.CheckpointBlocked). Msg("XChat reconnect catch-up completed") return nil } + +func runXChatPriorityCatchup(ctx context.Context, state xchatInboxCatchupState, ops xchatInboxCatchupOps, prepare func(context.Context, []response.XChatInboxPage) error, recoveryPending func(string) bool, drain twittermeow.XChatLiveDrain) (xchatInboxCatchupResult, error) { + var pages []response.XChatInboxPage + bufferedBytes, bufferedItems := 0, 0 + fallback := drain == nil + collect := ops + collect.Checkpoint = nil + seen := map[payload.XChatCursor]bool{} + collect.FetchNext = func(ctx context.Context, vars *payload.GetInboxPageRequestQueryVariables) (response.XChatInboxPage, error) { + if !fallback { + key := *vars.ContinueCursor + if seen[key] { + return response.XChatInboxPage{}, fmt.Errorf("XChat inbox cursor cycle") + } + seen[key] = true + } + return ops.FetchNext(ctx, vars) + } + collect.ProcessPage = func(ctx context.Context, page response.XChatInboxPage) (xchatInboxPageProcessResult, error) { + if !fallback { + encoded, err := json.Marshal(page) + if err != nil { + return xchatInboxPageProcessResult{}, err + } + bufferedBytes += len(encoded) + bufferedItems += len(page.Items) + fallback = len(pages) >= 128 || bufferedBytes > 16<<20 || bufferedItems > 4096 || page.MessageEventsCursor != nil + for i := range page.Items { + fallback = fallback || !twittermeow.XChatInboxItemIsConversationScoped(&page.Items[i]) + } + } + if fallback { + seen = nil + for _, buffered := range pages { + if _, err := ops.ProcessPage(ctx, buffered); err != nil { + return xchatInboxPageProcessResult{}, err + } + } + pages = nil + processed, err := ops.ProcessPage(ctx, page) + // Ignore sequence progress from on-demand history. + processed.MaxSequenceID = "" + if page.MessageEventsCursor != nil { + processed.MaxSequenceID = page.MessageEventsCursor.MaxLocalSequenceID + for _, encoded := range page.EncodedMessageEvents { + if event, decodeErr := twittermeow.DecodeMessageEvent(encoded); decodeErr == nil && event.SequenceId != nil { + processed.MaxSequenceID = maxXChatSequenceID(processed.MaxSequenceID, *event.SequenceId) + } + } + } + return processed, err + } + pages = append(pages, page) + return xchatInboxPageProcessResult{}, nil + } + result, err := runXChatInboxCatchup(ctx, state, collect) + if err != nil { + return result, err + } + if !fallback { + if err = prepare(ctx, pages); err != nil { + return result, err + } + pending := map[string][]response.XChatInboxItem{} + known := map[string]bool{} + order := []string{} + for _, page := range pages { + for _, item := range page.Items { + id := item.ConversationDetail.ConversationID + known[id] = true + if _, ok := pending[id]; !ok { + order = append(order, id) + } + pending[id] = append(pending[id], item) + } + } + apply := func(ctx context.Context, items []response.XChatInboxItem) error { + for _, item := range items { + if _, err := ops.ProcessPage(ctx, response.XChatInboxPage{Items: []response.XChatInboxItem{item}}); err != nil { + return err + } + } + return nil + } + applyIDs := func(ids []string) error { + for _, id := range ids { + if err := apply(ctx, pending[id]); err != nil { + return err + } + delete(pending, id) + } + return nil + } + before := func(message *payload.Message) error { + ids := twittermeow.XChatMessageConversations(message) + if ids == nil { + return applyIDs(order) + } + if err := applyIDs(ids); err != nil { + return err + } + for _, id := range ids { + if !known[id] || recoveryPending(id) { + return applyIDs(order) + } + } + return nil + } + for offset := 0; offset < len(order); offset += 10 { + if err = drain(before); err != nil { + return result, err + } + group, groupCtx := errgroup.WithContext(ctx) + for _, id := range order[offset:min(offset+10, len(order))] { + items := pending[id] + delete(pending, id) + group.Go(func() error { return apply(groupCtx, items) }) + } + if err = group.Wait(); err != nil { + return result, err + } + } + } + if drain != nil { + err = drain(nil) + } + return result, err +} + +// Deferred rooms must use the newest profiles from the snapshot. +func (tc *TwitterClient) prepareXChatSnapshotProfiles(ctx context.Context, pages []response.XChatInboxPage) error { + var missing []string + strip := func(input []response.XChatUserResult) []response.XChatUserResult { + users := append([]response.XChatUserResult(nil), input...) + for i := range users { + if id, _ := xchatUserFromResult(users[i]); id != "" { + users[i].RestID = id + users[i].Result = nil + } + } + return users + } + for p := range pages { + if err := ctx.Err(); err != nil { + return err + } + pages[p].Items = append([]response.XChatInboxItem(nil), pages[p].Items...) + for i := range pages[p].Items { + missing = append(missing, tc.cacheUsersFromItem(&pages[p].Items[i])...) + detail := &pages[p].Items[i].ConversationDetail + detail.ParticipantsResults = strip(detail.ParticipantsResults) + detail.GroupMembersResults = strip(detail.GroupMembersResults) + detail.GroupAdminsResults = strip(detail.GroupAdminsResults) + } + } + if err := tc.ensureUsersInCacheByID(ctx, missing); err != nil { + zerolog.Ctx(ctx).Warn().Err(err).Msg("Failed to prefetch snapshot users") + } + return ctx.Err() +} diff --git a/pkg/twittermeow/client.go b/pkg/twittermeow/client.go index 33e3fa4..cebad84 100644 --- a/pkg/twittermeow/client.go +++ b/pkg/twittermeow/client.go @@ -38,10 +38,8 @@ type StreamEventHandler func(evt response.StreamEvent) // The callback receives the conversation ID and the inbox item containing the latest conversation data. type ConversationDataCallback func(ctx context.Context, conversationID string, item *response.XChatInboxItem) -// XChatConnectHandler repairs any events that may have been missed before a -// websocket connection. It runs after each socket is established and before -// that socket starts consuming live frames. -type XChatConnectHandler func(ctx context.Context) error +// XChatConnectHandler repairs missed events before normal live dispatch starts. +type XChatConnectHandler func(ctx context.Context, drain XChatLiveDrain) error // XChatTokenTTL is how long an XChat token is considered valid. const XChatTokenTTL = 5 * time.Minute diff --git a/pkg/twittermeow/websocket.go b/pkg/twittermeow/websocket.go index 0a57adf..1992506 100644 --- a/pkg/twittermeow/websocket.go +++ b/pkg/twittermeow/websocket.go @@ -16,6 +16,7 @@ import ( "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/endpoints" "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/payload" + "go.mau.fi/mautrix-twitter/pkg/twittermeow/data/response" ) // Reconnect configuration @@ -65,7 +66,7 @@ type xchatWebsocketClient struct { writeMu sync.Mutex tokenProvider func(context.Context, bool) (string, error) - connectionRunner func(context.Context, string, zerolog.Logger, func(context.Context) error) (bool, error) + connectionRunner func(context.Context, string, zerolog.Logger, func(context.Context, XChatLiveDrain) error) (bool, error) initialRetryDelay time.Duration maximumRetryDelay time.Duration } @@ -160,17 +161,17 @@ func (xc *xchatWebsocketClient) start(ctx context.Context) error { // Run connection (blocks until disconnect) readyThisAttempt := false var connectedAt time.Time - refreshToken, err := xc.runConnectionAttempt(ctx, token, log, func(connectionCtx context.Context) error { + refreshToken, err := xc.runConnectionAttempt(ctx, token, log, func(connectionCtx context.Context, drain XChatLiveDrain) error { + if xc.client.xchatProcessor != nil { + xc.client.xchatProcessor.MarkReconnected() + } if connectHandler := xc.client.getXChatConnectHandler(); connectHandler != nil { log.Info().Msg("Running XChat socket handoff catch-up") - if err := connectHandler(connectionCtx); err != nil { + if err := connectHandler(connectionCtx, drain); err != nil { return fmt.Errorf("socket handoff catch-up failed: %w", err) } log.Info().Msg("Finished XChat socket handoff catch-up") } - if xc.client.xchatProcessor != nil { - xc.client.xchatProcessor.MarkReconnected() - } readyThisAttempt = true connectedAt = time.Now() readyOnce.Do(func() { @@ -226,7 +227,7 @@ func (xc *xchatWebsocketClient) runConnectionAttempt( ctx context.Context, token string, log zerolog.Logger, - onConnected func(context.Context) error, + onConnected func(context.Context, XChatLiveDrain) error, ) (refreshToken bool, err error) { defer func() { if recovered := recover(); recovered != nil { @@ -299,7 +300,7 @@ func (xc *xchatWebsocketClient) runConnection( ctx context.Context, token string, log zerolog.Logger, - onConnected func(context.Context) error, + onConnected func(context.Context, XChatLiveDrain) error, ) (bool, error) { wsURL, err := url.Parse(endpoints.XCHAT_WEBSOCKET_URL) if err != nil { @@ -379,49 +380,59 @@ func (xc *xchatWebsocketClient) runConnection( } }() - if onConnected != nil { - if err := onConnected(pingCtx); err != nil { - return false, fmt.Errorf("xchat connected hook failed: %w", err) - } - } - - for { - msgType, data, err := conn.Read(pingCtx) - if err != nil { - status := websocket.CloseStatus(err) - closeReceived := status != -1 - if status == websocket.StatusNormalClosure || status == websocket.StatusGoingAway { - log.Debug().Err(err).Uint32("status", uint32(status)).Msg("XChat websocket closed by server") - } else if ctx.Err() != nil { - log.Debug().Err(err).Msg("XChat websocket read stopped by context") - } else { - log.Error().Err(err).Uint32("status", uint32(status)).Msg("XChat websocket read failed") + conn.SetReadLimit(32768) + readCtx, cancelRead := context.WithCancelCause(pingCtx) + frames := make(chan []byte, 32) + done := make(chan struct{}) + var readErr error + var refreshToken bool + go func() { + defer close(done) + defer close(frames) + for { + typ, data, err := conn.Read(readCtx) + if err != nil { + readErr = fmt.Errorf("read XChat websocket: %w", err) + status := websocket.CloseStatus(err) + refreshToken = status != -1 + if status == websocket.StatusNormalClosure || status == websocket.StatusGoingAway { + log.Debug().Err(err).Uint32("status", uint32(status)).Msg("XChat websocket closed by server") + } else if readCtx.Err() != nil { + log.Debug().Err(err).Msg("XChat websocket read stopped by context") + } else { + log.Error().Err(err).Uint32("status", uint32(status)).Msg("XChat websocket read failed") + } + cancelRead(readErr) + return + } + log.Debug().Str("type", typ.String()).Int("bytes", len(data)).Msg("Received XChat websocket message") + if typ != websocket.MessageBinary || len(data) == 0 { + continue + } + select { + case frames <- data: + case <-readCtx.Done(): + return } - return closeReceived, fmt.Errorf("read failed: %w", err) } - - // Handle message. Currently we just trace-log; hook processing here as formats become known. - log.Debug(). - Str("type", msgType.String()). - Int("bytes", len(data)). - Msg("Received XChat websocket message") - - if msgType != websocket.MessageBinary { - log.Debug().Int("bytes", len(data)).Msg("Skipping non-binary XChat websocket frame") - continue + }() + defer func() { cancelRead(context.Canceled); <-done }() + result := func(err error) (bool, error) { + cancelRead(err) + <-done + if readErr != nil && (refreshToken || !errors.Is(readErr, context.Canceled)) { + return refreshToken, readErr } - if len(data) == 0 { - log.Debug().Msg("Skipping empty XChat websocket frame") - continue + return false, err + } + process := func(data []byte, before func(*payload.Message) error) error { + if err := readCtx.Err(); err != nil { + return err } - decoded, err := decodeXChatPayload(data) if err != nil { - log.Warn(). - Err(err). - Int("bytes", len(data)). - Msg("Failed to decode XChat websocket payload") - return false, fmt.Errorf("decode XChat websocket payload: %w", err) + log.Warn().Err(err).Int("bytes", len(data)).Msg("Failed to decode XChat websocket payload") + return err } batchedEventCount := 0 if decoded.BatchedMessageEvents != nil { @@ -432,11 +443,44 @@ func (xc *xchatWebsocketClient) runConnection( Bool("has_instruction", decoded.MessageInstruction != nil). Int("batched_event_count", batchedEventCount). Msg("Decoded XChat websocket payload") - - // Process the message through the XChat processor - if err := xc.client.xchatProcessor.ProcessMessage(pingCtx, decoded); err != nil { + if before != nil { + if err = before(decoded); err != nil { + return err + } + } + err = xc.client.xchatProcessor.ProcessMessage(readCtx, decoded) + if err != nil { log.Err(err).Msg("Failed to process XChat message") - return false, fmt.Errorf("process XChat websocket payload: %w", err) + } + return err + } + receive := func(before func(*payload.Message) error) error { + select { + case data, ok := <-frames: + if !ok { + return context.Cause(readCtx) + } + return process(data, before) + case <-readCtx.Done(): + return context.Cause(readCtx) + } + } + drain := func(before func(*payload.Message) error) error { + for range len(frames) { + if err := receive(before); err != nil { + return err + } + } + return readCtx.Err() + } + if onConnected != nil { + if err := onConnected(readCtx, drain); err != nil { + return result(err) + } + } + for { + if err := receive(nil); err != nil { + return result(err) } } } @@ -446,3 +490,80 @@ func (xc *xchatWebsocketClient) stop() { (*cancel)() } } + +// XChatLiveDrain calls before, then dispatches, for a finite FIFO prefix. +type XChatLiveDrain func(before func(*payload.Message) error) error + +// XChatMessageConversations returns nil when a payload needs a global barrier. +func XChatMessageConversations(message *payload.Message) []string { + return xchatMessageConversations(message, "") +} + +func xchatMessageConversations(message *payload.Message, snapshotConversation string) []string { + if message == nil { + return nil + } + ids := []string{} + add := func(id *string) bool { + if id == nil || *id == "" || (snapshotConversation != "" && *id != snapshotConversation) { + return false + } + ids = append(ids, *id) + return true + } + events := []*payload.MessageEvent{} + if message.MessageEvent != nil { + events = append(events, message.MessageEvent) + } + if message.BatchedMessageEvents != nil { + events = append(events, message.BatchedMessageEvents.MessageEvents...) + } + if len(events) == 0 && message.MessageInstruction == nil { + return nil + } + for _, event := range events { + if event == nil || !add(event.ConversationId) { + return nil + } + if event.Detail == nil { + continue + } + if deletion := event.Detail.ConversationDeleteEvent; deletion != nil && (snapshotConversation == "" || !add(deletion.ConversationId)) { + return nil + } + if typing := event.Detail.MessageTypingEvent; typing != nil && typing.ConversationId != nil && !add(typing.ConversationId) { + return nil + } + } + return ids +} + +func XChatInboxItemIsConversationScoped(item *response.XChatInboxItem) bool { + id := item.ConversationDetail.ConversationID + if id == "" { + return false + } + encoded := append([]string{}, item.LatestMessageEvents...) + encoded = append(encoded, item.EncodedMessageEvents...) + encoded = append(encoded, item.LatestConversationKeyChangeEvents...) + encoded = append(encoded, item.LatestNotifiableMessageCreateEvent) + for _, read := range item.LatestReadEventsPerParticipant { + encoded = append(encoded, read.LatestMarkConversationReadEvent) + } + for _, raw := range encoded { + if raw == "" { + continue + } + event, err := DecodeMessageEvent(raw) + if err != nil { + return false + } + if event.ConversationId == nil || *event.ConversationId == "" { + event.ConversationId = &id + } + if xchatMessageConversations(&payload.Message{MessageEvent: event}, id) == nil { + return false + } + } + return true +} diff --git a/pkg/twittermeow/xchat_processor.go b/pkg/twittermeow/xchat_processor.go index 039006d..049be93 100644 --- a/pkg/twittermeow/xchat_processor.go +++ b/pkg/twittermeow/xchat_processor.go @@ -149,7 +149,7 @@ func (p *XChatEventProcessor) SetSequenceIDCallback(callback SequenceIDCallback) } // MaxHandledSequenceID returns the highest event sequence that completed -// handling during this connection, even while a gap blocks publication. +// handling, capped at the handoff cut while queued live frames remain. func (p *XChatEventProcessor) MaxHandledSequenceID() string { p.sequenceStateLock.Lock() defer p.sequenceStateLock.Unlock() @@ -231,6 +231,28 @@ func (p *XChatEventProcessor) ConversationGapUnresolved(conversationID string) b return unresolved } +func (p *XChatEventProcessor) CapHandledSequenceID(limit string) { + p.sequenceStateLock.Lock() + defer p.sequenceStateLock.Unlock() + if compareXChatSequenceIDs(p.maxHandledSequenceID, limit) > 0 { + p.maxHandledSequenceID = limit + } +} + +func (p *XChatEventProcessor) ConversationRecoveryPending(conversationID string) bool { + p.sequenceStateLock.Lock() + defer p.sequenceStateLock.Unlock() + if _, pending := p.unresolvedConversationGaps[conversationID]; pending { + return true + } + for _, id := range p.failedEvents { + if id == conversationID { + return true + } + } + return false +} + // ResetSequenceState clears connection-local ordering state before a full // reconnect. The persisted inbox checkpoint remains the source of truth. func (p *XChatEventProcessor) ResetSequenceState() { From eccaa4713c3cdd94f6758612c44385df72bc1a91 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 17 Sep 2026 16:27:55 -0400 Subject: [PATCH 2/2] websocket: fixed linting --- pkg/twittermeow/websocket.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/twittermeow/websocket.go b/pkg/twittermeow/websocket.go index 1992506..636ea24 100644 --- a/pkg/twittermeow/websocket.go +++ b/pkg/twittermeow/websocket.go @@ -405,7 +405,7 @@ func (xc *xchatWebsocketClient) runConnection( cancelRead(readErr) return } - log.Debug().Str("type", typ.String()).Int("bytes", len(data)).Msg("Received XChat websocket message") + log.Debug().Stringer("type", typ).Int("bytes", len(data)).Msg("Received XChat websocket message") if typ != websocket.MessageBinary || len(data) == 0 { continue }