Only send out device list updates for our own users

changelog.d/12465.feature

@@ -0,0 +1 @@
+Enable processing of device list updates asynchronously.

synapse/handlers/device.py

@@ -649,9 +649,13 @@ async def _handle_new_device_update_async(self) -> None:
                         return
 
                 for user_id, device_id, room_id, stream_id, opentracing_context in rows:
-                    joined_user_ids = await self.store.get_users_in_room(room_id)
-                    hosts = {get_domain_from_id(u) for u in joined_user_ids}
-                    hosts.discard(self.server_name)
+                    hosts = set()
+
+                    # Ignore any users that aren't ours
+                    if self.hs.is_mine_id(user_id):
+                        joined_user_ids = await self.store.get_users_in_room(room_id)
+                        hosts = {get_domain_from_id(u) for u in joined_user_ids}
+                        hosts.discard(self.server_name)
 
                     # Check if we've already sent this update to some hosts
                     if current_stream_id == stream_id:

synapse/storage/databases/main/devices.py

@@ -1703,7 +1703,9 @@ def _add_device_outbound_poke_to_stream_txn(
                     next(stream_id_iterator),
                     user_id,
                     device_id,
-                    False,
+                    not self.hs.is_mine_id(
+                        user_id
+                    ),  # We only need to send out update for *our* users
                     now,
                     encoded_context if whitelisted_homeserver(destination) else "{}",
                 )

tests/storage/test_devices.py

@@ -118,7 +118,7 @@ def test_get_device_updates_by_remote(self):
         device_ids = ["device_id1", "device_id2"]
 
         # Add two device updates with sequential `stream_id`s
-        self.add_device_change("user_id", device_ids, "somehost")
+        self.add_device_change("@user_id:test", device_ids, "somehost")
 
         # Get all device updates ever meant for this remote
         now_stream_id, device_updates = self.get_success(
@@ -142,7 +142,7 @@ def test_get_device_updates_by_remote_can_limit_properly(self):
             "device_id4",
             "device_id5",
         ]
-        self.add_device_change("user_id", device_ids, "somehost")
+        self.add_device_change("@user_id:test", device_ids, "somehost")
 
         # Get device updates meant for this remote
         next_stream_id, device_updates = self.get_success(
@@ -162,7 +162,7 @@ def test_get_device_updates_by_remote_can_limit_properly(self):
 
         # Add some more device updates to ensure it still resumes properly
         device_ids = ["device_id6", "device_id7"]
-        self.add_device_change("user_id", device_ids, "somehost")
+        self.add_device_change("@user_id:test", device_ids, "somehost")
 
         # Get the next batch of device updates
         next_stream_id, device_updates = self.get_success(