Track device list updates per room.

synapse/handlers/device.py

@@ -485,6 +485,7 @@ async def notify_device_update(
             user_id,
             device_ids,
             hosts=hosts,
+            room_ids=room_ids,
         )
 
         if not position:

synapse/replication/slave/storage/devices.py

@@ -44,6 +44,7 @@ def __init__(
             extra_tables=[
                 ("user_signature_stream", "stream_id"),
                 ("device_lists_outbound_pokes", "stream_id"),
+                ("device_lists_changes_in_room", "stream_id"),
             ],
         )
         device_list_max = self._device_list_id_gen.get_current_token()

synapse/storage/databases/main/__init__.py

@@ -146,6 +146,7 @@ def __init__(
             extra_tables=[
                 ("user_signature_stream", "stream_id"),
                 ("device_lists_outbound_pokes", "stream_id"),
+                ("device_lists_changes_in_room", "stream_id"),
             ],
         )
 

synapse/storage/databases/main/devices.py

@@ -1506,7 +1506,11 @@ def _update_remote_device_list_cache_txn(
         )
 
     async def add_device_change_to_streams(
-        self, user_id: str, device_ids: Collection[str], hosts: Collection[str]
+        self,
+        user_id: str,
+        device_ids: Collection[str],
+        hosts: Collection[str],
+        room_ids: Collection[str],
     ) -> Optional[int]:
         """Persist that a user's devices have been updated, and which hosts
         (if any) should be poked.
@@ -1516,6 +1520,7 @@ async def add_device_change_to_streams(
             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.
+            room_ids: The rooms that the user is in
 
         Returns:
             The maximum stream ID of device list updates that were added to the database, or
@@ -1535,10 +1540,26 @@ async def add_device_change_to_streams(
                 stream_ids,
             )
 
-        if not hosts:
+        if not room_ids:
             return stream_ids[-1]
 
         context = get_active_span_text_map()
+        async with self._device_list_id_gen.get_next_mult(
+            len(room_ids) * len(device_ids)
+        ) as stream_ids:
+            await self.db_pool.runInteraction(
+                "_add_device_outbound_room_poke_txn",
+                self._add_device_outbound_room_poke_txn,
+                user_id,
+                device_ids,
+                room_ids,
+                stream_ids,
+                context,
+            )
+
+        if not hosts:
+            return stream_ids[-1]
+
         async with self._device_list_id_gen.get_next_mult(
             len(hosts) * len(device_ids)
         ) as stream_ids:
@@ -1636,3 +1657,37 @@ def _add_device_outbound_poke_to_stream_txn(
                 for device_id in device_ids
             ],
         )
+
+    def _add_device_outbound_room_poke_txn(
+        self,
+        txn: LoggingTransaction,
+        user_id: str,
+        device_ids: Iterable[str],
+        room_ids: Collection[str],
+        stream_ids: List[str],
+        context: Dict[str, str],
+    ) -> None:
+        next_stream_id = iter(stream_ids)
+
+        self.db_pool.simple_insert_many_txn(
+            txn,
+            table="device_lists_changes_in_room",
+            keys=(
+                "stream_id",
+                "room_id",
+                "user_id",
+                "device_id",
+                "opentracing_context",
+            ),
+            values=[
+                (
+                    next(next_stream_id),
+                    room_id,
+                    user_id,
+                    device_id,
+                    json_encoder.encode(context),
+                )
+                for room_id in room_ids
+                for device_id in device_ids
+            ],
+        )

synapse/storage/schema/__init__.py

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-SCHEMA_VERSION = 68  # remember to update the list below when updating
+SCHEMA_VERSION = 69  # remember to update the list below when updating
 """Represents the expectations made by the codebase about the database schema
 
 This should be incremented whenever the codebase changes its requirements on the
@@ -58,6 +58,9 @@
     - event_reference_hashes is no longer read.
     - `events` has `state_key` and `rejection_reason` columns, which are populated for
       new events.
+
+Changes in SCHEMA_VERSION = 69:
+    - We now write to `device_lists_changes_in_room` table.
 """
 
 

synapse/storage/schema/main/delta/69/01device_list_oubound_by_room.sql

@@ -0,0 +1,24 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+CREATE TABLE device_lists_changes_in_room (
+    stream_id BIGINT NOT NULL,
+    room_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    device_id TEXT NOT NULL,
+    opentracing_context TEXT
+);
+
+CREATE UNIQUE INDEX device_lists_changes_in_stream_id ON device_lists_changes_in_room(stream_id);