diff --git a/.ci/scripts/schema_diff.py b/.ci/scripts/schema_diff.py index 9b7af72434..8354b64233 100755 --- a/.ci/scripts/schema_diff.py +++ b/.ci/scripts/schema_diff.py @@ -193,7 +193,7 @@ def main() -> None: # Refresh dependencies print("Installing dependencies for base commit...", file=sys.stderr) subprocess.run( - ["poetry", "install", "--no-root", "--extras", "postgres"], + ["poetry", "install", "--extras", "postgres"], cwd=REPO_ROOT, check=True, # Poetry install is noisy, so pipe its stdout to stderr diff --git a/changelog.d/20098.misc b/changelog.d/20098.misc new file mode 100644 index 0000000000..0504be872d --- /dev/null +++ b/changelog.d/20098.misc @@ -0,0 +1 @@ +Speed up the conversion of device list changes into outbound federation pokes, and add a metric for how far behind the conversion is. diff --git a/changelog.d/20117.misc b/changelog.d/20117.misc new file mode 100644 index 0000000000..92d421117e --- /dev/null +++ b/changelog.d/20117.misc @@ -0,0 +1 @@ +Fix the schema diff CI breaking when the Rust module was changed. \ No newline at end of file diff --git a/synapse/handlers/device.py b/synapse/handlers/device.py index 2225466648..ae61bc24bf 100644 --- a/synapse/handlers/device.py +++ b/synapse/handlers/device.py @@ -30,6 +30,8 @@ from typing import ( cast, ) +from prometheus_client import Gauge + from synapse.api import errors from synapse.api.constants import EduTypes, EventTypes, Membership from synapse.api.errors import ( @@ -41,6 +43,7 @@ from synapse.api.errors import ( SynapseError, ) from synapse.logging.opentracing import log_kv, set_tag, trace +from synapse.metrics import SERVER_NAME_LABEL from synapse.metrics.background_process_metrics import ( wrap_as_background_process, ) @@ -89,6 +92,21 @@ DELETE_DEVICE_MSGS_TASK_NAME = "delete_device_messages" MAX_DEVICE_DISPLAY_NAME_LEN = 100 DELETE_STALE_DEVICES_INTERVAL = Duration(days=1) +device_list_conversion_lag_gauge = Gauge( + "synapse_device_lists_changes_conversion_lag_seconds", + "Age of the oldest device list change that has yet to be converted to outbound federation pokes", + labelnames=[SERVER_NAME_LABEL], +) + +device_list_conversion_stream_lag_gauge = Gauge( + "synapse_device_lists_changes_conversion_stream_lag", + "Number of stream IDs between the current device lists stream position and the position converted to outbound federation pokes", + labelnames=[SERVER_NAME_LABEL], +) + +# How often to update the device list conversion lag gauges. +DEVICE_LIST_CONVERSION_LAG_GAUGE_METRIC_UPDATE_INTERVAL = Duration(seconds=30) + def _check_device_name_length(name: str | None) -> None: """ @@ -960,6 +978,13 @@ class DeviceWriterHandler(DeviceHandler): self.device_list_updater.incoming_device_list_update, ) + # Report how far behind we are at converting device list changes + # into outbound pokes. + self.clock.looping_call( + self._report_device_list_conversion_lag, + DEVICE_LIST_CONVERSION_LAG_GAUGE_METRIC_UPDATE_INTERVAL, + ) + @trace @measure_func("notify_device_update") async def notify_device_update( @@ -1033,6 +1058,35 @@ class DeviceWriterHandler(DeviceHandler): self._handle_new_device_update_async() return + @wrap_as_background_process("_report_device_list_conversion_lag") + async def _report_device_list_conversion_lag(self) -> None: + """Report how far behind we are at converting rows in + `device_lists_changes_in_room` to `device_lists_outbound_pokes`. + """ + ( + oldest_ts, + last_converted_pos, + ) = await self.store.get_device_list_conversion_lag() + + if oldest_ts is None: + device_list_conversion_lag_ms = 0 + else: + device_list_conversion_lag_ms = max(0, self.clock.time_msec() - oldest_ts) + + device_list_conversion_lag_gauge.labels( + **{SERVER_NAME_LABEL: self.server_name} + ).set(device_list_conversion_lag_ms / 1000.0) # convert to seconds + + # The stream ID lag is only an approximation of the conversion + # 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. + current_pos = self.store.get_device_stream_token().stream + + device_list_conversion_stream_lag_gauge.labels( + **{SERVER_NAME_LABEL: self.server_name} + ).set(max(0, current_pos - last_converted_pos)) + @wrap_as_background_process("_handle_new_device_update_async") async def _handle_new_device_update_async(self) -> None: """Called when we have a new local device list update that we need to diff --git a/synapse/storage/background_updates.py b/synapse/storage/background_updates.py index 311534c5e7..8137b02036 100644 --- a/synapse/storage/background_updates.py +++ b/synapse/storage/background_updates.py @@ -262,6 +262,10 @@ class BackgroundUpdater: # enum? self._all_done = False + # A set of background updates that we have queried the database for and + # found to be completed. + self._completed_background_updates: set[str] = set() + # Whether we're currently running updates self._running = False @@ -394,9 +398,15 @@ class BackgroundUpdater: return perf def start_doing_background_updates(self) -> None: + """Start doing background updates in the background. + + This gets called both on startup and when the admin API is used to + reschedule background updates. + """ if self.enabled: # if we start a new background update, not all updates are done. self._all_done = False + self._completed_background_updates.clear() sleep = self.sleep_enabled self.hs.run_as_background_process( "background_updates", @@ -478,6 +488,9 @@ class BackgroundUpdater: if update_name == self._current_background_update: return False + if update_name in self._completed_background_updates: + return True + update_exists = await self.db_pool.simple_select_one_onecol( "background_updates", keyvalues={"update_name": update_name}, @@ -486,6 +499,9 @@ class BackgroundUpdater: allow_none=True, ) + if not update_exists: + self._completed_background_updates.add(update_name) + return not update_exists async def have_completed_background_updates( diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py index 0e4c8ac491..69470ae6e6 100644 --- a/synapse/storage/databases/main/devices.py +++ b/synapse/storage/databases/main/devices.py @@ -83,6 +83,10 @@ BG_UPDATE_REMOVE_DUP_OUTBOUND_POKES = "remove_dup_outbound_pokes" # `device_lists_changes_in_room.inserted_ts`. BG_UPDATE_ADD_INSERTED_TS_INDEX = "device_lists_changes_in_room_inserted_ts_idx" +# Background update name for adding an index on unconverted rows in +# `device_lists_changes_in_room`. +BG_UPDATE_ADD_UNCONVERTED_IDX = "device_lists_changes_in_room_unconverted_idx" + # Prunes entries out of the `device_lists_changes_in_room` table that are more # than this old. @@ -2204,7 +2208,22 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore): converted_upto_stream_id: int, ) -> None: """If we've calculated the outbound pokes for a given room/device list - update, mark any subsequent changes as already converted""" + update, mark any subsequent changes as already converted. + + This is an optimization only. Skipping it is always safe, and just + means the subsequent changes get converted individually. + """ + + # Without the index added by `BG_UPDATE_ADD_UNCONVERTED_IDX`, the + # UPDATE below scans the unconverted backlog on every call, getting + # slower the further behind we are. Skip it until the index exists. + unconverted_idx_ready = ( + await self.db_pool.updates.has_completed_background_update( + BG_UPDATE_ADD_UNCONVERTED_IDX + ) + ) + if not unconverted_idx_ready: + return sql = """ UPDATE device_lists_changes_in_room @@ -2446,17 +2465,69 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore): `FALSE` have not been converted. """ + return await self.db_pool.runInteraction( + desc="get_device_change_last_converted_pos", + func=self.get_device_change_last_converted_pos_txn, + db_autocommit=True, + ) + + def get_device_change_last_converted_pos_txn( + self, txn: LoggingTransaction + ) -> tuple[int, str]: + """Get the position of the last row in `device_list_changes_in_room` that has been + converted to `device_lists_outbound_pokes`. + + Rows with a strictly greater position where `converted_to_destinations` is + `FALSE` have not been converted.""" + # There should be only one row in this table, though we want to # future-proof ourselves for when we have multiple rows (one for each # instance). So to handle that case we take the minimum of all rows. - rows = await self.db_pool.simple_select_list( + rows = self.db_pool.simple_select_list_txn( + txn, table="device_lists_changes_converted_stream_position", keyvalues={}, retcols=["stream_id", "room_id"], - desc="get_device_change_last_converted_pos", ) return cast(tuple[int, str], min(rows)) + 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: + 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. + """ + + # Rows for one device list update share a `stream_id` (and insertion + # time), so ordering by `stream_id` alone is fine. + sql = """ + SELECT inserted_ts FROM device_lists_changes_in_room + WHERE + (stream_id, room_id) > (?, ?) AND + NOT converted_to_destinations + ORDER BY stream_id ASC + LIMIT 1 + """ + + def get_device_list_conversion_lag_txn( + txn: LoggingTransaction, + ) -> tuple[int | None, int]: + stream_id, room_id = self.get_device_change_last_converted_pos_txn(txn) + + txn.execute(sql, (stream_id, room_id)) + row = txn.fetchone() + return row[0] if row else None, stream_id + + return await self.db_pool.runInteraction( + "get_device_list_conversion_lag", + get_device_list_conversion_lag_txn, + ) + async def set_device_change_last_converted_pos( self, stream_id: int, @@ -2699,6 +2770,15 @@ class DeviceBackgroundUpdateStore(SQLBaseStore): where_clause="inserted_ts IS NOT NULL", ) + # Add an index to speed up `mark_redundant_device_lists_pokes`. + self.db_pool.updates.register_background_index_update( + BG_UPDATE_ADD_UNCONVERTED_IDX, + index_name="device_lists_changes_in_room_unconverted_idx", + table="device_lists_changes_in_room", + columns=["user_id", "device_id", "room_id", "stream_id"], + where_clause="NOT converted_to_destinations", + ) + async def _drop_device_list_streams_non_unique_indexes( self, progress: JsonDict, batch_size: int ) -> int: diff --git a/synapse/storage/schema/main/delta/94/08_device_lists_changes_in_room_unconverted_idx.sql b/synapse/storage/schema/main/delta/94/08_device_lists_changes_in_room_unconverted_idx.sql new file mode 100644 index 0000000000..6bebcab11d --- /dev/null +++ b/synapse/storage/schema/main/delta/94/08_device_lists_changes_in_room_unconverted_idx.sql @@ -0,0 +1,22 @@ +-- +-- This file is licensed under the Affero General Public License (AGPL) version 3. +-- +-- Copyright (C) 2026 Element Creations Ltd +-- +-- This program is free software: you can redistribute it and/or modify +-- it under the terms of the GNU Affero General Public License as +-- published by the Free Software Foundation, either version 3 of the +-- License, or (at your option) any later version. +-- +-- See the GNU Affero General Public License for more details: +-- . + + +-- Add an index on `device_lists_changes_in_room(user_id, device_id, room_id, +-- stream_id)` for unconverted rows, so that marking redundant rows as +-- converted (in `mark_redundant_device_lists_pokes`) does not require a scan +-- of the unconverted backlog. +-- +-- This is a partial index as we only ever query for unconverted rows. +INSERT INTO background_updates (ordering, update_name, progress_json) VALUES + (9408, 'device_lists_changes_in_room_unconverted_idx', '{}');