From 340ae9b459cb324dc7ddc099e279a2ce1efc8388 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 28 Mar 2025 16:20:16 +0000 Subject: [PATCH 01/31] WIP This handles membership changes for users that are not in current state event delta stream table (e.g. rejecting remote invites). --- synapse/handlers/sliding_sync/__init__.py | 10 ++ synapse/handlers/sliding_sync/room_lists.py | 20 +++ synapse/storage/databases/main/events.py | 1 + synapse/storage/databases/main/stream.py | 165 ++++++++++++++++++ .../client/sliding_sync/test_rooms_invites.py | 72 +++++++- 5 files changed, 267 insertions(+), 1 deletion(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 459d3c3e248..72973703930 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -245,12 +245,16 @@ async def current_sync_for_user( to_token=to_token, ) + print("interested_rooms:", interested_rooms) + lists = interested_rooms.lists relevant_room_map = interested_rooms.relevant_room_map all_rooms = interested_rooms.all_rooms room_membership_for_user_map = interested_rooms.room_membership_for_user_map relevant_rooms_to_send_map = interested_rooms.relevant_rooms_to_send_map + print("relevant_rooms_to_send_map:", relevant_rooms_to_send_map) + # Fetch room data rooms: Dict[str, SlidingSyncResult.RoomResult] = {} @@ -274,6 +278,8 @@ async def handle_room(room_id: str) -> None: is_dm=room_id in interested_rooms.dm_room_ids, ) + print("room_sync_result:", room_id, room_sync_result) + # Filter out empty room results during incremental sync if room_sync_result or not from_token: rooms[room_id] = room_sync_result @@ -856,6 +862,10 @@ async def get_room_sync_data( # TODO: Limit the number of state events we're about to send down # the room, if its too many we should change this to an # `initial=True`? + + # TODO: We need to pull out membership changes if the user isn't in + # the room, i.e. to deal with rejecting remote invites. + deltas = await self.get_current_state_deltas_for_room( room_id=room_id, room_membership_for_user_at_to_token=room_membership_for_user_at_to_token, diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index a1730b7e05b..7d88d01144b 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -299,6 +299,8 @@ async def _compute_interested_rooms_new_tables( ) dm_room_ids = await self._get_dm_rooms_for_user(user_id) + print("newly left:", newly_left_room_map) + # Add back `newly_left` rooms (rooms left in the from -> to token range). # # We do this because `get_sliding_sync_rooms_for_user(...)` doesn't include @@ -1110,6 +1112,24 @@ async def _get_newly_joined_and_left_rooms( newly_joined_room_ids: Set[str] = set() newly_left_room_map: Dict[str, RoomsForUserStateReset] = {} + if not from_token: + return (), {} + + changes = await self.store.get_sliding_sync_membership_changes( + user_id, + from_key=from_token.room_key, + to_key=to_token.room_key, + excluded_room_ids=self.rooms_to_exclude_globally, + ) + + for room_id, entry in changes.items(): + if entry.membership == Membership.JOIN: + newly_joined_room_ids.add(room_id) + elif entry.membership == Membership.LEAVE: + newly_left_room_map[room_id] = entry + + return newly_joined_room_ids, newly_left_room_map + # We need to figure out the # # - 1) Figure out which rooms are `newly_left` rooms (> `from_token` and <= `to_token`) diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 26fbc1a483f..852e5e2300c 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -410,6 +410,7 @@ async def _calculate_sliding_sync_table_changes( Returns: SlidingSyncTableChanges """ + print("Changes:", delta_state) to_insert = delta_state.to_insert to_delete = delta_state.to_delete diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 00e52086743..8d6205c8761 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -80,6 +80,7 @@ ) from synapse.storage.databases.main.events_worker import EventsWorkerStore from synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine +from synapse.storage.roommember import RoomsForUserStateReset from synapse.storage.util.id_generators import MultiWriterIdGenerator from synapse.types import PersistedEventPosition, RoomStreamToken, StrCollection from synapse.util.caches.descriptors import cached, cachedList @@ -1136,6 +1137,170 @@ def f(txn: LoggingTransaction) -> List[CurrentStateDeltaMembership]: if membership_change.room_id not in room_ids_to_exclude ] + @trace + async def get_sliding_sync_membership_changes( + self, + user_id: str, + from_key: RoomStreamToken, + to_key: RoomStreamToken, + excluded_room_ids: Optional[List[str]] = None, + ) -> Dict[str, RoomsForUserStateReset]: + # Start by ruling out cases where a DB query is not necessary. + if from_key == to_key: + return [] + + if from_key: + has_changed = self._membership_stream_cache.has_entity_changed( + user_id, int(from_key.stream) + ) + if not has_changed: + return [] + + room_ids_to_exclude: AbstractSet[str] = set() + if excluded_room_ids is not None: + room_ids_to_exclude = set(excluded_room_ids) + + def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: + # To handle tokens with a non-empty instance_map we fetch more + # results than necessary and then filter down + min_from_id = from_key.stream + max_to_id = to_key.get_max_stream_pos() + + # This query looks at membership changes in + # `sliding_sync_membership_snapshots`. These will not include where + # users get state reset out of rooms, so we need to look for that + # case in `current_state_delta_stream`. + # + # TODO: Add an index a better index on sliding_sync_membership_snapshots + sql = """ + SELECT + room_id, + membership_event_id, + event_instance_name, + event_stream_ordering, + membership, + sender, + prev_membership, + room_version + FROM + ( + SELECT + room_id, + membership_event_id, + event_instance_name, + event_stream_ordering, + membership, + sender, + null AS prev_membership + FROM sliding_sync_membership_snapshots + + UNION + + SELECT + s.room_id, + e.event_id, + s.instance_name, + s.stream_id, + m.membership, + e.sender, + m_prev.membership AS prev_membership + FROM current_state_delta_stream AS s + LEFT JOIN events AS e ON e.event_id = s.event_id + LEFT JOIN room_memberships AS m ON m.event_id = s.event_id + LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id + WHERE + s.type = ? + AND s.state_key = ? + AND s.event_id IS NULL + ) AS c + INNER JOIN rooms USING (room_id) + WHERE event_stream_ordering > ? AND event_stream_ordering <= ? + ORDER BY event_stream_ordering ASC + """ + + txn.execute( + sql, + (EventTypes.Member, user_id, min_from_id, max_to_id), + ) + + membership_changes: Dict[str, RoomsForUserStateReset] = {} + for ( + room_id, + membership_event_id, + event_instance_name, + event_stream_ordering, + membership, + sender, + prev_membership, + room_version_id, + ) in txn: + assert room_id is not None + assert event_stream_ordering is not None + + if room_id in room_ids_to_exclude: + continue + + print( + room_id, + membership_event_id, + event_instance_name, + event_stream_ordering, + membership, + sender, + prev_membership, + room_version_id, + ) + + if _filter_results_by_stream( + from_key, + to_key, + event_instance_name, + event_stream_ordering, + ): + # When the server leaves a room, it will insert new rows into the + # `current_state_delta_stream` table with `event_id = null` for all + # current state. This means we might already have a row for the + # leave event and then another for the same leave where the + # `event_id=null` but the `prev_event_id` is pointing back at the + # earlier leave event. We don't want to report the leave, if we + # already have a leave event. + if ( + membership_event_id is None + and prev_membership == Membership.LEAVE + ): + continue + + if membership_event_id is None and room_id in membership_changes: + continue + + if membership is None: + membership = Membership.LEAVE + + # TODO: If we see a JOIN we need to check if the user newly + # joined the room (instead of just changing their display + # name) + + membership_change = RoomsForUserStateReset( + room_id=room_id, + sender=sender, + membership=membership, + event_id=membership_event_id, + event_pos=PersistedEventPosition( + event_instance_name, event_stream_ordering + ), + room_version_id=room_version_id, + ) + + membership_changes[room_id] = membership_change + + return membership_changes + + membership_changes = await self.db_pool.runInteraction( + "get_sliding_sync_membership_changes", f + ) + + return membership_changes + @cancellable async def get_membership_changes_for_user( self, diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index 882762ca29a..a3d0765771c 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -18,13 +18,18 @@ from twisted.test.proto_helpers import MemoryReactor import synapse.rest.admin -from synapse.api.constants import EventTypes, HistoryVisibility +from synapse.api.constants import EventTypes, HistoryVisibility, Membership +from synapse.api.room_versions import RoomVersions +from synapse.events import make_event_from_dict +from synapse.handlers.room import EventContext from synapse.rest.client import login, room, sync from synapse.server import HomeServer from synapse.types import UserID +from synapse.types.handlers.sliding_sync import StateValues from synapse.util import Clock from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase +from tests.unittest import override_config logger = logging.getLogger(__name__) @@ -64,6 +69,7 @@ class SlidingSyncRoomsInvitesTestCase(SlidingSyncBase): def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.store = hs.get_datastores().main self.storage_controllers = hs.get_storage_controllers() + self.federation_handler = hs.get_federation_handler() super().prepare(reactor, clock, hs) @@ -526,3 +532,67 @@ def test_rooms_invite_world_readable_history_incremental_sync(self) -> None: ], response_body["rooms"][room_id1]["invite_state"], ) + + @override_config({"federation_domain_whitelist": []}) + def test_reject_invite(self) -> None: + """Test that rejecting an invite gets sent down sliding sync""" + + user_id = self.register_user("user1", "pass") + user_tok = self.login(user_id, "pass") + + room_id = "!room:remote.server" + self._create_remote_invite_room_for_user(room_id, user_id) + + # Make the Sliding Sync request + sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [(EventTypes.Member, StateValues.ME)], + "timeline_limit": 3, + } + } + } + response_body, from_token = self.do_sync(sync_body, tok=user_tok) + + self.assertIn(room_id, response_body["rooms"]) + print(response_body["rooms"][room_id]) + + self.helper.leave(room_id, user_id, tok=user_tok) + + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) + + print(response_body["rooms"][room_id]) + self.assertIn(room_id, response_body["rooms"]) + + raise NotImplementedError() + + def _create_remote_invite_room_for_user( + self, + room_id: str, + user_id: str, + ) -> None: + invite_event_dict = { + "room_id": room_id, + "sender": "@inviter:remote.server", + "state_key": user_id, + "depth": 1, + "origin_server_ts": 1, + "type": EventTypes.Member, + "content": {"membership": Membership.INVITE}, + "auth_events": [], + "prev_events": [], + } + + invite_event = make_event_from_dict( + invite_event_dict, + room_version=RoomVersions.V10, + ) + + self.get_success( + self.federation_handler.on_invite_request( + "remote.server", + invite_event, + room_version=invite_event.room_version, + ) + ) From 74cd6d51d3b9f0bcff366403555c13b199b0db37 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Mon, 7 Apr 2025 16:49:00 -0600 Subject: [PATCH 02/31] Improve reject invite logic --- synapse/handlers/sliding_sync/__init__.py | 13 +++++++++++++ synapse/handlers/sliding_sync/room_lists.py | 2 +- synapse/storage/databases/main/stream.py | 14 +++++++++----- .../client/sliding_sync/test_rooms_invites.py | 16 +++++++++++----- 4 files changed, 34 insertions(+), 11 deletions(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 72973703930..c875ea11b48 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -275,6 +275,7 @@ async def handle_room(room_id: str) -> None: from_token=from_token, to_token=to_token, newly_joined=room_id in interested_rooms.newly_joined_rooms, + newly_left=room_id in interested_rooms.newly_left_rooms, is_dm=room_id in interested_rooms.dm_room_ids, ) @@ -548,6 +549,7 @@ async def get_room_sync_data( from_token: Optional[SlidingSyncStreamToken], to_token: StreamToken, newly_joined: bool, + newly_left: bool, is_dm: bool, ) -> SlidingSyncResult.RoomResult: """ @@ -866,6 +868,17 @@ async def get_room_sync_data( # TODO: We need to pull out membership changes if the user isn't in # the room, i.e. to deal with rejecting remote invites. + if newly_left and room_membership_for_user_at_to_token.event_id is not None: + membership_changed = True + leave_event = await self.store.get_event( + room_membership_for_user_at_to_token.event_id + ) + state_key = leave_event.get_state_key() + if state_key is not None: + room_state_delta_id_map[(leave_event.type, state_key)] = ( + room_membership_for_user_at_to_token.event_id + ) + deltas = await self.get_current_state_deltas_for_room( room_id=room_id, room_membership_for_user_at_to_token=room_membership_for_user_at_to_token, diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 7d88d01144b..24f33f9e44a 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -1113,7 +1113,7 @@ async def _get_newly_joined_and_left_rooms( newly_left_room_map: Dict[str, RoomsForUserStateReset] = {} if not from_token: - return (), {} + return newly_joined_room_ids, newly_left_room_map changes = await self.store.get_sliding_sync_membership_changes( user_id, diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 8d6205c8761..24587e47aed 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1147,14 +1147,14 @@ async def get_sliding_sync_membership_changes( ) -> Dict[str, RoomsForUserStateReset]: # Start by ruling out cases where a DB query is not necessary. if from_key == to_key: - return [] + return {} if from_key: has_changed = self._membership_stream_cache.has_entity_changed( user_id, int(from_key.stream) ) if not has_changed: - return [] + return {} room_ids_to_exclude: AbstractSet[str] = set() if excluded_room_ids is not None: @@ -1276,9 +1276,13 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: if membership is None: membership = Membership.LEAVE - # TODO: If we see a JOIN we need to check if the user newly - # joined the room (instead of just changing their display - # name) + if ( + membership is Membership.JOIN + and prev_membership is Membership.JOIN + ): + # The user was previously joined so this it not a new join. + # This happens when the user changes their display name. + continue membership_change = RoomsForUserStateReset( room_id=room_id, diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index a3d0765771c..e2a3e8f6646 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -21,7 +21,6 @@ from synapse.api.constants import EventTypes, HistoryVisibility, Membership from synapse.api.room_versions import RoomVersions from synapse.events import make_event_from_dict -from synapse.handlers.room import EventContext from synapse.rest.client import login, room, sync from synapse.server import HomeServer from synapse.types import UserID @@ -541,7 +540,7 @@ def test_reject_invite(self) -> None: user_tok = self.login(user_id, "pass") room_id = "!room:remote.server" - self._create_remote_invite_room_for_user(room_id, user_id) + self._create_remote_invite_room_for_user_using_federation_handler(room_id, user_id) # Make the Sliding Sync request sync_body = { @@ -556,7 +555,6 @@ def test_reject_invite(self) -> None: response_body, from_token = self.do_sync(sync_body, tok=user_tok) self.assertIn(room_id, response_body["rooms"]) - print(response_body["rooms"][room_id]) self.helper.leave(room_id, user_id, tok=user_tok) @@ -565,9 +563,17 @@ def test_reject_invite(self) -> None: print(response_body["rooms"][room_id]) self.assertIn(room_id, response_body["rooms"]) - raise NotImplementedError() + actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][0] + self.assertEqual( + { + f'{actual_requred_state_leave["content"]["membership"]} ("{actual_requred_state_leave["type"]}", "{actual_requred_state_leave["state_key"]}")' + }, + { + f'leave ("{EventTypes.Member}", "{user_id}")' + }, + ) - def _create_remote_invite_room_for_user( + def _create_remote_invite_room_for_user_using_federation_handler( self, room_id: str, user_id: str, From 17cdd4ffc743b5151a2a4cc8aa6ea4f2830d66a0 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 12:13:10 +0100 Subject: [PATCH 03/31] simplify query: left joins with e.event_id = NULL never match --- synapse/storage/databases/main/stream.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 24587e47aed..419256bd26b 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1198,15 +1198,13 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: SELECT s.room_id, - e.event_id, + NULL AS event_id, s.instance_name, s.stream_id, - m.membership, - e.sender, + NULL AS membership, + NULL AS sender, m_prev.membership AS prev_membership FROM current_state_delta_stream AS s - LEFT JOIN events AS e ON e.event_id = s.event_id - LEFT JOIN room_memberships AS m ON m.event_id = s.event_id LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id WHERE s.type = ? From 5ee188a6872e4725ebee7c117656d055ed99715b Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 12:13:39 +0100 Subject: [PATCH 04/31] optimise query: the unioned subqueries can not overlap and don't need to be made unique --- synapse/storage/databases/main/stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 419256bd26b..056463220b3 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1194,7 +1194,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: null AS prev_membership FROM sliding_sync_membership_snapshots - UNION + UNION ALL SELECT s.room_id, From de7217f264cb1394641ebde8270ba3d65bce933e Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 12:14:12 +0100 Subject: [PATCH 05/31] add missing `user_id` filter to first half of the query --- synapse/storage/databases/main/stream.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 056463220b3..4aa5637ea3d 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1193,6 +1193,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: sender, null AS prev_membership FROM sliding_sync_membership_snapshots + WHERE user_id = ? UNION ALL @@ -1218,7 +1219,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: txn.execute( sql, - (EventTypes.Member, user_id, min_from_id, max_to_id), + (user_id, EventTypes.Member, user_id, min_from_id, max_to_id), ) membership_changes: Dict[str, RoomsForUserStateReset] = {} From 9e41abdfdfa7fbf657acb6d45fc16069a3b3f918 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 12:18:26 +0100 Subject: [PATCH 06/31] don't use `is` for comparing strings returned from the DB; they may not have the same identity --- synapse/storage/databases/main/stream.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 4aa5637ea3d..f93e193a19f 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1276,8 +1276,8 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: membership = Membership.LEAVE if ( - membership is Membership.JOIN - and prev_membership is Membership.JOIN + membership == Membership.JOIN + and prev_membership == Membership.JOIN ): # The user was previously joined so this it not a new join. # This happens when the user changes their display name. From af7a4af30d2fbb17bf8511aaab41528e4285e5a9 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 14:07:00 +0100 Subject: [PATCH 07/31] TODO: suspicion --- synapse/storage/databases/main/stream.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index f93e193a19f..3364b322e14 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1270,6 +1270,9 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: continue if membership_event_id is None and room_id in membership_changes: + # SUSPICIOUS: if we join a room and get state reset out of it + # in the same queried window, + # won't this ignore the 'state reset out of it' part? continue if membership is None: From fd4a633d7f320f045088d8d87064de8b9956b7d9 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Wed, 16 Apr 2025 16:13:40 +0100 Subject: [PATCH 08/31] TODO: note what indices we need --- synapse/storage/databases/main/stream.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 3364b322e14..7afc125837e 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1172,6 +1172,10 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: # case in `current_state_delta_stream`. # # TODO: Add an index a better index on sliding_sync_membership_snapshots + # probably want: + # - sliding_sync_membership_snapshots (user_id, event_stream_ordering) + # replacing the existing index on only (user_id) + # - MAYBE current_state_delta_stream(state_key, stream_id) WHERE type = 'm.room.member' AND event_id IS NULL sql = """ SELECT room_id, From e0aad4be501d35d96fab06931cac0d0ffebaf77d Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Mon, 28 Apr 2025 15:54:29 -0600 Subject: [PATCH 09/31] Revert "add missing `user_id` filter to first half of the query" This reverts commit de7217f264cb1394641ebde8270ba3d65bce933e. --- synapse/storage/databases/main/stream.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 7afc125837e..d0d9c0aeeaf 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1197,7 +1197,6 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: sender, null AS prev_membership FROM sliding_sync_membership_snapshots - WHERE user_id = ? UNION ALL @@ -1223,7 +1222,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: txn.execute( sql, - (user_id, EventTypes.Member, user_id, min_from_id, max_to_id), + (EventTypes.Member, user_id, min_from_id, max_to_id), ) membership_changes: Dict[str, RoomsForUserStateReset] = {} From 1519d4f25efbbf6f03ddf99713610cc95f869e33 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Mon, 28 Apr 2025 17:43:30 -0600 Subject: [PATCH 10/31] Reapply "add missing `user_id` filter to first half of the query" This reverts commit e0aad4be501d35d96fab06931cac0d0ffebaf77d. --- synapse/storage/databases/main/stream.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index d0d9c0aeeaf..7afc125837e 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1197,6 +1197,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: sender, null AS prev_membership FROM sliding_sync_membership_snapshots + WHERE user_id = ? UNION ALL @@ -1222,7 +1223,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: txn.execute( sql, - (EventTypes.Member, user_id, min_from_id, max_to_id), + (user_id, EventTypes.Member, user_id, min_from_id, max_to_id), ) membership_changes: Dict[str, RoomsForUserStateReset] = {} From 7ed2a12b3d23617f1ddee71f0067903f2897932e Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Mon, 28 Apr 2025 18:37:53 -0600 Subject: [PATCH 11/31] Get tests passing again --- synapse/handlers/sliding_sync/room_lists.py | 110 +----------------- synapse/storage/databases/main/stream.py | 36 +++--- .../client/sliding_sync/test_rooms_invites.py | 12 +- 3 files changed, 28 insertions(+), 130 deletions(-) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 24f33f9e44a..05e7db79c9d 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -1109,6 +1109,7 @@ async def _get_newly_joined_and_left_rooms( was state reset out of the room. To actually check for a state reset, you need to check if a membership still exists in the room. """ + newly_joined_room_ids: Set[str] = set() newly_left_room_map: Dict[str, RoomsForUserStateReset] = {} @@ -1130,115 +1131,6 @@ async def _get_newly_joined_and_left_rooms( return newly_joined_room_ids, newly_left_room_map - # We need to figure out the - # - # - 1) Figure out which rooms are `newly_left` rooms (> `from_token` and <= `to_token`) - # - 2) Figure out which rooms are `newly_joined` (> `from_token` and <= `to_token`) - - # 1) Fetch membership changes that fall in the range from `from_token` up to `to_token` - current_state_delta_membership_changes_in_from_to_range = [] - if from_token: - current_state_delta_membership_changes_in_from_to_range = ( - await self.store.get_current_state_delta_membership_changes_for_user( - user_id, - from_key=from_token.room_key, - to_key=to_token.room_key, - excluded_room_ids=self.rooms_to_exclude_globally, - ) - ) - - # 1) Assemble a list of the last membership events in some given ranges. Someone - # could have left and joined multiple times during the given range but we only - # care about end-result so we grab the last one. - last_membership_change_by_room_id_in_from_to_range: Dict[ - str, CurrentStateDeltaMembership - ] = {} - # We also want to assemble a list of the first membership events during the token - # range so we can step backward to the previous membership that would apply to - # before the token range to see if we have `newly_joined` the room. - first_membership_change_by_room_id_in_from_to_range: Dict[ - str, CurrentStateDeltaMembership - ] = {} - # Keep track if the room has a non-join event in the token range so we can later - # tell if it was a `newly_joined` room. If the last membership event in the - # token range is a join and there is also some non-join in the range, we know - # they `newly_joined`. - has_non_join_event_by_room_id_in_from_to_range: Dict[str, bool] = {} - for ( - membership_change - ) in current_state_delta_membership_changes_in_from_to_range: - room_id = membership_change.room_id - - last_membership_change_by_room_id_in_from_to_range[room_id] = ( - membership_change - ) - # Only set if we haven't already set it - first_membership_change_by_room_id_in_from_to_range.setdefault( - room_id, membership_change - ) - - if membership_change.membership != Membership.JOIN: - has_non_join_event_by_room_id_in_from_to_range[room_id] = True - - # 1) Fixup - # - # 2) We also want to assemble a list of possibly newly joined rooms. Someone - # could have left and joined multiple times during the given range but we only - # care about whether they are joined at the end of the token range so we are - # working with the last membership even in the token range. - possibly_newly_joined_room_ids = set() - for ( - last_membership_change_in_from_to_range - ) in last_membership_change_by_room_id_in_from_to_range.values(): - room_id = last_membership_change_in_from_to_range.room_id - - # 2) - if last_membership_change_in_from_to_range.membership == Membership.JOIN: - possibly_newly_joined_room_ids.add(room_id) - - # 1) Figure out newly_left rooms (> `from_token` and <= `to_token`). - if last_membership_change_in_from_to_range.membership == Membership.LEAVE: - # 1) Mark this room as `newly_left` - newly_left_room_map[room_id] = RoomsForUserStateReset( - room_id=room_id, - sender=last_membership_change_in_from_to_range.sender, - membership=Membership.LEAVE, - event_id=last_membership_change_in_from_to_range.event_id, - event_pos=last_membership_change_in_from_to_range.event_pos, - room_version_id=await self.store.get_room_version_id(room_id), - ) - - # 2) Figure out `newly_joined` - for room_id in possibly_newly_joined_room_ids: - has_non_join_in_from_to_range = ( - has_non_join_event_by_room_id_in_from_to_range.get(room_id, False) - ) - # If the last membership event in the token range is a join and there is - # also some non-join in the range, we know they `newly_joined`. - if has_non_join_in_from_to_range: - # We found a `newly_joined` room (we left and joined within the token range) - newly_joined_room_ids.add(room_id) - else: - prev_event_id = first_membership_change_by_room_id_in_from_to_range[ - room_id - ].prev_event_id - prev_membership = first_membership_change_by_room_id_in_from_to_range[ - room_id - ].prev_membership - - if prev_event_id is None: - # We found a `newly_joined` room (we are joining the room for the - # first time within the token range) - newly_joined_room_ids.add(room_id) - # Last resort, we need to step back to the previous membership event - # just before the token range to see if we're joined then or not. - elif prev_membership != Membership.JOIN: - # We found a `newly_joined` room (we left before the token range - # and joined within the token range) - newly_joined_room_ids.add(room_id) - - return newly_joined_room_ids, newly_left_room_map - @trace async def _get_dm_rooms_for_user( self, diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 7afc125837e..67a976b1230 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1189,32 +1189,35 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: FROM ( SELECT - room_id, - membership_event_id, - event_instance_name, - event_stream_ordering, - membership, - sender, - null AS prev_membership - FROM sliding_sync_membership_snapshots - WHERE user_id = ? - - UNION ALL + s.room_id, + s.membership_event_id, + s.event_instance_name, + s.event_stream_ordering, + s.membership, + s.sender, + m_prev.membership AS prev_membership + FROM sliding_sync_membership_snapshots as s + LEFT JOIN event_edges AS e ON e.event_id = s.membership_event_id + LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = e.prev_event_id + WHERE s.user_id = ? + + UNION SELECT s.room_id, - NULL AS event_id, + e.event_id, s.instance_name, s.stream_id, - NULL AS membership, - NULL AS sender, + m.membership, + e.sender, m_prev.membership AS prev_membership FROM current_state_delta_stream AS s + LEFT JOIN events AS e ON e.event_id = s.event_id + LEFT JOIN room_memberships AS m ON m.event_id = s.event_id LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = s.prev_event_id WHERE s.type = ? AND s.state_key = ? - AND s.event_id IS NULL ) AS c INNER JOIN rooms USING (room_id) WHERE event_stream_ordering > ? AND event_stream_ordering <= ? @@ -1243,7 +1246,8 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: if room_id in room_ids_to_exclude: continue - print( + logger.warning( + "%s %s %s %s %s %s %s %s", room_id, membership_event_id, event_instance_name, diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index e2a3e8f6646..b86dd27b40e 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -540,7 +540,9 @@ def test_reject_invite(self) -> None: user_tok = self.login(user_id, "pass") room_id = "!room:remote.server" - self._create_remote_invite_room_for_user_using_federation_handler(room_id, user_id) + self._create_remote_invite_room_for_user_using_federation_handler( + room_id, user_id + ) # Make the Sliding Sync request sync_body = { @@ -563,14 +565,14 @@ def test_reject_invite(self) -> None: print(response_body["rooms"][room_id]) self.assertIn(room_id, response_body["rooms"]) - actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][0] + actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][ + 0 + ] self.assertEqual( { f'{actual_requred_state_leave["content"]["membership"]} ("{actual_requred_state_leave["type"]}", "{actual_requred_state_leave["state_key"]}")' }, - { - f'leave ("{EventTypes.Member}", "{user_id}")' - }, + {f'leave ("{EventTypes.Member}", "{user_id}")'}, ) def _create_remote_invite_room_for_user_using_federation_handler( From 223c87edddd972a22c4d8d2f99ac5872efe6841a Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Tue, 29 Apr 2025 10:50:25 -0600 Subject: [PATCH 12/31] Cleanup printing --- synapse/handlers/sliding_sync/__init__.py | 10 +--------- tests/rest/client/sliding_sync/test_rooms_invites.py | 1 - 2 files changed, 1 insertion(+), 10 deletions(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index c875ea11b48..b3179a988ab 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -245,16 +245,12 @@ async def current_sync_for_user( to_token=to_token, ) - print("interested_rooms:", interested_rooms) - lists = interested_rooms.lists relevant_room_map = interested_rooms.relevant_room_map all_rooms = interested_rooms.all_rooms room_membership_for_user_map = interested_rooms.room_membership_for_user_map relevant_rooms_to_send_map = interested_rooms.relevant_rooms_to_send_map - print("relevant_rooms_to_send_map:", relevant_rooms_to_send_map) - # Fetch room data rooms: Dict[str, SlidingSyncResult.RoomResult] = {} @@ -279,8 +275,6 @@ async def handle_room(room_id: str) -> None: is_dm=room_id in interested_rooms.dm_room_ids, ) - print("room_sync_result:", room_id, room_sync_result) - # Filter out empty room results during incremental sync if room_sync_result or not from_token: rooms[room_id] = room_sync_result @@ -567,6 +561,7 @@ async def get_room_sync_data( from_token: The point in the stream to sync from. to_token: The point in the stream to sync up to. newly_joined: If the user has newly joined the room + newly_left: If the user has newly left the room is_dm: Whether the room is a DM room """ user = sync_config.user @@ -865,9 +860,6 @@ async def get_room_sync_data( # the room, if its too many we should change this to an # `initial=True`? - # TODO: We need to pull out membership changes if the user isn't in - # the room, i.e. to deal with rejecting remote invites. - if newly_left and room_membership_for_user_at_to_token.event_id is not None: membership_changed = True leave_event = await self.store.get_event( diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index b86dd27b40e..869daf137da 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -562,7 +562,6 @@ def test_reject_invite(self) -> None: response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) - print(response_body["rooms"][room_id]) self.assertIn(room_id, response_body["rooms"]) actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][ From b8d1b5eb79e50cc06fdbcf329be1f1fb69ccb493 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Tue, 29 Apr 2025 10:52:15 -0600 Subject: [PATCH 13/31] More printing cleanup --- synapse/handlers/sliding_sync/room_lists.py | 2 -- synapse/storage/databases/main/events.py | 1 - synapse/storage/databases/main/stream.py | 12 ------------ 3 files changed, 15 deletions(-) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 05e7db79c9d..70489031ef7 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -299,8 +299,6 @@ async def _compute_interested_rooms_new_tables( ) dm_room_ids = await self._get_dm_rooms_for_user(user_id) - print("newly left:", newly_left_room_map) - # Add back `newly_left` rooms (rooms left in the from -> to token range). # # We do this because `get_sliding_sync_rooms_for_user(...)` doesn't include diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 852e5e2300c..26fbc1a483f 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -410,7 +410,6 @@ async def _calculate_sliding_sync_table_changes( Returns: SlidingSyncTableChanges """ - print("Changes:", delta_state) to_insert = delta_state.to_insert to_delete = delta_state.to_delete diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 67a976b1230..f7f822b7bc3 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1246,18 +1246,6 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: if room_id in room_ids_to_exclude: continue - logger.warning( - "%s %s %s %s %s %s %s %s", - room_id, - membership_event_id, - event_instance_name, - event_stream_ordering, - membership, - sender, - prev_membership, - room_version_id, - ) - if _filter_results_by_stream( from_key, to_key, From 7640c59623240865a0f9336fd72aa23841b7e473 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Tue, 29 Apr 2025 11:15:19 -0600 Subject: [PATCH 14/31] Readd union optimization --- synapse/storage/databases/main/stream.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index f7f822b7bc3..cb633329665 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1201,7 +1201,7 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: LEFT JOIN room_memberships AS m_prev ON m_prev.event_id = e.prev_event_id WHERE s.user_id = ? - UNION + UNION ALL SELECT s.room_id, From 241ec9af6bf2d73ecb7137449384a3801efb2d00 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Tue, 29 Apr 2025 12:01:35 -0600 Subject: [PATCH 15/31] Add changelog entry --- changelog.d/18375.bugfix | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/18375.bugfix diff --git a/changelog.d/18375.bugfix b/changelog.d/18375.bugfix new file mode 100644 index 00000000000..faebe6f0468 --- /dev/null +++ b/changelog.d/18375.bugfix @@ -0,0 +1 @@ +Pass leave from remote invite rejection down Sliding Sync. From 79fc8888481d2f87f1f93d5d244440141d603d77 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 08:46:54 -0600 Subject: [PATCH 16/31] Comment newly_left room logic --- synapse/handlers/sliding_sync/__init__.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index b3179a988ab..7e69efca8ed 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -860,6 +860,15 @@ async def get_room_sync_data( # the room, if its too many we should change this to an # `initial=True`? + + # For the case of rejecting remote invites, the leave event won't be + # returned by `get_current_state_deltas_for_room`. This is due to the current + # state only being filled out for rooms the server is in, and so doesn't pick + # up out-of-band leaves (including locally rejected invites) as these events + # are outliers and not added to the `current_state_delta_stream`. + # + # We rely on being explicitly told that the room has been `newly_left` to + # ensure we extract the out-of-band leave. if newly_left and room_membership_for_user_at_to_token.event_id is not None: membership_changed = True leave_event = await self.store.get_event( From 3e353b654e2f3d498bad052ab5ea3565a70c0f73 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 09:06:32 -0600 Subject: [PATCH 17/31] Move test & add comments --- synapse/handlers/sliding_sync/__init__.py | 1 - .../client/sliding_sync/test_rooms_invites.py | 78 +------------------ .../client/sliding_sync/test_sliding_sync.py | 42 ++++++++++ 3 files changed, 43 insertions(+), 78 deletions(-) diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 7e69efca8ed..cb56eb53fc7 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -860,7 +860,6 @@ async def get_room_sync_data( # the room, if its too many we should change this to an # `initial=True`? - # For the case of rejecting remote invites, the leave event won't be # returned by `get_current_state_deltas_for_room`. This is due to the current # state only being filled out for rooms the server is in, and so doesn't pick diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index 869daf137da..04264847d7a 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -18,17 +18,13 @@ from twisted.test.proto_helpers import MemoryReactor import synapse.rest.admin -from synapse.api.constants import EventTypes, HistoryVisibility, Membership -from synapse.api.room_versions import RoomVersions -from synapse.events import make_event_from_dict +from synapse.api.constants import EventTypes, HistoryVisibility from synapse.rest.client import login, room, sync from synapse.server import HomeServer from synapse.types import UserID -from synapse.types.handlers.sliding_sync import StateValues from synapse.util import Clock from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase -from tests.unittest import override_config logger = logging.getLogger(__name__) @@ -531,75 +527,3 @@ def test_rooms_invite_world_readable_history_incremental_sync(self) -> None: ], response_body["rooms"][room_id1]["invite_state"], ) - - @override_config({"federation_domain_whitelist": []}) - def test_reject_invite(self) -> None: - """Test that rejecting an invite gets sent down sliding sync""" - - user_id = self.register_user("user1", "pass") - user_tok = self.login(user_id, "pass") - - room_id = "!room:remote.server" - self._create_remote_invite_room_for_user_using_federation_handler( - room_id, user_id - ) - - # Make the Sliding Sync request - sync_body = { - "lists": { - "foo-list": { - "ranges": [[0, 1]], - "required_state": [(EventTypes.Member, StateValues.ME)], - "timeline_limit": 3, - } - } - } - response_body, from_token = self.do_sync(sync_body, tok=user_tok) - - self.assertIn(room_id, response_body["rooms"]) - - self.helper.leave(room_id, user_id, tok=user_tok) - - response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) - - self.assertIn(room_id, response_body["rooms"]) - - actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][ - 0 - ] - self.assertEqual( - { - f'{actual_requred_state_leave["content"]["membership"]} ("{actual_requred_state_leave["type"]}", "{actual_requred_state_leave["state_key"]}")' - }, - {f'leave ("{EventTypes.Member}", "{user_id}")'}, - ) - - def _create_remote_invite_room_for_user_using_federation_handler( - self, - room_id: str, - user_id: str, - ) -> None: - invite_event_dict = { - "room_id": room_id, - "sender": "@inviter:remote.server", - "state_key": user_id, - "depth": 1, - "origin_server_ts": 1, - "type": EventTypes.Member, - "content": {"membership": Membership.INVITE}, - "auth_events": [], - "prev_events": [], - } - - invite_event = make_event_from_dict( - invite_event_dict, - room_version=RoomVersions.V10, - ) - - self.get_success( - self.federation_handler.on_invite_request( - "remote.server", - invite_event, - room_version=invite_event.room_version, - ) - ) diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py index f3cf2111ec1..194dfe76395 100644 --- a/tests/rest/client/sliding_sync/test_sliding_sync.py +++ b/tests/rest/client/sliding_sync/test_sliding_sync.py @@ -790,6 +790,48 @@ def test_invited_to_forgotten_remote_room(self) -> None: exact=True, ) + def test_reject_remote_invite(self) -> None: + """Test that rejecting a remote invite gets sent down sliding sync""" + + user_id = self.register_user("user1", "pass") + user_tok = self.login(user_id, "pass") + + # Create a remote room invite (out-of-band membership) + room_id = "!room:remote.server" + self._create_remote_invite_room_for_user(user_id, None, room_id) + + # Make the Sliding Sync request + sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [(EventTypes.Member, StateValues.ME)], + "timeline_limit": 3, + } + } + } + response_body, from_token = self.do_sync(sync_body, tok=user_tok) + self.assertIn(room_id, response_body["rooms"]) + + # Reject the remote room invite + self.helper.leave(room_id, user_id, tok=user_tok) + + # Sync again after rejecting the invite + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) + + # We should see the leave for the room so clients don't end up with stuck + # invites + self.assertIn(room_id, response_body["rooms"]) + actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][ + 0 + ] + self.assertEqual( + { + f'{actual_requred_state_leave["content"]["membership"]} ("{actual_requred_state_leave["type"]}", "{actual_requred_state_leave["state_key"]}")' + }, + {f'leave ("{EventTypes.Member}", "{user_id}")'}, + ) + def test_ignored_user_invites_initial_sync(self) -> None: """ Make sure we ignore invites if they are from one of the `m.ignored_user_list` on From a7bc0caf47dc269e2e62697019af77299e84a9dc Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 21:01:11 +0000 Subject: [PATCH 18/31] Update tests/rest/client/sliding_sync/test_sliding_sync.py Co-authored-by: Eric Eastwood --- .../client/sliding_sync/test_sliding_sync.py | 33 +++++++++++++------ 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py index 194dfe76395..1a5a152f23d 100644 --- a/tests/rest/client/sliding_sync/test_sliding_sync.py +++ b/tests/rest/client/sliding_sync/test_sliding_sync.py @@ -791,7 +791,7 @@ def test_invited_to_forgotten_remote_room(self) -> None: ) def test_reject_remote_invite(self) -> None: - """Test that rejecting a remote invite gets sent down sliding sync""" + """Test that rejecting a remote invite comes down incremental sync""" user_id = self.register_user("user1", "pass") user_tok = self.login(user_id, "pass") @@ -811,7 +811,12 @@ def test_reject_remote_invite(self) -> None: } } response_body, from_token = self.do_sync(sync_body, tok=user_tok) - self.assertIn(room_id, response_body["rooms"]) + # We should see the room (like normal) + self.assertIncludes( + set(response_body["lists"]["foo-list"]["ops"][0]["room_ids"]), + {room_id}, + exact=True, + ) # Reject the remote room invite self.helper.leave(room_id, user_id, tok=user_tok) @@ -819,17 +824,25 @@ def test_reject_remote_invite(self) -> None: # Sync again after rejecting the invite response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) - # We should see the leave for the room so clients don't end up with stuck + # We should see the newly_left room + self.assertIncludes( + set(response_body["lists"]["foo-list"]["ops"][0]["room_ids"]), + {room_id}, + exact=True, + ) + # We should see the leave state for the room so clients don't end up with stuck # invites - self.assertIn(room_id, response_body["rooms"]) - actual_requred_state_leave = response_body["rooms"][room_id]["required_state"][ - 0 - ] - self.assertEqual( + self.assertIncludes( { - f'{actual_requred_state_leave["content"]["membership"]} ("{actual_requred_state_leave["type"]}", "{actual_requred_state_leave["state_key"]}")' + ( + state["type"], + state["state_key"], + state["content"].get("membership"), + ) + for state in response_body["rooms"][room_id]["required_state"] }, - {f'leave ("{EventTypes.Member}", "{user_id}")'}, + {(EventTypes.Member, user_id, Membership.LEAVE)}, + exact=True, ) def test_ignored_user_invites_initial_sync(self) -> None: From 93b00c3d6dfbec61edcedce104e93f639377d2f9 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 15:02:11 -0600 Subject: [PATCH 19/31] Remove leftover code --- tests/rest/client/sliding_sync/test_rooms_invites.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/rest/client/sliding_sync/test_rooms_invites.py b/tests/rest/client/sliding_sync/test_rooms_invites.py index 04264847d7a..882762ca29a 100644 --- a/tests/rest/client/sliding_sync/test_rooms_invites.py +++ b/tests/rest/client/sliding_sync/test_rooms_invites.py @@ -64,7 +64,6 @@ class SlidingSyncRoomsInvitesTestCase(SlidingSyncBase): def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: self.store = hs.get_datastores().main self.storage_controllers = hs.get_storage_controllers() - self.federation_handler = hs.get_federation_handler() super().prepare(reactor, clock, hs) From 720f85eea347fb060ee548daf3da3f2502892bd7 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 15:05:42 -0600 Subject: [PATCH 20/31] Make function take set --- synapse/handlers/sliding_sync/room_lists.py | 2 +- synapse/storage/databases/main/stream.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 70489031ef7..e4eec771438 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -1118,7 +1118,7 @@ async def _get_newly_joined_and_left_rooms( user_id, from_key=from_token.room_key, to_key=to_token.room_key, - excluded_room_ids=self.rooms_to_exclude_globally, + excluded_room_ids=set(self.rooms_to_exclude_globally), ) for room_id, entry in changes.items(): diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index cb633329665..0c8dd9a0b03 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1143,7 +1143,7 @@ async def get_sliding_sync_membership_changes( user_id: str, from_key: RoomStreamToken, to_key: RoomStreamToken, - excluded_room_ids: Optional[List[str]] = None, + excluded_room_ids: Optional[AbstractSet[str]] = None, ) -> Dict[str, RoomsForUserStateReset]: # Start by ruling out cases where a DB query is not necessary. if from_key == to_key: @@ -1158,7 +1158,7 @@ async def get_sliding_sync_membership_changes( room_ids_to_exclude: AbstractSet[str] = set() if excluded_room_ids is not None: - room_ids_to_exclude = set(excluded_room_ids) + room_ids_to_exclude = excluded_room_ids def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: # To handle tokens with a non-empty instance_map we fetch more From 06b9a3476fbbed86e8ad1534e46c2a1490c70816 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 21:06:44 +0000 Subject: [PATCH 21/31] Update synapse/storage/databases/main/stream.py Co-authored-by: Eric Eastwood --- synapse/storage/databases/main/stream.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 0c8dd9a0b03..fea48851497 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1167,8 +1167,8 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: max_to_id = to_key.get_max_stream_pos() # This query looks at membership changes in - # `sliding_sync_membership_snapshots`. These will not include where - # users get state reset out of rooms, so we need to look for that + # `sliding_sync_membership_snapshots` which will not include users + # that were state reset out of rooms; so we need to look for that # case in `current_state_delta_stream`. # # TODO: Add an index a better index on sliding_sync_membership_snapshots From db486c7d32da0110b48611d4aba12dd8127afd89 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 15:27:16 -0600 Subject: [PATCH 22/31] Add comment for why no membership means leave --- synapse/storage/databases/main/stream.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index fea48851497..16e12a6b924 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1271,8 +1271,14 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: # won't this ignore the 'state reset out of it' part? continue - if membership is None: - membership = Membership.LEAVE + # When `s.event_id = null`, we won't be able to get respective + # `room_membership` but can assume the user has left the room + # because this only happens when the server leaves a room + # (meaning everyone locally left) or a state reset which removed + # the person from the room. + membership = ( + membership if membership is not None else Membership.LEAVE + ) if ( membership == Membership.JOIN From 06209fc1cc73d23f19ffd17b445db0dfcb828262 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 15:35:03 -0600 Subject: [PATCH 23/31] Filter out all membership non-changes --- synapse/storage/databases/main/stream.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 16e12a6b924..9c7736417c2 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1280,12 +1280,10 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: membership if membership is not None else Membership.LEAVE ) - if ( - membership == Membership.JOIN - and prev_membership == Membership.JOIN - ): - # The user was previously joined so this it not a new join. - # This happens when the user changes their display name. + if membership == prev_membership: + # If `membership` and `prev_membership` are the same then this + # is not a meaningful change so we can skip it. + # An example of this happening is when the user changes their display name. continue membership_change = RoomsForUserStateReset( From c9a6a0bca0f1f45368bcdd782b16df1f81f390c5 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 16:37:44 -0600 Subject: [PATCH 24/31] Add better index on sliding_sync_membership_snapshots --- synapse/storage/databases/main/stream.py | 6 ------ .../delta/91/02_ss_membership_snapshot_idx.sql | 16 ++++++++++++++++ 2 files changed, 16 insertions(+), 6 deletions(-) create mode 100644 synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 9c7736417c2..7e4c4537473 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1170,12 +1170,6 @@ def f(txn: LoggingTransaction) -> Dict[str, RoomsForUserStateReset]: # `sliding_sync_membership_snapshots` which will not include users # that were state reset out of rooms; so we need to look for that # case in `current_state_delta_stream`. - # - # TODO: Add an index a better index on sliding_sync_membership_snapshots - # probably want: - # - sliding_sync_membership_snapshots (user_id, event_stream_ordering) - # replacing the existing index on only (user_id) - # - MAYBE current_state_delta_stream(state_key, stream_id) WHERE type = 'm.room.member' AND event_id IS NULL sql = """ SELECT room_id, diff --git a/synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql b/synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql new file mode 100644 index 00000000000..c694203f95e --- /dev/null +++ b/synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql @@ -0,0 +1,16 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2025 New Vector, 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: +-- . + +-- So we can fetch all rooms for a given user sorted by stream order +DROP INDEX IF EXISTS sliding_sync_membership_snapshots_user_id; +CREATE INDEX IF NOT EXISTS sliding_sync_membership_snapshots_user_id ON sliding_sync_membership_snapshots(user_id, event_stream_ordering); From 096bbfa1ea5caccd8fb17539e8f44d73fd6e8340 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 17:00:33 -0600 Subject: [PATCH 25/31] Add docstring to new func --- synapse/storage/databases/main/stream.py | 30 ++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 7e4c4537473..4a8d33c7915 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1145,6 +1145,36 @@ async def get_sliding_sync_membership_changes( to_key: RoomStreamToken, excluded_room_ids: Optional[AbstractSet[str]] = None, ) -> Dict[str, RoomsForUserStateReset]: + """ + Fetch membership events that result in a meaningful membership change for a + given user. + + A meaningful membership changes is one where the `membership` value actually + changes. This means memberships changes from `join` to `join` will be filtered + out since they result in no meaningful change. + + Note: This function only works with "live" tokens with `stream_ordering` only. + + We're looking for membership changes in the token range (> `from_key` and <= + `to_key`). + + Args: + user_id: The user ID to fetch membership events for. + from_key: The point in the stream to sync from (fetching events > this point). + to_key: The token to fetch rooms up to (fetching events <= this point). + excluded_room_ids: Optional list of room IDs to exclude from the results. + + Returns: + All meaningful membership changes to the current state in the token range. + Events are sorted by `stream_ordering` ascending. + + `event_id`/`sender` can be `None` when the server leaves a room (meaning + everyone locally left) or a state reset which removed the person from the + room. We can't tell the difference between the two cases with what's + available in the `current_state_delta_stream` table. To actually check for a + state reset, you need to check if a membership still exists in the room. + """ + # Start by ruling out cases where a DB query is not necessary. if from_key == to_key: return {} From ed7c5807738dcbcbf7d4d19a30edf477e4bcd73d Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Thu, 1 May 2025 17:20:39 -0600 Subject: [PATCH 26/31] Readd old impl for fallback tables case --- synapse/handlers/sliding_sync/room_lists.py | 136 +++++++++++++++++- .../client/sliding_sync/test_sliding_sync.py | 43 +++--- 2 files changed, 158 insertions(+), 21 deletions(-) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index e4eec771438..8d3687529bc 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -1040,7 +1040,7 @@ async def get_room_membership_for_user_at_to_token( ( newly_joined_room_ids, newly_left_room_map, - ) = await self._get_newly_joined_and_left_rooms( + ) = await self._get_newly_joined_and_left_rooms_fallback( user_id, to_token=to_token, from_token=from_token ) @@ -1129,6 +1129,140 @@ async def _get_newly_joined_and_left_rooms( return newly_joined_room_ids, newly_left_room_map + @trace + async def _get_newly_joined_and_left_rooms_fallback( + self, + user_id: str, + to_token: StreamToken, + from_token: Optional[StreamToken], + ) -> Tuple[AbstractSet[str], Mapping[str, RoomsForUserStateReset]]: + """Fetch the sets of rooms that the user newly joined or left in the + given token range. + Note: there may be rooms in the newly left rooms where the user was + "state reset" out of the room, and so that room would not be part of the + "current memberships" of the user. + Returns: + A 2-tuple of newly joined room IDs and a map of newly_left room + IDs to the `RoomsForUserStateReset` entry. + We're using `RoomsForUserStateReset` but that doesn't necessarily mean the + user was state reset of the rooms. It's just that the `event_id`/`sender` + are optional and we can't tell the difference between the server leaving the + room when the user was the last person participating in the room and left or + was state reset out of the room. To actually check for a state reset, you + need to check if a membership still exists in the room. + """ + newly_joined_room_ids: Set[str] = set() + newly_left_room_map: Dict[str, RoomsForUserStateReset] = {} + + # We need to figure out the + # + # - 1) Figure out which rooms are `newly_left` rooms (> `from_token` and <= `to_token`) + # - 2) Figure out which rooms are `newly_joined` (> `from_token` and <= `to_token`) + + # 1) Fetch membership changes that fall in the range from `from_token` up to `to_token` + current_state_delta_membership_changes_in_from_to_range = [] + if from_token: + current_state_delta_membership_changes_in_from_to_range = ( + await self.store.get_current_state_delta_membership_changes_for_user( + user_id, + from_key=from_token.room_key, + to_key=to_token.room_key, + excluded_room_ids=self.rooms_to_exclude_globally, + ) + ) + + # 1) Assemble a list of the last membership events in some given ranges. Someone + # could have left and joined multiple times during the given range but we only + # care about end-result so we grab the last one. + last_membership_change_by_room_id_in_from_to_range: Dict[ + str, CurrentStateDeltaMembership + ] = {} + # We also want to assemble a list of the first membership events during the token + # range so we can step backward to the previous membership that would apply to + # before the token range to see if we have `newly_joined` the room. + first_membership_change_by_room_id_in_from_to_range: Dict[ + str, CurrentStateDeltaMembership + ] = {} + # Keep track if the room has a non-join event in the token range so we can later + # tell if it was a `newly_joined` room. If the last membership event in the + # token range is a join and there is also some non-join in the range, we know + # they `newly_joined`. + has_non_join_event_by_room_id_in_from_to_range: Dict[str, bool] = {} + for ( + membership_change + ) in current_state_delta_membership_changes_in_from_to_range: + room_id = membership_change.room_id + + last_membership_change_by_room_id_in_from_to_range[room_id] = ( + membership_change + ) + # Only set if we haven't already set it + first_membership_change_by_room_id_in_from_to_range.setdefault( + room_id, membership_change + ) + + if membership_change.membership != Membership.JOIN: + has_non_join_event_by_room_id_in_from_to_range[room_id] = True + + # 1) Fixup + # + # 2) We also want to assemble a list of possibly newly joined rooms. Someone + # could have left and joined multiple times during the given range but we only + # care about whether they are joined at the end of the token range so we are + # working with the last membership even in the token range. + possibly_newly_joined_room_ids = set() + for ( + last_membership_change_in_from_to_range + ) in last_membership_change_by_room_id_in_from_to_range.values(): + room_id = last_membership_change_in_from_to_range.room_id + + # 2) + if last_membership_change_in_from_to_range.membership == Membership.JOIN: + possibly_newly_joined_room_ids.add(room_id) + + # 1) Figure out newly_left rooms (> `from_token` and <= `to_token`). + if last_membership_change_in_from_to_range.membership == Membership.LEAVE: + # 1) Mark this room as `newly_left` + newly_left_room_map[room_id] = RoomsForUserStateReset( + room_id=room_id, + sender=last_membership_change_in_from_to_range.sender, + membership=Membership.LEAVE, + event_id=last_membership_change_in_from_to_range.event_id, + event_pos=last_membership_change_in_from_to_range.event_pos, + room_version_id=await self.store.get_room_version_id(room_id), + ) + + # 2) Figure out `newly_joined` + for room_id in possibly_newly_joined_room_ids: + has_non_join_in_from_to_range = ( + has_non_join_event_by_room_id_in_from_to_range.get(room_id, False) + ) + # If the last membership event in the token range is a join and there is + # also some non-join in the range, we know they `newly_joined`. + if has_non_join_in_from_to_range: + # We found a `newly_joined` room (we left and joined within the token range) + newly_joined_room_ids.add(room_id) + else: + prev_event_id = first_membership_change_by_room_id_in_from_to_range[ + room_id + ].prev_event_id + prev_membership = first_membership_change_by_room_id_in_from_to_range[ + room_id + ].prev_membership + + if prev_event_id is None: + # We found a `newly_joined` room (we are joining the room for the + # first time within the token range) + newly_joined_room_ids.add(room_id) + # Last resort, we need to step back to the previous membership event + # just before the token range to see if we're joined then or not. + elif prev_membership != Membership.JOIN: + # We found a `newly_joined` room (we left before the token range + # and joined within the token range) + newly_joined_room_ids.add(room_id) + + return newly_joined_room_ids, newly_left_room_map + @trace async def _get_dm_rooms_for_user( self, diff --git a/tests/rest/client/sliding_sync/test_sliding_sync.py b/tests/rest/client/sliding_sync/test_sliding_sync.py index 1a5a152f23d..dcec5b4cf05 100644 --- a/tests/rest/client/sliding_sync/test_sliding_sync.py +++ b/tests/rest/client/sliding_sync/test_sliding_sync.py @@ -824,26 +824,29 @@ def test_reject_remote_invite(self) -> None: # Sync again after rejecting the invite response_body, _ = self.do_sync(sync_body, since=from_token, tok=user_tok) - # We should see the newly_left room - self.assertIncludes( - set(response_body["lists"]["foo-list"]["ops"][0]["room_ids"]), - {room_id}, - exact=True, - ) - # We should see the leave state for the room so clients don't end up with stuck - # invites - self.assertIncludes( - { - ( - state["type"], - state["state_key"], - state["content"].get("membership"), - ) - for state in response_body["rooms"][room_id]["required_state"] - }, - {(EventTypes.Member, user_id, Membership.LEAVE)}, - exact=True, - ) + # The fix to add the leave event to incremental sync when rejecting a remote + # invite relies on the new tables to work. + if self.use_new_tables: + # We should see the newly_left room + self.assertIncludes( + set(response_body["lists"]["foo-list"]["ops"][0]["room_ids"]), + {room_id}, + exact=True, + ) + # We should see the leave state for the room so clients don't end up with stuck + # invites + self.assertIncludes( + { + ( + state["type"], + state["state_key"], + state["content"].get("membership"), + ) + for state in response_body["rooms"][room_id]["required_state"] + }, + {(EventTypes.Member, user_id, Membership.LEAVE)}, + exact=True, + ) def test_ignored_user_invites_initial_sync(self) -> None: """ From 14c4e4de78b30b33c87a86d9d31f0caf892b8225 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Fri, 2 May 2025 10:56:31 -0600 Subject: [PATCH 27/31] Add newlines to docstring --- synapse/handlers/sliding_sync/room_lists.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/synapse/handlers/sliding_sync/room_lists.py b/synapse/handlers/sliding_sync/room_lists.py index 8d3687529bc..2272bcff1f2 100644 --- a/synapse/handlers/sliding_sync/room_lists.py +++ b/synapse/handlers/sliding_sync/room_lists.py @@ -1138,12 +1138,15 @@ async def _get_newly_joined_and_left_rooms_fallback( ) -> Tuple[AbstractSet[str], Mapping[str, RoomsForUserStateReset]]: """Fetch the sets of rooms that the user newly joined or left in the given token range. + Note: there may be rooms in the newly left rooms where the user was "state reset" out of the room, and so that room would not be part of the "current memberships" of the user. + Returns: A 2-tuple of newly joined room IDs and a map of newly_left room IDs to the `RoomsForUserStateReset` entry. + We're using `RoomsForUserStateReset` but that doesn't necessarily mean the user was state reset of the rooms. It's just that the `event_id`/`sender` are optional and we can't tell the difference between the server leaving the From 2c3f74f04327a497f199939810f2591fe74ab6e1 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Fri, 2 May 2025 10:58:10 -0600 Subject: [PATCH 28/31] Expand docstring --- synapse/storage/databases/main/stream.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 4a8d33c7915..363c42664a5 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -1150,8 +1150,8 @@ async def get_sliding_sync_membership_changes( given user. A meaningful membership changes is one where the `membership` value actually - changes. This means memberships changes from `join` to `join` will be filtered - out since they result in no meaningful change. + changes. This means memberships changes from `join` to `join` (like a display + name change) will be filtered out since they result in no meaningful change. Note: This function only works with "live" tokens with `stream_ordering` only. From 1f74e918efb9a2830155b4b695bdd4d78fb12c7d Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Fri, 2 May 2025 14:21:00 -0600 Subject: [PATCH 29/31] Assert topological tokens not used --- synapse/storage/databases/main/stream.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/synapse/storage/databases/main/stream.py b/synapse/storage/databases/main/stream.py index 363c42664a5..daf2fbc756c 100644 --- a/synapse/storage/databases/main/stream.py +++ b/synapse/storage/databases/main/stream.py @@ -992,6 +992,10 @@ async def get_current_state_delta_membership_changes_for_user( available in the `current_state_delta_stream` table. To actually check for a state reset, you need to check if a membership still exists in the room. """ + + assert from_key.topological is None + assert to_key.topological is None + # Start by ruling out cases where a DB query is not necessary. if from_key == to_key: return [] @@ -1175,6 +1179,9 @@ async def get_sliding_sync_membership_changes( state reset, you need to check if a membership still exists in the room. """ + assert from_key.topological is None + assert to_key.topological is None + # Start by ruling out cases where a DB query is not necessary. if from_key == to_key: return {} From 5ad092ee308d55486fb9e61681441c84512fbb00 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Fri, 2 May 2025 16:17:19 -0600 Subject: [PATCH 30/31] Add FIXME comment to update tests to run against new tables --- tests/handlers/test_sliding_sync.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/handlers/test_sliding_sync.py b/tests/handlers/test_sliding_sync.py index 5b7e2937f83..5ea0c9ff1ef 100644 --- a/tests/handlers/test_sliding_sync.py +++ b/tests/handlers/test_sliding_sync.py @@ -578,6 +578,12 @@ class GetRoomMembershipForUserAtToTokenTestCase(HomeserverTestCase): the correct list of rooms IDs. """ + # FIXME: We should refactor these tests to run against `compute_interested_rooms(...)` + # instead of just `get_room_membership_for_user_at_to_token(...)` which is only used + # in the fallback path (`_compute_interested_rooms_fallback(...)`). These scenarios do + # well to stress that logic and we shouldn't remove them just because we're removing + # the fallback path (tracked by https://github.com/element-hq/synapse/issues/17623). + servlets = [ admin.register_servlets, knock.register_servlets, @@ -2381,6 +2387,12 @@ class GetRoomMembershipForUserAtToTokenShardTestCase(BaseMultiWorkerStreamTestCa sharded event stream_writers enabled """ + # FIXME: We should refactor these tests to run against `compute_interested_rooms(...)` + # instead of just `get_room_membership_for_user_at_to_token(...)` which is only used + # in the fallback path (`_compute_interested_rooms_fallback(...)`). These scenarios do + # well to stress that logic and we shouldn't remove them just because we're removing + # the fallback path (tracked by https://github.com/element-hq/synapse/issues/17623). + servlets = [ admin.register_servlets_for_client_rest_resource, room.register_servlets, From ea3c9fc88aa6e4411bc595c93e3c62600867548e Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Wed, 7 May 2025 09:13:24 -0600 Subject: [PATCH 31/31] Move db delta to new schema version --- .../03_ss_membership_snapshot_idx.sql} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename synapse/storage/schema/main/delta/{91/02_ss_membership_snapshot_idx.sql => 92/03_ss_membership_snapshot_idx.sql} (100%) diff --git a/synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql b/synapse/storage/schema/main/delta/92/03_ss_membership_snapshot_idx.sql similarity index 100% rename from synapse/storage/schema/main/delta/91/02_ss_membership_snapshot_idx.sql rename to synapse/storage/schema/main/delta/92/03_ss_membership_snapshot_idx.sql