Merge pull request #418 from matrix-org/kegan/account-data-dupe

bugfix: don't send dupe room account data

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 {
@@ -50,10 +53,17 @@ func (r *AccountDataRequest) AppendLive(ctx context.Context, res *Response, extC
 		}
 	case caches.RoomUpdate:
 		if !r.RoomInScope(update.RoomID(), extCtx) {
-			break
+			return
 		}
 		// 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.
+				return
+			}
 			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
 			}
 		}
 	}

tests-e2e/account_data_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/matrix-org/complement/b"
 	"github.com/matrix-org/sliding-sync/sync3"
 	"github.com/matrix-org/sliding-sync/sync3/extensions"
 	"github.com/matrix-org/sliding-sync/testutils"
@@ -210,6 +211,69 @@ func TestAccountDataDoesntDupe(t *testing.T) {
 	}
 }
 
+// Regression test for https://github.com/matrix-org/sliding-sync/issues/417
+func TestAccountDataDoesntDupeWithUpdates(t *testing.T) {
+	alice := registerNewUser(t)
+	roomID := alice.MustCreateRoom(t, map[string]interface{}{
+		"preset": "public_chat",
+	})
+	roomAccountData1 := putRoomAccountData(t, alice, roomID, "foo", map[string]interface{}{"a": "b"})
+	roomAccountData2 := putRoomAccountData(t, alice, roomID, "bar", map[string]interface{}{"a2": "b2"})
+
+	// all_rooms with timeline limit 1 initially
+	res := alice.SlidingSync(t, sync3.Request{
+		Extensions: extensions.Request{
+			AccountData: &extensions.AccountDataRequest{
+				Core: extensions.Core{
+					Enabled: &boolTrue,
+				},
+			},
+		},
+		Lists: map[string]sync3.RequestList{
+			"all_rooms": {
+				Ranges: sync3.SliceRanges{{0, 10}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 1,
+				},
+			},
+		},
+	})
+	m.MatchResponse(t, res, m.MatchList("all_rooms", m.MatchV3Count(1)), m.MatchAccountData(nil, map[string][]json.RawMessage{
+		roomID: {roomAccountData1, roomAccountData2},
+	}))
+
+	// now send some events into the room
+	alice.SendTyping(t, roomID, true, 5000)
+	alice.SendEventSynced(t, roomID, b.Event{
+		Type: "custom",
+		Content: map[string]interface{}{
+			"foo": "bar",
+		},
+	})
+
+	// now sync with a visible_rooms list, we should not see duplicate account data (though do expect 2 as we're changing the subscription)
+	res = alice.SlidingSync(t, sync3.Request{
+		Lists: map[string]sync3.RequestList{
+			"all_rooms": {
+				Ranges: sync3.SliceRanges{{0, 10}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 1,
+				},
+			},
+			"visible_rooms": {
+				Ranges: sync3.SliceRanges{{0, 10}},
+				RoomSubscription: sync3.RoomSubscription{
+					TimelineLimit: 10,
+				},
+			},
+		},
+	}, WithPos(res.Pos))
+	m.MatchResponse(t, res, m.MatchList("all_rooms", m.MatchV3Count(1)), m.MatchAccountData(nil, map[string][]json.RawMessage{
+		roomID: {roomAccountData1, roomAccountData2},
+	}))
+
+}
+
 // putAccountData is a wrapper around SetGlobalAccountData. It returns the account data
 // event as a json.RawMessage, automatically including top-level `type` and `content`
 // fields. This is useful because it can be compared with account data events in a