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

@@ -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)

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/<user_id>
-
-        {
-            "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/<user_id>
+
+        {
+            "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)

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, {}