Calculate device list conversion lag in a single transaction

Read the converted stream position and the oldest unconverted row in
one transaction, rather than one (or two) each.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Thn52seV72QdxwDx5etBPM

synapse/handlers/device.py

@@ -1063,7 +1063,7 @@ class DeviceWriterHandler(DeviceHandler):
         """Report how far behind we are at converting rows in
         `device_lists_changes_in_room` to `device_lists_outbound_pokes`.
         """
-        oldest_ts = await self.store.get_oldest_unconverted_device_list_change_ts()
+        oldest_ts, converted_pos = await self.store.get_device_list_conversion_lag()
 
         if oldest_ts is None:
             device_list_conversion_lag_ms = 0
@@ -1078,7 +1078,6 @@ class DeviceWriterHandler(DeviceHandler):
         # backlog: the converted position only advances when the conversion
         # loop runs, and stream IDs in the gap may not have rows needing
         # conversion at all.
-        converted_pos, _ = await self.store.get_device_change_last_converted_pos()
         current_pos = self.store.get_device_stream_token().stream
 
         device_list_conversion_stream_lag_gauge.labels(

synapse/storage/databases/main/devices.py

@@ -2476,19 +2476,18 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
         )
         return cast(tuple[int, str], min(rows))
 
-    async def get_oldest_unconverted_device_list_change_ts(self) -> int | None:
-        """Get the timestamp of the oldest row in `device_lists_changes_in_room`
-        that has yet to be converted to `device_lists_outbound_pokes`, i.e. how
-        far behind the conversion loop is.
+    async def get_device_list_conversion_lag(self) -> tuple[int | None, int]:
+        """Get how far behind we are at converting rows in
+        `device_lists_changes_in_room` to `device_lists_outbound_pokes`.
 
         Returns:
-            The timestamp (ms) at which the oldest unconverted change was
-            inserted. None if there is nothing to convert, or if the oldest
-            row predates the `inserted_ts` column.
+            A tuple of:
+                1. the timestamp (ms) at which the oldest unconverted change
+                   was inserted. None if there is nothing to convert, or if
+                   the oldest row predates the `inserted_ts` column.
+                2. the stream ID of the last converted position.
         """
 
-        stream_id, room_id = await self.get_device_change_last_converted_pos()
-
         # Rows for one device list update share a `stream_id` (and insertion
         # time), so ordering by `stream_id` alone is fine.
         sql = """
@@ -2500,16 +2499,26 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
             LIMIT 1
         """
 
-        def get_oldest_unconverted_device_list_change_ts_txn(
+        def get_device_list_conversion_lag_txn(
             txn: LoggingTransaction,
-        ) -> int | None:
+        ) -> tuple[int | None, int]:
+            # As in `get_device_change_last_converted_pos`, take the minimum
+            # of all rows in case there is more than one.
+            pos_rows = self.db_pool.simple_select_list_txn(
+                txn,
+                table="device_lists_changes_converted_stream_position",
+                keyvalues={},
+                retcols=["stream_id", "room_id"],
+            )
+            stream_id, room_id = cast(tuple[int, str], min(pos_rows))
+
             txn.execute(sql, (stream_id, room_id))
             row = txn.fetchone()
-            return row[0] if row else None
+            return row[0] if row else None, stream_id
 
         return await self.db_pool.runInteraction(
-            "get_oldest_unconverted_device_list_change_ts",
-            get_oldest_unconverted_device_list_change_ts_txn,
+            "get_device_list_conversion_lag",
+            get_device_list_conversion_lag_txn,
         )
 
     async def set_device_change_last_converted_pos(