Remove profile_updates dedicates stream writer worker

Instead, all profile updates stream writes should go via the event persisters, via replication if not directly coming out of events.

docker/complement/conf/start_for_complement.sh

@@ -60,13 +60,12 @@ if [[ -n "$SYNAPSE_COMPLEMENT_USE_WORKERS" ]]; then
       federation_inbound, \
       federation_reader, \
       federation_sender, \
-      profile_updates, \
       synchrotron, \
       client_reader, \
       appservice, \
       pusher, \
       device_lists:2, \
-      stream_writers=account_data+presence+profile_updates+receipts+to_device+typing"
+      stream_writers=account_data+presence+receipts+to_device+typing"
 
   fi
   log "Workers requested: $SYNAPSE_WORKER_TYPES"

docker/configure_workers_and_start.py

@@ -269,10 +269,7 @@ WORKERS_CONFIG: dict[str, dict[str, Any]] = {
             "^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/(join|invite|leave|ban|unban|kick)$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/join/",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/knock/",
-            # The [^/] differentiates this endpoint from
-            # `ProfileRestFieldsServlet`, which we want to instead go to the
-            # `profile_updates` worker below.
-            "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/[^/]+",
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/profile/",
         ],
         "shared_extra_conf": {},
         "worker_extra_conf": "",
@@ -311,15 +308,6 @@ WORKERS_CONFIG: dict[str, dict[str, Any]] = {
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
-    "profile_updates": {
-        "app": "synapse.app.generic_worker",
-        "listener_resources": ["client", "replication"],
-        "endpoint_patterns": [
-            "^/_matrix/client/(unstable/uk.tcpip.msc4133|api/v1|r0|v3|unstable)/profile/.+/"
-        ],
-        "shared_extra_conf": {},
-        "worker_extra_conf": "",
-    },
     "device_lists": {
         "app": "synapse.app.generic_worker",
         "listener_resources": ["client", "replication"],
@@ -529,7 +517,6 @@ def add_worker_roles_to_shared_config(
         "typing",
         "push_rules",
         "thread_subscriptions",
-        "profile_updates",
     }
 
     # Worker-type specific sharding config. Now a single worker can fulfill multiple

docs/usage/configuration/config_documentation.md

@@ -4514,8 +4514,6 @@ This setting has the following sub-options:
 
 * `device_lists` (string): Name of a worker assigned to the `device_lists` stream.
 
-* `profile_updates` (string): Name of a worker assigned to the `profile_updates` stream.
-
 Example configuration:
 ```yaml
 stream_writers:

docs/workers.md

@@ -577,7 +577,6 @@ configured as stream writer for the `quarantined_media_changes` stream:
 
     ^/_synapse/admin/v1/quarantine_media/.*$
 
-
 #### Restrict outbound federation traffic to a specific set of workers
 
 The

schema/synapse-config.schema.yaml

@@ -5592,9 +5592,6 @@ properties:
       device_lists:
         type: string
         description: Name of a worker assigned to the `device_lists` stream.
-      profile_updates:
-        type: string
-        description: Name of a worker assigned to the `profile_updates` stream.
     default: {}
     examples:
       - events: worker1

synapse/config/workers.py

@@ -127,9 +127,10 @@ class WriterLocations:
     """Specifies the instances that write various streams.
 
     Attributes:
-        events: The instances that write to the event, backfill and `sticky_events` streams.
-            (`sticky_events` is written to during event persistence so must be handled by the
-            same stream writers.)
+        events: The instances that write to the event, backfill, `sticky_events` and
+            `profile_updates` streams.
+            (`sticky_events` and `profile_updates` are written to during event
+            persistence so must be handled by the same stream writers.)
         typing: The instances that write to the typing stream. Currently
             can only be a single instance.
         to_device: The instances that write to the to_device stream. Currently
@@ -185,10 +186,6 @@ class WriterLocations:
         default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
-    profile_updates: list[str] = attr.ib(
-        default=[MAIN_PROCESS_INSTANCE_NAME],
-        converter=_instance_to_list_converter,
-    )
     quarantined_media_changes: list[str] = attr.ib(
         default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
@@ -379,7 +376,6 @@ class WorkerConfig(Config):
             "push_rules",
             "device_lists",
             "thread_subscriptions",
-            "profile_updates",
         ):
             instances = _instance_to_list_converter(getattr(self.writers, stream))
             for instance in instances:
@@ -435,11 +431,6 @@ class WorkerConfig(Config):
                 "Must specify at least one instance to handle `thread_subscriptions` messages."
             )
 
-        if len(self.writers.profile_updates) == 0:
-            raise ConfigError(
-                "Must specify at least one instance to handle `profile_updates` messages."
-            )
-
         self.events_shard_config = RoutableShardedWorkerHandlingConfig(
             self.writers.events
         )

synapse/handlers/profile.py

@@ -82,7 +82,6 @@ class ProfileHandler:
         self.store = hs.get_datastores().main
         self.hs = hs
         self._notifier = hs.get_notifier()
-        self._msc4429_enabled = hs.config.server.include_profile_updates_in_sync
 
         self.federation = hs.get_federation_client()
         hs.get_federation_registry().register_query_handler(
@@ -106,8 +105,11 @@ 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
+
+        # Profile updates stream
+        self._msc4429_enabled = hs.config.server.include_profile_updates_in_sync
+        self._is_events_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.events
         )
         self._delete_profile_field_client = ReplicationProfileDeleteField.make_client(
             self.hs
@@ -116,9 +118,7 @@ class ProfileHandler:
             ReplicationProfileDeleteUponDeactivation.make_client(self.hs)
         )
         self._set_profile_field_client = ReplicationProfileSetField.make_client(self.hs)
-        self._profile_updates_writer_instance = (
-            self.hs.config.worker.writers.profile_updates[0]
-        )
+        self._profile_updates_writer_instance = self.hs.config.worker.writers.events[0]
 
     async def get_targets_for_profile_updates(
         self, user_id: UserID
@@ -482,7 +482,7 @@ class ProfileHandler:
         if not self._msc4429_enabled:
             return
         # We should never be called without being a profile worker
-        assert self._is_profile_worker
+        assert self._is_events_writer
 
         user_id_str = user_id.to_string()
 
@@ -524,7 +524,7 @@ class ProfileHandler:
         if not self._msc4429_enabled:
             return
         # We should never be called without being a profile worker
-        assert self._is_profile_worker
+        assert self._is_events_writer
 
         user_id_str = user_id.to_string()
 
@@ -553,14 +553,15 @@ class ProfileHandler:
     ) -> None:
         """
         Dispatch profile deletion upon deactivation to the right instance that
-        can write to profiles, or handle it ourselves, if we're a profile worker.
+        can write to the profile updates stream, or handle it ourselves,
+        if we're an events stream writer.
 
         Args:
             target_user: User ID who'se profile is being deactivated.
             requester: The requesting user.
             by_admin: Whether the action is being done by an admin.
         """
-        if self._is_profile_worker:
+        if self._is_events_writer:
             await self.delete_profile_upon_deactivation(
                 target_user=target_user,
                 requester=requester,
@@ -670,7 +671,7 @@ class ProfileHandler:
             by_admin: Whether this change was made by an administrator.
             propagate: Whether this change also applies to the user's membership events.
         """
-        if self._is_profile_worker:
+        if self._is_events_writer:
             await self.set_field(
                 target_user=target_user,
                 requester=requester,
@@ -934,7 +935,7 @@ class ProfileHandler:
             by_admin: Whether this change was made by an administrator.
         """
         assert field_name not in (ProfileFields.DISPLAYNAME, ProfileFields.AVATAR_URL)
-        if self._is_profile_worker:
+        if self._is_events_writer:
             await self.delete_profile_field(
                 target_user=target_user,
                 requester=requester,

synapse/handlers/room_member.py

@@ -206,13 +206,13 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
         )
         self._push_writer = hs.config.worker.writers.push_rules[0]
         self._copy_push_client = ReplicationCopyPusherRestServlet.make_client(hs)
+
+        # Profile updates stream
         self._msc4429_enabled = hs.config.server.include_profile_updates_in_sync
-        self._is_profile_worker = (
-            hs.get_instance_name() in hs.config.worker.writers.profile_updates
-        )
-        self._profile_updates_writer_instance = (
-            self.hs.config.worker.writers.profile_updates[0]
+        self._is_events_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.events
         )
+        self._profile_updates_writer_instance = self.hs.config.worker.writers.events[0]
         self._profile_user_room_membership_change_client = (
             ReplicationProfileUserRoomMembershipChange.make_client(self.hs)
         )
@@ -541,7 +541,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                                 # Notify the profile handler. We only want to do this once
                                 # in a multi-worker setup, so we can't listen on the dispatched
                                 # event above.
-                                if self._is_profile_worker:
+                                if self._is_events_writer:
                                     await self.profile_handler.user_left_room(
                                         target, room_id
                                     )
@@ -566,7 +566,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                     ):
                         # Notify the profile handler. We only want to do this once
                         # in a multi-worker setup, so we can't dispatch a hook to all workers.
-                        if self._is_profile_worker:
+                        if self._is_events_writer:
                             await self.profile_handler.user_joined_room(target, room_id)
                         else:
                             # Offload to the right worker via http replication
@@ -1595,7 +1595,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
                         # Notify the profile handler. We only want to do this once
                         # in a multi-worker setup, so we can't listen on the dispatched
                         # event above.
-                        if self._is_profile_worker:
+                        if self._is_events_writer:
                             await self.profile_handler.user_left_room(
                                 target_user, room_id
                             )
@@ -1614,7 +1614,7 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
             if not prev_member_event or prev_member_event.membership != Membership.JOIN:
                 # Notify the profile handler. We only want to do this once
                 # in a multi-worker setup, so we can't dispatch a hook to all workers.
-                if self._is_profile_worker:
+                if self._is_events_writer:
                     await self.profile_handler.user_joined_room(target_user, room_id)
                 else:
                     # Offload to the right worker via http replication

synapse/replication/tcp/handler.py

@@ -220,7 +220,7 @@ class ReplicationCommandHandler:
                 continue
 
             if isinstance(stream, ProfileUpdatesStream):
-                if hs.get_instance_name() in hs.config.worker.writers.profile_updates:
+                if hs.get_instance_name() in hs.config.worker.writers.events:
                     self._streams_to_replicate.append(stream)
 
                 continue

synapse/rest/client/profile.py

@@ -109,9 +109,6 @@ class ProfileFieldRestServlet(RestServlet):
         self.hs = hs
         self.profile_handler = hs.get_profile_handler()
         self.auth = hs.get_auth()
-        self._is_profile_worker = (
-            hs.get_instance_name() in hs.config.worker.writers.profile_updates
-        )
         if hs.config.experimental.msc4133_enabled:
             self.PATTERNS.append(
                 re.compile(

synapse/storage/databases/main/profile.py

@@ -80,9 +80,7 @@ class ProfileWorkerStore(SQLBaseStore):
         )
 
         self._msc4429_enabled = hs.config.server.include_profile_updates_in_sync
-        self._can_write_to_profile_updates = (
-            self._instance_name in hs.config.worker.writers.profile_updates
-        )
+        self._is_events_writer = self._instance_name in hs.config.worker.writers.events
         self._profile_updates_id_gen: MultiWriterIdGenerator = MultiWriterIdGenerator(
             db_conn=db_conn,
             db=database,
@@ -94,7 +92,7 @@ class ProfileWorkerStore(SQLBaseStore):
                 ("profile_updates", "instance_name", "stream_id"),
             ],
             sequence_name="profile_updates_sequence",
-            writers=hs.config.worker.writers.profile_updates,
+            writers=hs.config.worker.writers.events,
         )
 
     async def populate_full_user_id_profiles(
@@ -684,7 +682,7 @@ class ProfileWorkerStore(SQLBaseStore):
             The profile updates stream ID that was created in this transaction
         """
         if self._msc4429_enabled:
-            assert self._can_write_to_profile_updates
+            assert self._is_events_writer
 
         self._check_profile_size(txn, user_id, field_name, new_value)
 
@@ -890,7 +888,7 @@ class ProfileWorkerStore(SQLBaseStore):
         """
 
         if self._msc4429_enabled:
-            assert self._can_write_to_profile_updates
+            assert self._is_events_writer
 
         def delete_profile_field(txn: LoggingTransaction) -> int | None:
             if isinstance(self.database_engine, PostgresEngine):
@@ -995,7 +993,7 @@ class ProfileWorkerStore(SQLBaseStore):
         """
         if not users_to_update:
             return None
-        assert self._can_write_to_profile_updates
+        assert self._is_events_writer
 
         def _record_profile_updates_for_user_left_room_txn(
             txn: LoggingTransaction,
@@ -1063,7 +1061,7 @@ class ProfileWorkerStore(SQLBaseStore):
         """
         if not users_to_update:
             return None
-        assert self._can_write_to_profile_updates
+        assert self._is_events_writer
 
         def _record_profile_updates_for_user_joined_room_txn(
             txn: LoggingTransaction,
@@ -1104,7 +1102,7 @@ class ProfileWorkerStore(SQLBaseStore):
         Returns:
             None
         """
-        assert self._can_write_to_profile_updates
+        assert self._is_events_writer
         if not users_to_remove:
             return