Move device changes off the main process (#18581)

The main goal of this PR is to handle device list changes onto multiple
writers, off the main process, so that we can have logins happening
whilst Synapse is rolling-restarting.

This is quite an intrusive change, so I would advise to review this
commit by commit; I tried to keep the history as clean as possible.

There are a few things to consider:

- the `device_list_key` in stream tokens becomes a
`MultiWriterStreamToken`, which has a few implications in sync and on
the storage layer
- we had a split between `DeviceHandler` and `DeviceWorkerHandler` for
master vs. worker process. I've kept this split, but making it rather
writer vs. non-writer worker, using method overrides for doing
replication calls when needed
- there are a few operations that need to happen on a single worker at a
time. Instead of using cross-worker locks, for now I made them run on
the first writer on the list

---------

Co-authored-by: Eric Eastwood <[email protected]>

Author: sandhose

changelog.d/18581.feature

@@ -0,0 +1 @@
+Enable workers to write directly to the device lists stream and handle device list updates, reducing load on the main process.

docker/complement/conf/start_for_complement.sh

@@ -54,7 +54,6 @@ if [[ -n "$SYNAPSE_COMPLEMENT_USE_WORKERS" ]]; then
     export SYNAPSE_WORKER_TYPES="\
       event_persister:2, \
       background_worker, \
-      frontend_proxy, \
       event_creator, \
       user_dir, \
       media_repository, \
@@ -65,6 +64,7 @@ if [[ -n "$SYNAPSE_COMPLEMENT_USE_WORKERS" ]]; then
       client_reader, \
       appservice, \
       pusher, \
+      device_lists:2, \
       stream_writers=account_data+presence+receipts+to_device+typing"
 
   fi

docker/configure_workers_and_start.py

@@ -178,6 +178,8 @@
             "^/_matrix/client/(api/v1|r0|v3|unstable)/login$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/account/3pid$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/account/whoami$",
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/devices(/|$)",
+            "^/_matrix/client/(r0|v3)/delete_devices$",
             "^/_matrix/client/versions$",
             "^/_matrix/client/(api/v1|r0|v3|unstable)/voip/turnServer$",
             "^/_matrix/client/(r0|v3|unstable)/register$",
@@ -194,6 +196,9 @@
             "^/_matrix/client/(api/v1|r0|v3|unstable)/directory/room/.*$",
             "^/_matrix/client/(r0|v3|unstable)/capabilities$",
             "^/_matrix/client/(r0|v3|unstable)/notifications$",
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/upload",
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/device_signing/upload$",
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$",
         ],
         "shared_extra_conf": {},
         "worker_extra_conf": "",
@@ -265,13 +270,6 @@
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
-    "frontend_proxy": {
-        "app": "synapse.app.generic_worker",
-        "listener_resources": ["client", "replication"],
-        "endpoint_patterns": ["^/_matrix/client/(api/v1|r0|v3|unstable)/keys/upload"],
-        "shared_extra_conf": {},
-        "worker_extra_conf": "",
-    },
     "account_data": {
         "app": "synapse.app.generic_worker",
         "listener_resources": ["client", "replication"],
@@ -306,6 +304,13 @@
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
+    "device_lists": {
+        "app": "synapse.app.generic_worker",
+        "listener_resources": ["client", "replication"],
+        "endpoint_patterns": [],
+        "shared_extra_conf": {},
+        "worker_extra_conf": "",
+    },
     "typing": {
         "app": "synapse.app.generic_worker",
         "listener_resources": ["client", "replication"],
@@ -412,16 +417,17 @@ def add_worker_roles_to_shared_config(
     # streams
     instance_map = shared_config.setdefault("instance_map", {})
 
-    # This is a list of the stream_writers that there can be only one of. Events can be
-    # sharded, and therefore doesn't belong here.
-    singular_stream_writers = [
+    # This is a list of the stream_writers.
+    stream_writers = {
         "account_data",
+        "events",
+        "device_lists",
         "presence",
         "receipts",
         "to_device",
         "typing",
         "push_rules",
-    ]
+    }
 
     # Worker-type specific sharding config. Now a single worker can fulfill multiple
     # roles, check each.
@@ -431,28 +437,11 @@ def add_worker_roles_to_shared_config(
     if "federation_sender" in worker_types_set:
         shared_config.setdefault("federation_sender_instances", []).append(worker_name)
 
-    if "event_persister" in worker_types_set:
-        # Event persisters write to the events stream, so we need to update
-        # the list of event stream writers
-        shared_config.setdefault("stream_writers", {}).setdefault("events", []).append(
-            worker_name
-        )
-
-        # Map of stream writer instance names to host/ports combos
-        if os.environ.get("SYNAPSE_USE_UNIX_SOCKET", False):
-            instance_map[worker_name] = {
-                "path": f"/run/worker.{worker_port}",
-            }
-        else:
-            instance_map[worker_name] = {
-                "host": "localhost",
-                "port": worker_port,
-            }
     # Update the list of stream writers. It's convenient that the name of the worker
     # type is the same as the stream to write. Iterate over the whole list in case there
     # is more than one.
     for worker in worker_types_set:
-        if worker in singular_stream_writers:
+        if worker in stream_writers:
             shared_config.setdefault("stream_writers", {}).setdefault(
                 worker, []
             ).append(worker_name)
@@ -876,6 +865,13 @@ def generate_worker_files(
         else:
             healthcheck_urls.append("http://localhost:%d/health" % (worker_port,))
 
+        # Special case for event_persister: those are just workers that write to
+        # the `events` stream. For other workers, the worker name is the same
+        # name of the stream they write to, but for some reason it is not the
+        # case for event_persister.
+        if "event_persister" in worker_types_set:
+            worker_types_set.add("events")
+
         # Update the shared config with sharding-related options if necessary
         add_worker_roles_to_shared_config(
             shared_config, worker_types_set, worker_name, worker_port

docs/usage/configuration/config_documentation.md

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

docs/workers.md

@@ -238,7 +238,8 @@ information.
     ^/_matrix/client/unstable/im.nheko.summary/summary/.*$
     ^/_matrix/client/(r0|v3|unstable)/account/3pid$
     ^/_matrix/client/(r0|v3|unstable)/account/whoami$
-    ^/_matrix/client/(r0|v3|unstable)/devices$
+    ^/_matrix/client/(r0|v3)/delete_devices$
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/devices(/|$)
     ^/_matrix/client/versions$
     ^/_matrix/client/(api/v1|r0|v3|unstable)/voip/turnServer$
     ^/_matrix/client/(api/v1|r0|v3|unstable)/rooms/.*/event/
@@ -257,7 +258,9 @@ information.
     ^/_matrix/client/(r0|v3|unstable)/keys/changes$
     ^/_matrix/client/(r0|v3|unstable)/keys/claim$
     ^/_matrix/client/(r0|v3|unstable)/room_keys/
-    ^/_matrix/client/(r0|v3|unstable)/keys/upload$
+    ^/_matrix/client/(r0|v3|unstable)/keys/upload
+    ^/_matrix/client/(api/v1|r0|v3|unstable/keys/device_signing/upload$
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$
 
     # Registration/login requests
     ^/_matrix/client/(api/v1|r0|v3|unstable)/login$
@@ -282,7 +285,6 @@ Additionally, the following REST endpoints can be handled for GET requests:
 
     ^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/
     ^/_matrix/client/unstable/org.matrix.msc4140/delayed_events
-    ^/_matrix/client/(api/v1|r0|v3|unstable)/devices/
 
     # Account data requests
     ^/_matrix/client/(r0|v3|unstable)/.*/tags
@@ -329,7 +331,6 @@ set to `true`), the following endpoints can be handled by the worker:
     ^/_synapse/admin/v2/users/[^/]+$
     ^/_synapse/admin/v1/username_available$
     ^/_synapse/admin/v1/users/[^/]+/_allow_cross_signing_replacement_without_uia$
-    # Only the GET method:
     ^/_synapse/admin/v1/users/[^/]+/devices$
 
 Note that a [HTTP listener](usage/configuration/config_documentation.md#listeners)
@@ -550,6 +551,18 @@ the stream writer for the `push_rules` stream:
 
     ^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/
 
+##### The `device_lists` stream
+
+The `device_lists` stream supports multiple writers. The following endpoints
+can be handled by any worker, but should be routed directly one of the workers
+configured as stream writer for the `device_lists` stream:
+
+    ^/_matrix/client/(r0|v3)/delete_devices$
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/devices/
+    ^/_matrix/client/(r0|v3|unstable)/keys/upload
+    ^/_matrix/client/(api/v1|r0|v3|unstable/keys/device_signing/upload$
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$
+
 #### Restrict outbound federation traffic to a specific set of workers
 
 The

schema/synapse-config.schema.yaml

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

synapse/config/workers.py

@@ -134,40 +134,44 @@ class WriterLocations:
             can only be a single instance.
         account_data: The instances that write to the account data streams. Currently
             can only be a single instance.
-        receipts: The instances that write to the receipts stream. Currently
-            can only be a single instance.
+        receipts: The instances that write to the receipts stream.
         presence: The instances that write to the presence stream. Currently
             can only be a single instance.
         push_rules: The instances that write to the push stream. Currently
             can only be a single instance.
+        device_lists: The instances that write to the device list stream.
     """
 
     events: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     typing: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     to_device: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     account_data: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     receipts: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     presence: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
     push_rules: List[str] = attr.ib(
-        default=["master"],
+        default=[MAIN_PROCESS_INSTANCE_NAME],
+        converter=_instance_to_list_converter,
+    )
+    device_lists: List[str] = attr.ib(
+        default=[MAIN_PROCESS_INSTANCE_NAME],
         converter=_instance_to_list_converter,
     )
 
@@ -358,7 +362,10 @@ def read_config(
         ):
             instances = _instance_to_list_converter(getattr(self.writers, stream))
             for instance in instances:
-                if instance != "master" and instance not in self.instance_map:
+                if (
+                    instance != MAIN_PROCESS_INSTANCE_NAME
+                    and instance not in self.instance_map
+                ):
                     raise ConfigError(
                         "Instance %r is configured to write %s but does not appear in `instance_map` config."
                         % (instance, stream)
@@ -397,6 +404,11 @@ def read_config(
                 "Must only specify one instance to handle `push` messages."
             )
 
+        if len(self.writers.device_lists) == 0:
+            raise ConfigError(
+                "Must specify at least one instance to handle `device_lists` messages."
+            )
+
         self.events_shard_config = RoutableShardedWorkerHandlingConfig(
             self.writers.events
         )
@@ -419,9 +431,12 @@ def read_config(
         #
         # No effort is made to ensure only a single instance of these tasks is
         # running.
-        background_tasks_instance = config.get("run_background_tasks_on") or "master"
+        background_tasks_instance = (
+            config.get("run_background_tasks_on") or MAIN_PROCESS_INSTANCE_NAME
+        )
         self.run_background_tasks = (
-            self.worker_name is None and background_tasks_instance == "master"
+            self.worker_name is None
+            and background_tasks_instance == MAIN_PROCESS_INSTANCE_NAME
         ) or self.worker_name == background_tasks_instance
 
         self.should_notify_appservices = self._should_this_worker_perform_duty(
@@ -493,9 +508,10 @@ def _should_this_worker_perform_duty(
         # 'don't run here'.
         new_option_should_run_here = None
         if new_option_name in config:
-            designated_worker = config[new_option_name] or "master"
+            designated_worker = config[new_option_name] or MAIN_PROCESS_INSTANCE_NAME
             new_option_should_run_here = (
-                designated_worker == "master" and self.worker_name is None
+                designated_worker == MAIN_PROCESS_INSTANCE_NAME
+                and self.worker_name is None
             ) or designated_worker == self.worker_name
 
         legacy_option_should_run_here = None
@@ -592,7 +608,7 @@ def _worker_names_performing_this_duty(
             # If no worker instances are set we check if the legacy option
             # is set, which means use the main process.
             if legacy_option:
-                worker_instances = ["master"]
+                worker_instances = [MAIN_PROCESS_INSTANCE_NAME]
 
             if self.worker_app == legacy_app_name:
                 if legacy_option:

synapse/handlers/appservice.py

@@ -638,7 +638,8 @@ async def _get_device_list_summary(
 
         # Fetch the users who have modified their device list since then.
         users_with_changed_device_lists = await self.store.get_all_devices_changed(
-            from_key, to_key=new_key
+            MultiWriterStreamToken(stream=from_key),
+            to_key=MultiWriterStreamToken(stream=new_key),
         )
 
         # Filter out any users the application service is not interested in

synapse/handlers/deactivate_account.py

@@ -24,7 +24,6 @@
 
 from synapse.api.constants import Membership
 from synapse.api.errors import SynapseError
-from synapse.handlers.device import DeviceHandler
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.types import Codes, Requester, UserID, create_requester
 
@@ -84,10 +83,6 @@ async def deactivate_account(
         Returns:
             True if identity server supports removing threepids, otherwise False.
         """
-
-        # This can only be called on the main process.
-        assert isinstance(self._device_handler, DeviceHandler)
-
         # Check if this user can be deactivated
         if not await self._third_party_rules.check_can_deactivate_user(
             user_id, by_admin