diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index 1dd114c601..2b91f1f67c 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -1530,12 +1530,13 @@ class PersistEventsStore: room_id, # Even though `Mapping`/`Dict` have no guaranteed order, some # implementations may preserve insertion order so we're just going to - # choose the best possible answer by using the "last" event ID which we + # choose the best possible answer by using the "first" event ID which we # will assume will have the greatest `stream_ordering`. We really just # need *some* answer in case we are the first ones inserting into the - # table and this will resolve itself when we update this field in the - # persist events loop. - list(to_insert.values())[-1], + # table and in reality, `_store_event_txn()` is run before this function + # so it will already have the correct value. This is just to account for + # things changing in the future. + next(iter(to_insert.values())), ] # If we have a `bump_event_id`, let's update the `bump_stamp` column bump_stamp_column = "" @@ -1977,6 +1978,64 @@ class PersistEventsStore: ], ) + # Handle updating `sliding_sync_joined_rooms` + room_id_to_stream_ordering_map: Dict[str, int] = {} + room_id_to_bump_stamp_map: Dict[str, int] = {} + for event, _ in events_and_contexts: + existing_stream_ordering = room_id_to_stream_ordering_map.get(event.room_id) + # This should exist at this point because we're inserting events here which require it + assert event.internal_metadata.stream_ordering is not None + if ( + existing_stream_ordering is None + or existing_stream_ordering < event.internal_metadata.stream_ordering + ): + room_id_to_stream_ordering_map[event.room_id] = ( + event.internal_metadata.stream_ordering + ) + + if event.type in SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES: + existing_bump_stamp = room_id_to_bump_stamp_map.get(event.room_id) + # This should exist at this point because we're inserting events here which require it + assert event.internal_metadata.stream_ordering is not None + if ( + existing_bump_stamp is None + or existing_bump_stamp < event.internal_metadata.stream_ordering + ): + room_id_to_bump_stamp_map[event.room_id] = ( + event.internal_metadata.stream_ordering + ) + + # `_store_event_txn` is run before `_update_current_state_txn` which handles + # deleting the rows if we are no longer in the room so we don't need to worry + # about inserting something that will be orphaned. + self.db_pool.simple_upsert_many_txn( + txn, + table="sliding_sync_joined_rooms", + key_names=("room_id",), + key_values=[ + (room_id,) for room_id in room_id_to_stream_ordering_map.keys() + ], + value_names=("event_stream_ordering",), + value_values=[ + (room_id_to_stream_ordering_map[room_id],) + for room_id in room_id_to_stream_ordering_map.keys() + ], + ) + # This has to be separate from the upsert above because we won't have a + # `bump_stamp` for every event and we don't want to overwrite the existing value + # with `None`. + self.db_pool.simple_upsert_many_txn( + txn, + table="sliding_sync_joined_rooms", + key_names=("room_id",), + key_values=[(room_id,) for room_id in room_id_to_bump_stamp_map.keys()], + value_names=("bump_stamp",), + value_values=[ + (room_id_to_bump_stamp_map[room_id],) + for room_id in room_id_to_bump_stamp_map.keys() + ], + ) + def _store_rejected_events_txn( self, txn: LoggingTransaction, diff --git a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql index 61ea65aba2..a60fabb586 100644 --- a/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql +++ b/synapse/storage/schema/main/delta/87/01_sliding_sync_memberships.sql @@ -31,7 +31,16 @@ CREATE TABLE IF NOT EXISTS sliding_sync_joined_rooms( PRIMARY KEY (room_id) ); - +-- Store the user's non-join room memberships. Only stores the latest membership event +-- for a given user which matches `local_current_membership` (except we don't store +-- joins). +-- +-- FIXME: It might be easier to just store any membership here but just indicate that +-- the state is a snapshot of the current state at the time of the membership event. +-- That way we don't have to worry about clearing out +-- `sliding_sync_non_join_memberships` when the the user joins a room. And we always +-- have the full picture. Perhaps we can call it `sliding_sync_membership_snapshots`. +-- -- We don't include `bump_stamp` here because we can just use the `stream_ordering` from -- the membership event itself as the `bump_stamp`. CREATE TABLE IF NOT EXISTS sliding_sync_non_join_memberships( @@ -50,6 +59,8 @@ CREATE TABLE IF NOT EXISTS sliding_sync_non_join_memberships( -- `m.room.encryption` -> `content.algorithm` (according to the current state at the -- time of the membership) is_encrypted BOOLEAN DEFAULT 0 NOT NULL, + -- FIXME: Maybe we want to add `tombstone_successor_room_id` here to help with `include_old_rooms` + -- (tracked by https://github.com/element-hq/synapse/issues/17540) PRIMARY KEY (room_id, user_id) ); diff --git a/tests/storage/test_events.py b/tests/storage/test_events.py index 9543123473..b8302a7928 100644 --- a/tests/storage/test_events.py +++ b/tests/storage/test_events.py @@ -617,8 +617,9 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): room_id1 = self.helper.create_room_as(user1_id, tok=user1_tok) - # User1 joins the room - self.helper.join(room_id1, user1_id, tok=user1_tok) + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() self.assertIncludes( @@ -630,15 +631,20 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): sliding_sync_joined_rooms_results[room_id1], _SlidingSyncJoinedRoomResult( room_id=room_id1, - # TODO - event_stream_ordering=None, - bump_stamp=None, + # History visibility just happens to be the last event sent in the room + event_stream_ordering=state_map[ + (EventTypes.RoomHistoryVisibility, "") + ].internal_metadata.stream_ordering, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, room_type=None, room_name=None, is_encrypted=False, ), ) + # No one is non-joined to this room so we shouldn't see anything sliding_sync_non_join_memberships_results = ( self._get_sliding_sync_non_join_memberships() ) @@ -674,7 +680,14 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): ) # User1 joins the room - self.helper.join(room_id1, user1_id, tok=user1_tok) + user1_join_response = self.helper.join(room_id1, user1_id, tok=user1_tok) + user1_join_event_pos = self.get_success( + self.store.get_position_for_event(user1_join_response["event_id"]) + ) + + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() self.assertIncludes( @@ -686,9 +699,10 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): sliding_sync_joined_rooms_results[room_id1], _SlidingSyncJoinedRoomResult( room_id=room_id1, - # TODO - event_stream_ordering=None, - bump_stamp=None, + event_stream_ordering=user1_join_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, room_type=None, room_name="my super duper room", is_encrypted=True, @@ -704,100 +718,6 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): exact=True, ) - def test_joined_room_with_info_updated(self) -> None: - """ - Test info in `sliding_sync_joined_rooms` is updated when the current state is - updated. - """ - user1_id = self.register_user("user1", "pass") - user1_tok = self.login(user1_id, "pass") - user2_id = self.register_user("user2", "pass") - user2_tok = self.login(user2_id, "pass") - - room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok) - # Add a room name - self.helper.send_state( - room_id1, - EventTypes.Name, - {"name": "my super duper room"}, - tok=user2_tok, - ) - - # User1 joins the room - self.helper.join(room_id1, user1_id, tok=user1_tok) - - sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() - self.assertIncludes( - set(sliding_sync_joined_rooms_results.keys()), - {room_id1}, - exact=True, - ) - self.assertEqual( - sliding_sync_joined_rooms_results[room_id1], - _SlidingSyncJoinedRoomResult( - room_id=room_id1, - # TODO - event_stream_ordering=None, - bump_stamp=None, - room_type=None, - room_name="my super duper room", - is_encrypted=False, - ), - ) - - sliding_sync_non_join_memberships_results = ( - self._get_sliding_sync_non_join_memberships() - ) - self.assertIncludes( - set(sliding_sync_non_join_memberships_results.keys()), - set(), - exact=True, - ) - - # Update the room name - self.helper.send_state( - room_id1, - EventTypes.Name, - {"name": "my super duper room was renamed"}, - tok=user2_tok, - ) - # Encrypt the room - self.helper.send_state( - room_id1, - EventTypes.RoomEncryption, - {EventContentFields.ENCRYPTION_ALGORITHM: "m.megolm.v1.aes-sha2"}, - tok=user2_tok, - ) - - sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() - self.assertIncludes( - set(sliding_sync_joined_rooms_results.keys()), - {room_id1}, - exact=True, - ) - # Make sure we see the new room name - self.assertEqual( - sliding_sync_joined_rooms_results[room_id1], - _SlidingSyncJoinedRoomResult( - room_id=room_id1, - # TODO - event_stream_ordering=None, - bump_stamp=None, - room_type=None, - room_name="my super duper room was renamed", - is_encrypted=True, - ), - ) - - sliding_sync_non_join_memberships_results = ( - self._get_sliding_sync_non_join_memberships() - ) - self.assertIncludes( - set(sliding_sync_non_join_memberships_results.keys()), - set(), - exact=True, - ) - def test_joined_space_room_with_info(self) -> None: """ Test joined space room with name shows up in `sliding_sync_joined_rooms`. @@ -823,7 +743,14 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): ) # User1 joins the room - self.helper.join(space_room_id, user1_id, tok=user1_tok) + user1_join_response = self.helper.join(space_room_id, user1_id, tok=user1_tok) + user1_join_event_pos = self.get_success( + self.store.get_position_for_event(user1_join_response["event_id"]) + ) + + state_map = self.get_success( + self.storage_controllers.state.get_current_state(space_room_id) + ) sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() self.assertIncludes( @@ -835,9 +762,10 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): sliding_sync_joined_rooms_results[space_room_id], _SlidingSyncJoinedRoomResult( room_id=space_room_id, - # TODO - event_stream_ordering=None, - bump_stamp=None, + event_stream_ordering=user1_join_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, room_type=RoomTypes.SPACE, room_name="my super duper space", is_encrypted=False, @@ -853,6 +781,207 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): exact=True, ) + def test_joined_room_with_state_updated(self) -> None: + """ + Test state derived info in `sliding_sync_joined_rooms` is updated when the + current state is updated. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok) + # Add a room name + self.helper.send_state( + room_id1, + EventTypes.Name, + {"name": "my super duper room"}, + tok=user2_tok, + ) + + # User1 joins the room + user1_join_response = self.helper.join(room_id1, user1_id, tok=user1_tok) + user1_join_event_pos = self.get_success( + self.store.get_position_for_event(user1_join_response["event_id"]) + ) + + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) + + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + {room_id1}, + exact=True, + ) + self.assertEqual( + sliding_sync_joined_rooms_results[room_id1], + _SlidingSyncJoinedRoomResult( + room_id=room_id1, + event_stream_ordering=user1_join_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, + room_type=None, + room_name="my super duper room", + is_encrypted=False, + ), + ) + + sliding_sync_non_join_memberships_results = ( + self._get_sliding_sync_non_join_memberships() + ) + self.assertIncludes( + set(sliding_sync_non_join_memberships_results.keys()), + set(), + exact=True, + ) + + # Update the room name + self.helper.send_state( + room_id1, + EventTypes.Name, + {"name": "my super duper room was renamed"}, + tok=user2_tok, + ) + # Encrypt the room + encrypt_room_response = self.helper.send_state( + room_id1, + EventTypes.RoomEncryption, + {EventContentFields.ENCRYPTION_ALGORITHM: "m.megolm.v1.aes-sha2"}, + tok=user2_tok, + ) + encrypt_room_event_pos = self.get_success( + self.store.get_position_for_event(encrypt_room_response["event_id"]) + ) + + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + {room_id1}, + exact=True, + ) + # Make sure we see the new room name + self.assertEqual( + sliding_sync_joined_rooms_results[room_id1], + _SlidingSyncJoinedRoomResult( + room_id=room_id1, + event_stream_ordering=encrypt_room_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, + room_type=None, + room_name="my super duper room was renamed", + is_encrypted=True, + ), + ) + + sliding_sync_non_join_memberships_results = ( + self._get_sliding_sync_non_join_memberships() + ) + self.assertIncludes( + set(sliding_sync_non_join_memberships_results.keys()), + set(), + exact=True, + ) + + def test_joined_room_is_bumped(self) -> None: + """ + Test that `event_stream_ordering` and `bump_stamp` is updated when a new bump + event is sent (`sliding_sync_joined_rooms`). + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id1 = self.helper.create_room_as(user2_id, tok=user2_tok) + # Add a room name + self.helper.send_state( + room_id1, + EventTypes.Name, + {"name": "my super duper room"}, + tok=user2_tok, + ) + + # User1 joins the room + user1_join_response = self.helper.join(room_id1, user1_id, tok=user1_tok) + user1_join_event_pos = self.get_success( + self.store.get_position_for_event(user1_join_response["event_id"]) + ) + + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) + + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + {room_id1}, + exact=True, + ) + self.assertEqual( + sliding_sync_joined_rooms_results[room_id1], + _SlidingSyncJoinedRoomResult( + room_id=room_id1, + event_stream_ordering=user1_join_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, + room_type=None, + room_name="my super duper room", + is_encrypted=False, + ), + ) + + sliding_sync_non_join_memberships_results = ( + self._get_sliding_sync_non_join_memberships() + ) + self.assertIncludes( + set(sliding_sync_non_join_memberships_results.keys()), + set(), + exact=True, + ) + + # Send a new message to bump the room + event_response = self.helper.send(room_id1, "some message", tok=user1_tok) + event_pos = self.get_success( + self.store.get_position_for_event(event_response["event_id"]) + ) + + sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() + self.assertIncludes( + set(sliding_sync_joined_rooms_results.keys()), + {room_id1}, + exact=True, + ) + # Make sure we see the new room name + self.assertEqual( + sliding_sync_joined_rooms_results[room_id1], + _SlidingSyncJoinedRoomResult( + room_id=room_id1, + # Updated `event_stream_ordering` + event_stream_ordering=event_pos.stream, + # And since the event was a bump event, the `bump_stamp` should be updated + bump_stamp=event_pos.stream, + # The state is still the same (it didn't change) + room_type=None, + room_name="my super duper room", + is_encrypted=False, + ), + ) + + sliding_sync_non_join_memberships_results = ( + self._get_sliding_sync_non_join_memberships() + ) + self.assertIncludes( + set(sliding_sync_non_join_memberships_results.keys()), + set(), + exact=True, + ) + # TODO: test_joined_room_state_reset def test_non_join_space_room_with_info(self) -> None: @@ -896,12 +1025,19 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): # Update the room name after we are invited just to make sure # we don't update non-join memberships when the room name changes. - self.helper.send_state( + rename_response = self.helper.send_state( space_room_id, EventTypes.Name, {"name": "my super duper space was renamed"}, tok=user2_tok, ) + rename_event_pos = self.get_success( + self.store.get_position_for_event(rename_response["event_id"]) + ) + + state_map = self.get_success( + self.storage_controllers.state.get_current_state(space_room_id) + ) # User2 is still joined to the room so we should still have an entry in the # `sliding_sync_joined_rooms` table. @@ -915,9 +1051,10 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): sliding_sync_joined_rooms_results[space_room_id], _SlidingSyncJoinedRoomResult( room_id=space_room_id, - # TODO - event_stream_ordering=None, - bump_stamp=None, + event_stream_ordering=rename_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, room_type=RoomTypes.SPACE, room_name="my super duper space was renamed", is_encrypted=True, @@ -981,6 +1118,10 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): self.store.get_position_for_event(user3_ban_response["event_id"]) ) + state_map = self.get_success( + self.storage_controllers.state.get_current_state(room_id1) + ) + # User2 is still joined to the room so we should still have an entry # in the `sliding_sync_joined_rooms` table. sliding_sync_joined_rooms_results = self._get_sliding_sync_joined_rooms() @@ -993,9 +1134,10 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): sliding_sync_joined_rooms_results[room_id1], _SlidingSyncJoinedRoomResult( room_id=room_id1, - # TODO - event_stream_ordering=None, - bump_stamp=None, + event_stream_ordering=user3_ban_event_pos.stream, + bump_stamp=state_map[ + (EventTypes.Create, "") + ].internal_metadata.stream_ordering, room_type=None, room_name=None, is_encrypted=False, @@ -1042,6 +1184,8 @@ class SlidingSyncPrePopulatedTablesTestCase(HomeserverTestCase): # TODO: Test remote invite + # TODO Test for non-join membership changing + def test_non_join_server_left_room(self) -> None: """ Test everyone local leaves the room but their leave membership still shows up in