Merge remote-tracking branch 'origin/develop' into jaywink/msc4262

.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

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.

changelog.d/20117.misc

@@ -0,0 +1 @@
+Fix the schema diff CI breaking when the Rust module was changed.

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

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(

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:

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:
+-- <https://www.gnu.org/licenses/agpl-3.0.html>.
+
+
+-- 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', '{}');