From fd569dba9cc113cdd2a874f57cbe309325bd74be Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 13 Aug 2026 09:09:43 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01Thn52seV72QdxwDx5etBPM --- synapse/handlers/device.py | 3 +- synapse/storage/databases/main/devices.py | 37 ++++++++++++++--------- 2 files changed, 24 insertions(+), 16 deletions(-) diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index 70d6fb72fd..bd2171a6bb 100644 --- a/synapse/handlers/device.py +++ b/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( diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py index 21d73f78cb..202747018a 100644 --- a/synapse/storage/databases/main/devices.py +++ b/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(