Fix a bug causing device list pruning to skip some rows when the transaction gets retried. (#19947)

Introduced in: #19473

Noticed in:
https://github.com/element-hq/synapse/pull/19556#discussion_r3505783541

I have not experienced the bug in the real world, it's just something I
noticed by reading.

--

Fix bug in `_prune_device_lists_changes_in_room` when transaction is
retried
The `nonlocal` variable is a footgun as it increments the counter even
though the transaction did not commit yet and may still be retried.

---------

Signed-off-by: Olivier 'reivilibre <oliverw@matrix.org>

changelog.d/19947.bugfix

@@ -0,0 +1 @@
+Fix a bug causing device list pruning to skip some rows when the transaction gets retried.

synapse/storage/databases/main/devices.py

@@ -2561,9 +2561,14 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
         # We default to 0 here as that is less than all possible stream IDs.
         min_stream_id = 0
 
-        def prune_device_lists_changes_in_room_txn(txn: LoggingTransaction) -> int:
-            nonlocal min_stream_id
-
+        def prune_device_lists_changes_in_room_txn(
+            txn: LoggingTransaction, min_stream_id: int
+        ) -> tuple[int, int]:
+            """
+            Returns tuple of:
+                - number of rows deleted
+                - new `min_stream_id` for the next iteration
+            """
             delete_sql = """
                 DELETE FROM device_lists_changes_in_room
                 WHERE stream_id IN (
@@ -2596,13 +2601,14 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
                     updatevalues={"stream_id": min_stream_id},
                 )
 
-            return num_deleted
+            return num_deleted, min_stream_id
 
         progress_num_rows_deleted = 0
         while True:
-            batch_deleted = await self.db_pool.runInteraction(
+            batch_deleted, min_stream_id = await self.db_pool.runInteraction(
                 "prune_device_lists_changes_in_room",
                 prune_device_lists_changes_in_room_txn,
+                min_stream_id,
             )
 
             finished = batch_deleted < PRUNE_DEVICE_LISTS_BATCH_SIZE