Sliding sync: various fixups to the background update

changelog.d/17652.misc

@@ -0,0 +1 @@
+Pre-populate room data used in experimental [MSC3575](https://github.com/matrix-org/matrix-spec-proposals/pull/3575) Sliding Sync `/sync` endpoint for quick filtering/sorting.

synapse/storage/databases/main/events.py

@@ -1977,8 +1977,11 @@ def _get_sliding_sync_insert_values_from_state_map(
                 sliding_sync_insert_map["is_encrypted"] = is_encrypted
             elif state_key == (EventTypes.Name, ""):
                 room_name = event.content.get(EventContentFields.ROOM_NAME)
-                # Scrutinize JSON values
-                if room_name is None or isinstance(room_name, str):
+                # Scrutinize JSON values. We ignore values with nulls as
+                # postgres doesn't allow null bytes in text columns.
+                if room_name is None or (
+                    isinstance(room_name, str) and "\0" not in room_name
+                ):
                     sliding_sync_insert_map["room_name"] = room_name
             elif state_key == (EventTypes.Tombstone, ""):
                 successor_room_id = event.content.get(
@@ -2068,6 +2071,14 @@ def _get_sliding_sync_insert_values_from_stripped_state(
                     else None
                 )
 
+                if (
+                    sliding_sync_insert_map["room_name"] is not None
+                    and "\0" in sliding_sync_insert_map["room_name"]
+                ):
+                    # We ignore values with nulls as
+                    # postgres doesn't allow null bytes in text columns.
+                    sliding_sync_insert_map.pop("room_name")
+
                 # Find the tombstone_successor_room_id
                 # Note: This isn't one of the stripped state events according to the spec
                 # but seems like there is no reason not to support this kind of thing.

synapse/storage/databases/main/events_bg_updates.py

@@ -44,6 +44,7 @@
 from synapse.storage.databases.main.events_worker import DatabaseCorruptionError
 from synapse.storage.databases.main.state_deltas import StateDeltasStore
 from synapse.storage.databases.main.stream import StreamWorkerStore
+from synapse.storage.engines import PostgresEngine
 from synapse.storage.types import Cursor
 from synapse.types import JsonDict, RoomStreamToken, StateMap, StrCollection
 from synapse.types.handlers import SLIDING_SYNC_DEFAULT_BUMP_EVENT_TYPES
@@ -1865,9 +1866,29 @@ async def _sliding_sync_membership_snapshots_bg_update(
         def _find_memberships_to_update_txn(
             txn: LoggingTransaction,
         ) -> List[
-            Tuple[str, Optional[str], str, str, str, str, int, Optional[str], bool]
+            Tuple[
+                str,
+                Optional[str],
+                Optional[str],
+                str,
+                str,
+                str,
+                str,
+                int,
+                Optional[str],
+                bool,
+            ]
         ]:
             # Fetch the set of event IDs that we want to update
+            #
+            # We skip over rows which we've already handled, i.e. have a
+            # matching row in `sliding_sync_membership_snapshots` with the same
+            # room, user and event ID.
+            #
+            # We also ignore rooms that the user has left themselves (i.e. not
+            # kicked). This is to avoid having to port lots of old rooms that we
+            # will never send down sliding sync (as we exclude such rooms from
+            # initial syncs).
 
             if initial_phase:
                 # There are some old out-of-band memberships (before
@@ -1880,6 +1901,7 @@ def _find_memberships_to_update_txn(
                     SELECT
                         c.room_id,
                         r.room_id,
+                        r.room_version,
                         c.user_id,
                         e.sender,
                         c.event_id,
@@ -1888,13 +1910,16 @@ def _find_memberships_to_update_txn(
                         e.instance_name,
                         e.outlier
                     FROM local_current_membership AS c
+                    LEFT JOIN sliding_sync_membership_snapshots AS m USING (room_id, user_id)
                     INNER JOIN events AS e USING (event_id)
                     LEFT JOIN rooms AS r ON (c.room_id = r.room_id)
                     WHERE (c.room_id, c.user_id) > (?, ?)
+                        AND (m.user_id IS NULL OR c.event_id != m.membership_event_id)
+                        AND (c.membership != ? OR c.user_id != e.sender)
                     ORDER BY c.room_id ASC, c.user_id ASC
                     LIMIT ?
                     """,
-                    (last_room_id, last_user_id, batch_size),
+                    (last_room_id, last_user_id, Membership.LEAVE, batch_size),
                 )
             elif last_event_stream_ordering is not None:
                 # It's important to sort by `event_stream_ordering` *ascending* (oldest to
@@ -1910,7 +1935,8 @@ def _find_memberships_to_update_txn(
                     """
                     SELECT
                         c.room_id,
-                        c.room_id,
+                        r.room_id,
+                        r.room_version,
                         c.user_id,
                         e.sender,
                         c.event_id,
@@ -1919,20 +1945,33 @@ def _find_memberships_to_update_txn(
                         e.instance_name,
                         e.outlier
                     FROM local_current_membership AS c
+                    LEFT JOIN sliding_sync_membership_snapshots AS m USING (room_id, user_id)
                     INNER JOIN events AS e USING (event_id)
-                    WHERE event_stream_ordering > ?
-                    ORDER BY event_stream_ordering ASC
+                    LEFT JOIN rooms AS r ON (c.room_id = r.room_id)
+                    WHERE c.event_stream_ordering > ?
+                        AND (m.user_id IS NULL OR c.event_id != m.membership_event_id)
+                        AND (c.membership != ? OR c.user_id != e.sender)
+                    ORDER BY c.event_stream_ordering ASC
                     LIMIT ?
                     """,
-                    (last_event_stream_ordering, batch_size),
+                    (last_event_stream_ordering, Membership.LEAVE, batch_size),
                 )
             else:
                 raise Exception("last_event_stream_ordering should not be None")
 
             memberships_to_update_rows = cast(
                 List[
                     Tuple[
-                        str, Optional[str], str, str, str, str, int, Optional[str], bool
+                        str,
+                        Optional[str],
+                        Optional[str],
+                        str,
+                        str,
+                        str,
+                        str,
+                        int,
+                        Optional[str],
+                        bool,
                     ]
                 ],
                 txn.fetchall(),
@@ -1965,7 +2004,7 @@ def _find_memberships_to_update_txn(
 
         def _find_previous_membership_txn(
             txn: LoggingTransaction, room_id: str, user_id: str, event_id: str
-        ) -> Tuple[str, str]:
+        ) -> Optional[Tuple[str, str]]:
             # Find the previous invite/knock event before the leave event
             #
             # Here are some notes on how we landed on this query:
@@ -2011,8 +2050,13 @@ def _find_previous_membership_txn(
             )
             row = txn.fetchone()
 
-            # We should see a corresponding previous invite/knock event
-            assert row is not None
+            if row is None:
+                # Generally we should have an invite or knock event for leaves
+                # that are outliers, however this may not always be the case
+                # (e.g. a local user got kicked but the kick event got pulled in
+                # as an outlier).
+                return None
+
             event_id, membership = row
 
             return event_id, membership
@@ -2027,6 +2071,7 @@ def _find_previous_membership_txn(
         for (
             room_id,
             room_id_from_rooms_table,
+            room_version_id,
             user_id,
             sender,
             membership_event_id,
@@ -2045,6 +2090,14 @@ def _find_previous_membership_txn(
                 Membership.BAN,
             )
 
+            if (
+                room_version_id is not None
+                and room_version_id not in KNOWN_ROOM_VERSIONS
+            ):
+                # Ignore rooms with unknown room versions (these were
+                # experimental rooms, that we no longer support).
+                continue
+
             # There are some old out-of-band memberships (before
             # https://github.com/matrix-org/synapse/issues/6983) where we don't have the
             # corresponding room stored in the `rooms` table`. We have a `FOREIGN KEY`
@@ -2132,14 +2185,17 @@ def _find_previous_membership_txn(
                     # in the events table though. We'll just say that we don't
                     # know the state for these rooms and continue on with our
                     # day.
-                    sliding_sync_membership_snapshots_insert_map["has_known_state"] = (
-                        False
-                    )
+                    sliding_sync_membership_snapshots_insert_map = {
+                        "has_known_state": False,
+                        "room_type": None,
+                        "room_name": None,
+                        "is_encrypted": False,
+                    }
             elif membership in (Membership.INVITE, Membership.KNOCK) or (
                 membership in (Membership.LEAVE, Membership.BAN) and is_outlier
             ):
-                invite_or_knock_event_id = membership_event_id
-                invite_or_knock_membership = membership
+                invite_or_knock_event_id = None
+                invite_or_knock_membership = None
 
                 # If the event is an `out_of_band_membership` (special case of
                 # `outlier`), we never had historical state so we have to pull from
@@ -2148,35 +2204,55 @@ def _find_previous_membership_txn(
                 # membership (i.e. the room shouldn't disappear if your using the
                 # `is_encrypted` filter and you leave).
                 if membership in (Membership.LEAVE, Membership.BAN) and is_outlier:
-                    (
-                        invite_or_knock_event_id,
-                        invite_or_knock_membership,
-                    ) = await self.db_pool.runInteraction(
+                    previous_membership = await self.db_pool.runInteraction(
                         "sliding_sync_membership_snapshots_bg_update._find_previous_membership",
                         _find_previous_membership_txn,
                         room_id,
                         user_id,
                         membership_event_id,
                     )
+                    if previous_membership is not None:
+                        (
+                            invite_or_knock_event_id,
+                            invite_or_knock_membership,
+                        ) = previous_membership
+                else:
+                    invite_or_knock_event_id = membership_event_id
+                    invite_or_knock_membership = membership
 
-                # Pull from the stripped state on the invite/knock event
-                invite_or_knock_event = await self.get_event(invite_or_knock_event_id)
-
-                raw_stripped_state_events = None
-                if invite_or_knock_membership == Membership.INVITE:
-                    invite_room_state = invite_or_knock_event.unsigned.get(
-                        "invite_room_state"
-                    )
-                    raw_stripped_state_events = invite_room_state
-                elif invite_or_knock_membership == Membership.KNOCK:
-                    knock_room_state = invite_or_knock_event.unsigned.get(
-                        "knock_room_state"
+                if (
+                    invite_or_knock_event_id is not None
+                    and invite_or_knock_membership is not None
+                ):
+                    # Pull from the stripped state on the invite/knock event
+                    invite_or_knock_event = await self.get_event(
+                        invite_or_knock_event_id
                     )
-                    raw_stripped_state_events = knock_room_state
 
-                sliding_sync_membership_snapshots_insert_map = PersistEventsStore._get_sliding_sync_insert_values_from_stripped_state(
-                    raw_stripped_state_events
-                )
+                    raw_stripped_state_events = None
+                    if invite_or_knock_membership == Membership.INVITE:
+                        invite_room_state = invite_or_knock_event.unsigned.get(
+                            "invite_room_state"
+                        )
+                        raw_stripped_state_events = invite_room_state
+                    elif invite_or_knock_membership == Membership.KNOCK:
+                        knock_room_state = invite_or_knock_event.unsigned.get(
+                            "knock_room_state"
+                        )
+                        raw_stripped_state_events = knock_room_state
+
+                    sliding_sync_membership_snapshots_insert_map = PersistEventsStore._get_sliding_sync_insert_values_from_stripped_state(
+                        raw_stripped_state_events
+                    )
+                else:
+                    # We couldn't find any state for the membership, so we just have to
+                    # leave it as empty.
+                    sliding_sync_membership_snapshots_insert_map = {
+                        "has_known_state": False,
+                        "room_type": None,
+                        "room_name": None,
+                        "is_encrypted": False,
+                    }
 
                 # We should have some insert values for each room, even if no
                 # stripped state is on the event because we still want to record
@@ -2295,19 +2371,42 @@ def _fill_table_txn(txn: LoggingTransaction) -> None:
                 )
                 # We need to find the `forgotten` value during the transaction because
                 # we can't risk inserting stale data.
-                txn.execute(
-                    """
-                    UPDATE sliding_sync_membership_snapshots
-                    SET
-                        forgotten = (SELECT forgotten FROM room_memberships WHERE event_id = ?)
-                    WHERE room_id = ? and user_id = ?
-                    """,
-                    (
-                        membership_event_id,
-                        room_id,
-                        user_id,
-                    ),
-                )
+                if isinstance(txn.database_engine, PostgresEngine):
+                    txn.execute(
+                        """
+                        UPDATE sliding_sync_membership_snapshots
+                        SET
+                            forgotten = m.forgotten
+                        FROM room_memberships AS m
+                        WHERE sliding_sync_membership_snapshots.room_id = ?
+                            AND sliding_sync_membership_snapshots.user_id = ?
+                            AND membership_event_id = ?
+                            AND membership_event_id = m.event_id
+                            AND m.event_id IS NOT NULL
+                        """,
+                        (
+                            room_id,
+                            user_id,
+                            membership_event_id,
+                        ),
+                    )
+                else:
+                    # SQLite doesn't support UPDATE FROM before 3.33.0, so we do
+                    # this via sub-selects.
+                    txn.execute(
+                        """
+                        UPDATE sliding_sync_membership_snapshots
+                        SET
+                            forgotten = (SELECT forgotten FROM room_memberships WHERE event_id = ?)
+                        WHERE room_id = ? and user_id = ? AND membership_event_id = ?
+                        """,
+                        (
+                            membership_event_id,
+                            room_id,
+                            user_id,
+                            membership_event_id,
+                        ),
+                    )
 
         await self.db_pool.runInteraction(
             "sliding_sync_membership_snapshots_bg_update", _fill_table_txn
@@ -2317,6 +2416,7 @@ def _fill_table_txn(txn: LoggingTransaction) -> None:
         (
             room_id,
             _room_id_from_rooms_table,
+            _room_version_id,
             user_id,
             _sender,
             _membership_event_id,