Fix sending dupe room account data

Fixes #417

sync3/extensions/account_data.go

@@ -3,6 +3,7 @@ package extensions
 import (
 	"context"
 	"encoding/json"
+
 	"github.com/matrix-org/sliding-sync/internal"
 	"github.com/matrix-org/sliding-sync/state"
 	"github.com/matrix-org/sliding-sync/sync3/caches"
@@ -21,6 +22,8 @@ func (r *AccountDataRequest) Name() string {
 type AccountDataResponse struct {
 	Global []json.RawMessage            `json:"global,omitempty"`
 	Rooms  map[string][]json.RawMessage `json:"rooms,omitempty"`
+	// which rooms have had account data loaded from the DB in this response
+	loadedRooms map[string]bool
 }
 
 func (r *AccountDataResponse) HasData(isInitial bool) bool {
@@ -54,6 +57,13 @@ func (r *AccountDataRequest) AppendLive(ctx context.Context, res *Response, extC
 		}
 		// if this is a room update which is included in the response, send account data for this room
 		if _, exists := extCtx.RoomIDToTimeline[update.RoomID()]; exists {
+			// ..but only if we haven't before
+			if res.AccountData != nil && res.AccountData.loadedRooms[update.RoomID()] {
+				// we've loaded this room before, don't do it again
+				// this can happen when we consume lots of items in the buffer. If many of them are room updates
+				// for the same room, we could send dupe room account data if we didn't do this check.
+				break
+			}
 			roomAccountData, err := extCtx.Store.AccountDatas(extCtx.UserID, update.RoomID())
 			if err != nil {
 				logger.Err(err).Str("user", extCtx.UserID).Str("room", update.RoomID()).Msg("failed to fetch room account data")
@@ -70,12 +80,14 @@ func (r *AccountDataRequest) AppendLive(ctx context.Context, res *Response, extC
 	}
 	if res.AccountData == nil {
 		res.AccountData = &AccountDataResponse{
-			Rooms: make(map[string][]json.RawMessage),
+			Rooms:       make(map[string][]json.RawMessage),
+			loadedRooms: make(map[string]bool),
 		}
 	}
 	res.AccountData.Global = append(res.AccountData.Global, globalMsgs...)
 	for roomID, roomAccountData := range roomToMsgs {
 		res.AccountData.Rooms[roomID] = append(res.AccountData.Rooms[roomID], roomAccountData...)
+		res.AccountData.loadedRooms[roomID] = true
 	}
 }
 
@@ -89,7 +101,8 @@ func (r *AccountDataRequest) ProcessInitial(ctx context.Context, res *Response,
 		}
 	}
 	extRes := &AccountDataResponse{
-		Rooms: make(map[string][]json.RawMessage),
+		Rooms:       make(map[string][]json.RawMessage),
+		loadedRooms: make(map[string]bool),
 	}
 	// room account data needs to be sent every time the user scrolls the list to get new room IDs
 	// TODO: remember which rooms the client has been told about
@@ -102,6 +115,7 @@ func (r *AccountDataRequest) ProcessInitial(ctx context.Context, res *Response,
 			extRes.Rooms = make(map[string][]json.RawMessage)
 			for _, ad := range roomsAccountData {
 				extRes.Rooms[ad.RoomID] = append(extRes.Rooms[ad.RoomID], ad.Data)
+				extRes.loadedRooms[ad.RoomID] = true
 			}
 		}
 	}