From adb601b2d10099623beee73a684131f14b695ef0 Mon Sep 17 00:00:00 2001 From: Kegan Dougal <7190048+kegsay@users.noreply.github.com> Date: Wed, 8 Oct 2025 09:11:32 +0100 Subject: [PATCH] Add chunking for /sync, not SSS --- synapse/api/constants.py | 1 + synapse/handlers/sliding_sync/extensions.py | 4 + synapse/handlers/sync.py | 88 +++++++++---------- .../storage/databases/main/sticky_events.py | 17 ++-- 4 files changed, 60 insertions(+), 50 deletions(-) diff --git a/synapse/api/constants.py b/synapse/api/constants.py index 95ec369bc2..84f316eb23 100644 --- a/synapse/api/constants.py +++ b/synapse/api/constants.py @@ -372,3 +372,4 @@ class StickyEvent: QUERY_PARAM_NAME: Final = "org.matrix.msc4354.sticky_duration_ms" FIELD_NAME: Final = "msc4354_sticky" MAX_DURATION_MS: Final = 3600000 # 1 hour + MAX_EVENTS_IN_SYNC: Final = 100 diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 1bb7ff0d87..813b0ddcc3 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -1008,6 +1008,10 @@ class SlidingSyncExtensionHandler: from_id, to_token.sticky_events_key, now, + # We set no limit here because the client can control when they get sticky events. + # Furthermore, it doesn't seem possible to set a limit with the internal API shape + # as given, as we cannot manipulate the to_token.sticky_events_key sent to the client... + limit=0, ) all_sticky_event_ids = { ev_id for evs in room_to_event_ids.values() for ev_id in evs diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py index f1997070b8..6b139609ce 100644 --- a/synapse/handlers/sync.py +++ b/synapse/handlers/sync.py @@ -45,6 +45,7 @@ from synapse.api.constants import ( EventTypes, JoinRules, Membership, + StickyEvent, ) from synapse.api.filtering import FilterCollection from synapse.api.presence import UserPresenceState @@ -165,7 +166,11 @@ class JoinedSyncResult: 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. ) @@ -639,43 +644,12 @@ class SyncHandler: from_id, now_token.sticky_events_key, now, + 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 _generate_sticky_events( - self, - sync_result_builder: "SyncResultBuilder", - sticky_by_room: Dict[str, Set[str]], - ) -> None: - """Generate sticky events to put into the sync response. - - The builder should already be populated with timeline events for joined rooms, so we can - duplicate suppress sticky events that are already going to be returned in the timeline section - of the sync response. - - Args: - sync_result_builder - sticky_by_room: Map of room ID to sticky event IDs. - """ - for joined_room in sync_result_builder.joined: - sticky_event_ids = sticky_by_room.get(joined_room.room_id, set()) - if len(sticky_event_ids) == 0: - continue - # remove sticky events that are in the timeline - timeline = {ev.event_id for ev in joined_room.timeline.events} - sticky_event_ids = sticky_event_ids.difference(timeline) - if len(sticky_event_ids) == 0: - continue - event_map = await self.store.get_events(sticky_event_ids) - joined_room.sticky = await filter_events_for_client( - self._storage_controllers, - sync_result_builder.sync_config.user.to_string(), - list(event_map.values()), - always_include_ids=frozenset(sticky_event_ids), - ) - async def _load_filtered_recents( self, room_id: str, @@ -2253,6 +2227,13 @@ class SyncHandler: ) sync_result_builder.now_token = now_token + sticky_by_room: Dict[str, Set[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 not sync_result_builder.full_state: @@ -2263,7 +2244,7 @@ class SyncHandler: tags_by_room = await self.store.get_updated_tags( user_id, since_token.account_data_key ) - if not tags_by_room: + if not tags_by_room and not sticky_by_room: logger.debug("no-oping sync") return set(), set() @@ -2283,7 +2264,6 @@ class SyncHandler: tags_by_room = await self.store.get_tags_for_user(user_id) log_kv({"rooms_changed": len(room_changes.room_entries)}) - room_entries = room_changes.room_entries invited = room_changes.invited knocked = room_changes.knocked @@ -2301,6 +2281,7 @@ class SyncHandler: 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, set()), always_include=sync_result_builder.full_state, ) logger.debug("Generated room entry for %s", room_entry.room_id) @@ -2311,14 +2292,6 @@ class SyncHandler: sync_result_builder.invited.extend(invited) sync_result_builder.knocked.extend(knocked) - 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 - ) - if sticky_by_room: - await self._generate_sticky_events(sync_result_builder, sticky_by_room) - sync_result_builder.now_token = now_token - return set(newly_joined_rooms), set(newly_left_rooms) async def _have_rooms_changed( @@ -2695,6 +2668,7 @@ class SyncHandler: ephemeral: List[JsonDict], tags: Optional[Mapping[str, JsonMapping]], account_data: Mapping[str, JsonMapping], + sticky_event_ids: Set[str], always_include: bool = False, ) -> None: """Populates the `joined` and `archived` section of `sync_result_builder` @@ -2724,6 +2698,7 @@ class SyncHandler: 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. always_include: Always include this room in the sync response, even if empty. """ @@ -2734,7 +2709,13 @@ class SyncHandler: 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 @@ -2818,6 +2799,7 @@ class SyncHandler: or account_data_events or ephemeral or full_state + or sticky_event_ids ): return @@ -2864,6 +2846,22 @@ class SyncHandler: if room_builder.rtype == "joined": unread_notifications: Dict[str, int] = {} + sticky_events: List[EventBase] = [] + if sticky_event_ids: + # remove sticky events that are in the timeline, else we will needlessly duplicate + # events. 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 = {ev.event_id for ev in batch.events} + sticky_event_ids = sticky_event_ids.difference(timeline) + if sticky_event_ids: + sticky_event_map = await self.store.get_events(sticky_event_ids) + sticky_events = await filter_events_for_client( + self._storage_controllers, + sync_result_builder.sync_config.user.to_string(), + list(sticky_event_map.values()), + always_include_ids=frozenset(sticky_event_ids), + ) room_sync = JoinedSyncResult( room_id=room_id, timeline=batch, @@ -2874,7 +2872,7 @@ class SyncHandler: unread_thread_notifications={}, summary=summary, unread_count=0, - sticky=[], + sticky=sticky_events, ) if room_sync or always_include: diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index d22e6099f8..398bf3598a 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/synapse/storage/databases/main/sticky_events.py @@ -126,6 +126,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor from_id: int, to_id: int, now: int, + limit: int, ) -> Tuple[int, Dict[str, Set[str]]]: """ Fetch all the sticky events in the given rooms, from the given sticky stream ID. @@ -135,6 +136,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor 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. If <= 0, no limit is applied. Returns: A tuple of (to_id, map[room_id, event_ids]) """ @@ -145,6 +147,7 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor from_id, to_id, now, + limit, ) new_to_id = from_id room_to_events: Dict[str, Set[str]] = {} @@ -162,19 +165,23 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor from_id: int, to_id: int, now: int, + limit: int, ) -> List[Tuple[int, str, str]]: if len(room_ids) == 0: return [] clause, room_id_values = make_in_list_sql_clause( txn.database_engine, "room_id", room_ids ) - txn.execute( - f""" + query = f""" SELECT stream_id, room_id, event_id FROM sticky_events WHERE soft_failed != ? AND expires_at > ? AND stream_id > ? AND stream_id <= ? AND {clause} - """, - (True, now, from_id, to_id, *room_id_values), - ) + ORDER BY stream_id ASC + """ + params = (True, now, from_id, to_id, *room_id_values) + if limit > 0: + query += "LIMIT ?" + params += (limit,) + txn.execute(query, params) return cast(List[Tuple[int, str, str]], txn.fetchall()) async def get_updated_sticky_events(