Record max pruned ID in a separate table

Rather than trying to infer it from the minimum ID in the table.

We were seeing issues in CI due to not all device list updates having
entries in the `device_lists_changes_in_rooms` table due to the user not
being in any rooms. This meant that the returned minimum ID was larger
than expected, causing failures when calling
`get_all_device_list_changes(..)`.

By explicitly tracking the max pruned ID, we don't have to worry about
problems with trying to infer the actual max pruned ID.

Author: erikjohnston

synapse/storage/databases/main/devices.py

@@ -1745,16 +1745,21 @@ def get_devices_not_accessed_since_txn(
 
         return devices
 
-    def _get_min_device_lists_changes_in_room_txn(self, txn: LoggingTransaction) -> int:
-        """Returns the minimum stream ID that we have entries for
-        `device_lists_changes_in_room`
+    def _get_max_pruned_device_lists_changes_in_room_txn(
+        self, txn: LoggingTransaction
+    ) -> int:
+        """Returns the maximum stream ID that has been pruned from
+        `device_lists_changes_in_room`.
+
+        Any queries for stream IDs less than this value cannot be answered
+        completely, as the data has been deleted.
         """
 
         return self.db_pool.simple_select_one_onecol_txn(
             txn,
-            table="device_lists_changes_in_room",
+            table="device_lists_changes_in_room_max_pruned_stream_id",
             keyvalues={},
-            retcol="COALESCE(MIN(stream_id), 0)",
+            retcol="stream_id",
             allow_none=False,
         )
 
@@ -1783,9 +1788,11 @@ async def get_device_list_changes_in_rooms(
         def _get_device_list_changes_in_rooms_txn(
             txn: LoggingTransaction,
         ) -> set[str] | None:
-            # Check if the from_token is too old.
-            lowest_known_stream_id = self._get_min_device_lists_changes_in_room_txn(txn)
-            if lowest_known_stream_id > from_token.stream:
+            # Check if the from_token is too old (i.e. data has been pruned).
+            max_pruned_stream_id = (
+                self._get_max_pruned_device_lists_changes_in_room_txn(txn)
+            )
+            if max_pruned_stream_id > from_token.stream:
                 return None
 
             changes: set[str] = set()
@@ -1831,10 +1838,16 @@ async def get_all_device_list_changes(self, from_id: int, to_id: int) -> set[str
         def _get_all_device_list_changes_txn(
             txn: LoggingTransaction,
         ) -> set[str] | None:
-            # Check if the from_token is too old. We do this each time as we may
-            # prune the table in between runs.
-            lowest_known_stream_id = self._get_min_device_lists_changes_in_room_txn(txn)
-            if lowest_known_stream_id > from_id:
+            # Check if the from_token is too old (i.e. data has been pruned).
+            max_pruned_stream_id = (
+                self._get_max_pruned_device_lists_changes_in_room_txn(txn)
+            )
+            if max_pruned_stream_id > from_id:
+                logger.warning(
+                    "Given stream ID is too old %d < %d",
+                    from_id,
+                    max_pruned_stream_id,
+                )
                 return None
 
             sql = """
@@ -1851,7 +1864,7 @@ def _get_all_device_list_changes_txn(
         )
 
         if room_ids is None:
-            raise Exception("Given stream ID is too old")
+            raise Exception(f"Given stream ID is too old {from_id}")
 
         return room_ids
 
@@ -1870,9 +1883,11 @@ async def get_device_list_changes_in_room(
         def get_device_list_changes_in_room_txn(
             txn: LoggingTransaction,
         ) -> Collection[tuple[str, str]] | None:
-            # Check if the from_token is too old.
-            lowest_known_stream_id = self._get_min_device_lists_changes_in_room_txn(txn)
-            if lowest_known_stream_id > min_stream_id:
+            # Check if the from_token is too old (i.e. data has been pruned).
+            max_pruned_stream_id = (
+                self._get_max_pruned_device_lists_changes_in_room_txn(txn)
+            )
+            if max_pruned_stream_id > min_stream_id:
                 return None
 
             sql = """
@@ -2511,10 +2526,9 @@ def get_prune_before_stream_id_txn(txn: LoggingTransaction) -> int | None:
         if prune_before_stream_id is None:
             return
 
-        # Get the max stream ID that we have in the table, so that we avoid
-        # deleting it. We want to keep the max stream ID so that the minimum
-        # stream ID can be calculated in
-        # `_get_min_device_lists_changes_in_room`.
+        # Get the max stream ID in the table so we avoid deleting it. We need
+        # to keep the latest row so that we can calculate the maximum stream ID
+        # used.
         max_stream_id = await self.db_pool.simple_select_one_onecol(
             table="device_lists_changes_in_room",
             keyvalues={},
@@ -2586,6 +2600,15 @@ def prune_device_lists_changes_in_room_txn(txn: LoggingTransaction) -> int:
             await self.clock.sleep(Duration(milliseconds=100))
 
         if num_rows_deleted:
+            # Update the max pruned stream ID tracking table so that the
+            # safety check knows data up to this point has been deleted.
+            await self.db_pool.simple_update_one(
+                table="device_lists_changes_in_room_max_pruned_stream_id",
+                keyvalues={},
+                updatevalues={"stream_id": prune_before_stream_id},
+                desc="prune_device_lists_changes_in_room_update_max_pruned",
+            )
+
             logger.info(
                 "Pruned %d rows from device_lists_changes_in_room", num_rows_deleted
             )

synapse/storage/schema/main/delta/94/04_device_lists_changes_max_pruned.sql

@@ -0,0 +1,31 @@
+--
+-- 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>.
+
+
+-- Tracks the maximum stream_id that has been deleted (pruned) from the
+-- device_lists_changes_in_room table. This is used to determine whether
+-- it's safe to read from that table for a given stream_id — if the
+-- requested stream_id is <= the value here, the data has been pruned and
+-- the table cannot provide a complete answer.
+--
+-- This replaces the previous approach of using MIN(stream_id) on the
+-- device_lists_changes_in_room table, which incorrectly returned 0 when
+-- the table was empty.
+CREATE TABLE IF NOT EXISTS device_lists_changes_in_room_max_pruned_stream_id (
+    Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE,
+    stream_id BIGINT NOT NULL
+);
+
+-- Seed with the current minimum stream_id minus 1, or 0 if the table is empty.
+INSERT INTO device_lists_changes_in_room_max_pruned_stream_id (stream_id)
+    SELECT COALESCE(MIN(stream_id) - 1, 0) FROM device_lists_changes_in_room;

tests/storage/test_devices.py

@@ -399,12 +399,12 @@ def get_devices_in_room_status() -> tuple[int, str]:
         self.assertEqual(count, 10 * len(room_ids))
         self.assertEqual(min_device_id, "device_id0")
 
-        # Record the minimum stream ID before pruning, so we can check that this
-        # correctly updates after pruning (as it is cached).
-        starting_min_device_lists_id = self.get_success(
+        # Record the max pruned stream ID before pruning, so we can check
+        # that this correctly updates after pruning.
+        starting_max_pruned_id = self.get_success(
             self.store.db_pool.runInteraction(
-                "get_min_device_lists_changes_in_room",
-                self.store._get_min_device_lists_changes_in_room_txn,
+                "get_max_pruned_device_lists_changes_in_room",
+                self.store._get_max_pruned_device_lists_changes_in_room_txn,
             )
         )
 
@@ -442,14 +442,16 @@ def get_devices_in_room_status() -> tuple[int, str]:
             self.reactor.advance(Duration(milliseconds=110).as_secs())
 
         count, min_device_id = get_devices_in_room_status()
+        # We should always keep the most recent entries so that we can
+        # calculate the maximum stream ID used.
         self.assertEqual(count, len(room_ids))
         self.assertEqual(min_device_id, "device_id19")
 
-        # Check that the minimum stream ID cache has been advanced after pruning.
-        min_device_lists_id = self.get_success(
+        # Check that the max pruned stream ID has been advanced after pruning.
+        max_pruned_id = self.get_success(
             self.store.db_pool.runInteraction(
-                "get_min_device_lists_changes_in_room",
-                self.store._get_min_device_lists_changes_in_room_txn,
+                "get_max_pruned_device_lists_changes_in_room",
+                self.store._get_max_pruned_device_lists_changes_in_room_txn,
             )
         )
-        self.assertGreater(min_device_lists_id, starting_min_device_lists_id)
+        self.assertGreater(max_pruned_id, starting_max_pruned_id)