Fix sliding sync performance slow down for long lived connections.

changelog.d/19206.bugfix

@@ -0,0 +1 @@
+Fix sliding sync performance slow down for long lived connections.

scripts-dev/check_schema_delta.py

@@ -12,7 +12,7 @@
 
 SCHEMA_FILE_REGEX = re.compile(r"^synapse/storage/schema/(.*)/delta/(.*)/(.*)$")
 INDEX_CREATION_REGEX = re.compile(
-    r"CREATE .*INDEX .*ON ([a-z_0-9]+)", flags=re.IGNORECASE
+    r"CREATE .*INDEX .*ON ([a-z_0-9]+)\s+\(", flags=re.IGNORECASE
 )
 INDEX_DELETION_REGEX = re.compile(r"DROP .*INDEX ([a-z_0-9]+)", flags=re.IGNORECASE)
 TABLE_CREATION_REGEX = re.compile(

synapse/storage/databases/main/sliding_sync.py

@@ -14,7 +14,7 @@
 
 
 import logging
-from typing import TYPE_CHECKING, Mapping, cast
+from typing import TYPE_CHECKING, AbstractSet, Mapping, cast
 
 import attr
 
@@ -26,12 +26,14 @@
     LoggingDatabaseConnection,
     LoggingTransaction,
 )
+from synapse.storage.engines import PostgresEngine
 from synapse.types import MultiWriterStreamToken, RoomStreamToken
 from synapse.types.handlers.sliding_sync import (
     HaveSentRoom,
     HaveSentRoomFlag,
     MutablePerConnectionState,
     PerConnectionState,
+    RoomLazyMembershipChanges,
     RoomStatusMap,
     RoomSyncConfig,
 )
@@ -349,6 +351,13 @@ def persist_per_connection_state_txn(
             value_values=values,
         )
 
+        self._persist_sliding_sync_connection_lazy_members_txn(
+            txn,
+            connection_key,
+            connection_position,
+            per_connection_state.room_lazy_membership,
+        )
+
         return connection_position
 
     @cached(iterable=True, max_entries=100000)
@@ -406,6 +415,19 @@ def _get_and_clear_connection_positions_txn(
         """
         txn.execute(sql, (connection_key, connection_position))
 
+        # Move any lazy membership entries for this connection position to have
+        # `NULL` connection position, indicating that it applies to all future
+        # positions on this connecetion.
+        self.db_pool.simple_update_txn(
+            txn,
+            table="sliding_sync_connection_lazy_members",
+            keyvalues={
+                "connection_key": connection_key,
+                "connection_position": connection_position,
+            },
+            updatevalues={"connection_position": None},
+        )
+
         # Fetch and create a mapping from required state ID to the actual
         # required state for the connection.
         rows = self.db_pool.simple_select_list_txn(
@@ -484,8 +506,131 @@ def _get_and_clear_connection_positions_txn(
             receipts=RoomStatusMap(receipts),
             account_data=RoomStatusMap(account_data),
             room_configs=room_configs,
+            room_lazy_membership={},
         )
 
+    async def get_sliding_sync_connection_lazy_members(
+        self,
+        connection_position: int,
+        room_id: str,
+        user_ids: AbstractSet[str],
+    ) -> Mapping[str, int]:
+        """Get which user IDs in the room we have previously sent lazy
+        membership for.
+
+        Args:
+            connection_position: The sliding sync connection position.
+            room_id: The room ID to get lazy members for.
+            user_ids: The user IDs to check for lazy membership.
+
+        Returns:
+            The mapping of user IDs to the last seen timestamp for those user
+            IDs.
+        """
+
+        def get_sliding_sync_connection_lazy_members_txn(
+            txn: LoggingTransaction,
+        ) -> Mapping[str, int]:
+            sql = """
+                SELECT user_id, mem.connection_position, last_seen_ts
+                FROM sliding_sync_connection_positions AS pos
+                INNER JOIN sliding_sync_connection_lazy_members AS mem USING (connection_key)
+                WHERE pos.connection_position = ?
+            """
+
+            txn.execute(sql, (connection_position,))
+
+            # Filter out any cache entries that only apply to forked connection
+            # positions. Entries with `NULL` connection position apply to all
+            # positions on the connection.
+            return {
+                user_id: last_seen_ts
+                for user_id, db_connection_position, last_seen_ts in txn
+                if db_connection_position == connection_position
+                or db_connection_position is None
+            }
+
+        return await self.db_pool.runInteraction(
+            "sliding_sync_connection_lazy_members",
+            get_sliding_sync_connection_lazy_members_txn,
+            db_autocommit=True,  # Avoid transaction for single read
+        )
+
+    def _persist_sliding_sync_connection_lazy_members_txn(
+        self,
+        txn: LoggingTransaction,
+        connection_key: int,
+        new_connection_position: int,
+        all_changes: dict[str, RoomLazyMembershipChanges],
+    ) -> None:
+        """Persist that we have sent lazy membership for the given user IDs."""
+
+        now = self.clock.time_msec()
+
+        # Figure out which cache entries to add or update.
+        #
+        # These are either a) new entries we've never sent before (i.e. with a
+        # None last_seen_ts), or b) where the `last_seen_ts` is old enough that
+        # we want to update it.
+        to_update: list[tuple[str, str]] = []
+        for room_id, room_changes in all_changes.items():
+            for user_id, last_seen_ts in room_changes.returned.items():
+                if last_seen_ts is None:
+                    # We've never sent this user before, so we need to record that
+                    # we've sent it at the new connection position.
+                    to_update.append((room_id, user_id))
+                elif last_seen_ts + 60 * 60 * 1000 < now:
+                    # We last saw this user over an hour ago, so we update the
+                    # timestamp.
+                    to_update.append((room_id, user_id))
+
+        if to_update:
+            # Upsert the new/updated entries.
+            #
+            # Ignore conflicts where the existing entry has a different
+            # connection position (i.e. from a forked connection position). This
+            # may mean that we lose some updates, but that's acceptable as this
+            # is a cache and its fine for it to *not* include rows.
+            sql = """
+                INSERT INTO sliding_sync_connection_lazy_members
+                (connection_key, connection_position, room_id, user_id, last_seen_ts)
+                VALUES {value_placeholder}
+                ON CONFLICT (connection_key, room_id, user_id)
+                DO UPDATE SET last_seen_ts = EXCLUDED.last_seen_ts
+                WHERE sliding_sync_connection_lazy_members.connection_position IS NULL
+                    OR sliding_sync_connection_lazy_members.connection_position = EXCLUDED.connection_position
+            """
+
+            args = [
+                (connection_key, new_connection_position, room_id, user_id, now)
+                for room_id, user_id in to_update
+            ]
+
+            if isinstance(self.database_engine, PostgresEngine):
+                sql = sql.format(value_placeholder="?")
+                txn.execute_values(sql, args, fetch=False)
+            else:
+                sql = sql.format(value_placeholder="(?, ?, ?, ?, ?)")
+                txn.execute_batch(sql, args)
+
+        # Remove any invlalidated entries.
+        to_remove: list[tuple[str, str]] = []
+        for room_id, room_changes in all_changes.items():
+            for user_id in room_changes.invalidated:
+                to_remove.append((room_id, user_id))
+
+        if to_remove:
+            # We don't try and match on connection position here: it's fine to
+            # remove it from all forks.
+            self.db_pool.simple_delete_many_batch_txn(
+                txn,
+                table="sliding_sync_connection_lazy_members",
+                keys=("connection_key", "room_id", "user_id"),
+                values=[
+                    (connection_key, room_id, user_id) for room_id, user_id in to_remove
+                ],
+            )
+
 
 @attr.s(auto_attribs=True, frozen=True)
 class PerConnectionStateDB:
@@ -496,6 +641,8 @@ class PerConnectionStateDB:
     serialized to strings.
 
     When persisting this *only* contains updates to the state.
+
+    The `room_lazy_membership` field is only used when persisting.
     """
 
     rooms: "RoomStatusMap[str]"
@@ -504,6 +651,8 @@ class PerConnectionStateDB:
 
     room_configs: Mapping[str, "RoomSyncConfig"]
 
+    room_lazy_membership: dict[str, RoomLazyMembershipChanges]
+
     @staticmethod
     async def from_state(
         per_connection_state: "MutablePerConnectionState", store: "DataStore"
@@ -557,6 +706,7 @@ async def from_state(
             receipts=RoomStatusMap(receipts),
             account_data=RoomStatusMap(account_data),
             room_configs=per_connection_state.room_configs.maps[0],
+            room_lazy_membership=per_connection_state.room_lazy_membership,
         )
 
     async def to_state(self, store: "DataStore") -> "PerConnectionState":

synapse/storage/schema/main/delta/93/02_sliding_sync_members.sql

@@ -0,0 +1,48 @@
+--
+-- This file is licensed under the Affero General Public License (AGPL) version 3.
+--
+-- Copyright (C) 2025 Element Creations 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>.
+
+
+-- Tracks which member states have been sent to the client for lazy-loaded
+-- members in sliding sync. This is a *cache* as it doesn't matter if we send
+-- down members we've previously sent down.
+--
+-- We track a *rough* `last_seen_ts` for each user in each room which indicates
+-- when we last would've sent their member state to the client. This is used so
+-- that we can remove members which haven't been seen for a while to save space.
+--
+-- Care must be taken when handling "forked" positions, i.e. we have responded
+-- to a request with a position and then get another different request using the
+-- previous position as a base. We track this by including a
+-- `connection_position` for newly inserted rows. When we advance the position
+-- we set this to NULL for all rows which were present at that position, and
+-- delete all other rows. When reading rows we can then filter out any rows
+-- which have a non-NULL `connection_position` which is not the current
+-- position.
+--
+-- I.e. `connection_position` is NULL for rows which are valid for *all*
+-- positions on the connection, and is non-NULL for rows which are only valid
+-- for a specific position.
+--
+-- When invalidating rows, we can just delete them. Technically this could
+-- incorrectly invalidate for a forked position, but this is acceptable as it's
+-- just a cache.
+CREATE TABLE sliding_sync_connection_lazy_members (
+    connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
+    connection_position BIGINT REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE,
+    room_id TEXT NOT NULL,
+    user_id TEXT NOT NULL,
+    last_seen_ts BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX sliding_sync_connection_lazy_members_idx ON sliding_sync_connection_lazy_members (connection_key, room_id, user_id);
+CREATE INDEX sliding_sync_connection_lazy_members_pos_idx ON sliding_sync_connection_lazy_members (connection_key, connection_position) WHERE connection_position IS NOT NULL;

synapse/types/handlers/sliding_sync.py

@@ -885,6 +885,36 @@ def __len__(self) -> int:
         return len(self.rooms) + len(self.receipts) + len(self.room_configs)
 
 
+@attr.s(auto_attribs=True)
+class RoomLazyMembershipChanges:
+    """Changes to lazily-loaded room memberships for a given room.
+
+    Attributes:
+        returned: Map from user ID to timestamp for users whose membership we
+            have lazily loaded. The timestamp indicates the time we previously
+            saw the membership if we have sent it down previously, or None if
+            we sent it down for the first time.
+
+            Note: this will include users whose membership we would have sent
+                down but didn't due to us having previously sent them.
+        invalidated: Set of user IDs whose latest membership we have *not* sent
+            down
+    """
+
+    # A map from user ID -> timestamp. Indicates that those memberships have
+    # been lazily loaded. I.e. that either a) we sent those memberships down, or
+    # b) we did so previously. The timestamp indicates the time we previously
+    # saw the membership.
+    returned: Mapping[str, int | None] = attr.Factory(dict)
+
+    # A set of user IDs whose membership change we have *not* sent
+    # down
+    invalidated: AbstractSet[str] = attr.Factory(set)
+
+    def __bool__(self) -> bool:
+        return bool(self.returned or self.invalidated)
+
+
 @attr.s(auto_attribs=True)
 class MutablePerConnectionState(PerConnectionState):
     """A mutable version of `PerConnectionState`"""
@@ -895,12 +925,19 @@ class MutablePerConnectionState(PerConnectionState):
 
     room_configs: typing.ChainMap[str, RoomSyncConfig]
 
+    # A map from room ID -> user ID -> timestamp. Indicates that those
+    # memberships have been lazily loaded. I.e. that either a) we sent those
+    # memberships down, or b) we did so previously. The timestamp indicates the
+    # time we previously saw the membership.
+    room_lazy_membership: dict[str, RoomLazyMembershipChanges] = attr.Factory(dict)
+
     def has_updates(self) -> bool:
         return (
             bool(self.rooms.get_updates())
             or bool(self.receipts.get_updates())
             or bool(self.account_data.get_updates())
             or bool(self.get_room_config_updates())
+            or bool(self.room_lazy_membership)
         )
 
     def get_room_config_updates(self) -> Mapping[str, RoomSyncConfig]: