Expose MSC4354 Sticky Events over the legacy (v3) /sync API. (#19487)

Follows: #19365

Part of: MSC4354 whose experimental feature tracking issue is #19409

Partially supersedes: #18968

---------

Signed-off-by: Olivier 'reivilibre' <oliverw@matrix.org>

Author: reivilibre

changelog.d/19487.feature

@@ -0,0 +1 @@
+Expose [MSC4354 Sticky Events](https://github.com/matrix-org/matrix-spec-proposals/pull/4354) over the legacy (v3) /sync API.

synapse/handlers/sync.py

@@ -37,6 +37,7 @@
     EventContentFields,
     EventTypes,
     Membership,
+    StickyEvent,
 )
 from synapse.api.filtering import FilterCollection
 from synapse.api.presence import UserPresenceState
@@ -147,6 +148,7 @@ class JoinedSyncResult:
     state: StateMap[EventBase]
     ephemeral: list[JsonDict]
     account_data: list[JsonDict]
+    sticky: list[EventBase]
     unread_notifications: JsonDict
     unread_thread_notifications: JsonDict
     summary: JsonDict | None
@@ -157,7 +159,11 @@ def __bool__(self) -> bool:
         to tell if room needs to be part of the sync result.
         """
         return bool(
-            self.timeline or self.state or self.ephemeral or self.account_data
+            self.timeline
+            or self.state
+            or self.ephemeral
+            or self.account_data
+            or self.sticky
             # nb the notification count does not, er, count: if there's nothing
             # else in the result, we don't need to send it.
         )
@@ -601,6 +607,41 @@ async def ephemeral_by_room(
 
         return now_token, ephemeral_by_room
 
+    async def sticky_events_by_room(
+        self,
+        sync_result_builder: "SyncResultBuilder",
+        now_token: StreamToken,
+        since_token: StreamToken | None = None,
+    ) -> tuple[StreamToken, dict[str, list[str]]]:
+        """Get the sticky events for each room the user is in
+        Args:
+            sync_result_builder
+            now_token: Where the server is currently up to.
+            since_token: Where the server was when the client last synced.
+        Returns:
+            A tuple of the now StreamToken, updated to reflect the which sticky
+            events are included, and a dict mapping from room_id to a list
+            of sticky event IDs for that room (in sticky event stream order).
+        """
+        now = self.clock.time_msec()
+        with Measure(
+            self.clock, name="sticky_events_by_room", server_name=self.server_name
+        ):
+            from_id = since_token.sticky_events_key if since_token else 0
+
+            room_ids = sync_result_builder.joined_room_ids
+
+            to_id, sticky_by_room = await self.store.get_sticky_events_in_rooms(
+                room_ids,
+                from_id=from_id,
+                to_id=now_token.sticky_events_key,
+                now=now,
+                limit=StickyEvent.MAX_EVENTS_IN_SYNC,
+            )
+            now_token = now_token.copy_and_replace(StreamKeyType.STICKY_EVENTS, to_id)
+
+        return now_token, sticky_by_room
+
     async def _load_filtered_recents(
         self,
         room_id: str,
@@ -2177,11 +2218,43 @@ async def _generate_sync_entry_for_rooms(
             )
             sync_result_builder.now_token = now_token
 
+        sticky_by_room: dict[str, list[str]] = {}
+        if self.hs_config.experimental.msc4354_enabled:
+            now_token, sticky_by_room = await self.sticky_events_by_room(
+                sync_result_builder, now_token, since_token
+            )
+            sync_result_builder.now_token = now_token
+
         # 2. We check up front if anything has changed, if it hasn't then there is
         # no point in going further.
+        #
+        # If this is an initial sync (no since_token), then of course we can't skip
+        # the sync entry, as we have no base to use as a comparison for the question
+        # 'has anything changed' (this is the client's first time 'seeing' anything).
+        #
+        # Otherwise, for incremental syncs, we consider skipping the sync entry,
+        # doing cheap checks first:
+        #
+        # - are there any per-room EDUs;
+        # - is there any Room Account Data; or
+        # - are there any sticky events in the rooms; or
+        # - might the rooms have changed
+        #   (using in-memory event stream change caches, which can
+        #   only answer either 'Not changed' or 'Possibly changed')
+        #
+        # If none of those cheap checks give us a reason to continue generating the sync entry,
+        # we finally query the database to check for changed room tags.
+        # If there are also no changed tags, we can short-circuit return an empty sync entry.
         if not sync_result_builder.full_state:
-            if since_token and not ephemeral_by_room and not account_data_by_room:
-                have_changed = await self._have_rooms_changed(sync_result_builder)
+            # Cheap checks first
+            if (
+                since_token
+                and not ephemeral_by_room
+                and not account_data_by_room
+                and not sticky_by_room
+            ):
+                # This is also a cheap check, but we log the answer
+                have_changed = self._may_have_rooms_changed(sync_result_builder)
                 log_kv({"rooms_have_changed": have_changed})
                 if not have_changed:
                     tags_by_room = await self.store.get_updated_tags(
@@ -2225,6 +2298,7 @@ async def handle_room_entries(room_entry: "RoomSyncResultBuilder") -> None:
                 ephemeral=ephemeral_by_room.get(room_entry.room_id, []),
                 tags=tags_by_room.get(room_entry.room_id),
                 account_data=account_data_by_room.get(room_entry.room_id, {}),
+                sticky_event_ids=sticky_by_room.get(room_entry.room_id, []),
                 always_include=sync_result_builder.full_state,
             )
             logger.debug("Generated room entry for %s", room_entry.room_id)
@@ -2237,11 +2311,9 @@ async def handle_room_entries(room_entry: "RoomSyncResultBuilder") -> None:
 
         return set(newly_joined_rooms), set(newly_left_rooms)
 
-    async def _have_rooms_changed(
-        self, sync_result_builder: "SyncResultBuilder"
-    ) -> bool:
+    def _may_have_rooms_changed(self, sync_result_builder: "SyncResultBuilder") -> bool:
         """Returns whether there may be any new events that should be sent down
-        the sync. Returns True if there are.
+        the sync. Returns True if there **may** be.
 
         Does not modify the `sync_result_builder`.
         """
@@ -2611,6 +2683,7 @@ async def _generate_room_entry(
         ephemeral: list[JsonDict],
         tags: Mapping[str, JsonMapping] | None,
         account_data: Mapping[str, JsonMapping],
+        sticky_event_ids: list[str],
         always_include: bool = False,
     ) -> None:
         """Populates the `joined` and `archived` section of `sync_result_builder`
@@ -2640,6 +2713,8 @@ async def _generate_room_entry(
             tags: List of *all* tags for room, or None if there has been
                 no change.
             account_data: List of new account data for room
+            sticky_event_ids: MSC4354 sticky events in the room, if any.
+                In sticky event stream order.
             always_include: Always include this room in the sync response,
                 even if empty.
         """
@@ -2650,7 +2725,13 @@ async def _generate_room_entry(
         events = room_builder.events
 
         # We want to shortcut out as early as possible.
-        if not (always_include or account_data or ephemeral or full_state):
+        if not (
+            always_include
+            or account_data
+            or ephemeral
+            or full_state
+            or sticky_event_ids
+        ):
             if events == [] and tags is None:
                 return
 
@@ -2742,6 +2823,7 @@ async def _generate_room_entry(
                 or account_data_events
                 or ephemeral
                 or full_state
+                or sticky_event_ids
             ):
                 return
 
@@ -2788,6 +2870,32 @@ async def _generate_room_entry(
 
             if room_builder.rtype == "joined":
                 unread_notifications: dict[str, int] = {}
+                sticky_events: list[EventBase] = []
+                if sticky_event_ids:
+                    # As per MSC4354:
+                    # Remove sticky events that are already in the timeline, else we will needlessly duplicate
+                    # events.
+                    # There is no purpose in including sticky events in the sticky section if they're already in
+                    # the timeline, as either way the client becomes aware of them.
+                    # This is particularly important given the risk of sticky events spam since
+                    # anyone can send sticky events, so halving the bandwidth on average for each sticky
+                    # event is helpful.
+                    timeline_event_id_set = {ev.event_id for ev in batch.events}
+                    # Must preserve sticky event stream order
+                    sticky_event_ids = [
+                        e for e in sticky_event_ids if e not in timeline_event_id_set
+                    ]
+                    if sticky_event_ids:
+                        # Fetch and filter the sticky events
+                        sticky_events = await filter_and_transform_events_for_client(
+                            self._storage_controllers,
+                            sync_result_builder.sync_config.user.to_string(),
+                            await self.store.get_events_as_list(sticky_event_ids),
+                            # As per MSC4354:
+                            # > History visibility checks MUST NOT be applied to sticky events.
+                            # > Any joined user is authorised to see sticky events for the duration they remain sticky.
+                            always_include_ids=frozenset(sticky_event_ids),
+                        )
                 room_sync = JoinedSyncResult(
                     room_id=room_id,
                     timeline=batch,
@@ -2798,6 +2906,7 @@ async def _generate_room_entry(
                     unread_thread_notifications={},
                     summary=summary,
                     unread_count=0,
+                    sticky=sticky_events,
                 )
 
                 if room_sync or always_include:

synapse/rest/client/sync.py

@@ -619,6 +619,14 @@ async def encode_room(
             ephemeral_events = room.ephemeral
             result["ephemeral"] = {"events": ephemeral_events}
             result["unread_notifications"] = room.unread_notifications
+            if room.sticky:
+                # The sticky events have already been deduplicated so that events
+                # appearing in the timeline won't appear again here
+                result["msc4354_sticky"] = {
+                    "events": await self._event_serializer.serialize_events(
+                        room.sticky, time_now, config=serialize_options
+                    )
+                }
             if room.unread_thread_notifications:
                 result["unread_thread_notifications"] = room.unread_thread_notifications
                 if self._msc3773_enabled:

synapse/storage/databases/main/sticky_events.py

@@ -13,9 +13,7 @@
 import logging
 import random
 from dataclasses import dataclass
-from typing import (
-    TYPE_CHECKING,
-)
+from typing import TYPE_CHECKING, Collection, cast
 
 from twisted.internet.defer import Deferred
 
@@ -25,6 +23,7 @@
     DatabasePool,
     LoggingDatabaseConnection,
     LoggingTransaction,
+    make_in_list_sql_clause,
 )
 from synapse.storage.databases.main.cache import CacheInvalidationWorkerStore
 from synapse.storage.databases.main.state import StateGroupWorkerStore
@@ -138,6 +137,98 @@ def get_max_sticky_events_stream_id(self) -> int:
     def get_sticky_events_stream_id_generator(self) -> MultiWriterIdGenerator:
         return self._sticky_events_id_gen
 
+    async def get_sticky_events_in_rooms(
+        self,
+        room_ids: Collection[str],
+        *,
+        from_id: int,
+        to_id: int,
+        now: int,
+        limit: int | None,
+    ) -> tuple[int, dict[str, list[str]]]:
+        """
+        Fetch all the sticky events' IDs in the given rooms, with sticky stream IDs satisfying
+        from_id < sticky stream ID <= to_id.
+
+        The events are returned ordered by the sticky events stream.
+
+        Args:
+            room_ids: The room IDs to return sticky events in.
+            from_id: The sticky stream ID that sticky events should be returned from (exclusive).
+            to_id: The sticky stream ID that sticky events should end at (inclusive).
+            now: The current time in unix millis, used for skipping expired events.
+            limit: Max sticky events to return, or None to apply no limit.
+        Returns:
+            to_id, dict[room_id, list[event_ids]]
+        """
+        sticky_events_rows = await self.db_pool.runInteraction(
+            "get_sticky_events_in_rooms",
+            self._get_sticky_events_in_rooms_txn,
+            room_ids,
+            from_id=from_id,
+            to_id=to_id,
+            now=now,
+            limit=limit,
+        )
+
+        if not sticky_events_rows:
+            return to_id, {}
+
+        # Get stream_id of the last row, which is the highest
+        new_to_id, _, _ = sticky_events_rows[-1]
+
+        # room ID -> event IDs
+        room_id_to_event_ids: dict[str, list[str]] = {}
+        for _, room_id, event_id in sticky_events_rows:
+            events = room_id_to_event_ids.setdefault(room_id, [])
+            events.append(event_id)
+
+        return (new_to_id, room_id_to_event_ids)
+
+    def _get_sticky_events_in_rooms_txn(
+        self,
+        txn: LoggingTransaction,
+        room_ids: Collection[str],
+        *,
+        from_id: int,
+        to_id: int,
+        now: int,
+        limit: int | None,
+    ) -> list[tuple[int, str, str]]:
+        if len(room_ids) == 0:
+            return []
+        room_id_in_list_clause, room_id_in_list_values = make_in_list_sql_clause(
+            txn.database_engine, "se.room_id", room_ids
+        )
+        limit_clause = ""
+        limit_params: tuple[int, ...] = ()
+        if limit is not None:
+            limit_clause = "LIMIT ?"
+            limit_params = (limit,)
+
+        if isinstance(self.database_engine, PostgresEngine):
+            expr_soft_failed = "COALESCE(((ej.internal_metadata::jsonb)->>'soft_failed')::boolean, FALSE)"
+        else:
+            expr_soft_failed = "COALESCE(ej.internal_metadata->>'soft_failed', FALSE)"
+
+        txn.execute(
+            f"""
+            SELECT se.stream_id, se.room_id, event_id
+            FROM sticky_events se
+            INNER JOIN event_json ej USING (event_id)
+            WHERE
+                NOT {expr_soft_failed}
+                AND ? < expires_at
+                AND ? < stream_id
+                AND stream_id <= ?
+                AND {room_id_in_list_clause}
+            ORDER BY stream_id ASC
+            {limit_clause}
+            """,
+            (now, from_id, to_id, *room_id_in_list_values, *limit_params),
+        )
+        return cast(list[tuple[int, str, str]], txn.fetchall())
+
     async def get_updated_sticky_events(
         self, *, from_id: int, to_id: int, limit: int
     ) -> list[StickyEventUpdate]: