diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index 2c2eb6c753..5aedf1ebb9 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -20,7 +20,6 @@ # import logging import random -from bisect import bisect_right from typing import TYPE_CHECKING from twisted.internet.defer import CancelledError @@ -875,6 +874,27 @@ class ProfileHandler: target_user_str = target_user.to_string() + # Compute the ordered list of rooms upfront to ensure consistency across restarts + all_room_ids = await self.store.get_rooms_for_user(target_user_str) + + # Get the user's latest read receipts for all rooms + user_receipts = await self.store.get_receipts_for_user_with_orderings( + target_user_str, + [ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE], + ) + + # Sort rooms by most recent read receipt (highest stream_ordering first), + # with fallback to alphabetical ordering for rooms without receipts + def sort_key(room_id: str) -> tuple[int, str]: + if room_id in user_receipts: + # Rooms with receipts: sort by stream_ordering (descending) then by room_id + return (-user_receipts[room_id]["stream_ordering"], room_id) + else: + # Rooms without receipts: sort alphabetically after all rooms with receipts + return (0, room_id) + + room_ids = sorted(all_room_ids, key=sort_key) + # Cancel any ongoing profile membership updates for this user, # and start a new one. async with self._worker_locks.acquire_lock( @@ -895,6 +915,7 @@ class ProfileHandler: resource_id=target_user_str, params={ "requester_authenticated_entity": requester.authenticated_entity, + "ordered_room_ids": room_ids, }, ) @@ -906,33 +927,16 @@ class ProfileHandler: assert task.params target_user = UserID.from_string(task.resource_id) - all_room_ids = await self.store.get_rooms_for_user(target_user.to_string()) - # Get the user's latest read receipts for all rooms - user_receipts = await self.store.get_receipts_for_user_with_orderings( - target_user.to_string(), - [ReceiptTypes.READ, ReceiptTypes.READ_PRIVATE], - ) - - # Sort rooms by most recent read receipt (highest stream_ordering first), - # with fallback to alphabetical ordering for rooms without receipts - def sort_key(room_id: str) -> tuple[int, str]: - if room_id in user_receipts: - # Rooms with receipts: sort by stream_ordering (descending) then by room_id - return (-user_receipts[room_id]["stream_ordering"], room_id) - else: - # Rooms without receipts: sort alphabetically after all rooms with receipts - return (0, room_id) - - room_ids = sorted(all_room_ids, key=sort_key) + # Use the precomputed room ordering from task params to ensure consistency + room_ids = task.params.get("ordered_room_ids", []) last_room_id = task.result.get("last_room_id", None) if task.result else None if last_room_id: # Filter out room IDs that have already been handled - # by finding the first room ID greater than the last handled room ID - # and slicing the list from that point onwards. - room_ids = room_ids[bisect_right(room_ids, last_room_id) :] + last_index = room_ids.index(last_room_id, 0) + room_ids = room_ids[last_index + 1 :] requester = create_requester( user_id=target_user, diff --git a/tests/handlers/test_profile.py b/tests/handlers/test_profile.py index e439f5a0c7..5aa91916c3 100644 --- a/tests/handlers/test_profile.py +++ b/tests/handlers/test_profile.py @@ -764,12 +764,46 @@ class ProfileTestCase(unittest.HomeserverTestCase): room_id_2 = self.helper.create_room_as( self.frank.to_string(), tok=self.frank_token ) + room_id_3 = self.helper.create_room_as( + self.frank.to_string(), tok=self.frank_token + ) - # Ensure `room_id_1` comes before `room_id_2` alphabetically - if room_id_1 > room_id_2: - room_id_1, room_id_2 = room_id_2, room_id_1 - # Without read receipts, both rooms should be processed in alphabetical order + # Set read receipts with different timestamps (simulate different read times) + # Room 1 should be most recent, then room 2, then room 3 + event_3 = self.helper.send(room_id_3, "Hello 3", tok=self.frank_token) + event_2 = self.helper.send(room_id_2, "Hello 2", tok=self.frank_token) + event_1 = self.helper.send(room_id_1, "Hello 1", tok=self.frank_token) + self.get_success( + self.store.insert_receipt( + room_id_3, + ReceiptTypes.READ, + user_id=self.frank.to_string(), + event_ids=[event_3["event_id"]], + thread_id=None, + data={"ts": 100}, + ) + ) + self.get_success( + self.store.insert_receipt( + room_id_2, + ReceiptTypes.READ, + user_id=self.frank.to_string(), + event_ids=[event_2["event_id"]], + thread_id=None, + data={"ts": 200}, + ) + ) + self.get_success( + self.store.insert_receipt( + room_id_1, + ReceiptTypes.READ, + user_id=self.frank.to_string(), + event_ids=[event_1["event_id"]], + thread_id=None, + data={"ts": 300}, + ) + ) original_update_membership = self.hs.get_room_member_handler().update_membership @@ -778,7 +812,7 @@ class ProfileTestCase(unittest.HomeserverTestCase): async def potentially_slow_update_membership( *args: Any, **kwargs: Any ) -> tuple[str, int]: - if args[2] == room_id_2: + if args[2] == room_id_2 or args[2] == room_id_3: await self.clock.sleep(Duration(milliseconds=10)) if args[2] == room_id_1: nonlocal room_1_updated @@ -858,6 +892,15 @@ class ProfileTestCase(unittest.HomeserverTestCase): self.assertEqual( membership[state_tuple].content["displayname"], "Frank Jr." ) + membership = self.get_success( + self.storage_controllers.state.get_current_state( + room_id_3, StateFilter.from_types([state_tuple]) + ) + ) + self.assertEqual( + membership[state_tuple].content["displayname"], "Frank Jr." + ) + def test_room_update_ordering_by_read_receipt(self) -> None: """Test that rooms are updated in order of most recent read receipt."""