dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-08-29 05:38:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add chunking for /sync, not SSS

Browse Source
This commit is contained in:
Kegan Dougal
2025-10-08 09:11:32 +01:00
parent 4def40414e
commit adb601b2d1
4 changed files with 60 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/api/constants.py
+1
View File
@@ -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
synapse/handlers/sliding_sync/extensions.py
+4
View File
@@ -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
synapse/handlers/sync.py
+43 -45
View File
@@ -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:
synapse/storage/databases/main/sticky_events.py
+12 -5
View File
@@ -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(