Track device list updates per room.

synapse/config/server.py

@@ -680,6 +680,14 @@ def read_config(self, config, **kwargs):
             config.get("use_account_validity_in_account_status") or False
         )
 
+        # This is a temporary option that enables fully using the new
+        # `device_lists_changes_in_room` without the backwards compat code. This
+        # is primarily for testing. If enabled the server should *not* be
+        # downgraded, as it may lead to missing device list updates.
+        self.use_new_device_lists_changes_in_room = (
+            config.get("use_new_device_lists_changes_in_room") or False
+        )
+
     def has_tls_listener(self) -> bool:
         return any(listener.tls for listener in self.listeners)
 

synapse/handlers/device.py

@@ -291,6 +291,12 @@ def __init__(self, hs: "HomeServer"):
         # On start up check if there are any updates pending.
         hs.get_reactor().callWhenRunning(self._handle_new_device_update_async)
 
+        # Used to decide if we calculate outbound pokes up front or not. By
+        # default we do to allow safely downgrading Synapse.
+        self.use_new_device_lists_changes_in_room = (
+            hs.config.server.use_new_device_lists_changes_in_room
+        )
+
     def _check_device_name_length(self, name: Optional[str]) -> None:
         """
         Checks whether a device name is longer than the maximum allowed length.
@@ -484,15 +490,18 @@ async def notify_device_update(
 
         room_ids = await self.store.get_rooms_for_user(user_id)
 
-        hosts: Set[str] = set()
-        if self.hs.is_mine_id(user_id):
-            for room_id in room_ids:
-                joined_users = await self.store.get_users_in_room(room_id)
-                hosts.update(get_domain_from_id(u) for u in joined_users)
+        hosts: Optional[Set[str]] = None
+        if self.use_new_device_lists_changes_in_room:
+            hosts = set()
+
+            if self.hs.is_mine_id(user_id):
+                for room_id in room_ids:
+                    joined_users = await self.store.get_users_in_room(room_id)
+                    hosts.update(get_domain_from_id(u) for u in joined_users)
 
-        hosts.discard(self.server_name)
+                set_tag("target_hosts", hosts)
 
-        set_tag("target_hosts", hosts)
+                hosts.discard(self.server_name)
 
         position = await self.store.add_device_change_to_streams(
             user_id,

synapse/storage/databases/main/devices.py

@@ -1509,7 +1509,7 @@ async def add_device_change_to_streams(
         self,
         user_id: str,
         device_ids: Collection[str],
-        hosts: Collection[str],
+        hosts: Optional[Collection[str]],
         room_ids: Collection[str],
     ) -> Optional[int]:
         """Persist that a user's devices have been updated, and which hosts
@@ -1519,7 +1519,9 @@ async def add_device_change_to_streams(
             user_id: The ID of the user whose device changed.
             device_ids: The IDs of any changed devices. If empty, this function will
                 return None.
-            hosts: The remote destinations that should be notified of the change.
+            hosts: The remote destinations that should be notified of the change. If
+                None then the set of hosts have *not* been calculated, and will be
+                calculated later by a background task.
             room_ids: The rooms that the user is in
 
         Returns:
@@ -1529,10 +1531,9 @@ async def add_device_change_to_streams(
         if not device_ids:
             return None
 
-        num_stream_ids = max(
-            len(device_ids),
-            len(hosts) * len(device_ids),
-        )
+        num_stream_ids = len(device_ids)
+        if hosts:
+            num_stream_ids = len(hosts) * len(device_ids)
 
         context = get_active_span_text_map()
 
@@ -1554,6 +1555,7 @@ def add_device_changes_txn(txn, stream_ids):
                 room_ids,
                 stream_ids,
                 context,
+                hosts_have_been_calculated=hosts is not None,
             )
 
             if not hosts:
@@ -1668,7 +1670,20 @@ def _add_device_outbound_room_poke_txn(
         room_ids: Collection[str],
         stream_ids: List[str],
         context: Dict[str, str],
+        hosts_have_been_calculated: bool,
     ) -> None:
+        """Record the user in the room has updated their device.
+
+        Args:
+            hosts_have_been_calculated: True if `device_lists_outbound_pokes`
+                has been updated already with the updates.
+        """
+
+        # We only need to convert to outbound pokes if they are our user.
+        converted_to_destinations = (
+            hosts_have_been_calculated or not self.hs.is_mine_id(user_id)
+        )
+
         self.db_pool.simple_insert_many_txn(
             txn,
             table="device_lists_changes_in_room",
@@ -1686,7 +1701,7 @@ def _add_device_outbound_room_poke_txn(
                     device_id,
                     room_id,
                     stream_id,
-                    True,  # As we're updating `device_lists_outbound_pokes` at the same time.
+                    converted_to_destinations,
                     json_encoder.encode(context),
                 )
                 for room_id in room_ids