From ddc00d76f52ed5a6e90a492b2d4d0ee819a854ce Mon Sep 17 00:00:00 2001 From: Quentin Gliech Date: Wed, 26 Aug 2026 15:02:45 +0200 Subject: [PATCH] Allocate device-list stream IDs per batch, not per full device list `add_device_change_to_streams` chunks `device_ids` into batches of 1000, but the transaction allocated `len(device_ids)` stream IDs on every batch instead of `len(batch_device_ids)`. Both `_add_device_change_to_stream_txn` and `_add_device_outbound_room_poke_txn` zip the IDs against the batch they were handed, so the surplus was silently discarded. For a user with N devices this allocated N * ceil(N / 1000) IDs to write N rows, so the `device_lists` stream ID space advanced quadratically in the number of devices a user has. Accounts holding ~1.1M devices burned over a billion IDs per change, pushing the stream head far enough ahead that readers could not catch up -- the catch-up path in `get_all_device_list_changes_for_remotes` fetches 100 rows at a time and advances only as far as the last row's stream ID. --- changelog.d/20151.bugfix | 1 + synapse/storage/databases/main/devices.py | 2 +- tests/storage/test_devices.py | 32 +++++++++++++++++++++++ 3 files changed, 34 insertions(+), 1 deletion(-) create mode 100644 changelog.d/20151.bugfix diff --git a/changelog.d/20151.bugfix b/changelog.d/20151.bugfix new file mode 100644 index 00000000000..874dc5c3057 --- /dev/null +++ b/changelog.d/20151.bugfix @@ -0,0 +1 @@ +Fix a bug where a device list change for a user with a large number of devices would consume far more `device_lists` stream IDs than necessary, causing replication to fall behind. diff --git a/synapse/storage/databases/main/devices.py b/synapse/storage/databases/main/devices.py index 6cf9270ff28..4a35a2ceb93 100644 --- a/synapse/storage/databases/main/devices.py +++ b/synapse/storage/databases/main/devices.py @@ -2085,7 +2085,7 @@ def add_device_changes_txn( batch_device_ids: StrCollection, ) -> int: stream_ids = self._device_list_id_gen.get_next_mult_txn( - txn, len(device_ids) + txn, len(batch_device_ids) ) self._add_device_change_to_stream_txn( diff --git a/tests/storage/test_devices.py b/tests/storage/test_devices.py index b153c749808..9b063e80531 100644 --- a/tests/storage/test_devices.py +++ b/tests/storage/test_devices.py @@ -358,6 +358,38 @@ def test_update_unknown_device(self) -> None: ) self.assertEqual(404, exc.value.code) + def test_add_device_change_to_streams_allocates_one_id_per_device(self) -> None: + """Adding more devices than fit in a single batch should still only consume + one stream ID per device. + """ + user_id = "@user_id:test" + # Enough devices to span more than one `batch_iter` batch. + device_ids = [f"device_id{i}" for i in range(1500)] + + self.get_success( + self.store.add_device_change_to_streams( + user_id=user_id, + device_ids=device_ids, + room_ids=["!some:room"], + ) + ) + + stream_ids = self.get_success( + self.store.db_pool.simple_select_onecol( + table="device_lists_stream", + keyvalues={"user_id": user_id}, + retcol="stream_id", + ) + ) + + self.assertEqual(len(stream_ids), len(device_ids)) + # The allocated IDs should be contiguous: a gap means IDs were allocated + # and then thrown away. + self.assertEqual( + max(stream_ids) - min(stream_ids) + 1, + len(device_ids), + ) + @patch("synapse.storage.databases.main.devices.PRUNE_DEVICE_LISTS_BATCH_SIZE", 5) def test_prune_old_device_lists_changes_in_room(self) -> None: """Test that old entries in the `device_lists_changes_in_room` table are pruned properly."""