Sliding sync: Store the per-connection state in the database.

changelog.d/17599.misc

@@ -0,0 +1 @@
+Store sliding sync per-connection state in the database.

changelog.d/17600.misc

@@ -0,0 +1 @@
+Make the sliding sync `PerConnectionState` class immutable.

scripts-dev/mypy_synapse_plugin.py

@@ -38,6 +38,7 @@
     NoneType,
     TupleType,
     TypeAliasType,
+    TypeVarType,
     UninhabitedType,
     UnionType,
 )
@@ -233,6 +234,7 @@ def check_is_cacheable(
     "synapse.synapse_rust.push.FilteredPushRules",
     # This is technically not immutable, but close enough.
     "signedjson.types.VerifyKey",
+    "synapse.types.StrCollection",
 }
 
 # Immutable containers only if the values are also immutable.
@@ -298,7 +300,7 @@ def is_cacheable(
 
         elif rt.type.fullname in MUTABLE_CONTAINER_TYPES:
             # Mutable containers are mutable regardless of their underlying type.
-            return False, None
+            return False, f"container {rt.type.fullname} is mutable"
 
         elif "attrs" in rt.type.metadata:
             # attrs classes are only cachable iff it is frozen (immutable itself)
@@ -318,6 +320,9 @@ def is_cacheable(
             else:
                 return False, "non-frozen attrs class"
 
+        elif rt.type.is_enum:
+            # We assume Enum values are immutable
+            return True, None
         else:
             # Ensure we fail for unknown types, these generally means that the
             # above code is not complete.
@@ -326,6 +331,18 @@ def is_cacheable(
                 f"Don't know how to handle {rt.type.fullname} return type instance",
             )
 
+    elif isinstance(rt, TypeVarType):
+        # We consider TypeVars immutable if they are bound to a set of immutable
+        # types.
+        if rt.values:
+            for value in rt.values:
+                ok, note = is_cacheable(value, signature, verbose)
+                if not ok:
+                    return False, f"TypeVar bound not cacheable {value}"
+            return True, None
+
+        return False, "TypeVar is unbound"
+
     elif isinstance(rt, NoneType):
         # None is cachable.
         return True, None

synapse/app/generic_worker.py

@@ -98,6 +98,7 @@
 from synapse.storage.databases.main.search import SearchStore
 from synapse.storage.databases.main.session import SessionStore
 from synapse.storage.databases.main.signatures import SignatureWorkerStore
+from synapse.storage.databases.main.sliding_sync import SlidingSyncStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
 from synapse.storage.databases.main.stats import StatsStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
@@ -159,6 +160,7 @@ class GenericWorkerStore(
     SessionStore,
     TaskSchedulerWorkerStore,
     ExperimentalFeaturesStore,
+    SlidingSyncStore,
 ):
     # Properties that multiple storage classes define. Tell mypy what the
     # expected type is.

synapse/handlers/sliding_sync/__init__.py

@@ -29,13 +29,6 @@
     _RoomMembershipForUser,
 )
 from synapse.handlers.sliding_sync.store import SlidingSyncConnectionStore
-from synapse.handlers.sliding_sync.types import (
-    HaveSentRoomFlag,
-    MutablePerConnectionState,
-    PerConnectionState,
-    RoomSyncConfig,
-    StateValues,
-)
 from synapse.logging.opentracing import (
     SynapseTags,
     log_kv,
@@ -57,7 +50,15 @@
     StreamKeyType,
     StreamToken,
 )
-from synapse.types.handlers import SlidingSyncConfig, SlidingSyncResult
+from synapse.types.handlers.sliding_sync import (
+    HaveSentRoomFlag,
+    MutablePerConnectionState,
+    PerConnectionState,
+    RoomSyncConfig,
+    SlidingSyncConfig,
+    SlidingSyncResult,
+    StateValues,
+)
 from synapse.types.state import StateFilter
 from synapse.util.async_helpers import concurrently_execute
 from synapse.visibility import filter_events_for_client
@@ -99,7 +100,7 @@ def __init__(self, hs: "HomeServer"):
         self.rooms_to_exclude_globally = hs.config.server.rooms_to_exclude_from_sync
         self.is_mine_id = hs.is_mine_id
 
-        self.connection_store = SlidingSyncConnectionStore()
+        self.connection_store = SlidingSyncConnectionStore(self.store)
         self.extensions = SlidingSyncExtensionHandler(hs)
         self.room_lists = SlidingSyncRoomLists(hs)
 
@@ -220,16 +221,11 @@ async def current_sync_for_user(
         # amount of time (more with round-trips and re-processing) in the end to
         # get everything again.
         previous_connection_state = (
-            await self.connection_store.get_per_connection_state(
+            await self.connection_store.get_and_clear_connection_positions(
                 sync_config, from_token
             )
         )
 
-        await self.connection_store.mark_token_seen(
-            sync_config=sync_config,
-            from_token=from_token,
-        )
-
         # Get all of the room IDs that the user should be able to see in the sync
         # response
         has_lists = sync_config.lists is not None and len(sync_config.lists) > 0

synapse/handlers/sliding_sync/extensions.py

@@ -20,11 +20,6 @@
 
 from synapse.api.constants import AccountDataTypes, EduTypes
 from synapse.handlers.receipts import ReceiptEventSource
-from synapse.handlers.sliding_sync.types import (
-    HaveSentRoomFlag,
-    MutablePerConnectionState,
-    PerConnectionState,
-)
 from synapse.logging.opentracing import trace
 from synapse.storage.databases.main.receipts import ReceiptInRoom
 from synapse.types import (
@@ -35,7 +30,14 @@
     StrCollection,
     StreamToken,
 )
-from synapse.types.handlers import OperationType, SlidingSyncConfig, SlidingSyncResult
+from synapse.types.handlers.sliding_sync import (
+    HaveSentRoomFlag,
+    MutablePerConnectionState,
+    OperationType,
+    PerConnectionState,
+    SlidingSyncConfig,
+    SlidingSyncResult,
+)
 
 if TYPE_CHECKING:
     from synapse.server import HomeServer

synapse/handlers/sliding_sync/room_lists.py

@@ -40,11 +40,6 @@
 )
 from synapse.events import StrippedStateEvent
 from synapse.events.utils import parse_stripped_state_event
-from synapse.handlers.sliding_sync.types import (
-    HaveSentRoomFlag,
-    PerConnectionState,
-    RoomSyncConfig,
-)
 from synapse.logging.opentracing import start_active_span, trace
 from synapse.storage.databases.main.state import (
     ROOM_UNKNOWN_SENTINEL,
@@ -61,7 +56,14 @@
     StreamToken,
     UserID,
 )
-from synapse.types.handlers import OperationType, SlidingSyncConfig, SlidingSyncResult
+from synapse.types.handlers.sliding_sync import (
+    HaveSentRoomFlag,
+    OperationType,
+    PerConnectionState,
+    RoomSyncConfig,
+    SlidingSyncConfig,
+    SlidingSyncResult,
+)
 from synapse.types.state import StateFilter
 
 if TYPE_CHECKING:
@@ -279,15 +281,11 @@ async def compute_interested_rooms(
                                     room_id
                                 )
                                 if existing_room_sync_config is not None:
-                                    existing_room_sync_config.combine_room_sync_config(
+                                    room_sync_config = existing_room_sync_config.combine_room_sync_config(
                                         room_sync_config
                                     )
-                                else:
-                                    # Make a copy so if we modify it later, it doesn't
-                                    # affect all references.
-                                    relevant_room_map[room_id] = (
-                                        room_sync_config.deep_copy()
-                                    )
+
+                                relevant_room_map[room_id] = room_sync_config
 
                                 room_ids_in_list.append(room_id)
 
@@ -351,11 +349,13 @@ async def compute_interested_rooms(
                     # and need to fetch more info about.
                     existing_room_sync_config = relevant_room_map.get(room_id)
                     if existing_room_sync_config is not None:
-                        existing_room_sync_config.combine_room_sync_config(
-                            room_sync_config
+                        room_sync_config = (
+                            existing_room_sync_config.combine_room_sync_config(
+                                room_sync_config
+                            )
                         )
-                    else:
-                        relevant_room_map[room_id] = room_sync_config
+
+                    relevant_room_map[room_id] = room_sync_config
 
         # Filtered subset of `relevant_room_map` for rooms that may have updates
         # (in the event stream)