dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-09-01 18:08:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Store precomputed room ordering in profile update tasks

Browse Source
This commit is contained in:
Neil Johnson
2026-08-14 15:38:13 +00:00
committed by Neil Johnson
parent a45d4ef0a4
commit 83130b64f8
2 changed files with 74 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/handlers/profile.py
+26 -22
View File
@@ -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,
tests/handlers/test_profile.py
+48 -5
View File
@@ -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."""