Add per connection state tracking for profile updates for sliding sync

Also make `test_tracking_of_sent_fields_per_sliding_sync_connection` failing now.

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,
         )

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+-- 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);

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]:

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: