From 82f593c6b15bed455d9a0a0276e707f1ee9aa36d Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Sat, 15 Nov 2025 10:56:10 +0000 Subject: [PATCH] WIP --- synapse/handlers/sliding_sync/__init__.py | 73 +++++++---- .../storage/databases/main/sliding_sync.py | 123 +++++++++++++++++- .../main/delta/92/10_sliding_sync_members.sql | 24 ++++ synapse/types/handlers/sliding_sync.py | 7 + 4 files changed, 203 insertions(+), 24 deletions(-) create mode 100644 synapse/storage/schema/main/delta/92/10_sliding_sync_members.sql diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 6a5d5c7b3c..9ea5d82951 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/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 diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index 2b67e75ac4..3d766efc41 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/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": diff --git a/synapse/storage/schema/main/delta/92/10_sliding_sync_members.sql b/synapse/storage/schema/main/delta/92/10_sliding_sync_members.sql new file mode 100644 index 0000000000..b949f31869 --- /dev/null +++ b/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: +-- . + + +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; diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index 494e3570d0..8f63680249 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/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]: