From 7a60657dd11c27e54ca9c242d90f7a36950f8ee3 Mon Sep 17 00:00:00 2001 From: Jason Robinson Date: Fri, 26 Jun 2026 13:25:53 +0300 Subject: [PATCH] Clean up recording profile updates on changes Instead of using replication when setting field values, remove that replication, keep writing the field via the profile handler in the current instance, and then dispatch recording the profile update over replication, if needed. This is cleaner as it ensures we don't need to wrap various places outside the profile handler that set things like the displayname in the "if profile worker else replication" logic. --- synapse/handlers/profile.py | 51 ++++++++++-- synapse/replication/http/profile.py | 125 +++++++++++----------------- synapse/rest/client/profile.py | 71 ++++------------ 3 files changed, 111 insertions(+), 136 deletions(-) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index c9203b915f..8bc724ad35 100644 --- a/synapse/handlers/profile.py +++ b/synapse/handlers/profile.py @@ -34,6 +34,7 @@ from synapse.api.errors import ( StoreError, SynapseError, ) +from synapse.replication.http.profile import ReplicationProfileRecordFieldUpdates from synapse.storage.databases.main.media_repository import LocalMedia, RemoteMedia from synapse.storage.roommember import ProfileInfo from synapse.types import ( @@ -101,8 +102,17 @@ class ProfileHandler: self._update_join_states_task, UPDATE_JOIN_STATES_ACTION_NAME ) self._worker_locks = hs.get_worker_locks_handler() + self._is_profile_worker = ( + hs.get_instance_name() in hs.config.worker.writers.profile_updates + ) + self._record_profile_updates_client = ( + ReplicationProfileRecordFieldUpdates.make_client(self.hs) + ) + self._profile_updates_writer_instance = ( + self.hs.config.worker.writers.profile_updates[0] + ) - async def _record_profile_updates( + async def record_profile_updates( self, user_id: UserID, updated_fields: set[str] ) -> None: """ @@ -298,7 +308,7 @@ class ProfileHandler: ) await self.store.set_profile_displayname(target_user, displayname_to_set) - await self._record_profile_updates( + await self._dispatch_record_profile_updates( target_user, {ProfileFields.DISPLAYNAME}, ) @@ -411,7 +421,7 @@ class ProfileHandler: ) await self.store.set_profile_avatar_url(target_user, avatar_url_to_set) - await self._record_profile_updates( + await self._dispatch_record_profile_updates( target_user, {ProfileFields.AVATAR_URL}, ) @@ -544,7 +554,9 @@ class ProfileHandler: profile_updates.append((field_name, None)) await self.store.delete_profile(target_user) - await self._record_profile_updates( + + # Record profile updates for the profile update stream + await self._dispatch_record_profile_updates( target_user, {field_name for field_name, _value in profile_updates} ) @@ -555,6 +567,33 @@ class ProfileHandler: deactivation=True, ) + async def _dispatch_record_profile_updates( + self, user_id: UserID, updated_fields: set[str] + ) -> None: + """ + Dispatch the recording of profile updates, either directly via the current + instance, if we're a profile worker, otherwise push via replication. + + Args: + user_id: The user whose profile has had updates. + updated_fields: A set of the names of the fields that were updated. + + Returns: + None + """ + if self._is_profile_worker: + await self.record_profile_updates( + user_id, + updated_fields, + ) + else: + # Offload to the right worker via http replication + await self._record_profile_updates_client( + instance_name=self._profile_updates_writer_instance, + user_id=user_id.to_string(), + updated_fields=updated_fields, + ) + @cached() async def check_avatar_size_and_mime_type(self, mxc: str) -> bool: """Check that the size and content type of the avatar at the given MXC URI are @@ -754,7 +793,7 @@ class ProfileHandler: raise AuthError(403, "Cannot set another user's profile") await self.store.set_profile_field(target_user, field_name, new_value) - await self._record_profile_updates(target_user, {field_name}) + await self._dispatch_record_profile_updates(target_user, {field_name}) # Custom fields do not propagate into the user directory *or* rooms. profile = await self.store.get_profileinfo(target_user) @@ -790,7 +829,7 @@ class ProfileHandler: raise AuthError(400, "Cannot set another user's profile") await self.store.delete_profile_field(target_user, field_name) - await self._record_profile_updates(target_user, {field_name}) + await self._dispatch_record_profile_updates(target_user, {field_name}) # Custom fields do not propagate into the user directory *or* rooms. profile = await self.store.get_profileinfo(target_user) diff --git a/synapse/replication/http/profile.py b/synapse/replication/http/profile.py index a692836489..8b27b08873 100644 --- a/synapse/replication/http/profile.py +++ b/synapse/replication/http/profile.py @@ -21,7 +21,7 @@ from twisted.web.server import Request from synapse.api.constants import Membership from synapse.http.server import HttpServer from synapse.replication.http._base import ReplicationEndpoint -from synapse.types import JsonDict, JsonValue, UserID, create_requester +from synapse.types import JsonDict, UserID if TYPE_CHECKING: from synapse.server import HomeServer @@ -29,79 +29,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -class ReplicationProfileSetFieldValue(ReplicationEndpoint): - """Set profile field for a user. - - The POST looks like: - - POST /_synapse/replication/profile_set_field_value/ - - { - "requester_id": "@user:domain.tld", - "field_name": "displayname", - "new_value": "User Display Name", - "by_admin": False, - "propagate": False, - "authenticated_entity": "@admin:domain.tld", - } - - 200 OK - - {} - """ - - NAME = "profile_set_field_value" - PATH_ARGS = ("user_id",) - METHOD = "POST" - CACHE = False - - def __init__(self, hs: "HomeServer"): - super().__init__(hs) - - self._profile_handler = hs.get_profile_handler() - - @staticmethod - async def _serialize_payload( # type: ignore[override] - user_id: str, - requester_id: str, - field_name: str, - new_value: JsonValue | dict[str, JsonValue], - by_admin: bool = False, - propagate: bool = False, - authenticated_entity: str | None = None, - ) -> JsonDict: - return { - "requester_id": requester_id, - "field_name": field_name, - "new_value": new_value, - "by_admin": by_admin, - "propagate": propagate, - "authenticated_entity": authenticated_entity, - } - - async def _handle_request( # type: ignore[override] - self, request: Request, content: JsonDict, user_id: str - ) -> tuple[int, JsonDict]: - # Create a requester object with potentially an authenticated_entity, - # ie an admin who has done the request on behalf of the user. - requester = create_requester( - user_id=user_id, - authenticated_entity=content["authenticated_entity"] - if content["by_admin"] - else None, - ) - await self._profile_handler.set_field( - target_user=UserID.from_string(user_id), - requester=requester, - field_name=content["field_name"], - new_value=content["new_value"], - by_admin=content["by_admin"], - propagate=content["propagate"], - ) - - return (200, {}) - - class ReplicationProfileUserRoomMembershipChange(ReplicationEndpoint): """Store user profile update action regarding membership changes. @@ -159,6 +86,54 @@ class ReplicationProfileUserRoomMembershipChange(ReplicationEndpoint): return (200, {}) +class ReplicationProfileRecordFieldUpdates(ReplicationEndpoint): + """Record user profile field updates for the profile updates stream. + + The POST looks like: + + POST /_synapse/replication/profile_record_field_updates/ + + { + "updated_fields": ["list", "of", "fields"] + } + + 200 OK + + {} + """ + + NAME = "profile_record_field_updates" + PATH_ARGS = ("user_id",) + METHOD = "POST" + CACHE = False + + def __init__(self, hs: "HomeServer"): + super().__init__(hs) + + self._profile_handler = hs.get_profile_handler() + + @staticmethod + async def _serialize_payload( # type: ignore[override] + user_id: str, + updated_fields: set[str], + ) -> JsonDict: + assert len(updated_fields) > 0 + return { + "updated_fields": list(updated_fields), + } + + async def _handle_request( # type: ignore[override] + self, request: Request, content: JsonDict, user_id: str + ) -> tuple[int, JsonDict]: + assert len(content["updated_fields"]) > 0 + await self._profile_handler.record_profile_updates( + user_id=UserID.from_string(user_id), + updated_fields=set(content["updated_fields"]), + ) + + return (200, {}) + + def register_servlets(hs: "HomeServer", http_server: HttpServer) -> None: - ReplicationProfileSetFieldValue(hs).register(http_server) ReplicationProfileUserRoomMembershipChange(hs).register(http_server) + ReplicationProfileRecordFieldUpdates(hs).register(http_server) diff --git a/synapse/rest/client/profile.py b/synapse/rest/client/profile.py index 1767b13eb6..daab2c6651 100644 --- a/synapse/rest/client/profile.py +++ b/synapse/rest/client/profile.py @@ -35,7 +35,6 @@ from synapse.http.servlet import ( parse_json_object_from_request, ) from synapse.http.site import SynapseRequest -from synapse.replication.http.profile import ReplicationProfileSetFieldValue from synapse.rest.client._base import client_patterns from synapse.types import JsonDict, JsonValue, UserID from synapse.util.stringutils import is_namedspaced_grammar @@ -210,33 +209,14 @@ class ProfileFieldRestServlet(RestServlet): Codes.USER_ACCOUNT_SUSPENDED, ) - if self._is_profile_worker: - await self.profile_handler.set_field( - target_user=user, - requester=requester, - field_name=field_name, - new_value=new_value, - by_admin=is_admin, - propagate=propagate, - ) - else: - # Offload to the right worker via http replication - set_profile_data_client = ReplicationProfileSetFieldValue.make_client( - self.hs - ) - profile_updates_writer_instance = ( - self.hs.config.worker.writers.profile_updates[0] - ) - await set_profile_data_client( - instance_name=profile_updates_writer_instance, - user_id=user.to_string(), - requester_id=requester.user.to_string(), - field_name=field_name, - new_value=new_value, - by_admin=is_admin, - propagate=propagate, - authenticated_entity=requester.authenticated_entity, - ) + await self.profile_handler.set_field( + target_user=user, + requester=requester, + field_name=field_name, + new_value=new_value, + by_admin=is_admin, + propagate=propagate, + ) return 200, {} @@ -282,33 +262,14 @@ class ProfileFieldRestServlet(RestServlet): Codes.USER_ACCOUNT_SUSPENDED, ) - if self._is_profile_worker: - await self.profile_handler.set_field( - target_user=user, - requester=requester, - field_name=field_name, - new_value="", - by_admin=is_admin, - propagate=propagate, - ) - else: - # Offload to the right worker via http replication - set_profile_data_client = ReplicationProfileSetFieldValue.make_client( - self.hs - ) - profile_updates_writer_instance = ( - self.hs.config.worker.writers.profile_updates[0] - ) - await set_profile_data_client( - instance_name=profile_updates_writer_instance, - user_id=user.to_string(), - requester_id=requester.user.to_string(), - field_name=field_name, - new_value="", - by_admin=is_admin, - propagate=propagate, - authenticated_entity=requester.authenticated_entity, - ) + await self.profile_handler.set_field( + target_user=user, + requester=requester, + field_name=field_name, + new_value="", + by_admin=is_admin, + propagate=propagate, + ) return 200, {}