dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-28 11:34:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Read from profile_updates_per_user when syncing

Browse Source 
No longer do we do the expensive lookup for which users share a room during /sync.
This commit is contained in:
Andrew Morgan
2026-06-21 22:40:52 +01:00
parent 029c4d81e4
commit 676ef4f819
3 changed files with 129 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/api/filtering.py
+2 -2
View File
@@ -227,7 +227,7 @@ class FilterCollection:
self.event_fields = filter_json.get("event_fields", [])
self.event_format = filter_json.get("event_format", "client")
self.profile_fields: list[str] = []
self.profile_fields: set[str] = set()
if hs.config.experimental.msc4429_enabled:
profile_fields_filter = filter_json.get("org.matrix.msc4429.profile_fields")
@@ -235,7 +235,7 @@ class FilterCollection:
ids = profile_fields_filter.get("ids", [])
if ids is None:
ids = []
self.profile_fields = list(ids)
self.profile_fields = set(ids)
def __repr__(self) -> str:
return "<FilterCollection %s>" % (json.dumps(self._filter_json),)
synapse/handlers/sync.py
+28 -34
View File
@@ -2142,7 +2142,7 @@ class SyncHandler:
*,
user_id: str,
sync_result_builder: "SyncResultBuilder",
profile_fields: list[str],
profile_fields: set[str],
include_users: set[str] | None,
) -> None:
"""
@@ -2159,21 +2159,35 @@ class SyncHandler:
for when we have calculated a list of users in our lazy loading
sync and want to only return those.
"""
# Currently, limited to only local profiles, so filter remote servers out
user_ids = await self.store.get_local_users_who_share_room_with_user(user_id)
updates = await self.store.get_profile_updates_for_user_and_fields(
from_id=0,
to_id=sync_result_builder.now_token.profile_updates_key,
user_id=user_id,
field_names=profile_fields,
include_users=include_users,
)
if include_users:
# Filter down to selected included users
user_ids = {user_id for user_id in user_ids if user_id in include_users}
user_fields: dict[str, set[str]] = {}
for update in updates:
if (
update.action != ProfileUpdateAction.UPDATE.value
# TODO: When would field_name be None?
or update.field_name is None
):
continue
if not user_ids:
user_fields.setdefault(update.user_id, set()).add(update.field_name)
if not user_fields:
return
profile_data_by_user = await self.store.get_profile_data_for_users(user_ids)
profile_data_by_user = await self.store.get_profile_data_for_users(
user_fields.keys()
)
# Serialise the profile updates into the sync response format.
profile_updates: dict[str, dict[str, JsonValue | None]] = {}
for other_user_id in user_ids:
for other_user_id, fields in user_fields.items():
profile_data = profile_data_by_user.get(other_user_id)
if profile_data is None:
# Don't generate anything for users with no profile data
@@ -2181,7 +2195,7 @@ class SyncHandler:
continue
per_user_updates: dict[str, JsonValue] = {}
for field_name in profile_fields:
for field_name in fields:
if profile_data.get(field_name):
per_user_updates[field_name] = cast(
JsonValue, profile_data[field_name]
@@ -2247,42 +2261,23 @@ class SyncHandler:
if since_token.profile_updates_key == now_token.profile_updates_key:
return
updates = await self.store.get_profile_updates_for_fields(
from_id=since_token.profile_updates_key,
to_id=now_token.profile_updates_key,
field_names=profile_fields,
)
interesting_updates = await self.store.get_profile_updates_per_user_for_user(
updates = await self.store.get_profile_updates_for_user_and_fields(
from_id=since_token.profile_updates_key,
to_id=now_token.profile_updates_key,
user_id=user_id,
field_names=profile_fields,
)
if include_users:
# Further filter down to selected included users
updates = [update for update in updates if update.user_id in include_users]
left_room_user_ids = {
update.user_id
for update in updates
if update.action == ProfileUpdateAction.LEFT_ROOM.value
}
# Only include updates we've got for us specifically for field updates
updates = [
update for update in updates if update.stream_id in interesting_updates
]
updated_user_ids = {
update.user_id
for update in updates
if update.action == ProfileUpdateAction.UPDATE.value
}
no_longer_sharing_rooms_user_ids: set[str] = set()
if left_room_user_ids:
shared_left_user_ids = await self.store.do_users_share_a_room(
user_id, left_room_user_ids
)
no_longer_sharing_rooms_user_ids = set(left_room_user_ids) - set(
shared_left_user_ids
)
if not updated_user_ids and not left_room_user_ids:
return
@@ -2293,7 +2288,6 @@ class SyncHandler:
# Process field updates
if updated_user_ids:
updated_user_ids.add(user_id)
user_fields: dict[str, set[str]] = {}
for update in updates:
if not update.field_name or update.user_id not in updated_user_ids:
@@ -2331,8 +2325,8 @@ class SyncHandler:
profile_updates[other_user_id] = per_user_updates
# Process left rooms
if no_longer_sharing_rooms_user_ids:
for other_user_id in no_longer_sharing_rooms_user_ids:
if left_room_user_ids:
for other_user_id in left_room_user_ids:
# Return an empty dictionary to the client
profile_updates[other_user_id] = {}
synapse/storage/databases/main/profile.py
+99 -1
View File
@@ -411,7 +411,7 @@ class ProfileWorkerStore(SQLBaseStore):
" OR action != ?) "
" ORDER BY stream_id ASC"
)
txn.execute(sql, (from_id, to_id, ProfileUpdateAction.UPDATE.value, *args))
txn.execute(sql, (from_id, to_id, *args, ProfileUpdateAction.UPDATE.value))
rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall())
updates: list[ProfileUpdate] = []
@@ -431,6 +431,104 @@ class ProfileWorkerStore(SQLBaseStore):
"get_profile_updates_for_fields", _get_profile_updates_for_fields_txn
)
async def get_profile_updates_for_user_and_fields(
self,
*,
from_id: int,
to_id: int,
user_id: str,
field_names: set[str],
include_users: set[str] | None = None,
) -> list[ProfileUpdate]:
"""Get profile update markers for a user in a stream range.
The returned profile update rows are restricted to those with a
corresponding `profile_updates_per_user` row for the syncing user.
Bounds: from_id < ... <= to_id
Args:
from_id: The starting stream ID (exclusive).
to_id: The ending stream ID (inclusive).
user_id: The full user ID to filter on.
field_names: Set of field names to filter update actions against.
include_users: If given, only include updates for these user IDs.
Returns:
A list of ProfileUpdates update rows.
"""
if from_id == to_id:
return []
if len(field_names) == 0:
return []
if include_users is not None and len(include_users) == 0:
# All updates have been filtered out by lazy-loading.
return []
def _get_profile_updates_for_user_and_fields_txn(
txn: LoggingTransaction,
) -> list[ProfileUpdate]:
field_clause, field_args = make_in_list_sql_clause(
txn.database_engine, "pu.field_name", field_names
)
user_clause = ""
user_args: list[str] = []
if include_users is not None:
# Filter out rows that aren't in `include_users`, if defined.
# This is only relevant when lazy-loading.
user_clause, user_args = make_in_list_sql_clause(
txn.database_engine, "pu.user_id", include_users
)
user_clause = f"AND {user_clause}"
# Retrieve profile updates where there's a corresponding row in
# `profile_updates_per_user` within the given `stream_id` bounds
# and the `user_id` and `field_names` match.
sql = f"""
SELECT pu.stream_id, pu.user_id, pu.action, pu.field_name
FROM profile_updates AS pu
INNER JOIN profile_updates_per_user AS puf
ON pu.stream_id = puf.stream_id
WHERE ? < pu.stream_id AND pu.stream_id <= ?
AND puf.user_id = ?
{user_clause}
AND ({field_clause} OR pu.action != ?)
ORDER BY pu.stream_id ASC
"""
txn.execute(
sql,
(
from_id,
to_id,
user_id,
*user_args,
*field_args,
ProfileUpdateAction.UPDATE.value,
),
)
rows = cast(list[tuple[int, str, str, str | None]], txn.fetchall())
updates: list[ProfileUpdate] = []
for stream_id, updated_user_id, action, field_name in rows:
updates.append(
ProfileUpdate(
stream_id=stream_id,
user_id=updated_user_id,
action=action,
field_name=field_name,
)
)
return updates
return await self.db_pool.runInteraction(
"get_profile_updates_for_user_and_fields",
_get_profile_updates_for_user_and_fields_txn,
)
async def get_profile_data_for_users(
self, user_ids: Collection[str]
) -> dict[str, dict[str, str | JsonDict | None]]: