Ensure that remote users' device list resyncing always happens on master (#9043)

Currently `DeviceMessageHandler` only ever exists on master, but that is about to change.

changelog.d/9043.feature

@@ -0,0 +1 @@
+Add experimental support for handling and persistence of to-device messages to happen on worker processes.

synapse/handlers/devicemessage.py

@@ -24,6 +24,7 @@ from synapse.logging.opentracing import (
     set_tag,
     start_active_span,
 )
+from synapse.replication.http.devices import ReplicationUserDevicesResyncRestServlet
 from synapse.types import JsonDict, UserID, get_domain_from_id
 from synapse.util import json_encoder
 from synapse.util.stringutils import random_string
@@ -50,7 +51,17 @@ class DeviceMessageHandler:
             "m.direct_to_device", self.on_direct_to_device_edu
         )
 
-        self._device_list_updater = hs.get_device_handler().device_list_updater
+        # 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
+            )
+        else:
+            self._user_device_resync = ReplicationUserDevicesResyncRestServlet.make_client(
+                hs
+            )
 
     async def on_direct_to_device_edu(self, origin: str, content: JsonDict) -> None:
         local_messages = {}
@@ -138,9 +149,7 @@ class DeviceMessageHandler:
             await self.store.mark_remote_user_device_cache_as_stale(sender_user_id)
 
             # Immediately attempt a resync in the background
-            run_in_background(
+            run_in_background(self._user_device_resync, sender_user_id)
-                self._device_list_updater.user_device_resync, sender_user_id
-            )
 
     async def send_device_message(
         self,