Sliding Sync: Update filters to be robust against remote invite rooms

changelog.d/17450.bugfix

@@ -0,0 +1 @@
+Update experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint to handle invite/knock rooms when filtering.

synapse/api/constants.py

@@ -225,6 +225,9 @@ class EventContentFields:
     # This is deprecated in MSC2175.
     ROOM_CREATOR: Final = "creator"
 
+    # The version of the room for `m.room.create` events.
+    ROOM_VERSION: Final = "room_version"
+
     # Used in m.room.guest_access events.
     GUEST_ACCESS: Final = "guest_access"
 
@@ -237,6 +240,9 @@ class EventContentFields:
     # an unspecced field added to to-device messages to identify them uniquely-ish
     TO_DEVICE_MSGID: Final = "org.matrix.msgid"
 
+    # `m.room.encryption`` algorithm field
+    ENCRYPTION_ALGORITHM: Final = "algorithm"
+
 
 class EventUnsignedContentFields:
     """Fields found inside the 'unsigned' data on events"""

synapse/handlers/sliding_sync.py

@@ -17,18 +17,26 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
+import enum
 import logging
 from itertools import chain
 from typing import TYPE_CHECKING, Any, Dict, Final, List, Mapping, Optional, Set, Tuple
 
 import attr
 from immutabledict import immutabledict
 
-from synapse.api.constants import AccountDataTypes, Direction, EventTypes, Membership
+from synapse.api.constants import (
+    AccountDataTypes,
+    Direction,
+    EventContentFields,
+    EventTypes,
+    Membership,
+)
 from synapse.events import EventBase
 from synapse.events.utils import strip_event
 from synapse.handlers.relations import BundledAggregations
 from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
+from synapse.storage.databases.main.state import ROOM_UNKNOWN_SENTINEL
 from synapse.storage.databases.main.stream import CurrentStateDeltaMembership
 from synapse.storage.roommember import MemberSummary
 from synapse.types import (
@@ -37,6 +45,7 @@
     Requester,
     RoomStreamToken,
     StateMap,
+    StrCollection,
     StreamKeyType,
     StreamToken,
     UserID,
@@ -51,6 +60,12 @@
 logger = logging.getLogger(__name__)
 
 
+class Sentinel(enum.Enum):
+    # defining a sentinel in this way allows mypy to correctly handle the
+    # type of a dictionary lookup and subsequent type narrowing.
+    UNSET_SENTINEL = object()
+
+
 # The event types that clients should consider as new activity.
 DEFAULT_BUMP_EVENT_TYPES = {
     EventTypes.Message,
@@ -1085,6 +1100,72 @@ async def filter_rooms(
             A filtered dictionary of room IDs along with membership information in the
             room at the time of `to_token`.
         """
+        room_id_to_stripped_state_map: Dict[str, Optional[List[Any]]] = {}
+
+        async def _bulk_get_stripped_state_for_rooms(
+            room_ids: StrCollection,
+        ) -> None:
+            """
+            Fetch stripped state for a list of room IDs. Stripped state is only
+            applicable to invite/knock rooms. Other rooms will have `None` as their
+            stripped state.
+
+            For invite rooms, we pull from `unsigned.invite_room_state`.
+            For knock rooms, we pull from `unsigned.knock_room_state`.
+            """
+            # Fetch what we haven't before
+            room_ids_to_fetch = [
+                room_id
+                for room_id in room_ids
+                if room_id not in room_id_to_stripped_state_map
+            ]
+
+            # Gather a list of event IDs we can grab stripped state from
+            invite_or_knock_event_ids: List[str] = []
+            for room_id in room_ids_to_fetch:
+                if sync_room_map[room_id].membership in (
+                    Membership.INVITE,
+                    Membership.KNOCK,
+                ):
+                    event_id = sync_room_map[room_id].event_id
+                    # If this is an invite/knock then there should be an event_id
+                    assert event_id is not None
+                    invite_or_knock_event_ids.append(event_id)
+                else:
+                    room_id_to_stripped_state_map[room_id] = None
+
+            invite_or_knock_events = await self.store.get_events(
+                invite_or_knock_event_ids
+            )
+            for invite_or_knock_event in invite_or_knock_events.values():
+                room_id = invite_or_knock_event.room_id
+                membership = invite_or_knock_event.membership
+
+                if membership == Membership.INVITE:
+                    # Scrutinize unsigned things
+                    invite_room_state = invite_or_knock_event.unsigned.get(
+                        "invite_room_state", None
+                    )
+                    # `invite_room_state` should be a list of stripped events
+                    if isinstance(invite_room_state, list):
+                        room_id_to_stripped_state_map[room_id] = invite_room_state
+                    else:
+                        room_id_to_stripped_state_map[room_id] = None
+                elif membership == Membership.KNOCK:
+                    # Scrutinize unsigned things
+                    knock_room_state = invite_or_knock_event.unsigned.get(
+                        "knock_room_state", None
+                    )
+                    # `knock_room_state` should be a list of stripped events
+                    if isinstance(knock_room_state, list):
+                        room_id_to_stripped_state_map[room_id] = knock_room_state
+                    else:
+                        room_id_to_stripped_state_map[room_id] = None
+                else:
+                    raise AssertionError(
+                        f"Unexpected membership {membership} (this is a problem with Synapse itself)"
+                    )
+
         filtered_room_id_set = set(sync_room_map.keys())
 
         # Filter for Direct-Message (DM) rooms
@@ -1104,31 +1185,72 @@ async def filter_rooms(
                     if not sync_room_map[room_id].is_dm
                 }
 
-        if filters.spaces:
+        if filters.spaces is not None:
             raise NotImplementedError()
 
         # Filter for encrypted rooms
         if filters.is_encrypted is not None:
+            # Lookup the encryption state from the database. Since this function is
+            # cached, we need to make a mutable copy via `dict(...)`.
+            room_id_to_encryption = dict(
+                await self.store.bulk_get_room_encryption(filtered_room_id_set)
+            )
+            room_ids_with_results = [
+                room_id
+                for room_id, encryption in room_id_to_encryption.items()
+                if encryption is not ROOM_UNKNOWN_SENTINEL
+            ]
+
+            # We might not have room state for remote invite/knocks if we are the first
+            # person on our server to see the room. The best we can do is look in the
+            # optional stripped state from the invite/knock event.
+            room_ids_without_results = filtered_room_id_set.difference(
+                room_ids_with_results
+            )
+            await _bulk_get_stripped_state_for_rooms(room_ids_without_results)
+
+            # Update our `room_id_to_encryption` map based on the stripped state
+            for room_id in room_ids_without_results:
+                stripped_state = room_id_to_stripped_state_map.get(
+                    room_id, Sentinel.UNSET_SENTINEL
+                )
+                assert stripped_state is not Sentinel.UNSET_SENTINEL, (
+                    f"Stripped state left unset for room {room_id}. "
+                    + "Make sure you're calling `_bulk_get_stripped_state_for_rooms(...)` "
+                    + "with that room_id. (this is a problem with Synapse itself)"
+                )
+
+                if stripped_state is not None:
+                    for event in stripped_state:
+                        event_type = event.get("type")
+                        event_content = event.get("content")
+                        # Scrutinize unsigned stripped state
+                        if isinstance(event_type, str) and isinstance(
+                            event_content, dict
+                        ):
+                            if event_type == EventTypes.RoomEncryption:
+                                room_id_to_encryption[room_id] = event_content.get(
+                                    EventContentFields.ENCRYPTION_ALGORITHM
+                                )
+                                break
+                    else:
+                        # Didn't see any encryption events in the stripped state so we
+                        # can assume the room is unencrypted.
+                        room_id_to_encryption[room_id] = None
+
             # Make a copy so we don't run into an error: `Set changed size during
             # iteration`, when we filter out and remove items
             for room_id in filtered_room_id_set.copy():
-                state_at_to_token = await self.storage_controllers.state.get_state_at(
-                    room_id,
-                    to_token,
-                    state_filter=StateFilter.from_types(
-                        [(EventTypes.RoomEncryption, "")]
-                    ),
-                    # Partially-stated rooms should have all state events except for the
-                    # membership events so we don't need to wait because we only care
-                    # about retrieving the `EventTypes.RoomEncryption` state event here.
-                    # Plus we don't want to block the whole sync waiting for this one
-                    # room.
-                    await_full_state=False,
-                )
-                is_encrypted = state_at_to_token.get((EventTypes.RoomEncryption, ""))
+                encryption = room_id_to_encryption.get(room_id, ROOM_UNKNOWN_SENTINEL)
+
+                # Just remove rooms if we can't determine their encryption status
+                if encryption is ROOM_UNKNOWN_SENTINEL:
+                    filtered_room_id_set.remove(room_id)
+                    continue
 
                 # If we're looking for encrypted rooms, filter out rooms that are not
                 # encrypted and vice versa
+                is_encrypted = encryption is not None
                 if (filters.is_encrypted and not is_encrypted) or (
                     not filters.is_encrypted and is_encrypted
                 ):
@@ -1154,15 +1276,60 @@ async def filter_rooms(
         # provided in the list. `None` is a valid type for rooms which do not have a
         # room type.
         if filters.room_types is not None or filters.not_room_types is not None:
-            room_to_type = await self.store.bulk_get_room_type(
-                {
-                    room_id
-                    for room_id in filtered_room_id_set
-                    # We only know the room types for joined rooms
-                    if sync_room_map[room_id].membership == Membership.JOIN
-                }
+            # Lookup the room type from the database. Since this function is
+            # cached, we need to make a mutable copy via `dict(...)`.
+            room_id_to_type = dict(
+                await self.store.bulk_get_room_type(filtered_room_id_set)
+            )
+            room_ids_with_results = [
+                room_id
+                for room_id, room_type in room_id_to_type.items()
+                if room_type is not ROOM_UNKNOWN_SENTINEL
+            ]
+
+            # We might not have room state for remote invite/knocks if we are the first
+            # person on our server to see the room. The best we can do is look in the
+            # optional stripped state from the invite/knock event.
+            room_ids_without_results = filtered_room_id_set.difference(
+                room_ids_with_results
             )
-            for room_id, room_type in room_to_type.items():
+            await _bulk_get_stripped_state_for_rooms(room_ids_without_results)
+
+            # Update our `room_id_to_type` map based on the stripped state
+            for room_id in room_ids_without_results:
+                stripped_state = room_id_to_stripped_state_map.get(
+                    room_id, Sentinel.UNSET_SENTINEL
+                )
+                assert stripped_state is not Sentinel.UNSET_SENTINEL, (
+                    f"Stripped state left unset for room {room_id}. "
+                    + "Make sure you're calling `_bulk_get_stripped_state_for_rooms(...)` "
+                    + "with that room_id. (this is a problem with Synapse itself)"
+                )
+
+                if stripped_state is not None:
+                    for event in stripped_state:
+                        event_type = event.get("type")
+                        event_content = event.get("content")
+                        # Scrutinize unsigned stripped state
+                        if isinstance(event_type, str) and isinstance(
+                            event_content, dict
+                        ):
+                            if event_type == EventTypes.Create:
+                                room_id_to_type[room_id] = event_content.get(
+                                    EventContentFields.ROOM_TYPE
+                                )
+                                break
+
+            # Make a copy so we don't run into an error: `Set changed size during
+            # iteration`, when we filter out and remove items
+            for room_id in filtered_room_id_set.copy():
+                room_type = room_id_to_type.get(room_id, ROOM_UNKNOWN_SENTINEL)
+
+                # Just remove rooms if we can't determine their type
+                if room_type is ROOM_UNKNOWN_SENTINEL:
+                    filtered_room_id_set.remove(room_id)
+                    continue
+
                 if (
                     filters.room_types is not None
                     and room_type not in filters.room_types
@@ -1175,13 +1342,13 @@ async def filter_rooms(
                 ):
                     filtered_room_id_set.remove(room_id)
 
-        if filters.room_name_like:
+        if filters.room_name_like is not None:
             raise NotImplementedError()
 
-        if filters.tags:
+        if filters.tags is not None:
             raise NotImplementedError()
 
-        if filters.not_tags:
+        if filters.not_tags is not None:
             raise NotImplementedError()
 
         # Assemble a new sync room map but only with the `filtered_room_id_set`

synapse/handlers/stats.py

@@ -293,7 +293,9 @@ async def _handle_deltas(
                     "history_visibility"
                 )
             elif delta.event_type == EventTypes.RoomEncryption:
-                room_state["encryption"] = event_content.get("algorithm")
+                room_state["encryption"] = event_content.get(
+                    EventContentFields.ENCRYPTION_ALGORITHM
+                )
             elif delta.event_type == EventTypes.Name:
                 room_state["name"] = event_content.get("name")
             elif delta.event_type == EventTypes.Topic:

synapse/storage/_base.py

@@ -124,6 +124,8 @@ def _invalidate_state_caches(
         # Purge other caches based on room state.
         self._attempt_to_invalidate_cache("get_room_summary", (room_id,))
         self._attempt_to_invalidate_cache("get_partial_current_state_ids", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
     def _invalidate_state_caches_all(self, room_id: str) -> None:
         """Invalidates caches that are based on the current state, but does
@@ -147,6 +149,8 @@ def _invalidate_state_caches_all(self, room_id: str) -> None:
         self._attempt_to_invalidate_cache("get_user_in_room_with_profile", None)
         self._attempt_to_invalidate_cache("get_rooms_for_user", None)
         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,))
 
     def _attempt_to_invalidate_cache(
         self, cache_name: str, key: Optional[Collection[Any]]

synapse/storage/databases/main/cache.py

@@ -269,12 +269,18 @@ def _process_event_stream_row(self, token: int, row: EventsStreamRow) -> None:
 
             if data.type == EventTypes.Member:
                 self.get_rooms_for_user.invalidate((data.state_key,))  # type: ignore[attr-defined]
+            elif data.type == EventTypes.RoomEncryption:
+                self.get_room_encryption.invalidate((data.room_id,))  # type: ignore[attr-defined]
+            elif data.type == EventTypes.Create:
+                self.get_room_type.invalidate((data.room_id,))  # type: ignore[attr-defined]
         elif row.type == EventsStreamAllStateRow.TypeId:
             assert isinstance(data, EventsStreamAllStateRow)
             # Similar to the above, but the entire caches are invalidated. This is
             # unfortunate for the membership caches, but should recover quickly.
             self._curr_state_delta_stream_cache.entity_has_changed(data.room_id, token)  # type: ignore[attr-defined]
             self.get_rooms_for_user.invalidate_all()  # type: ignore[attr-defined]
+            self.get_room_type.invalidate((data.room_id,))  # type: ignore[attr-defined]
+            self.get_room_encryption.invalidate((data.room_id,))  # type: ignore[attr-defined]
         else:
             raise Exception("Unknown events stream row type %s" % (row.type,))
 
@@ -342,6 +348,10 @@ def _invalidate_caches_for_event(
             self._attempt_to_invalidate_cache(
                 "get_forgotten_rooms_for_user", (state_key,)
             )
+        elif etype == EventTypes.Create:
+            self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        elif etype == EventTypes.RoomEncryption:
+            self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
         if relates_to:
             self._attempt_to_invalidate_cache(
@@ -399,6 +409,8 @@ def _invalidate_caches_for_room_events(self, room_id: str) -> None:
         self._attempt_to_invalidate_cache("get_thread_summary", None)
         self._attempt_to_invalidate_cache("get_thread_participated", None)
         self._attempt_to_invalidate_cache("get_threads", (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_state_group_for_event", None)
 
@@ -451,6 +463,8 @@ def _invalidate_caches_for_room(self, room_id: str) -> None:
         self._attempt_to_invalidate_cache("get_forgotten_rooms_for_user", None)
         self._attempt_to_invalidate_cache("_get_membership_from_event_id", None)
         self._attempt_to_invalidate_cache("get_room_version_id", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_type", (room_id,))
+        self._attempt_to_invalidate_cache("get_room_encryption", (room_id,))
 
         # And delete state caches.