diff --git a/changelog.d/20203.bugfix b/changelog.d/20203.bugfix new file mode 100644 index 0000000000..125afaad56 --- /dev/null +++ b/changelog.d/20203.bugfix @@ -0,0 +1 @@ +Tell clients to remove local profile information for users after leaving the last room shared with them. \ No newline at end of file diff --git a/synapse/storage/databases/main/events.py b/synapse/storage/databases/main/events.py index b628b7fed7..ace4582fd9 100644 --- a/synapse/storage/databases/main/events.py +++ b/synapse/storage/databases/main/events.py @@ -2219,7 +2219,13 @@ class PersistEventsStore: """ Record updates into the profile updates stream for when a user leaves a room. - If this was the last shared room with a set of users, clear all old rows from + This handles two distinct cases when a user leaves a room: + 1) we find users in the the room who no longer share rooms with the user that + left the room, and record a `LEFT_ROOM` action for them. + 2) we check for the user who left the room if they no longer share rooms with + some users of the room that was left, and do the same in reverse. + + In both cases, when recording a `LEFT_ROOM` action, we clear all old rows from the `profile_updates_per_user` table relating to those users, to avoid exposing any profile field changes past the point of not being in any common rooms with the user. @@ -2271,15 +2277,39 @@ class PersistEventsStore: (*user_args, user_id.to_string()), ) - # Now record the "left room" action in the stream + # Now record the "left room" action in the stream for each user + # in the room that no longer shares a room with the user who left the room. self.store.record_profile_updates_txn( txn=txn, - user_id=user_id, + users={user_id.to_string()}, action=ProfileUpdateAction.LEFT_ROOM, field_names=[], target_users=users_no_longer_sharing_rooms, ) + # We also need to record things in reverse. The user, who left the + # room, needs to get profile update rows for every user they no longer + # share a room with. + # First clear old rows between these users. + txn.execute( + f""" + DELETE FROM profile_updates_per_user + WHERE user_id = ? + AND stream_id IN ( + SELECT stream_id FROM profile_updates WHERE {user_clause} + ) + """, + (user_id.to_string(), *user_args), + ) + # Then add the left room action rows in the stream. + self.store.record_profile_updates_txn( + txn=txn, + users=users_no_longer_sharing_rooms, + action=ProfileUpdateAction.LEFT_ROOM, + field_names=[], + target_users={user_id.to_string()}, + ) + @classmethod def _get_relevant_sliding_sync_current_state_event_ids_txn( cls, txn: LoggingTransaction, room_id: str diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index 7f860258ee..8ed0d645bc 100644 --- a/synapse/storage/databases/main/profile.py +++ b/synapse/storage/databases/main/profile.py @@ -854,7 +854,7 @@ class ProfileWorkerStore(SQLBaseStore): # Record updates in the profile updates stream stream_id = self.record_profile_updates_txn( txn=txn, - user_id=user_id, + users={user_id.to_string()}, action=ProfileUpdateAction.UPDATE, field_names=[field_name], ) @@ -885,42 +885,57 @@ class ProfileWorkerStore(SQLBaseStore): # Ensure we're working with local users only users = {user_id for user_id in joined_users if self.hs.is_mine_id(user_id)} + # Get the members of the room + rows = self.db_pool.simple_select_list_txn( + txn=txn, + table="local_current_membership", + keyvalues={ + "room_id": room_id, + "membership": Membership.JOIN, + }, + retcols=("user_id",), + ) + target_users = {row[0] for row in rows} + # Record the profile updates for each user - for user_id in users: - self.record_profile_updates_txn( - txn=txn, - user_id=UserID.from_string(user_id), - action=ProfileUpdateAction.JOINED_ROOM, - field_names=None, - user_rooms={room_id}, - ) + self.record_profile_updates_txn( + txn=txn, + users=users, + action=ProfileUpdateAction.JOINED_ROOM, + target_users=target_users, + field_names=None, + ) def record_profile_updates_txn( self, *, txn: LoggingTransaction, - user_id: UserID, + users: set[str], action: ProfileUpdateAction, field_names: Collection[str] | None, - user_rooms: set[str] | None = None, target_users: set[str] | None = None, + user_rooms: set[str] | None = None, ) -> int | None: """ Record updates into the profile updates stream tables. + If `target`_users` is not given as a parameter, `users` must be a single user. + Currently, updates are only recorded for local users. Args: txn: Transaction to use - user_id: User ID that made the profile update + users: A set of user IDs to write the profile updates for. action: The profile update action, either `update`, `left_room` or `joined_room`. field_names: A list of fields that were set, if ProfileUpdateAction.UPDATE + target_users: Optionally, set of users to create per user profile update + stream rows for. If not given, and the length of `users` is only + a single user, a database lookup will be done based on + `user_rooms`, or if that is not set, the result of the rooms lookup. user_rooms: Optionally, a set of rooms that the update concerns. If not given, a database lookup will be done to fetch all the users rooms. - target_users: Optionally, set of users to create profile update stream rows - for. If not given, a database lookup will be done based on `user_rooms`, - or if that is not set, the result of the rooms lookup. + Only used if `target_users` is not given. Returns: The latest stream ID created in this transaction @@ -933,15 +948,20 @@ class ProfileWorkerStore(SQLBaseStore): else: assert not field_names - if not target_users: - if not user_rooms: + if target_users is None: + # This function must be called with one user only if it needs to + # compute the target users. This restriction mainly exists as a + # fail safe to ensure we don't abuse the loop of membership fetches here + # and ensure calling code makes the necessary optimizations. + assert len(users) == 1 + if user_rooms is None: rows = self.db_pool.simple_select_onecol_txn( txn=txn, table="current_state_events", keyvalues={ "type": EventTypes.Member, "membership": Membership.JOIN, - "state_key": user_id.to_string(), + "state_key": list(users)[0], }, retcol="room_id", ) @@ -959,40 +979,97 @@ class ProfileWorkerStore(SQLBaseStore): ) target_users = {row[0] for row in rows} + if action == ProfileUpdateAction.UPDATE: + # Always include ourselves when updating field values. + # We need to do this as the users updating their profile may not be + # in any rooms, and thus wont be collected above, but should still get the + # update pushed to their other devices. + target_users = target_users.union(users) + # Ensure we only write updates for local users - users = {user for user in target_users if self.hs.is_mine_id(user)} + target_users = {user for user in target_users if self.hs.is_mine_id(user)} - if action in (ProfileUpdateAction.JOINED_ROOM, ProfileUpdateAction.LEFT_ROOM): - users.discard(user_id.to_string()) - if not users: - # No point writing an update for ourselves, if a membership change and no - # other users interested - return None - elif action == ProfileUpdateAction.UPDATE: - # Always include ourselves when updating field values - users.add(user_id.to_string()) + if not target_users: + return None - # Record the profile update inserted_ts = self.clock.time_msec() - stream_id = self._profile_updates_id_gen.get_next_txn(txn) + profile_update_values = {} + sorted_field_names = ( + json_encoder.encode(sorted(field_names)) if field_names else None + ) - self.db_pool.simple_insert_txn( + # Collect profile updates to add + # We don't have stream ID's at this point, so just use a counter, and add in + # the stream ID's later. We do this here to avoid generating stream ID's we're + # not going to use, because here we'll be dropping any updates which only + # contain the user themselves, in a "joined room" or "left room" situation. If + # this call contains multiple users for "joined room" or "left room" situations, + # we'll filter the user out later in the per user updates. + # We also need to maintain a new users list, as it may shring. + final_users = [] + for counter, user_id in enumerate(users): + targets = ( + target_users - {user_id} + if action + in (ProfileUpdateAction.JOINED_ROOM, ProfileUpdateAction.LEFT_ROOM) + else target_users + ) + if not len(targets): + # No point writing a joined or left to the user themselves, skip. + continue + final_users.append(user_id) + profile_update_values[counter] = ( + self._instance_name, + user_id, + action.value, + sorted_field_names, + inserted_ts, + ) + + if not len(profile_update_values): + # We found nothing to update, abort. + return None + + # Now generate the stream ID's we want to use and add them to the list of + # updates. + stream_ids = self._profile_updates_id_gen.get_next_mult_txn( + txn, len(profile_update_values.keys()) + ) + profile_updates = [ + (stream_id, *values) + for stream_id, values in zip(stream_ids, profile_update_values.values()) + ] + # Maintain a map of stream_id to user_id, so we can later filter out rows + # when inserting into the per user updates table. + stream_ids_to_user_id = dict(list(zip(stream_ids, final_users))) + + self.db_pool.simple_insert_many_txn( txn, table="profile_updates", - values={ - "stream_id": stream_id, - "instance_name": self._instance_name, - "user_id": user_id.to_string(), - "action": action.value, - "affected_fields": json_encoder.encode(sorted(field_names)) - if field_names - else None, - "inserted_ts": inserted_ts, - }, + keys=[ + "stream_id", + "instance_name", + "user_id", + "action", + "affected_fields", + "inserted_ts", + ], + values=profile_updates, ) # Add per user tracking rows for each generated stream ID - per_user_values = [(stream_id, user_id, inserted_ts) for user_id in users] + per_user_values = [] + for stream_id in stream_ids: + per_user_values.extend( + # Filter out JOINED_ROOM/LEFT_ROOM updates to ourselves. + [ + (stream_id, user_id, inserted_ts) + for user_id in target_users + if stream_ids_to_user_id[stream_id] != user_id + or action == ProfileUpdateAction.UPDATE + ] + ) + self.db_pool.simple_insert_many_txn( txn, table="profile_updates_per_user", @@ -1003,7 +1080,7 @@ class ProfileWorkerStore(SQLBaseStore): ], values=per_user_values, ) - return stream_id + return stream_ids[-1] async def set_profile_field( self, @@ -1067,7 +1144,7 @@ class ProfileWorkerStore(SQLBaseStore): stream_id = self.record_profile_updates_txn( txn=txn, - user_id=user_id, + users={user_id.to_string()}, action=ProfileUpdateAction.UPDATE, field_names=[field_name], ) diff --git a/tests/handlers/test_profile.py b/tests/handlers/test_profile.py index 7af0345c87..8ea932c46e 100644 --- a/tests/handlers/test_profile.py +++ b/tests/handlers/test_profile.py @@ -659,7 +659,7 @@ class ProfileTestCase(unittest.HomeserverTestCase): affected_fields=None, ), ProfileUpdate( - stream_id=7, + stream_id=9, user_id="@gracie:test", action="left_room", affected_fields=None, @@ -688,6 +688,45 @@ class ProfileTestCase(unittest.HomeserverTestCase): ], ) + @override_config({"include_profile_updates_in_sync": True}) + def test_left_room_event_if_we_leave_the_last_shared_room( + self, + ) -> None: + """Test that when we leave a room, we get profile update rows with a "left room" + action for users we no longer share a room with. + """ + self.register_user("roger", "password") + roger_token = self.login("roger", "password") + room_id = self.helper.create_room_as( + room_creator=self.frank.to_string(), + tok=self.frank_token, + ) + self.helper.join(room_id, "@roger:test", tok=roger_token) + + # Make us leave the room + self.helper.leave(room_id, self.frank.to_string(), tok=self.frank_token) + per_user_updates = self.get_success( + self.store.get_profile_updates_for_user_and_fields( + from_id=0, + to_id=10, + user_id=self.frank.to_string(), + field_names={"m.status"}, + ) + ) + # We're no longer in any rooms with roger, and the profile update stream + # should have an update regarding that. + self.assertEqual( + per_user_updates, + [ + ProfileUpdate( + stream_id=4, + user_id="@roger:test", + action="left_room", + affected_fields=None, + ), + ], + ) + @parameterized.expand( [ ["displayname", "Frank"], diff --git a/tests/handlers/test_sync.py b/tests/handlers/test_sync.py index a74bd68df1..2f0107dd0a 100644 --- a/tests/handlers/test_sync.py +++ b/tests/handlers/test_sync.py @@ -2074,7 +2074,8 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): self, ) -> None: """Test that with `include_profile_updates_in_sync` enabled the incremental - sync response includes a 'null' for users who are no longer sharing rooms. + sync response includes a 'null' for users who are no longer sharing rooms, due + to the other user leaving the last room. """ requester = create_requester(self.user) initial_result = self.get_success( @@ -2119,6 +2120,59 @@ class SyncProfileUpdatesTestCase(tests.unittest.HomeserverTestCase): incremental_result.profile_updates["@other_user:test"], ) + @override_config({"include_profile_updates_in_sync": True}) + def test_incremental_sync_sends_down_null_profile_we_no_longer_sharing_rooms( + self, + ) -> None: + """Test that with `include_profile_updates_in_sync` enabled the incremental + sync response includes a 'null' for users who are no longer sharing rooms, due + us leaving the last shared room. + """ + requester = create_requester(self.user) + initial_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json={ + "org.matrix.msc4429.profile_fields": { + "ids": ["m.status", "displayname", "avatar_url"] + } + }, + ), + ), + request_key=generate_request_key(), + ) + ) + self.helper.leave( + room=self.joined_room, + user=self.user, + tok=self.tok, + ) + incremental_result = self.get_success( + self.sync_handler.wait_for_sync_for_user( + requester, + since_token=initial_result.next_batch, + sync_config=generate_sync_config( + user_id=self.user, + filter_collection=FilterCollection( + hs=self.hs, + filter_json={ + "org.matrix.msc4429.profile_fields": { + "ids": ["m.status", "displayname", "avatar_url"] + } + }, + ), + ), + request_key=generate_request_key(), + ) + ) + self.assertIsNone( + incremental_result.profile_updates["@other_user:test"], + ) + @parameterized.expand( [ True, diff --git a/tests/rest/client/sliding_sync/test_extension_profiles.py b/tests/rest/client/sliding_sync/test_extension_profiles.py index 5c2d17547c..2e76d6923c 100644 --- a/tests/rest/client/sliding_sync/test_extension_profiles.py +++ b/tests/rest/client/sliding_sync/test_extension_profiles.py @@ -525,7 +525,7 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): ) -> None: """ Test that profile extension response returns a null for the user in - incremental sync. + incremental sync, if the user left the last shared room. """ # Make an initial Sliding Sync request with the profiles extension enabled profiles_config: dict = { @@ -552,6 +552,46 @@ class SlidingSyncProfilesTestCase(SlidingSyncBase): ], ) + @parameterized.expand( + [ + True, + False, + ] + ) + @override_config({"include_profile_updates_in_sync": True}) + def test_null_profile_returned_we_left_all_rooms( + self, + request_fields: bool, + ) -> None: + """ + Test that profile extension response returns a null for the user in + incremental sync, if we left the last shared room with a user. + """ + # Make an initial Sliding Sync request with the profiles extension enabled + profiles_config: dict = { + "enabled": True, + } + if request_fields: + profiles_config["fields"] = ["field"] + sync_body = { + "lists": {}, + "extensions": { + "org.matrix.msc4262.profiles": profiles_config, + }, + } + response_body, from_token = self.do_sync(sync_body, tok=self.tok) + + self.helper.leave(self.joined_room, self.user, tok=self.tok) + + # Make an incremental Sliding Sync request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=self.tok) + # We should see a null profile + self.assertIsNone( + response_body["extensions"]["org.matrix.msc4262.profiles"]["users"][ + "@other_user:test" + ], + ) + @override_config({"include_profile_updates_in_sync": True}) def test_profile_returned_if_user_left_then_rejoined(self) -> None: """