WIP

synapse/handlers/sliding_sync/__init__.py

@@ -983,14 +983,10 @@ class SlidingSyncHandler:
         #
         # Calculate the `StateFilter` based on the `required_state` for the room
         required_state_filter = StateFilter.none()
-        # The requested `required_state_map` with the lazy membership expanded and
-        # `$ME` replaced with the user's ID. This allows us to see what membership we've
-        # sent down to the client in the next request.
-        #
-        # Make a copy so we can modify it. Still need to be careful to make a copy of
-        # the state key sets if we want to add/remove from them. We could make a deep
-        # copy but this saves us some work.
-        expanded_required_state_map = dict(room_sync_config.required_state_map)
+
+        # Keep track of which users' state we may need to fetch if lazy-loaded
+        # members is used.
+        required_user_state = set()
         if room_membership_for_user_at_to_token.membership not in (
             Membership.INVITE,
             Membership.KNOCK,
@@ -1082,12 +1078,7 @@ class SlidingSyncHandler:
                             # Make a new set or copy of the state key set so we can
                             # modify it without affecting the original
                             # `required_state_map`
-                            expanded_required_state_map[EventTypes.Member] = (
-                                expanded_required_state_map.get(
-                                    EventTypes.Member, set()
-                                )
-                                | timeline_membership
-                            )
+                            required_user_state.update(timeline_membership)
                         elif state_key == StateValues.ME:
                             num_others += 1
                             required_state_types.append((state_type, user.to_string()))
@@ -1098,12 +1089,7 @@ class SlidingSyncHandler:
                             # Make a new set or copy of the state key set so we can
                             # modify it without affecting the original
                             # `required_state_map`
-                            expanded_required_state_map[EventTypes.Member] = (
-                                expanded_required_state_map.get(
-                                    EventTypes.Member, set()
-                                )
-                                | {user.to_string()}
-                            )
+                            required_user_state.add(user.to_string())
                         else:
                             num_others += 1
                             required_state_types.append((state_type, state_key))
@@ -1151,6 +1137,11 @@ class SlidingSyncHandler:
         # We can return all of the state that was requested if this was the first
         # time we've sent the room down this connection.
         room_state: StateMap[EventBase] = {}
+
+        new_connection_state.room_lazy_membership[room_id] = dict.fromkeys(
+            required_user_state
+        )
+
         if initial:
             room_state = await self.get_current_state_at(
                 room_id=room_id,
@@ -1159,16 +1150,34 @@ class SlidingSyncHandler:
                 to_token=to_token,
             )
         else:
+            assert from_token is not None
             assert from_bound is not None
 
             if prev_room_sync_config is not None:
+                # Fetch which of the needed lazy-loaded members we have already sent.
+                if required_user_state:
+                    previously_returned_user_state = (
+                        await self.store.get_sliding_sync_connection_lazy_members(
+                            connection_position=from_token.connection_position,
+                            room_id=room_id,
+                            user_ids=required_user_state,
+                        )
+                    )
+                    new_connection_state.room_lazy_membership[room_id].update(
+                        previously_returned_user_state
+                    )
+                else:
+                    previously_returned_user_state = {}
+
                 # Check if there are any changes to the required state config
                 # that we need to handle.
                 changed_required_state_map, added_state_filter = (
                     _required_state_changes(
                         user.to_string(),
                         prev_required_state_map=prev_room_sync_config.required_state_map,
-                        request_required_state_map=expanded_required_state_map,
+                        request_required_state_map=room_sync_config.required_state_map,
+                        previously_returned_user_state=previously_returned_user_state.keys(),
+                        required_user_state=required_user_state,
                         state_deltas=room_state_delta_id_map,
                     )
                 )
@@ -1283,7 +1292,7 @@ class SlidingSyncHandler:
             bump_stamp = 0
 
         room_sync_required_state_map_to_persist: Mapping[str, AbstractSet[str]] = (
-            expanded_required_state_map
+            room_sync_config.required_state_map
         )
         if changed_required_state_map:
             room_sync_required_state_map_to_persist = changed_required_state_map
@@ -1478,6 +1487,8 @@ def _required_state_changes(
     *,
     prev_required_state_map: Mapping[str, AbstractSet[str]],
     request_required_state_map: Mapping[str, AbstractSet[str]],
+    previously_returned_user_state: AbstractSet[str],
+    required_user_state: AbstractSet[str],
     state_deltas: StateMap[str],
 ) -> tuple[Mapping[str, AbstractSet[str]] | None, StateFilter]:
     """Calculates the changes between the required state room config from the
@@ -1499,7 +1510,15 @@ def _required_state_changes(
         return.
     """
     if prev_required_state_map == request_required_state_map:
-        # There has been no change. Return immediately.
+        # There has been no change in state, just need to check lazy members.
+        if required_user_state - previously_returned_user_state:
+            # There are some new lazy members we need to fetch.
+            added_types: list[tuple[str, str | None]] = []
+            for user_id in required_user_state - previously_returned_user_state:
+                added_types.append((EventTypes.Member, user_id))
+
+            return None, StateFilter.from_types(added_types)
+
         return None, StateFilter.none()
 
     prev_wildcard = prev_required_state_map.get(StateValues.WILDCARD, set())
@@ -1622,9 +1641,17 @@ def _required_state_changes(
                     # LAZY values should also be ignore for event types that are
                     # not membership.
                     pass
+                elif event_type == EventTypes.Member:
+                    if state_key not in previously_returned_user_state:
+                        # Only add *explicit* members we haven't previously sent
+                        # down.
+                        added.append((event_type, state_key))
                 else:
                     added.append((event_type, state_key))
 
+    for user_id in required_user_state - previously_returned_user_state:
+        added.append((EventTypes.Member, user_id))
+
     added_state_filter = StateFilter.from_types(added)
 
     # Figure out what changes we need to apply to the effective required state

synapse/storage/databases/main/sliding_sync.py

@@ -14,7 +14,7 @@
 
 
 import logging
-from typing import TYPE_CHECKING, Mapping, cast
+from typing import TYPE_CHECKING, AbstractSet, Mapping, cast
 
 import attr
 
@@ -26,6 +26,7 @@ from synapse.storage.database import (
     LoggingDatabaseConnection,
     LoggingTransaction,
 )
+from synapse.storage.engines import PostgresEngine
 from synapse.types import MultiWriterStreamToken, RoomStreamToken
 from synapse.types.handlers.sliding_sync import (
     HaveSentRoom,
@@ -349,6 +350,16 @@ class SlidingSyncStore(SQLBaseStore):
             value_values=values,
         )
 
+        for room_id, user_ids in per_connection_state.room_lazy_membership.items():
+            self._persist_sliding_sync_connection_lazy_members_txn(
+                txn,
+                connection_key,
+                previous_connection_position,
+                connection_position,
+                room_id,
+                user_ids,
+            )
+
         return connection_position
 
     @cached(iterable=True, max_entries=100000)
@@ -406,6 +417,19 @@ class SlidingSyncStore(SQLBaseStore):
         """
         txn.execute(sql, (connection_key, connection_position))
 
+        # Move any lazy membership entries for this connection position to have
+        # `NULL` connection position, indicating that it applies to all future
+        # positions on this connecetion.
+        self.db_pool.simple_update_txn(
+            txn,
+            table="sliding_sync_connection_lazy_members",
+            keyvalues={
+                "connection_key": connection_key,
+                "connection_position": connection_position,
+            },
+            updatevalues={"connection_position": None},
+        )
+
         # Fetch and create a mapping from required state ID to the actual
         # required state for the connection.
         rows = self.db_pool.simple_select_list_txn(
@@ -484,8 +508,100 @@ class SlidingSyncStore(SQLBaseStore):
             receipts=RoomStatusMap(receipts),
             account_data=RoomStatusMap(account_data),
             room_configs=room_configs,
+            room_lazy_membership={},
         )
 
+    async def get_sliding_sync_connection_lazy_members(
+        self,
+        connection_position: int,
+        room_id: str,
+        user_ids: AbstractSet[str],
+    ) -> Mapping[str, int]:
+        """Get which user IDs in the room we have previously sent lazy
+        membership for.
+
+        Args:
+            connection_position: The sliding sync connection position.
+            room_id: The room ID to get lazy members for.
+            user_ids: The user IDs to check for lazy membership.
+
+        Returns:
+            The mapping of user IDs to the last seen timestamp for those user
+            IDs.
+        """
+
+        def get_sliding_sync_connection_lazy_members_txn(
+            txn: LoggingTransaction,
+        ) -> Mapping[str, int]:
+            sql = """
+                SELECT user_id, mem.connection_position, last_seen_ts
+                FROM sliding_sync_connection_positions AS pos
+                INNER JOIN sliding_sync_connection_lazy_members AS mem USING (connection_key)
+                WHERE pos.connection_position = ?
+            """
+
+            txn.execute(sql, (connection_position,))
+            return {
+                user_id: last_seen_ts
+                for user_id, db_connection_position, last_seen_ts in txn
+                if db_connection_position == connection_position
+                or db_connection_position is None
+            }
+
+        return await self.db_pool.runInteraction(
+            "sliding_sync_connection_lazy_members",
+            get_sliding_sync_connection_lazy_members_txn,
+        )
+
+    def _persist_sliding_sync_connection_lazy_members_txn(
+        self,
+        txn: LoggingTransaction,
+        connection_key: int,
+        previous_connection_position: int | None,
+        new_connection_position: int,
+        room_id: str,
+        user_ids: Mapping[str, int | None],
+    ) -> None:
+        """Persist that we have sent lazy membership for the given user IDs."""
+
+        now = self.clock.time_msec()
+
+        to_update: list[str] = []
+        for user_id, last_seen_ts in user_ids.items():
+            if last_seen_ts is None:
+                # We've never sent this user before, so we need to record that
+                # we've sent it at the new connection position.
+                to_update.append(user_id)
+            elif last_seen_ts + 60 * 60 * 1000 < now:
+                # We last saw this user over an hour ago, so we update the
+                # timestamp.
+                to_update.append(user_id)
+
+        if not to_update:
+            return
+
+        sql = """
+            INSERT INTO sliding_sync_connection_lazy_members
+            (connection_key, connection_position, room_id, user_id, last_seen_ts)
+            VALUES {value_placeholder}
+            ON CONFLICT (connection_key, room_id, user_id)
+            DO UPDATE SET last_seen_ts = EXCLUDED.last_seen_ts
+            WHERE sliding_sync_connection_lazy_members.connection_position IS NULL
+                OR sliding_sync_connection_lazy_members.connection_position = EXCLUDED.connection_position
+        """
+
+        args = [
+            (connection_key, new_connection_position, room_id, user_id, now)
+            for user_id in to_update
+        ]
+
+        if isinstance(self.database_engine, PostgresEngine):
+            sql = sql.format(value_placeholder="?")
+            txn.execute_values(sql, args, fetch=False)
+        else:
+            sql = sql.format(value_placeholder="(?, ?, ?, ?, ?)")
+            txn.execute_batch(sql, args)
+
 
 @attr.s(auto_attribs=True, frozen=True)
 class PerConnectionStateDB:
@@ -496,6 +612,8 @@ class PerConnectionStateDB:
     serialized to strings.
 
     When persisting this *only* contains updates to the state.
+
+    The `room_lazy_membership` field is only used when persisting.
     """
 
     rooms: "RoomStatusMap[str]"
@@ -504,6 +622,8 @@ class PerConnectionStateDB:
 
     room_configs: Mapping[str, "RoomSyncConfig"]
 
+    room_lazy_membership: dict[str, dict[str, int | None]]
+
     @staticmethod
     async def from_state(
         per_connection_state: "MutablePerConnectionState", store: "DataStore"
@@ -557,6 +677,7 @@ class PerConnectionStateDB:
             receipts=RoomStatusMap(receipts),
             account_data=RoomStatusMap(account_data),
             room_configs=per_connection_state.room_configs.maps[0],
+            room_lazy_membership=per_connection_state.room_lazy_membership,
         )
 
     async def to_state(self, store: "DataStore") -> "PerConnectionState":

synapse/storage/schema/main/delta/92/10_sliding_sync_members.sql

@@ -0,0 +1,24 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2025 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>.
+
+
+CREATE TABLE sliding_sync_connection_lazy_members (
+    connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
+    connection_position BIGINT REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE,
+    room_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    last_seen_ts BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX sliding_sync_connection_lazy_members_idx ON sliding_sync_connection_lazy_members (connection_key, room_id, user_id);
+CREATE INDEX sliding_sync_connection_lazy_members_pos_idx ON sliding_sync_connection_lazy_members (connection_key, connection_position) WHERE connection_position IS NOT NULL;

synapse/types/handlers/sliding_sync.py

@@ -895,12 +895,19 @@ class MutablePerConnectionState(PerConnectionState):
 
     room_configs: typing.ChainMap[str, RoomSyncConfig]
 
+    # A map from room ID -> user ID -> timestamp. Indicates that those
+    # memberships have been lazily loaded. I.e. that either a) we sent those
+    # memberships down, or b) we did so previously. The timestamp indicates the
+    # time we previously saw the membership.
+    room_lazy_membership: dict[str, dict[str, int | None]] = attr.Factory(dict)
+
     def has_updates(self) -> bool:
         return (
             bool(self.rooms.get_updates())
             or bool(self.receipts.get_updates())
             or bool(self.account_data.get_updates())
             or bool(self.get_room_config_updates())
+            or bool(self.room_lazy_membership)
         )
 
     def get_room_config_updates(self) -> Mapping[str, RoomSyncConfig]: