Merge remote-tracking branch 'origin/release-v1.109' into matrix-org-hotfixes

changelog.d/17275.bugfix

@@ -0,0 +1 @@
+Fix bug where OTKs were not always included in `/sync` response when using workers. Introduced v1.109.0rc1.

changelog.d/17292.bugfix

@@ -0,0 +1 @@
+Fix bug where `/sync` could get stuck due to edge case in device lists handling. Introduced in v1.109.0rc1.

synapse/handlers/sync.py

@@ -545,7 +545,7 @@ class SyncHandler:
                         )
                     )
 
-                cache_context.should_cache = False
+                cache_context.should_cache = False  # Don't cache empty responses
                 return SyncResult.empty(
                     since_token, one_time_keys_count, unused_fallback_key_types
                 )

synapse/storage/databases/main/devices.py

@@ -108,6 +108,11 @@ class DeviceWorkerStore(RoomMemberWorkerStore, EndToEndKeyWorkerStore):
                 ("device_lists_outbound_pokes", "instance_name", "stream_id"),
                 ("device_lists_changes_in_room", "instance_name", "stream_id"),
                 ("device_lists_remote_pending", "instance_name", "stream_id"),
+                (
+                    "device_lists_changes_converted_stream_position",
+                    "instance_name",
+                    "stream_id",
+                ),
             ],
             sequence_name="device_lists_sequence",
             writers=["master"],
@@ -2396,15 +2401,16 @@ class DeviceStore(DeviceWorkerStore, DeviceBackgroundUpdateStore):
         `FALSE` have not been converted.
         """
 
-        return cast(
+        # There should be only one row in this table, though we want to
-            Tuple[int, str],
+        # future-proof ourselves for when we have multiple rows (one for each
-            await self.db_pool.simple_select_one(
+        # instance). So to handle that case we take the minimum of all rows.
-                table="device_lists_changes_converted_stream_position",
+        rows = await self.db_pool.simple_select_list(
-                keyvalues={},
+            table="device_lists_changes_converted_stream_position",
-                retcols=["stream_id", "room_id"],
+            keyvalues={},
-                desc="get_device_change_last_converted_pos",
+            retcols=["stream_id", "room_id"],
-            ),
+            desc="get_device_change_last_converted_pos",
         )
+        return cast(Tuple[int, str], min(rows))
 
     async def set_device_change_last_converted_pos(
         self,
@@ -2419,6 +2425,10 @@ class DeviceStore(DeviceWorkerStore, DeviceBackgroundUpdateStore):
         await self.db_pool.simple_update_one(
             table="device_lists_changes_converted_stream_position",
             keyvalues={},
-            updatevalues={"stream_id": stream_id, "room_id": room_id},
+            updatevalues={
+                "stream_id": stream_id,
+                "instance_name": self._instance_name,
+                "room_id": room_id,
+            },
             desc="set_device_change_last_converted_pos",
         )

synapse/storage/schema/main/delta/85/05_add_instance_names_converted_pos.sql

@@ -0,0 +1,16 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2024 New Vector, 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 `instance_name` columns to stream tables to allow them to be used with
+-- `MultiWriterIdGenerator`
+ALTER TABLE device_lists_changes_converted_stream_position ADD COLUMN instance_name TEXT;