Track device list updates per room.

synapse/_scripts/synapse_port_db.py

@@ -97,6 +97,7 @@
     "users": ["shadow_banned"],
     "e2e_fallback_keys_json": ["used"],
     "access_tokens": ["used"],
+    "device_lists_changes_in_room": ["converted_to_destinations"],
 }
 
 

synapse/handlers/device.py

@@ -491,7 +491,7 @@ async def notify_device_update(
         room_ids = await self.store.get_rooms_for_user(user_id)
 
         hosts: Optional[Set[str]] = None
-        if self.use_new_device_lists_changes_in_room:
+        if not self.use_new_device_lists_changes_in_room:
             hosts = set()
 
             if self.hs.is_mine_id(user_id):

synapse/storage/databases/main/devices.py

@@ -1555,6 +1555,8 @@ async def add_device_change_to_streams(
 
         num_stream_ids = len(device_ids)
         if hosts:
+            # The `device_lists_outbound_pokes` wants a different stream ID for
+            # each row, which is a row per host per device update.
             num_stream_ids = len(hosts) * len(device_ids)
 
         context = get_active_span_text_map()
@@ -1567,9 +1569,6 @@ def add_device_changes_txn(txn, stream_ids):
                 stream_ids,
             )
 
-            if not room_ids:
-                return
-
             self._add_device_outbound_room_poke_txn(
                 txn,
                 user_id,
@@ -1580,6 +1579,9 @@ def add_device_changes_txn(txn, stream_ids):
                 hosts_have_been_calculated=hosts is not None,
             )
 
+            # If the set of hosts to send to has not been calculated yet (and so
+            # `hosts` is None) or there are no `hosts` to send to, then skip
+            # trying to persist them to the DB.
             if not hosts:
                 return
 
@@ -1653,8 +1655,9 @@ def _add_device_outbound_poke_to_stream_txn(
             )
 
         now = self._clock.time_msec()
-        next_stream_id = iter(stream_ids)
+        stream_id_iterator = iter(stream_ids)
 
+        encoded_context = json_encoder.encode(context)
         self.db_pool.simple_insert_many_txn(
             txn,
             table="device_lists_outbound_pokes",
@@ -1670,14 +1673,12 @@ def _add_device_outbound_poke_to_stream_txn(
             values=[
                 (
                     destination,
-                    next(next_stream_id),
+                    next(stream_id_iterator),
                     user_id,
                     device_id,
                     False,
                     now,
-                    json_encoder.encode(context)
-                    if whitelisted_homeserver(destination)
-                    else "{}",
+                    encoded_context if whitelisted_homeserver(destination) else "{}",
                 )
                 for destination in hosts
                 for device_id in device_ids
@@ -1706,6 +1707,12 @@ def _add_device_outbound_room_poke_txn(
             hosts_have_been_calculated or not self.hs.is_mine_id(user_id)
         )
 
+        encoded_context = json_encoder.encode(context)
+
+        # The `device_lists_changes_in_room.stream_id` column matches the
+        # corresponding `stream_id` of the update in the `device_lists_stream`
+        # table, i.e. all rows persisted for the same device update will have
+        # the same `stream_id` (but different room IDs).
         self.db_pool.simple_insert_many_txn(
             txn,
             table="device_lists_changes_in_room",
@@ -1724,7 +1731,7 @@ def _add_device_outbound_room_poke_txn(
                     room_id,
                     stream_id,
                     converted_to_destinations,
-                    json_encoder.encode(context),
+                    encoded_context,
                 )
                 for room_id in room_ids
                 for device_id, stream_id in zip(device_ids, stream_ids)

synapse/storage/schema/__init__.py

@@ -61,6 +61,7 @@
 
 Changes in SCHEMA_VERSION = 69:
     - We now write to `device_lists_changes_in_room` table.
+    - Use sequence to generate future `application_services_txns.txn_id`s
 """
 
 

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

@@ -17,12 +17,22 @@ CREATE TABLE device_lists_changes_in_room (
     user_id TEXT NOT NULL,
     device_id TEXT NOT NULL,
     room_id TEXT NOT NULL,
-    stream_id BIGINT NOT NULL,  -- This matches `device_lists_stream.stream_id`
-    -- We track if we've calculated the hosts in the room and updated
-    -- `device_lists_outbound_pokes`.
+
+    -- This initialy matches `device_lists_stream.stream_id`. Note that we
+    -- delete older values from `device_lists_stream`, so we can't use a foreign
+    -- constraint here.
+    --
+    -- The table will contain rows with the same `stream_id` but different
+    -- `room_id`, as for each device update we store a row per room the user is
+    -- joined to. Therefore `(stream_id, room_id)` gives a unique index.
+    stream_id BIGINT NOT NULL,
+
+    -- We have a background process which goes through this table and converts
+    -- entries into rows in `device_lists_outbound_pokes`. Once we have processed
+    -- a row, we mark it as such by setting `converted_to_destinations=TRUE`.
     converted_to_destinations BOOLEAN NOT NULL,
     opentracing_context TEXT
 );
 
-CREATE INDEX device_lists_changes_in_stream_id ON device_lists_changes_in_room(stream_id);
-CREATE INDEX device_lists_changes_in_stream_id_converted ON device_lists_changes_in_room(stream_id) WHERE NOT converted_to_destinations;
+CREATE UNIQUE INDEX device_lists_changes_in_stream_id ON device_lists_changes_in_room(stream_id, room_id);
+CREATE INDEX device_lists_changes_in_stream_id_unconverted ON device_lists_changes_in_room(stream_id) WHERE NOT converted_to_destinations;