From a42f3b77d431637cee04ad982e10d839a5e00bf7 Mon Sep 17 00:00:00 2001 From: Jason Robinson Date: Wed, 12 Aug 2026 16:20:55 +0300 Subject: [PATCH] Implement lazy loading based on timeline events If a room is configured for lazy loading, only members that had timeline events are included in the response, unless they would otherwise be included. --- synapse/handlers/sliding_sync/extensions.py | 158 +++++++++---- .../sliding_sync/test_extension_profiles.py | 216 +++++++++++++++++- 2 files changed, 329 insertions(+), 45 deletions(-) diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 6a74db6817..9f13054279 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -216,6 +216,7 @@ class SlidingSyncExtensionHandler: all_interested_room_ids=all_interested_room_ids, to_token=to_token, from_token=from_token, + actual_room_response_map=actual_room_response_map, ) ( @@ -1082,12 +1083,90 @@ class SlidingSyncExtensionHandler: ), ) + async def _get_profile_ids_for_profiles_extension( + self, + user_id: str, + rooms: set[str], + sync_config: SlidingSyncConfig, + actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], + ) -> tuple[set[str], set[str]]: + """ + Calculate target user profiles as candiates to include in the profile + extension sync response. + + This function looks at both the sync config and the already calculated + rooms response, and pieces together the full set of user ID's to include + profiles for, based on sync config rooms being lazy loading or not. + + For rooms with lazy loading, only profiles for those users who have sent events + into the timeline will be included, unless they would be included otherwise. + For other rooms, all members of the room will be included as candidates. + + Args: + user_id: The full user ID syncing. + rooms: A set of rooms that was already calculated as relevant for this + sync response. + sync_config: The Sliding Sync config object. + actual_room_response_map: A calculated map of responses per room. + + Returns: + Set of user ID's. + """ + users_in_timeline = set() + non_lazy_profile_user_ids = set() + if rooms: + # Separate rooms into lazy and non-lazy based on sync config. + lazy_rooms = ( + { + room_id + for room_id, room_config in sync_config.room_subscriptions.items() + if ("m.room.member", "$LAZY") in room_config.required_state + } + if sync_config.room_subscriptions + else set() + ) + + if lazy_rooms: + # For rooms configured as lazy, include users based on timeline events. + for room_id, room_data in actual_room_response_map.items(): + if room_id not in lazy_rooms: + continue + for timeline_event in room_data.timeline_events: + users_in_timeline.add(timeline_event.event.sender) + + non_lazy_rooms = rooms.difference(lazy_rooms) + # If we still have non-lazy rooms, get their members. + if non_lazy_rooms: + non_lazy_profile_user_ids = ( + await self.store.get_local_users_who_share_room_with_user( + user_id, + limit_to_rooms=non_lazy_rooms, + ) + ) + else: + # Get all members of all rooms the sync response is dealing with. + non_lazy_profile_user_ids = ( + await self.store.get_local_users_who_share_room_with_user( + user_id, + ) + ) + + # Unify the two lists + profile_user_ids = users_in_timeline.union(non_lazy_profile_user_ids) + + # Return a tuple containing the full list of user ID's and the lazy subset. + return ( + profile_user_ids, + users_in_timeline, + ) + async def _get_profiles_extension_initial_sync_response( self, user_id: UserID, fields: set[str], rooms: set[str], new_connection_state: MutablePerConnectionState, + profile_user_ids: set[str], ) -> dict[str, JsonDict | None]: """ Build an initial sync response for the profiles extension. @@ -1097,18 +1176,15 @@ class SlidingSyncExtensionHandler: fields: A set of fields to include in the response. rooms: A set of rooms to limit the user profiles for. new_connection_state: The new connection state to be modified. + profile_user_ids: Set of user profile ID's related to this sync response. Returns: A dictionary containing the profile updates in an `updated` dictionary. """ response: dict[str, JsonDict | None] = {} - profile_user_ids = await self.store.get_local_users_who_share_room_with_user( - user_id.to_string(), - limit_to_rooms=rooms, - ) # Ensure we're in the list even if we don't belong to any rooms - profile_user_ids.add(user_id.to_string()) + profile_user_ids = profile_user_ids.union({user_id.to_string()}) profile_data_by_user = await self.store.get_profile_data_for_users( profile_user_ids @@ -1151,6 +1227,7 @@ class SlidingSyncExtensionHandler: all_interested_room_ids: set[str], to_token: StreamToken, from_token: SlidingSyncStreamToken | None, + actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], ) -> SlidingSyncResult.Extensions.ProfilesExtension | None: """ Generate a response for the profiles extension. @@ -1163,6 +1240,7 @@ class SlidingSyncExtensionHandler: all_interested_room_ids: Set of rooms the sync request is interested in. to_token: The stream token to generate a response until. from_token: The stream token to generate a response from. + actual_room_response_map: A calculated map of responses per room. Returns: A SlidingSyncResult.Extensions.ProfilesExtension object containing @@ -1176,6 +1254,16 @@ class SlidingSyncExtensionHandler: response: dict[str, JsonDict | None] = {} + ( + profile_user_ids, + lazy_profile_user_ids, + ) = await self._get_profile_ids_for_profiles_extension( + user_id=user_id, + rooms=all_interested_room_ids, + sync_config=sync_config, + actual_room_response_map=actual_room_response_map, + ) + if from_token is None: # Initial sync return SlidingSyncResult.Extensions.ProfilesExtension( @@ -1184,6 +1272,7 @@ class SlidingSyncExtensionHandler: fields=fields, rooms=all_interested_room_ids, new_connection_state=new_connection_state, + profile_user_ids=profile_user_ids, ), ) @@ -1195,7 +1284,6 @@ class SlidingSyncExtensionHandler: field_names=fields, field_names_empty_means_all_fields=True, ) - profile_user_ids = set() left_room_user_ids = { update.user_id for update in updates @@ -1235,9 +1323,6 @@ class SlidingSyncExtensionHandler: profile_user_ids ) - # TODO lazy loading - is_lazy = False - # Serialise the profile updates into the sync response format. for profile_user_id in profile_user_ids: profile_data = profile_data_by_user.get(profile_user_id) @@ -1254,36 +1339,31 @@ class SlidingSyncExtensionHandler: user_fields = set(profile_data.keys()).intersection(fields) else: user_fields = set(profile_data.keys()) - if is_lazy: - # TODO lazy cache - for field_name in user_fields: - per_user_updates[field_name] = profile_data.get(field_name) - else: - # Include only the diff, unless the user recently joined, - # then send all the fields the client asked for. - # We don't use a cache here as for non-lazy sync we always - # send changes and/or fields the client asked for, if relevant - # as above joined condition. - updated_fields: set[str] = updated_user_fields.get( - profile_user_id, set() - ) - user_fields = ( - user_fields - if profile_user_id in joined_room_user_ids - else updated_fields.intersection(profile_data.keys()) - ) - for field_name in user_fields: - # Ensure we don't send the field unnecessarely to the client, if - # we've sent it down in this connection before, and it hasn't been - # updated. - if ( - field_name in updated_fields - or previous_connection_state.profile_updates.have_sent_field( - profile_user_id, field_name - ).status - != HaveSentFlag.LIVE - ): - per_user_updates[field_name] = profile_data[field_name] + + # TODO lazy cache + + # Include only the diff, unless the user recently joined, + # or the user is in a room that was lazy loaded, + # then send all the fields the client asked for. + updated_fields: set[str] = updated_user_fields.get(profile_user_id, set()) + user_fields = ( + user_fields + if profile_user_id in joined_room_user_ids + or profile_user_id in lazy_profile_user_ids + else updated_fields.intersection(profile_data.keys()) + ) + for field_name in user_fields: + # Ensure we don't send the field unnecessarely to the client, if + # we've sent it down in this connection before, and it hasn't been + # updated. + if ( + field_name in updated_fields + or previous_connection_state.profile_updates.have_sent_field( + profile_user_id, field_name + ).status + != HaveSentFlag.LIVE + ): + per_user_updates[field_name] = profile_data[field_name] if per_user_updates: # Record sending these fields to this connection and add to the response diff --git a/tests/rest/client/sliding_sync/test_extension_profiles.py b/tests/rest/client/sliding_sync/test_extension_profiles.py index 58871dfb3b..306b40a590 100644 --- a/tests/rest/client/sliding_sync/test_extension_profiles.py +++ b/tests/rest/client/sliding_sync/test_extension_profiles.py @@ -705,16 +705,220 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): > updates will then include changes to this field. """ - """ - Sliding Sync offers the lazy_members boolean flag on a per-room basis, which when true - will only send down membership information for a user if: - """ + @parameterized.expand( + [ + True, + False, + ] + ) + @override_config({"include_profile_updates_in_sync": True}) + def test_lazy_loading_sends_down_full_profile_if_events_in_timeline( + self, + is_initial: bool, + ) -> None: + """ + Test that when lazy loading, only those members who have events in + the timeline get their profiles sent down in the sync response, for + rooms configured with lazy loading. + + Rooms without lazy loading should include all the members in initial sync, + none in incremental. + """ + third_user = self.register_user("third_user", "password") + third_tok = self.login("third_user", "password") + fourth_user = self.register_user("fourth_user", "password") + fourth_tok = self.login("fourth_user", "password") + self.helper.join( + room=self.joined_room, + user=third_user, + tok=third_tok, + ) + new_room = self.helper.create_room_as(self.user, tok=self.tok) + self.helper.join( + room=new_room, + user=fourth_user, + tok=fourth_tok, + ) + if is_initial: + self.helper.send_messages( + room_id=self.joined_room, num_events=1, tok=self.other_tok + ) + self.helper.send_messages( + room_id=self.joined_room, num_events=10, tok=third_tok + ) + # Make an initial Sliding Sync request with the profiles extension enabled + sync_body: dict[str, dict] = { + "lists": {}, + "room_subscriptions": { + self.joined_room: { + "required_state": [], + "timeline_limit": 10, + }, + new_room: { + "required_state": [], + "timeline_limit": 10, + }, + }, + "extensions": { + "org.matrix.msc4262.profiles": { + "enabled": True, + }, + }, + } + if is_initial: + sync_body["room_subscriptions"][self.joined_room]["required_state"] = [ + ["m.room.member", "$LAZY"], + ["*", "*"], + ] + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + if is_initial: + self.assertIsNotNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) + # Other user should be filtered out. + self.assertIsNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@other_user:test" + ) + ) + # Third user has events in the timeline, so should be here. + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@third_user:test" + ) + ) + # Initial sync always includes ourselves + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@user:test" + ) + ) + # Fourth user is a member of a non-lazy configured room, so should be here. + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@fourth_user:test" + ) + ) + + if not is_initial: + # Clear up the sliding sync connection profile updates tracking rows + # as otherwise we won't re-send these unchanged fields in this connection. + self.get_success( + self.store.db_pool.simple_delete_many( + "sliding_sync_connection_profile_updates", + column="user_id", + iterable=[ + "@user:test", + "@other_user:test", + "@third_user:test", + "@fourth_user:test", + ], + keyvalues={}, + desc="clear_old_sliding_sync_connection_profile_updates", + ) + ) + + self.helper.send_messages( + room_id=self.joined_room, num_events=1, tok=self.other_tok + ) + self.helper.send_messages( + room_id=self.joined_room, num_events=10, tok=third_tok + ) + sync_body["room_subscriptions"][self.joined_room]["required_state"] = [ + ["m.room.member", "$LAZY"], + ["*", "*"], + ] + # Make an incremental Sliding Sync request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) + self.assertIsNotNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) + # Other user should be filtered out. + self.assertIsNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@other_user:test" + ) + ) + # Third user has events in the timeline, so should be here. + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@third_user:test" + ) + ) + # We are not included ourselves in incremental sync without updates. + self.assertIsNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@user:test" + ) + ) + # Fourth user is a member of a non-lazy configured room, but had no updates, + # so shouldn't be here. + self.assertIsNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@fourth_user:test" + ) + ) @override_config({"include_profile_updates_in_sync": True}) - def test_lazy_loading_sends_down_full_profile_if_events_in_timeline(self) -> None: + def test_lazy_loading_sends_full_profile_even_if_no_events_if_otherwise_included( + self, + ) -> None: """ - > the user is one of the senders of a timeline event included in the response + Test that when lazy loading, if a user is in both a lazy loading room + and a non-lazy configured room, even if there are no events in the timeline, + their profile is sent down. + + This test only makes sense for initial sync, as for incremental we would + not expect to see users without timeline events if they had no profile updates. """ + new_room = self.helper.create_room_as(self.user, tok=self.tok) + self.helper.join( + room=new_room, + user=self.other_user, + tok=self.other_tok, + ) + # Make an initial Sliding Sync request with the profiles extension enabled + sync_body: dict[str, dict] = { + "lists": {}, + "room_subscriptions": { + self.joined_room: { + "required_state": [ + ["m.room.member", "$LAZY"], + ["*", "*"], + ], + # Force zero timeline events in the response, otherwise + # this test wont work, as the timeline_events in the room + # response will contain all the create/join etc events too. + "timeline_limit": 0, + }, + new_room: { + "required_state": [], + "timeline_limit": 10, + }, + }, + "extensions": { + "org.matrix.msc4262.profiles": { + "enabled": True, + }, + }, + } + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + self.assertIsNotNone( + response_body["extensions"].get("org.matrix.msc4262.profiles") + ) + # Other user should be included as they are in a non-lazy room too, + # even though the lazy configured room had no events. + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@other_user:test" + ) + ) + # Initial sync always includes ourselves + self.assertIsNotNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"].get( + "@user:test" + ) + ) @override_config({"include_profile_updates_in_sync": True}) def test_lazy_loading_sends_down_full_profile_if_membership_events_that_are_returned(