From 8eb7bb536deab3f3dfa8662bc19fbde69b7c201a Mon Sep 17 00:00:00 2001 From: Jason Robinson Date: Thu, 23 Jul 2026 11:41:05 +0300 Subject: [PATCH] Add per connection state tracking for profile updates for sliding sync Also make `test_tracking_of_sent_fields_per_sliding_sync_connection` failing now. --- .../storage/databases/main/sliding_sync.py | 100 +++++++++++++++++ .../94/08_sliding_sync_profile_updates.sql | 26 +++++ synapse/types/handlers/sliding_sync.py | 105 +++++++++++++++++- .../sliding_sync/test_extension_profiles.py | 45 ++++++++ 4 files changed, 275 insertions(+), 1 deletion(-) create mode 100644 synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index a5d6cd2548..0fc3f7a436 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/synapse/storage/databases/main/sliding_sync.py @@ -35,6 +35,7 @@ from synapse.types.handlers.sliding_sync import ( HaveSentRoomFlag, MutablePerConnectionState, PerConnectionState, + ProfileFieldStatusMap, RoomLazyMembershipChanges, RoomStatusMap, RoomSyncConfig, @@ -350,6 +351,15 @@ class SlidingSyncStore(SQLBaseStore): """ txn.execute(sql, (connection_position, previous_connection_position)) + sql = """ + INSERT INTO sliding_sync_connection_profile_updates + (connection_position, user_id, field_name, field_status, last_token) + SELECT ?, user_id, field_name, field_status, last_token + FROM sliding_sync_connection_profile_updates + WHERE connection_position = ? + """ + txn.execute(sql, (connection_position, previous_connection_position)) + # We now upsert the changes to the various streams. key_values = [] value_values = [] @@ -390,6 +400,38 @@ class SlidingSyncStore(SQLBaseStore): value_values=value_values, ) + # Handle profile updates - nested structure: user_id -> field_name -> status + profile_update_key_values = [] + profile_update_value_values = [] + for ( + user_id, + field_statuses, + ) in per_connection_state.profile_updates._statuses.items(): + for field_name, have_sent_field in field_statuses.items(): + profile_update_key_values.append( + (connection_position, user_id, field_name) + ) + profile_update_value_values.append( + (have_sent_field.status.value, have_sent_field.last_token) + ) + + if profile_update_key_values: + self.db_pool.simple_upsert_many_txn( + txn, + table="sliding_sync_connection_profile_updates", + key_names=( + "connection_position", + "user_id", + "field_name", + ), + key_values=profile_update_key_values, + value_names=( + "field_status", + "last_token", + ), + value_values=profile_update_value_values, + ) + # ... and upsert changes to the room configs. keys = [] values = [] @@ -623,11 +665,34 @@ class SlidingSyncStore(SQLBaseStore): # future we want to be able to easily add more stream types. logger.warning("Unrecognized sliding sync stream in DB %r", stream) + # Now look up the per-profile field stream data. + profile_updates: dict[str, dict[str, HaveSentRoom[str]]] = {} + + profile_update_rows = self.db_pool.simple_select_list_txn( + txn, + table="sliding_sync_connection_profile_updates", + keyvalues={"connection_position": connection_position}, + retcols=( + "user_id", + "field_name", + "field_status", + "last_token", + ), + ) + for user_id, field_name, field_status, last_token in profile_update_rows: + have_sent_field: HaveSentRoom[str] = HaveSentRoom( + status=HaveSentRoomFlag(field_status), last_token=last_token + ) + if user_id not in profile_updates: + profile_updates[user_id] = {} + profile_updates[user_id][field_name] = have_sent_field + return PerConnectionStateDB( last_used_ts=last_used_ts, rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + profile_updates=ProfileFieldStatusMap(profile_updates), room_configs=room_configs, room_lazy_membership={}, ) @@ -810,6 +875,7 @@ class PerConnectionStateDB: rooms: "RoomStatusMap[str]" receipts: "RoomStatusMap[str]" account_data: "RoomStatusMap[str]" + profile_updates: "ProfileFieldStatusMap[str]" room_configs: Mapping[str, "RoomSyncConfig"] @@ -856,11 +922,29 @@ class PerConnectionStateDB: for room_id, status in per_connection_state.account_data.get_updates().items() } + profile_updates: dict[str, dict[str, HaveSentRoom[str]]] = {} + for ( + user_id, + field_statuses, + ) in per_connection_state.profile_updates.get_updates().items(): + profile_updates[user_id] = { + field_name: HaveSentRoom( + status=status.status, + last_token=( + await status.last_token.to_string(store) + if status.last_token is not None + else None + ), + ) + for field_name, status in field_statuses.items() + } + log_kv( { "rooms": rooms, "receipts": receipts, "account_data": account_data, + "profile_updates": profile_updates, "room_configs": per_connection_state.room_configs.maps[0], } ) @@ -870,6 +954,7 @@ class PerConnectionStateDB: rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + profile_updates=ProfileFieldStatusMap(profile_updates), room_configs=per_connection_state.room_configs.maps[0], room_lazy_membership=per_connection_state.room_lazy_membership, ) @@ -910,10 +995,25 @@ class PerConnectionStateDB: for room_id, status in self.account_data._statuses.items() } + profile_updates: dict[str, dict[str, HaveSentRoom[MultiWriterStreamToken]]] = {} + for user_id, field_statuses in self.profile_updates._statuses.items(): + profile_updates[user_id] = { + field_name: HaveSentRoom( + status=status.status, + last_token=( + await MultiWriterStreamToken.parse(store, status.last_token) + if status.last_token is not None + else None + ), + ) + for field_name, status in field_statuses.items() + } + return PerConnectionState( last_used_ts=self.last_used_ts, rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + profile_updates=ProfileFieldStatusMap(profile_updates), room_configs=self.room_configs, ) diff --git a/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql b/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql new file mode 100644 index 0000000000..124ff9e463 --- /dev/null +++ b/synapse/storage/schema/main/delta/94/08_sliding_sync_profile_updates.sql @@ -0,0 +1,26 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations Ltd. +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + +-- Stores what profile field data we have sent for given users down given connections. +-- +-- Similar to `sliding_sync_connection_streams`, but for user profile fields rather than rooms. +-- This tracks which profile fields we've sent for which users on a given connection position. +CREATE TABLE sliding_sync_connection_profile_updates( + connection_position BIGINT NOT NULL REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE, + user_id TEXT NOT NULL, + field_name TEXT NOT NULL, + field_status TEXT NOT NULL, -- "live" or "previously", i.e. the `HaveSentRoomFlag` value + last_token TEXT -- For "previously" the token for the stream we have sent up to. +); + +CREATE UNIQUE INDEX sliding_sync_connection_profile_updates_idx ON sliding_sync_connection_profile_updates(connection_position, user_id, field_name); \ No newline at end of file diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index d27793d9a7..0c62f9a4c7 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -911,6 +911,94 @@ class MutableRoomStatusMap(RoomStatusMap[T]): self._statuses[room_id] = HaveSentRoom.previously(from_token) +@attr.s(auto_attribs=True, slots=True, frozen=True) +class ProfileFieldStatusMap(Generic[T]): + """For a given profile field, records what we have or have not sent + down for that field in a given user profile.""" + + # `user_id` -> `field_name` -> `HaveSentRoom` + _statuses: Mapping[str, Mapping[str, HaveSentRoom[T]]] = attr.Factory(dict) + + def have_sent_field(self, user_id: str, field_name: str) -> HaveSentRoom[T]: + """Return whether we have previously sent the field for this user""" + return self._statuses.get(user_id, {}).get(field_name, HaveSentRoom.never()) + + def get_mutable(self) -> "MutableProfileFieldStatusMap[T]": + """Get a mutable copy of this state.""" + return MutableProfileFieldStatusMap( + statuses=self._statuses, + ) + + def copy(self) -> "ProfileFieldStatusMap[T]": + """Make a copy of the class. Useful for converting from a mutable to + immutable version.""" + return ProfileFieldStatusMap(statuses=dict(self._statuses)) + + def __len__(self) -> int: + return len(self._statuses) + + +class MutableProfileFieldStatusMap(ProfileFieldStatusMap[T]): + """A mutable version of `ProfileFieldStatusMap`""" + + # We use a ChainMap here so that we can easily track what has been updated + # and what hasn't. Note that when we persist the per connection state this + # will get flattened to a normal dict (via calling `.copy()`) + _statuses: ChainMap[str, Mapping[str, HaveSentRoom[T]]] + + def __init__( + self, + statuses: Mapping[str, Mapping[str, HaveSentRoom[T]]], + ) -> None: + # ChainMap requires a mutable mapping, but we're not actually going to + # mutate it. + statuses_mutable = cast(MutableMapping, statuses) + + super().__init__( + statuses=ChainMap({}, statuses_mutable), + ) + + def get_updates(self) -> Mapping[str, Mapping[str, HaveSentRoom[T]]]: + """Return only the changes that were made""" + return self._statuses.maps[0] + + def record_sent_fields(self, user_id: str, field_names: list[str]) -> None: + """Record that we have sent these fields for a user in the response""" + if user_id not in self._statuses: + self._statuses[user_id] = {} + + user_fields = cast( + MutableMapping[str, HaveSentRoom[T]], self._statuses[user_id] + ) + + for field_name in field_names: + current_status = user_fields.get(field_name, HaveSentRoom.never()) + if current_status.status == HaveSentRoomFlag.LIVE: + continue + + user_fields[field_name] = HaveSentRoom.live() + + def record_unsent_fields( + self, user_id: str, field_names: list[str], from_token: T + ) -> None: + """Record that we have not sent these fields for a user in the response, but there + have been updates. + """ + if user_id not in self._statuses: + return + + user_fields = cast( + MutableMapping[str, HaveSentRoom[T]], self._statuses[user_id] + ) + + for field_name in field_names: + current_status = user_fields.get(field_name, HaveSentRoom.never()) + if current_status.status != HaveSentRoomFlag.LIVE: + continue + + user_fields[field_name] = HaveSentRoom.previously(from_token) + + @attr.s(auto_attribs=True, frozen=True) class PerConnectionState: """The per-connection state. A snapshot of what we've sent down the @@ -944,6 +1032,10 @@ class PerConnectionState: room_configs: Mapping[str, RoomSyncConfig] = attr.Factory(dict) + profile_updates: ProfileFieldStatusMap[MultiWriterStreamToken] = attr.Factory( + ProfileFieldStatusMap + ) + def get_mutable(self) -> "MutablePerConnectionState": """Get a mutable copy of this state.""" room_configs = cast(MutableMapping[str, RoomSyncConfig], self.room_configs) @@ -954,6 +1046,7 @@ class PerConnectionState: receipts=self.receipts.get_mutable(), account_data=self.account_data.get_mutable(), room_configs=ChainMap({}, room_configs), + profile_updates=self.profile_updates.get_mutable(), ) def copy(self) -> "PerConnectionState": @@ -963,10 +1056,17 @@ class PerConnectionState: receipts=self.receipts.copy(), account_data=self.account_data.copy(), room_configs=dict(self.room_configs), + profile_updates=self.profile_updates.copy(), ) def __len__(self) -> int: - return len(self.rooms) + len(self.receipts) + len(self.room_configs) + # FIXME: this is missing self.account_data + return ( + len(self.rooms) + + len(self.receipts) + + len(self.room_configs) + + len(self.profile_updates) + ) @attr.s(auto_attribs=True) @@ -1044,6 +1144,8 @@ class MutablePerConnectionState(PerConnectionState): room_configs: ChainMap[str, RoomSyncConfig] + profile_updates: MutableProfileFieldStatusMap[MultiWriterStreamToken] + # A map from room ID to the lazily-loaded memberships needed for the # request in that room. room_lazy_membership: dict[str, RoomLazyMembershipChanges] = attr.Factory(dict) @@ -1066,6 +1168,7 @@ class MutablePerConnectionState(PerConnectionState): change.has_updates(clock) for change in self.room_lazy_membership.values() ) + or bool(self.profile_updates.get_updates()) ) def get_room_config_updates(self) -> Mapping[str, RoomSyncConfig]: diff --git a/tests/rest/client/sliding_sync/test_extension_profiles.py b/tests/rest/client/sliding_sync/test_extension_profiles.py index 1be9987efd..f028ad948e 100644 --- a/tests/rest/client/sliding_sync/test_extension_profiles.py +++ b/tests/rest/client/sliding_sync/test_extension_profiles.py @@ -539,6 +539,51 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): > enter the room subset, users who were already in rooms within the room subset > will not have their full profiles sent down a second time. """ + new_room = self.helper.create_room_as(self.user, tok=self.tok) + # Make an initial Sliding Sync request with the profiles extension enabled + profiles_config: dict = { + "enabled": True, + } + sync_body = { + "lists": {}, + "room_subscriptions": { + self.joined_room: { + "required_state": [], + "timeline_limit": 10, + }, + }, + "extensions": { + "org.matrix.msc4262.profiles": profiles_config, + }, + } + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + + # Starting situation, we get the full profile + self.assertEqual( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][ + "@other_user:test" + ], + { + "updated": { + "avatar_url": None, + "displayname": "other_user", + }, + }, + ) + + # Join other user to the new room and sync now asking for both + self.helper.join(new_room, self.other_user, tok=self.other_tok) + sync_body["room_subscriptions"][new_room] = { + "required_state": [], + "timeline_limit": 10, + } + + # Make an incremental Sliding Sync request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) + # We should not get other user re-sent + self.assertIsNone( + response_body["extensions"].get("org.matrix.msc4262.profiles"), + ) @override_config({"include_profile_updates_in_sync": True}) def test_removed_fields_get_sent_down_as_removed(self) -> None: