diff --git a/docker/complement/conf/start_for_complement.sh b/docker/complement/conf/start_for_complement.sh index 9d8fd93334..e0d30abed3 100755 --- a/docker/complement/conf/start_for_complement.sh +++ b/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" diff --git a/docker/configure_workers_and_start.py b/docker/configure_workers_and_start.py index b733384a5b..38f8649b44 100755 --- a/docker/configure_workers_and_start.py +++ b/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 diff --git a/docs/usage/configuration/config_documentation.md b/docs/usage/configuration/config_documentation.md index 5e154ccfa9..d8085e0e8a 100644 --- a/docs/usage/configuration/config_documentation.md +++ b/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: diff --git a/docs/workers.md b/docs/workers.md index 7401a4190b..92b606607c 100644 --- a/docs/workers.md +++ b/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 diff --git a/schema/synapse-config.schema.yaml b/schema/synapse-config.schema.yaml index 57eb8ba777..34754de734 100644 --- a/schema/synapse-config.schema.yaml +++ b/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 diff --git a/synapse/config/workers.py b/synapse/config/workers.py index a05843d900..3bfe300d49 100644 --- a/synapse/config/workers.py +++ b/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 ) diff --git a/synapse/handlers/profile.py b/synapse/handlers/profile.py index b17dffcded..0054978ec1 100644 --- a/synapse/handlers/profile.py +++ b/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, diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py index 19eb8728b1..d0e4ec69c7 100644 --- a/synapse/handlers/room_member.py +++ b/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 diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py index 65befb7706..1d0586abe0 100644 --- a/synapse/replication/tcp/handler.py +++ b/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 diff --git a/synapse/rest/client/profile.py b/synapse/rest/client/profile.py index dc632822c1..5fdf9ae0ec 100644 --- a/synapse/rest/client/profile.py +++ b/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( diff --git a/synapse/storage/databases/main/profile.py b/synapse/storage/databases/main/profile.py index f2d9130eaf..197e1353e6 100644 --- a/synapse/storage/databases/main/profile.py +++ b/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