Merge branch 'erikj/ss_new_tables' into erikj/ss_hacks2

Co-authored-by: Eric Eastwood <eric.eastwood@beta.gouv.fr>

changelog.d/17630.misc

@@ -0,0 +1 @@
+Use new database tables for sliding sync.

synapse/storage/_base.py

@@ -126,6 +126,9 @@ class SQLBaseStore(metaclass=ABCMeta):
             self._attempt_to_invalidate_cache(
                 "_get_rooms_for_local_user_where_membership_is_inner", (user_id,)
             )
+            self._attempt_to_invalidate_cache(
+                "get_sliding_sync_rooms_for_user", (user_id,)
+            )
 
         # Purge other caches based on room state.
         self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
@@ -160,6 +163,7 @@ class SQLBaseStore(metaclass=ABCMeta):
         self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
         self._attempt_to_invalidate_cache("get_room_type", (room_id,))
         self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
+        self._attempt_to_invalidate_cache("get_sliding_sync_rooms_for_user", None)
 
     def _attempt_to_invalidate_cache(
         self, cache_name: str, key: Optional[Collection[Any]]

synapse/storage/background_updates.py

@@ -44,7 +44,7 @@ from synapse._pydantic_compat import HAS_PYDANTIC_V2
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.engines import PostgresEngine
 from synapse.storage.types import Connection, Cursor
-from synapse.types import JsonDict
+from synapse.types import JsonDict, StrCollection
 from synapse.util import Clock, json_encoder
 
 from . import engines
@@ -487,6 +487,25 @@ class BackgroundUpdater:
 
         return not update_exists
 
+    async def have_completed_background_updates(
+        self, update_names: StrCollection
+    ) -> bool:
+        """Return the name of background updates that have not yet been
+        completed"""
+        if self._all_done:
+            return True
+
+        rows = await self.db_pool.simple_select_many_batch(
+            table="background_updates",
+            column="update_name",
+            iterable=update_names,
+            retcols=("update_name",),
+            desc="get_uncompleted_background_updates",
+        )
+
+        # If we find any rows then we've not completed the update.
+        return not bool(rows)
+
     async def do_next_background_update(self, sleep: bool = True) -> bool:
         """Does some amount of work on the next queued background update
 

synapse/storage/databases/main/events_bg_updates.py

@@ -2318,6 +2318,17 @@ class EventsBackgroundUpdatesStore(StreamWorkerStore, StateDeltasStore, SQLBaseS
 
         return len(memberships_to_update_rows)
 
+    async def have_finished_sliding_sync_background_jobs(self) -> bool:
+        """Return if its safe to use the sliding sync membership tables."""
+
+        return await self.db_pool.updates.have_completed_background_updates(
+            (
+                _BackgroundUpdates.SLIDING_SYNC_PREFILL_JOINED_ROOMS_TO_RECALCULATE_TABLE_BG_UPDATE,
+                _BackgroundUpdates.SLIDING_SYNC_JOINED_ROOMS_BG_UPDATE,
+                _BackgroundUpdates.SLIDING_SYNC_MEMBERSHIP_SNAPSHOTS_BG_UPDATE,
+            )
+        )
+
 
 def _resolve_stale_data_in_sliding_sync_tables(
     txn: LoggingTransaction,

synapse/storage/databases/main/roommember.py

@@ -53,7 +53,12 @@ from synapse.storage.database import (
 from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.events_worker import EventsWorkerStore
 from synapse.storage.engines import Sqlite3Engine
-from synapse.storage.roommember import MemberSummary, ProfileInfo, RoomsForUser
+from synapse.storage.roommember import (
+    MemberSummary,
+    ProfileInfo,
+    RoomsForUser,
+    RoomsForUserSlidingSync,
+)
 from synapse.types import (
     JsonDict,
     PersistedEventPosition,
@@ -1377,6 +1382,54 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
             desc="room_forgetter_stream_pos",
         )
 
+    @cached(iterable=True, max_entries=10000)
+    async def get_sliding_sync_rooms_for_user(
+        self,
+        user_id: str,
+    ) -> Mapping[str, RoomsForUserSlidingSync]:
+        """Get all the rooms for a user to handle a sliding sync request.
+
+        Ignores forgotten rooms and rooms that the user has been kicked from.
+
+        Returns:
+            Map from room ID to membership info
+        """
+
+        def get_sliding_sync_rooms_for_user_txn(
+            txn: LoggingTransaction,
+        ) -> Dict[str, RoomsForUserSlidingSync]:
+            sql = """
+                SELECT m.room_id, m.sender, m.membership, m.membership_event_id,
+                    r.room_version,
+                    m.event_instance_name, m.event_stream_ordering,
+                    COALESCE(j.room_type, m.room_type),
+                    COALESCE(j.is_encrypted, m.is_encrypted)
+                FROM sliding_sync_membership_snapshots AS m
+                INNER JOIN rooms AS r USING (room_id)
+                LEFT JOIN sliding_sync_joined_rooms AS j ON (j.room_id = m.room_id AND m.membership = 'join')
+                WHERE user_id = ?
+                    AND m.forgotten = 0
+            """
+            txn.execute(sql, (user_id,))
+            return {
+                row[0]: RoomsForUserSlidingSync(
+                    room_id=row[0],
+                    sender=row[1],
+                    membership=row[2],
+                    event_id=row[3],
+                    room_version_id=row[4],
+                    event_pos=PersistedEventPosition(row[5], row[6]),
+                    room_type=row[7],
+                    is_encrypted=row[8],
+                )
+                for row in txn
+            }
+
+        return await self.db_pool.runInteraction(
+            "get_sliding_sync_rooms_for_user",
+            get_sliding_sync_rooms_for_user_txn,
+        )
+
 
 class RoomMemberBackgroundUpdateStore(SQLBaseStore):
     def __init__(

synapse/storage/roommember.py

@@ -39,6 +39,19 @@ class RoomsForUser:
     room_version_id: str
 
 
+@attr.s(slots=True, frozen=True, weakref_slot=False, auto_attribs=True)
+class RoomsForUserSlidingSync:
+    room_id: str
+    sender: Optional[str]
+    membership: str
+    event_id: Optional[str]
+    event_pos: PersistedEventPosition
+    room_version_id: str
+
+    room_type: Optional[str]
+    is_encrypted: bool
+
+
 @attr.s(slots=True, frozen=True, weakref_slot=False, auto_attribs=True)
 class GetRoomsForUserWithStreamOrdering:
     room_id: str

tests/rest/client/sliding_sync/test_connection_tracking.py

@@ -13,7 +13,7 @@
 #
 import logging
 
-from parameterized import parameterized
+from parameterized import parameterized, parameterized_class
 
 from twisted.test.proto_helpers import MemoryReactor
 
@@ -28,6 +28,18 @@ from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncConnectionTrackingTestCase(SlidingSyncBase):
     """
     Test connection tracking in the Sliding Sync API.
@@ -44,6 +56,8 @@ class SlidingSyncConnectionTrackingTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.storage_controllers = hs.get_storage_controllers()
 
+        super().prepare(reactor, clock, hs)
+
     def test_rooms_required_state_incremental_sync_LIVE(self) -> None:
         """Test that we only get state updates in incremental sync for rooms
         we've already seen (LIVE).

tests/rest/client/sliding_sync/test_extension_account_data.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.server import TimedOutException
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncAccountDataExtensionTestCase(SlidingSyncBase):
     """Tests for the account_data sliding sync extension"""
 
@@ -43,6 +57,8 @@ class SlidingSyncAccountDataExtensionTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.account_data_handler = hs.get_account_data_handler()
 
+        super().prepare(reactor, clock, hs)
+
     def test_no_data_initial_sync(self) -> None:
         """
         Test that enabling the account_data extension works during an intitial sync,

tests/rest/client/sliding_sync/test_extension_e2ee.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -27,6 +29,18 @@ from tests.server import TimedOutException
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncE2eeExtensionTestCase(SlidingSyncBase):
     """Tests for the e2ee sliding sync extension"""
 
@@ -42,6 +56,8 @@ class SlidingSyncE2eeExtensionTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.e2e_keys_handler = hs.get_e2e_keys_handler()
 
+        super().prepare(reactor, clock, hs)
+
     def test_no_data_initial_sync(self) -> None:
         """
         Test that enabling e2ee extension works during an intitial sync, even if there

tests/rest/client/sliding_sync/test_extension_receipts.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.server import TimedOutException
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncReceiptsExtensionTestCase(SlidingSyncBase):
     """Tests for the receipts sliding sync extension"""
 
@@ -42,6 +56,8 @@ class SlidingSyncReceiptsExtensionTestCase(SlidingSyncBase):
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.store = hs.get_datastores().main
 
+        super().prepare(reactor, clock, hs)
+
     def test_no_data_initial_sync(self) -> None:
         """
         Test that enabling the receipts extension works during an intitial sync,

tests/rest/client/sliding_sync/test_extension_to_device.py

@@ -14,6 +14,8 @@
 import logging
 from typing import List
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.server import TimedOutException
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncToDeviceExtensionTestCase(SlidingSyncBase):
     """Tests for the to-device sliding sync extension"""
 
@@ -40,6 +54,7 @@ class SlidingSyncToDeviceExtensionTestCase(SlidingSyncBase):
 
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.store = hs.get_datastores().main
+        super().prepare(reactor, clock, hs)
 
     def _assert_to_device_response(
         self, response_body: JsonDict, expected_messages: List[JsonDict]

tests/rest/client/sliding_sync/test_extension_typing.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.server import TimedOutException
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncTypingExtensionTestCase(SlidingSyncBase):
     """Tests for the typing notification sliding sync extension"""
 
@@ -41,6 +55,8 @@ class SlidingSyncTypingExtensionTestCase(SlidingSyncBase):
     def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
         self.store = hs.get_datastores().main
 
+        super().prepare(reactor, clock, hs)
+
     def test_no_data_initial_sync(self) -> None:
         """
         Test that enabling the typing extension works during an intitial sync,

tests/rest/client/sliding_sync/test_extensions.py

@@ -14,7 +14,7 @@
 import logging
 from typing import Literal
 
-from parameterized import parameterized
+from parameterized import parameterized, parameterized_class
 from typing_extensions import assert_never
 
 from twisted.test.proto_helpers import MemoryReactor
@@ -30,6 +30,18 @@ from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncExtensionsTestCase(SlidingSyncBase):
     """
     Test general extensions behavior in the Sliding Sync API. Each extension has their
@@ -49,6 +61,8 @@ class SlidingSyncExtensionsTestCase(SlidingSyncBase):
         self.storage_controllers = hs.get_storage_controllers()
         self.account_data_handler = hs.get_account_data_handler()
 
+        super().prepare(reactor, clock, hs)
+
     # Any extensions that use `lists`/`rooms` should be tested here
     @parameterized.expand([("account_data",), ("receipts",), ("typing",)])
     def test_extensions_lists_rooms_relevant_rooms(

tests/rest/client/sliding_sync/test_room_subscriptions.py

@@ -14,6 +14,8 @@
 import logging
 from http import HTTPStatus
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -27,6 +29,18 @@ from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncRoomSubscriptionsTestCase(SlidingSyncBase):
     """
     Test `room_subscriptions` in the Sliding Sync API.
@@ -43,6 +57,8 @@ class SlidingSyncRoomSubscriptionsTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.storage_controllers = hs.get_storage_controllers()
 
+        super().prepare(reactor, clock, hs)
+
     def test_room_subscriptions_with_join_membership(self) -> None:
         """
         Test `room_subscriptions` with a joined room should give us timeline and current

tests/rest/client/sliding_sync/test_rooms_invites.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -27,6 +29,18 @@ from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncRoomsInvitesTestCase(SlidingSyncBase):
     """
     Test to make sure the `rooms` response looks good for invites in the Sliding Sync API.
@@ -49,6 +63,8 @@ class SlidingSyncRoomsInvitesTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.storage_controllers = hs.get_storage_controllers()
 
+        super().prepare(reactor, clock, hs)
+
     def test_rooms_invite_shared_history_initial_sync(self) -> None:
         """
         Test that `rooms` we are invited to have some stripped `invite_state` during an

tests/rest/client/sliding_sync/test_rooms_meta.py

@@ -13,6 +13,8 @@
 #
 import logging
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.test_utils.event_injection import create_event
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncRoomsMetaTestCase(SlidingSyncBase):
     """
     Test rooms meta info like name, avatar, joined_count, invited_count, is_dm,
@@ -49,6 +63,8 @@ class SlidingSyncRoomsMetaTestCase(SlidingSyncBase):
         assert persistence is not None
         self.persistence = persistence
 
+        super().prepare(reactor, clock, hs)
+
     def test_rooms_meta_when_joined(self) -> None:
         """
         Test that the `rooms` `name` and `avatar` are included in the response and

tests/rest/client/sliding_sync/test_rooms_required_state.py

@@ -13,7 +13,7 @@
 #
 import logging
 
-from parameterized import parameterized
+from parameterized import parameterized, parameterized_class
 
 from twisted.test.proto_helpers import MemoryReactor
 
@@ -30,6 +30,18 @@ from tests.test_utils.event_injection import mark_event_as_partial_state
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncRoomsRequiredStateTestCase(SlidingSyncBase):
     """
     Test `rooms.required_state` in the Sliding Sync API.
@@ -46,6 +58,8 @@ class SlidingSyncRoomsRequiredStateTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.storage_controllers = hs.get_storage_controllers()
 
+        super().prepare(reactor, clock, hs)
+
     def test_rooms_no_required_state(self) -> None:
         """
         Empty `rooms.required_state` should not return any state events in the room

tests/rest/client/sliding_sync/test_rooms_timeline.py

@@ -14,6 +14,8 @@
 import logging
 from typing import List, Optional
 
+from parameterized import parameterized_class
+
 from twisted.test.proto_helpers import MemoryReactor
 
 import synapse.rest.admin
@@ -28,6 +30,18 @@ from tests.rest.client.sliding_sync.test_sliding_sync import SlidingSyncBase
 logger = logging.getLogger(__name__)
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncRoomsTimelineTestCase(SlidingSyncBase):
     """
     Test `rooms.timeline` in the Sliding Sync API.
@@ -44,6 +58,8 @@ class SlidingSyncRoomsTimelineTestCase(SlidingSyncBase):
         self.store = hs.get_datastores().main
         self.storage_controllers = hs.get_storage_controllers()
 
+        super().prepare(reactor, clock, hs)
+
     def _assertListEqual(
         self,
         actual_items: StrSequence,

tests/rest/client/sliding_sync/test_sliding_sync.py

@@ -13,7 +13,9 @@
 #
 import logging
 from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple
+from unittest.mock import AsyncMock
 
+from parameterized import parameterized_class
 from typing_extensions import assert_never
 
 from twisted.test.proto_helpers import MemoryReactor
@@ -47,8 +49,16 @@ logger = logging.getLogger(__name__)
 class SlidingSyncBase(unittest.HomeserverTestCase):
     """Base class for sliding sync test cases"""
 
+    # Flag as to whether to use the new sliding sync tables or not
+    use_new_tables: bool = True
+
     sync_endpoint = "/_matrix/client/unstable/org.matrix.simplified_msc3575/sync"
 
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        hs.get_datastores().main.have_finished_sliding_sync_background_jobs = AsyncMock(  # type: ignore[method-assign]
+            return_value=self.use_new_tables
+        )
+
     def default_config(self) -> JsonDict:
         config = super().default_config()
         # Enable sliding sync
@@ -203,6 +213,18 @@ class SlidingSyncBase(unittest.HomeserverTestCase):
             )
 
 
+# FIXME: This can be removed once we bump `SCHEMA_COMPAT_VERSION` and run the
+# foreground update for
+# `sliding_sync_joined_rooms`/`sliding_sync_membership_snapshots` (tracked by
+# https://github.com/element-hq/synapse/issues/17623)
+@parameterized_class(
+    ("use_new_tables",),
+    [
+        (True,),
+        (False,),
+    ],
+    class_name_func=lambda cls, num, params_dict: f"{cls.__name__}_{'new' if params_dict['use_new_tables'] else 'fallback'}",
+)
 class SlidingSyncTestCase(SlidingSyncBase):
     """
     Tests regarding MSC3575 Sliding Sync `/sync` endpoint.
@@ -226,6 +248,8 @@ class SlidingSyncTestCase(SlidingSyncBase):
         self.storage_controllers = hs.get_storage_controllers()
         self.account_data_handler = hs.get_account_data_handler()
 
+        super().prepare(reactor, clock, hs)
+
     def _add_new_dm_to_global_account_data(
         self, source_user_id: str, target_user_id: str, target_room_id: str
     ) -> None: