Move device changes off the main process

changelog.d/18581.feature

@@ -0,0 +1 @@
+Temporary for CI to pass.

docker/complement/conf/start_for_complement.sh

@@ -65,6 +65,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

@@ -306,6 +306,15 @@
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
+    "device_lists": {
+        "app": "synapse.app.generic_worker",
+        "listener_resources": ["client", "replication"],
+        "endpoint_patterns": [
+            "^/_matrix/client/(api/v1|r0|v3|unstable)/keys/signatures/upload$"
+        ],
+        "shared_extra_conf": {},
+        "worker_extra_conf": "",
+    },
     "typing": {
         "app": "synapse.app.generic_worker",
         "listener_resources": ["client", "replication"],
@@ -412,16 +421,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.
@@ -434,25 +444,13 @@ def add_worker_roles_to_shared_config(
     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
-        )
+        worker_types_set.add("events")
 
-        # 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)

docs/usage/configuration/config_documentation.md

@@ -4291,6 +4291,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:

schema/synapse-config.schema.yaml

@@ -5323,6 +5323,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

@@ -158,12 +158,13 @@ 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. Currently
+            can only be a single instance.
     """
 
     events: List[str] = attr.ib(
@@ -194,6 +195,10 @@ class WriterLocations:
         default=["master"],
         converter=_instance_to_list_converter,
     )
+    device_lists: List[str] = attr.ib(
+        default=["master"],
+        converter=_instance_to_list_converter,
+    )
 
 
 @attr.s(auto_attribs=True)
@@ -415,6 +420,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
         )

synapse/handlers/appservice.py

@@ -631,7 +631,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