Allow running sendToDevice on workers

changelog.d/9043.feature

@@ -1 +1 @@
-Allow handling `/sendToDevice` API endpoints on workers.
+Add experimental support for handling and persistence of to-device messages to happen on worker processes.

changelog.d/9044.feature

@@ -1 +1 @@
-Allow handling `/sendToDevice` API endpoints on workers.
+Add experimental support for handling and persistence of to-device messages to happen on worker processes

scripts/synapse_port_db

@@ -629,6 +629,7 @@ class Porter(object):
             await self._setup_state_group_id_seq()
             await self._setup_user_id_seq()
             await self._setup_events_stream_seqs()
+            await self._setup_device_inbox_seq()
 
             # Step 3. Get tables.
             self.progress.set_state("Fetching tables")
@@ -911,6 +912,32 @@ class Porter(object):
             "_setup_events_stream_seqs", _setup_events_stream_seqs_set_pos,
         )
 
+    async def _setup_device_inbox_seq(self):
+        """Set the device inbox sequence to the correct value.
+        """
+        curr_local_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
+            table="device_inbox",
+            keyvalues={},
+            retcol="COALESCE(MAX(stream_id), 1)",
+            allow_none=True,
+        )
+
+        curr_federation_id = await self.sqlite_store.db_pool.simple_select_one_onecol(
+            table="device_federation_outbox",
+            keyvalues={},
+            retcol="COALESCE(MAX(stream_id), 1)",
+            allow_none=True,
+        )
+
+        next_id = max(curr_local_id, curr_federation_id) + 1
+
+        def r(txn):
+            txn.execute(
+                "ALTER SEQUENCE device_inbox_sequence RESTART WITH %s", (next_id,)
+            )
+
+        return self.postgres_store.db_pool.runInteraction("_setup_device_inbox_seq", r)
+
 
 ##############################################
 # The following is simply UI stuff

synapse/federation/federation_server.py

@@ -861,7 +861,9 @@ def __init__(self, hs: "HomeServer"):
         )  # type: Dict[str, Callable[[str, dict], Awaitable[None]]]
         self.query_handlers = {}  # type: Dict[str, Callable[[dict], Awaitable[None]]]
 
-        # Map from type to instance name that we should route EDU handling to.
+        # Map from type to instance names that we should route EDU handling to.
+        # We randomly choose one instance from the list to route to for each new
+        # EDU received.
         self._edu_type_to_instance = {}  # type: Dict[str, List[str]]
 
     def register_edu_handler(

synapse/handlers/devicemessage.py

@@ -47,7 +47,9 @@ def __init__(self, hs: "HomeServer"):
         self.is_mine = hs.is_mine
 
         # We only need to poke the federation sender explicitly if its on the
-        # same instance.
+        # same instance. Other federation sender instances will get notified by
+        # the `generic_worker` when it sees it in the to-device replication
+        # stream.
         self.federation_sender = None
         if hs.should_send_federation():
             self.federation_sender = hs.get_federation_sender()
@@ -63,6 +65,9 @@ def __init__(self, hs: "HomeServer"):
                 "m.direct_to_device", hs.config.worker.writers.to_device,
             )
 
+        # The handler to call when we think a user's device list might be out of
+        # sync. We do all device list resyncing on the master instance, so if
+        # we're on a worker we hit the device resync replication API.
         if hs.config.worker.worker_app is None:
             self._user_device_resync = (
                 hs.get_device_handler().device_list_updater.user_device_resync

synapse/storage/databases/main/__init__.py

@@ -127,9 +127,6 @@ def __init__(self, database: DatabasePool, db_conn, hs):
         self._presence_id_gen = StreamIdGenerator(
             db_conn, "presence_stream", "stream_id"
         )
-        self._device_inbox_id_gen = StreamIdGenerator(
-            db_conn, "device_inbox", "stream_id"
-        )
         self._public_room_id_gen = StreamIdGenerator(
             db_conn, "public_room_list_stream", "stream_id"
         )