From 0d28e93a7aa2265232accd0bd26335894525ea82 Mon Sep 17 00:00:00 2001 From: Johannes Marbach Date: Mon, 26 Oct 2026 15:02:22 +0100 Subject: [PATCH] Move the sticky events sliding sync extension to the global `pos` and MSC4508 scoping (#20363) This includes the following: - Implements ['Implicit Forward Pagination'](https://github.com/matrix-org/matrix-spec/issues/2378). Remove the extension's `since` and `next_batch` fields. The position is now tracked via the global `pos`, with per-room connection state recording how far each room has been sent so that `limit` paginates without holding `pos` back and lost responses are re-delivered. - Discussed in https://github.com/matrix-org/matrix-spec-proposals/pull/4480#discussion_r4165887827 - Honour the `lists` and `rooms` fields. - Send all unexpired sticky events when a room enters scope for the first time or is newly joined, and catch up on events sent while a room was out of scope when it re-enters. - Enabling the extension partway through a connection returns all unexpired sticky events for in-scope rooms, and re-enabling it after disabling it returns the unexpired sticky events sent in the meantime. I tried to split this up into standalone commits though the bulk of the change is in the third commit. --------- Co-authored-by: Olivier 'reivilibre --- changelog.d/20363.feature | 1 + synapse/handlers/sliding_sync/__init__.py | 2 +- synapse/handlers/sliding_sync/extensions.py | 357 +++++--- synapse/rest/client/sync.py | 1 - .../storage/databases/main/sliding_sync.py | 41 + .../storage/databases/main/sticky_events.py | 51 ++ synapse/types/handlers/sliding_sync.py | 28 +- synapse/types/rest/client/__init__.py | 157 +--- .../replication/test_sharded_sticky_events.py | 19 +- .../test_extension_sticky_events.py | 759 +++++++++++++++--- tests/storage/test_sticky_events.py | 68 ++ 11 files changed, 1140 insertions(+), 344 deletions(-) create mode 100644 changelog.d/20363.feature diff --git a/changelog.d/20363.feature b/changelog.d/20363.feature new file mode 100644 index 0000000000..6f86ebde27 --- /dev/null +++ b/changelog.d/20363.feature @@ -0,0 +1 @@ +Move the sticky events sliding sync extension to the global `pos` and MSC4508 scoping. diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index afb93e42b4..aa249fc710 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -328,7 +328,6 @@ class SlidingSyncHandler: actual_lists=lists, previous_connection_state=previous_connection_state, new_connection_state=new_connection_state, - all_interested_room_ids=interested_rooms.all_rooms, # We're purposely using `relevant_room_map` instead of # `relevant_rooms_to_send_map` here. This needs to be all room_ids we could # send regardless of whether they have an event update or not. The @@ -336,6 +335,7 @@ class SlidingSyncHandler: # account data, read receipts, typing indicators, to-device messages, etc). actual_room_ids=set(relevant_room_map.keys()), actual_room_response_map=rooms, + newly_joined_room_ids=interested_rooms.newly_joined_rooms, from_token=from_token, to_token=to_token, ) diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 0348f96115..23293fb8ea 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -32,7 +32,6 @@ from synapse.api.constants import ( ProfileUpdateAction, StickyEvent, ) -from synapse.api.errors import SlidingSyncUnknownPosition from synapse.events.utils import FilteredEvent from synapse.handlers.receipts import ReceiptEventSource from synapse.logging.opentracing import trace @@ -60,7 +59,6 @@ from synapse.types.handlers.sliding_sync import ( SlidingSyncResult, StateValues, ) -from synapse.types.rest.client import SlidingSyncStickyEventsToken from synapse.util.async_helpers import ( concurrently_execute, gather_optional_coroutines, @@ -100,10 +98,10 @@ class SlidingSyncExtensionHandler: sync_config: SlidingSyncConfig, previous_connection_state: "PerConnectionState", new_connection_state: "MutablePerConnectionState", - all_interested_room_ids: set[str], actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList], actual_room_ids: set[str], actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], + newly_joined_room_ids: AbstractSet[str], to_token: StreamToken, from_token: SlidingSyncStreamToken | None, ) -> SlidingSyncResult.Extensions: @@ -114,19 +112,30 @@ class SlidingSyncExtensionHandler: previous_connection_state: Snapshot of the current per-connection state new_connection_state: A mutable copy of the per-connection state, used to record updates to the state during this request. - all_interested_room_ids: The IDs of all rooms that the client is interested in, - even if they don't appear in the current limited window. - See `SlidingSyncInterestedRooms.all_rooms`. actual_lists: Sliding window API. A map of list key to list results in the Sliding Sync response. actual_room_ids: The actual room IDs in the the Sliding Sync response. actual_room_response_map: A map of room ID to room results in the the Sliding Sync response. + newly_joined_room_ids: The rooms the user newly joined in the token range. to_token: The latest point in the stream to sync up to. from_token: The point in the stream to sync from. """ if sync_config.extensions is None: + # If no extensions are enabled, shortcut out. + + if self._enable_sticky_events: + # The sticky events extension tracks per-room state even while it + # isn't enabled on the request, so that re-enabling it doesn't miss anything. + # (But only if the MSC4354 experimental feature is enabled) + await self._record_sticky_events_in_out_of_scope_rooms( + previous_connection_state=previous_connection_state, + new_connection_state=new_connection_state, + relevant_room_ids=set(), + to_token=to_token, + from_token=from_token, + ) return SlidingSyncResult.Extensions() to_device_coro = None @@ -198,14 +207,16 @@ class SlidingSyncExtensionHandler: ) sticky_events_coro = None - if ( - sync_config.extensions.sticky_events is not Absent - and self._enable_sticky_events - ): + if self._enable_sticky_events: sticky_events_coro = self.get_sticky_events_extension_response( sync_config=sync_config, + previous_connection_state=previous_connection_state, + new_connection_state=new_connection_state, + actual_lists=actual_lists, + actual_room_ids=actual_room_ids, + actual_room_response_map=actual_room_response_map, + newly_joined_room_ids=newly_joined_room_ids, sticky_events_request=sync_config.extensions.sticky_events, - all_interested_room_ids=all_interested_room_ids, to_token=to_token, from_token=from_token, ) @@ -255,8 +266,8 @@ class SlidingSyncExtensionHandler: def find_relevant_room_ids_for_extension( self, - requested_lists: StrCollection | None, - requested_room_ids: StrCollection | None, + requested_lists: StrCollection, + requested_room_ids: StrCollection, actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList], actual_room_ids: AbstractSet[str], ) -> set[str]: @@ -285,36 +296,23 @@ class SlidingSyncExtensionHandler: relevant_room_ids: set[str] = set() # See what rooms from the room subscriptions we should get account data for - if requested_room_ids is not None: - for room_id in requested_room_ids: - # A wildcard means we process all rooms from the room subscriptions - if room_id == "*": - relevant_room_ids.update(actual_room_ids) - break + for room_id in requested_room_ids: + # A wildcard means we process all rooms from the room subscriptions + if room_id == "*": + relevant_room_ids.update(actual_room_ids) + break - if room_id in actual_room_ids: - relevant_room_ids.add(room_id) + if room_id in actual_room_ids: + relevant_room_ids.add(room_id) # See what rooms from the sliding window lists we should get account data for - if requested_lists is not None: - for list_key in requested_lists: - # Just some typing because we share the variable name in multiple places - actual_list: SlidingSyncResult.SlidingWindowList | None = None + for list_key in requested_lists: + # Just some typing because we share the variable name in multiple places + actual_list: SlidingSyncResult.SlidingWindowList | None = None - # A wildcard means we process rooms from all lists - if list_key == "*": - for actual_list in actual_lists.values(): - # We only expect a single SYNC operation for any list - assert len(actual_list.ops) == 1 - sync_op = actual_list.ops[0] - assert sync_op.op == OperationType.SYNC - - relevant_room_ids.update(sync_op.room_ids) - - break - - actual_list = actual_lists.get(list_key) - if actual_list is not None: + # A wildcard means we process rooms from all lists + if list_key == "*": + for actual_list in actual_lists.values(): # We only expect a single SYNC operation for any list assert len(actual_list.ops) == 1 sync_op = actual_list.ops[0] @@ -322,6 +320,17 @@ class SlidingSyncExtensionHandler: relevant_room_ids.update(sync_op.room_ids) + break + + actual_list = actual_lists.get(list_key) + if actual_list is not None: + # We only expect a single SYNC operation for any list + assert len(actual_list.ops) == 1 + sync_op = actual_list.ops[0] + assert sync_op.op == OperationType.SYNC + + relevant_room_ids.update(sync_op.room_ids) + return relevant_room_ids @trace @@ -1025,71 +1034,167 @@ class SlidingSyncExtensionHandler: async def get_sticky_events_extension_response( self, sync_config: SlidingSyncConfig, - sticky_events_request: SlidingSyncConfig.Extensions.StickyEventsExtension, - all_interested_room_ids: set[str], + previous_connection_state: "PerConnectionState", + new_connection_state: "MutablePerConnectionState", + actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList], + actual_room_ids: set[str], + actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], + newly_joined_room_ids: AbstractSet[str], + sticky_events_request: ( + SlidingSyncConfig.Extensions.StickyEventsExtension | AbsentType + ), to_token: StreamToken, from_token: SlidingSyncStreamToken | None, ) -> SlidingSyncResult.Extensions.StickyEventsExtension | None: - if not sticky_events_request.enabled: + """Handle Sticky Events extension (MSC4354 / MSC4480) + + Args: + sync_config: Sync configuration + previous_connection_state: The current per-connection state + new_connection_state: A mutable copy of the per-connection + state, used to record updates to the state. + actual_lists: Sliding window API. A map of list key to list results in the + Sliding Sync response. + actual_room_ids: The actual room IDs in the the Sliding Sync response. + actual_room_response_map: A map of room ID to room results in the the + Sliding Sync response. + newly_joined_room_ids: The rooms the user newly joined in the token range. + sticky_events_request: The sticky_events extension from the request, + or absent if the request didn't include it. + to_token: The point in the stream to sync up to. + from_token: The point in the stream to sync from. + """ + if sticky_events_request is Absent or not sticky_events_request.enabled: + # While the extension is disabled, no room is in scope for it. We still + # keep track of which rooms gain sticky events in the meantime, so that + # when the extension is re-enabled the client receives the (unexpired) + # sticky events sent while it was disabled, rather than only those + # sent after the request's `pos`. + await self._record_sticky_events_in_out_of_scope_rooms( + previous_connection_state=previous_connection_state, + new_connection_state=new_connection_state, + relevant_room_ids=set(), + to_token=to_token, + from_token=from_token, + ) return None - now = self.clock.time_msec() - # If there is no `since` token specified, start from the beginning of the stream - # to make sure the client receives all visible (unexpired) sticky events - since_token = sticky_events_request.since or SlidingSyncStickyEventsToken.START - since_token_as_stream_token = await since_token.to_stream_token(self.store) - - if not since_token_as_stream_token.is_before_or_eq( - self.event_sources.get_current_token().sticky_events_key - ): - # The `since_token` is *after* the current position as seen on this worker. - # This either means that this worker is lagging, or the client has a token - # from the future. The client having a future token could happen maliciously - # where someone intentionally messes with the token or innocently if a - # database was rolled back by restoring from backup. - # - # Get the max allocated token out of the database to see which case it is. - # (For efficiency, we only do this after the non-database common case check.) - max_token = await self.store.get_sticky_events_stream_id_generator().get_max_allocated_token() - - # It may be tempting to wonder why we don't do anything about the 'worker lagging' - # case, such as waiting for streams to catch up. - # The case where this worker is lagging should not happen in practice because: - # assuming the client didn't mess with the `since` token, it must have been - # produced as a `to_token` in a previous request, where the response-wide `next_pos` - # is equal to or later than the token given by this extension. - # - # Then in this request, the client will present that `next_pos` as the `pos` for the - # entire request and `wait_for_stream_token` (which is called with the request-level pos tokens) - # will cause us to wait for worker lag on that token, including the sticky events stream. - # - # (It also doesn't actually matter since `get_sticky_events_in_rooms` will just return - # an empty result and is safe against rewinding the token.) - - if max_token < since_token_as_stream_token.get_max_stream_pos(): - # The client has a token from the future. - # - # We could wait until we reach the token but we might as well not waste - # our resources on an invalid state scenario. Reset the sliding sync - # connection. - raise SlidingSyncUnknownPosition( - "The `org.matrix.msc4354.sticky_events` extension `since` token is considered " - "invalid because it includes stream positions greater than the furthest " - "persisted position across all of the workers. This indicates either a Synapse " - "programming error (as we should never hand out invalid future tokens), database " - "was rolled back, or a fabricated `from` token. If you've modified the token, " - "you can try paginating from the beginning again.", - ) - ( - sticky_events_to_token, - room_to_event_ids, - ) = await self.store.get_sticky_events_in_rooms( - all_interested_room_ids, - from_token=since_token_as_stream_token, - to_token=to_token.sticky_events_key, - now=now, - limit=min(sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC), + relevant_room_ids = self.find_relevant_room_ids_for_extension( + requested_lists=sticky_events_request.lists, + requested_room_ids=sticky_events_request.rooms, + actual_lists=actual_lists, + actual_room_ids=actual_room_ids, ) + + if not relevant_room_ids: + # No rooms are relevant, so shortcut out after recording + # any sticky events we're missing in out-of-scope rooms. + await self._record_sticky_events_in_out_of_scope_rooms( + previous_connection_state=previous_connection_state, + new_connection_state=new_connection_state, + relevant_room_ids=set(), + to_token=to_token, + from_token=from_token, + ) + return SlidingSyncResult.Extensions.StickyEventsExtension( + room_id_to_sticky_events={}, + ) + + room_id_to_sticky_events: dict[str, list[FilteredEvent]] = {} + # We need to handle the different cases depending on if we have sent + # down sticky events previously or not, so we split the relevant rooms + # up into different collections based on status. + initial_rooms: set[str] = set() + previously_rooms: dict[str, MultiWriterStreamToken] = {} + live_rooms: set[str] = set() + + for room_id in relevant_room_ids: + if not from_token or room_id in newly_joined_room_ids: + initial_rooms.add(room_id) + continue + + # If we're sending down the room from scratch again for some + # reason, we should also resend all the sticky events. + room_result = actual_room_response_map.get(room_id) + if room_result is not None and room_result.initial: + initial_rooms.add(room_id) + continue + + room_status = previous_connection_state.sticky_events.have_sent_room( + room_id + ) + if room_status.status == HaveSentRoomFlag.LIVE: + live_rooms.add(room_id) + elif room_status.status == HaveSentRoomFlag.PREVIOUSLY: + assert room_status.last_token is not None + previously_rooms[room_id] = room_status.last_token + elif room_status.status == HaveSentRoomFlag.NEVER: + initial_rooms.add(room_id) + else: + assert_never(room_status.status) + + # Build the groups of rooms to read sticky events for and their stream + # position to read from. + # [(set of appliable rooms, from_token), ...] + groups: list[tuple[AbstractSet[str], MultiWriterStreamToken]] = [] + if initial_rooms: + # Read from the start of the stream to get all unexpired sticky + # events. + groups.append((initial_rooms, StreamToken.START.sticky_events_key)) + previously_rooms_by_token: dict[MultiWriterStreamToken, set[str]] = {} + for room_id, last_token in previously_rooms.items(): + previously_rooms_by_token.setdefault(last_token, set()).add(room_id) + groups.extend( + (room_ids, last_token) + for last_token, room_ids in previously_rooms_by_token.items() + ) + if live_rooms: + assert from_token is not None + groups.append((live_rooms, from_token.stream_token.sticky_events_key)) + + now = self.clock.time_msec() + + # Now build a list of events to send, limited by the requested limit + remaining_limit = min( + sticky_events_request.limit, StickyEvent.MAX_EVENTS_IN_SYNC + ) + room_to_event_ids: dict[str, list[str]] = {} + for room_ids, group_from_token in groups: + if remaining_limit <= 0: + # We've run out of budget. Record that these rooms still need + # to be read from `group_from_token` on a subsequent request. + new_connection_state.sticky_events.record_partially_sent_rooms( + room_ids, group_from_token + ) + continue + + ( + read_to_token, + group_room_to_event_ids, + ) = await self.store.get_sticky_events_in_rooms( + room_ids, + from_token=group_from_token, + to_token=to_token.sticky_events_key, + now=now, + limit=remaining_limit, + ) + num_read = 0 + for room_id, event_ids in group_room_to_event_ids.items(): + room_to_event_ids.setdefault(room_id, []).extend(event_ids) + num_read += len(event_ids) + + if num_read >= remaining_limit: + # We hit the limit, so there may be more sticky events to send + # for these rooms. Record where we got up to so that the next + # request continues from there. + new_connection_state.sticky_events.record_partially_sent_rooms( + room_ids, read_to_token + ) + else: + # We read everything up to `to_token` for these rooms. + new_connection_state.sticky_events.record_sent_rooms(room_ids) + remaining_limit -= num_read + # No need to preserve sticky event order here because we will # reassemble it in the right order after. all_sticky_event_ids = { @@ -1103,7 +1208,6 @@ class SlidingSyncExtensionHandler: ) filtered_event_map = {ev.event.event_id: ev for ev in filtered_events} - room_id_to_sticky_events: dict[str, list[FilteredEvent]] = {} for room_id, sticky_event_ids in room_to_event_ids.items(): filtered_events_for_room = [ filtered_event_map[event_id] @@ -1117,11 +1221,60 @@ class SlidingSyncExtensionHandler: room_id_to_sticky_events[room_id] = filtered_events_for_room + await self._record_sticky_events_in_out_of_scope_rooms( + previous_connection_state=previous_connection_state, + new_connection_state=new_connection_state, + relevant_room_ids=relevant_room_ids, + to_token=to_token, + from_token=from_token, + ) + return SlidingSyncResult.Extensions.StickyEventsExtension( room_id_to_sticky_events=room_id_to_sticky_events, - next_batch=await SlidingSyncStickyEventsToken.from_stream_token( - self.store, sticky_events_to_token - ), + ) + + async def _record_sticky_events_in_out_of_scope_rooms( + self, + previous_connection_state: "PerConnectionState", + new_connection_state: "MutablePerConnectionState", + relevant_room_ids: AbstractSet[str], + to_token: StreamToken, + from_token: SlidingSyncStreamToken | None, + ) -> None: + """Record which rooms (may) have sticky events that we're not sending + down because they are not in scope for the sticky events extension, so + that we can send them when the room comes back into scope. + + We only need to check rooms that are `LIVE`: `PREVIOUSLY` rooms keep + pointing at their previous position and `NEVER` rooms will get everything + when they come into scope. + + Args: + previous_connection_state: The current per-connection state + new_connection_state: A mutable copy of the per-connection + state, used to record updates to the state. + relevant_room_ids: The rooms that are in scope for the extension + (empty if the extension is disabled). + to_token: The point in the stream to sync up to. + from_token: The point in the stream to sync from. + """ + if from_token is None: + # No previous connection state, so nothing to record. + return + + rooms_out_of_scope = [ + room_id + for room_id, room_status in previous_connection_state.sticky_events._statuses.items() + if room_status.status == HaveSentRoomFlag.LIVE + and room_id not in relevant_room_ids + ] + changed_rooms = await self.store.get_rooms_with_sticky_events_between( + rooms_out_of_scope, + from_token=from_token.stream_token.sticky_events_key, + to_token=to_token.sticky_events_key, + ) + new_connection_state.sticky_events.record_unsent_rooms( + changed_rooms, from_token.stream_token.sticky_events_key ) async def _get_profile_ids_for_profiles_extension( diff --git a/synapse/rest/client/sync.py b/synapse/rest/client/sync.py index bf4b2c660a..532f2866f4 100644 --- a/synapse/rest/client/sync.py +++ b/synapse/rest/client/sync.py @@ -1248,7 +1248,6 @@ class SlidingSyncRestServlet(RestServlet): return { "rooms": rooms_out, - "next_batch": sticky_events.next_batch.serialise(), } diff --git a/synapse/storage/databases/main/sliding_sync.py b/synapse/storage/databases/main/sliding_sync.py index a5d6cd2548..1bce901543 100644 --- a/synapse/storage/databases/main/sliding_sync.py +++ b/synapse/storage/databases/main/sliding_sync.py @@ -374,6 +374,15 @@ class SlidingSyncStore(SQLBaseStore): (have_sent_room.status.value, have_sent_room.last_token) ) + for ( + room_id, + have_sent_room, + ) in per_connection_state.sticky_events._statuses.items(): + key_values.append((connection_position, "sticky_events", room_id)) + value_values.append( + (have_sent_room.status.value, have_sent_room.last_token) + ) + self.db_pool.simple_upsert_many_txn( txn, table="sliding_sync_connection_streams", @@ -596,6 +605,7 @@ class SlidingSyncStore(SQLBaseStore): rooms: dict[str, HaveSentRoom[str]] = {} receipts: dict[str, HaveSentRoom[str]] = {} account_data: dict[str, HaveSentRoom[str]] = {} + sticky_events: dict[str, HaveSentRoom[str]] = {} receipt_rows = self.db_pool.simple_select_list_txn( txn, @@ -618,6 +628,8 @@ class SlidingSyncStore(SQLBaseStore): receipts[room_id] = have_sent_room elif stream == "account_data": account_data[room_id] = have_sent_room + elif stream == "sticky_events": + sticky_events[room_id] = have_sent_room else: # For forwards compatibility we ignore unknown streams, as in # future we want to be able to easily add more stream types. @@ -628,6 +640,7 @@ class SlidingSyncStore(SQLBaseStore): rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + sticky_events=RoomStatusMap(sticky_events), room_configs=room_configs, room_lazy_membership={}, ) @@ -810,6 +823,7 @@ class PerConnectionStateDB: rooms: "RoomStatusMap[str]" receipts: "RoomStatusMap[str]" account_data: "RoomStatusMap[str]" + sticky_events: "RoomStatusMap[str]" room_configs: Mapping[str, "RoomSyncConfig"] @@ -856,11 +870,24 @@ class PerConnectionStateDB: for room_id, status in per_connection_state.account_data.get_updates().items() } + sticky_events = { + room_id: HaveSentRoom( + status=status.status, + last_token=( + await status.last_token.to_string(store) + if status.last_token is not None + else None + ), + ) + for room_id, status in per_connection_state.sticky_events.get_updates().items() + } + log_kv( { "rooms": rooms, "receipts": receipts, "account_data": account_data, + "sticky_events": sticky_events, "room_configs": per_connection_state.room_configs.maps[0], } ) @@ -870,6 +897,7 @@ class PerConnectionStateDB: rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + sticky_events=RoomStatusMap(sticky_events), room_configs=per_connection_state.room_configs.maps[0], room_lazy_membership=per_connection_state.room_lazy_membership, ) @@ -910,10 +938,23 @@ class PerConnectionStateDB: for room_id, status in self.account_data._statuses.items() } + sticky_events = { + room_id: HaveSentRoom( + status=status.status, + last_token=( + await MultiWriterStreamToken.parse(store, status.last_token) + if status.last_token is not None + else None + ), + ) + for room_id, status in self.sticky_events._statuses.items() + } + return PerConnectionState( last_used_ts=self.last_used_ts, rooms=RoomStatusMap(rooms), receipts=RoomStatusMap(receipts), account_data=RoomStatusMap(account_data), + sticky_events=RoomStatusMap(sticky_events), room_configs=self.room_configs, ) diff --git a/synapse/storage/databases/main/sticky_events.py b/synapse/storage/databases/main/sticky_events.py index 4f95736709..afb77c43fc 100644 --- a/synapse/storage/databases/main/sticky_events.py +++ b/synapse/storage/databases/main/sticky_events.py @@ -273,6 +273,57 @@ class StickyEventsWorkerStore(StateGroupWorkerStore, CacheInvalidationWorkerStor ) return cast(list[tuple[int, str, str]], txn.fetchall()) + async def get_rooms_with_sticky_events_between( + self, + room_ids: Collection[str], + *, + from_token: MultiWriterStreamToken, + to_token: MultiWriterStreamToken, + ) -> set[str]: + """Given a set of room IDs, find out which of them have sticky events + in the `sticky_events` table (which may be expired and awaiting cleanup) + with sticky stream positions after `from_token` and at or before `to_token`. + + Bounds: from_token < ... <= to_token + + Args: + room_ids: The room IDs to check. + from_token: The sticky stream position to check from (exclusive). + to_token: The sticky stream position to check to (inclusive). + Returns: + The subset of `room_ids` that have sticky events in the range. + """ + if len(room_ids) == 0 or to_token.is_before_or_eq(from_token): + return set() + + def _get_rooms_with_sticky_events_between_txn( + txn: LoggingTransaction, + ) -> set[str]: + room_id_in_list_clause, room_id_in_list_values = make_in_list_sql_clause( + txn.database_engine, "room_id", room_ids + ) + stream_clause, stream_values = make_multiwriter_sharded_token_bounds_sql( + self.database_engine, + stream_id_column="stream_id", + instance_name_column="instance_name", + from_token_exclusive=from_token, + to_token_inclusive=to_token, + ) + txn.execute( + f""" + SELECT DISTINCT room_id + FROM sticky_events + WHERE {stream_clause} AND {room_id_in_list_clause} + """, + (*stream_values, *room_id_in_list_values), + ) + return {room_id for (room_id,) in txn} + + return await self.db_pool.runInteraction( + "get_rooms_with_sticky_events_between", + _get_rooms_with_sticky_events_between_txn, + ) + async def get_updated_sticky_events( self, *, from_id: int, to_id: int, limit: int ) -> list[StickyEventUpdate]: diff --git a/synapse/types/handlers/sliding_sync.py b/synapse/types/handlers/sliding_sync.py index f6d30460b5..007cfb228f 100644 --- a/synapse/types/handlers/sliding_sync.py +++ b/synapse/types/handlers/sliding_sync.py @@ -48,7 +48,7 @@ from synapse.types import ( ThreadSubscriptionsToken, UserID, ) -from synapse.types.rest.client import SlidingSyncBody, SlidingSyncStickyEventsToken +from synapse.types.rest.client import SlidingSyncBody from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -426,7 +426,7 @@ class SlidingSyncResult: @attr.s(slots=True, frozen=True, auto_attribs=True) class StickyEventsExtension: - """The Sticky Events extension (MSC4354) + """The Sticky Events extension (MSC4354 / MSC4480) Attributes: room_id_to_sticky_events: map (room_id -> [unexpired_sticky_events]) @@ -437,7 +437,6 @@ class SlidingSyncResult: """ room_id_to_sticky_events: Mapping[str, list[FilteredEvent]] - next_batch: SlidingSyncStickyEventsToken def __bool__(self) -> bool: return bool(self.room_id_to_sticky_events) @@ -772,8 +771,10 @@ class HaveSentRoomFlag(Enum): The valid state changes here are: NEVER -> LIVE + NEVER -> PREVIOUSLY (when only part of the room's data was sent, e.g. due to a limit) LIVE -> PREVIOUSLY PREVIOUSLY -> LIVE + PREVIOUSLY -> PREVIOUSLY (with a later token, when more of the data was sent) """ # The room has never been sent down (or we have forgotten we have sent it @@ -889,6 +890,20 @@ class MutableRoomStatusMap(RoomStatusMap[T]): self._statuses[room_id] = HaveSentRoom.live() + def record_partially_sent_rooms( + self, room_ids: StrCollection, last_token: T + ) -> None: + """Record that we have sent these rooms in the response, but only up to + `last_token` (exclusive of anything after it). + + Used for Sticky Events, which are implicitly paginated by the server. + But this is not necessarily a good pattern to follow for future work, + so consider alternatives. + See: https://github.com/matrix-org/matrix-spec-proposals/pull/4480#discussion_r4165887827 + """ + for room_id in room_ids: + self._statuses[room_id] = HaveSentRoom.previously(last_token) + def record_unsent_rooms(self, room_ids: StrCollection, from_token: T) -> None: """Record that we have not sent these rooms in the response, but there have been updates. @@ -935,6 +950,7 @@ class PerConnectionState: room_configs: Map from room_id to the `RoomSyncConfig` of all rooms that we have previously sent down. account_data: The status of each room for the account_data stream. + sticky_events: The status of each room for the sticky events stream. """ last_used_ts: int | None = None @@ -942,6 +958,7 @@ class PerConnectionState: rooms: RoomStatusMap[RoomStreamToken] = attr.Factory(RoomStatusMap) receipts: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap) account_data: RoomStatusMap[int] = attr.Factory(RoomStatusMap) + sticky_events: RoomStatusMap[MultiWriterStreamToken] = attr.Factory(RoomStatusMap) room_configs: Mapping[str, RoomSyncConfig] = attr.Factory(dict) @@ -954,6 +971,7 @@ class PerConnectionState: rooms=self.rooms.get_mutable(), receipts=self.receipts.get_mutable(), account_data=self.account_data.get_mutable(), + sticky_events=self.sticky_events.get_mutable(), room_configs=ChainMap({}, room_configs), ) @@ -963,6 +981,7 @@ class PerConnectionState: rooms=self.rooms.copy(), receipts=self.receipts.copy(), account_data=self.account_data.copy(), + sticky_events=self.sticky_events.copy(), room_configs=dict(self.room_configs), ) @@ -971,6 +990,7 @@ class PerConnectionState: len(self.account_data) + len(self.rooms) + len(self.receipts) + + len(self.sticky_events) + len(self.room_configs) ) @@ -1047,6 +1067,7 @@ class MutablePerConnectionState(PerConnectionState): rooms: MutableRoomStatusMap[RoomStreamToken] receipts: MutableRoomStatusMap[MultiWriterStreamToken] account_data: MutableRoomStatusMap[int] + sticky_events: MutableRoomStatusMap[MultiWriterStreamToken] room_configs: ChainMap[str, RoomSyncConfig] @@ -1067,6 +1088,7 @@ class MutablePerConnectionState(PerConnectionState): bool(self.rooms.get_updates()) or bool(self.receipts.get_updates()) or bool(self.account_data.get_updates()) + or bool(self.sticky_events.get_updates()) or bool(self.get_room_config_updates()) or any( change.has_updates(clock) diff --git a/synapse/types/rest/client/__init__.py b/synapse/types/rest/client/__init__.py index 28a48127f1..aa5eaaa031 100644 --- a/synapse/types/rest/client/__init__.py +++ b/synapse/types/rest/client/__init__.py @@ -18,14 +18,10 @@ # [This file includes modifications made by New Vector Limited] # # -import re -from typing import TYPE_CHECKING, ClassVar - import pydantic_core.core_schema from pydantic import ( ConfigDict, Field, - GetCoreSchemaHandler, GetPydanticSchema, StrictBool, StrictInt, @@ -34,22 +30,12 @@ from pydantic import ( field_validator, model_validator, ) -from pydantic_core import CoreSchema from typing_extensions import Annotated, Self -from synapse.types import ( - Absent, - AbsentType, - MultiWriterStreamToken, - NonNegativeStrictInt, - UnresolvedMultiWriterStreamToken, -) +from synapse.types import Absent, AbsentType, NonNegativeStrictInt from synapse.types.rest import RequestBodyModel from synapse.util.threepids import validate_email -if TYPE_CHECKING: - from synapse.storage.databases.main import DataStore - class AuthenticationData(RequestBodyModel): """ @@ -142,104 +128,6 @@ class MsisdnRequestTokenBody(ThreepidRequestTokenBody): phone_number: StrictStr -class SlidingSyncStickyEventsToken: - """ - A token returned by `next_batch` of the MSC4354 Sticky Events extension to Sliding Sync - and then accepted as the `since` parameter in the requests of the same extension. - - Current format: - SlidingSyncStickyEventsToken ::= 'sticky_' UnresolvedMultiWriterStreamToken - - where `MultiWriterStreamToken` is the serialised form of a (potentially sharded) - `MultiWriterStreamToken` for the sticky events stream, e.g. `42` or `m42~1.45`. - - The `sticky_` prefix allows us to make sure it's not swapped for another token - or to evolve the type of token accepted with backwards compatibility in the future. - - Note that the inner stream token can only be interpreted with access to the - database (to resolve instance IDs to instance names), so this class holds the - serialised form and converts on demand. - """ - - PATTERN = re.compile(r"^sticky_([0-9]+|m[0-9]+(~[0-9]+\.[0-9]+)*)$") - START: ClassVar["SlidingSyncStickyEventsToken"] - - def __init__( - self, *, unresolved_stream_token: UnresolvedMultiWriterStreamToken - ) -> None: - self._unresolved_stream_token = unresolved_stream_token - - @classmethod - async def from_stream_token( - cls, store: "DataStore", token: MultiWriterStreamToken - ) -> "SlidingSyncStickyEventsToken": - return cls(unresolved_stream_token=await token.to_unresolved(store)) - - async def to_stream_token(self, store: "DataStore") -> MultiWriterStreamToken: - return await MultiWriterStreamToken.from_unresolved( - store, self._unresolved_stream_token - ) - - @classmethod - def __get_pydantic_core_schema__( - cls, source_type: object, handler: GetCoreSchemaHandler - ) -> CoreSchema: - """ - This function is checked for and used by Pydantic when - attempting to deserialise/validate a field of this type. - - This returns a schema that will parse a string into an - instance of `SlidingSyncStickyEventsToken`. - """ - - return pydantic_core.core_schema.no_info_plain_validator_function( - cls._validate, - serialization=pydantic_core.core_schema.plain_serializer_function_ser_schema( - cls.serialise, - info_arg=False, - ), - ) - - @classmethod - def _validate(cls, v: object) -> Self: - """ - Create an instance from serialised string form. - - The inverse of `serialise`. - """ - if isinstance(v, cls): - return v - if isinstance(v, str): - match = cls.PATTERN.match(v) - if match is None: - raise ValueError(f"Invalid SlidingSyncStickyEventsToken format: {v!r}") - - return cls( - unresolved_stream_token=UnresolvedMultiWriterStreamToken.parse( - match.group(1) - ) - ) - raise ValueError(f"Cannot parse SlidingSyncStickyEventsToken from {type(v)}") - - def serialise(self) -> str: - """ - Convert this instance to string. - - The inverse of `_validate`. - """ - return f"sticky_{self._unresolved_stream_token.to_string()}" - - def __repr__(self) -> str: - # Use the serialised form as debug output. - return self.serialise() - - -# Starting reading a stream at 0 ensures all stream fact rows will be read -SlidingSyncStickyEventsToken.START = SlidingSyncStickyEventsToken( - unresolved_stream_token=UnresolvedMultiWriterStreamToken(stream=0) -) - - class SlidingSyncBody(RequestBodyModel): """ Sliding Sync API request body. @@ -452,10 +340,10 @@ class SlidingSyncBody(RequestBodyModel): enabled """ - enabled: StrictBool | None = False + enabled: StrictBool = False class AccountDataExtension(RequestBodyModel): - """The Account Data extension (MSC3959) + """The Account Data extension (MSC4548, previously MSC3959) Attributes: enabled @@ -465,14 +353,14 @@ class SlidingSyncBody(RequestBodyModel): extension to. """ - enabled: StrictBool | None = False + enabled: StrictBool = False # Process all lists defined in the Sliding Window API. (This is the default.) - lists: list[StrictStr] | None = ["*"] + lists: list[StrictStr] = ["*"] # Process all room subscriptions defined in the Room Subscription API. (This is the default.) - rooms: list[StrictStr] | None = ["*"] + rooms: list[StrictStr] = ["*"] class ReceiptsExtension(RequestBodyModel): - """The Receipts extension (MSC3960) + """The Receipts extension (MSC4551, previously MSC3960) Attributes: enabled @@ -482,14 +370,14 @@ class SlidingSyncBody(RequestBodyModel): extension to. """ - enabled: StrictBool | None = False + enabled: StrictBool = False # Process all lists defined in the Sliding Window API. (This is the default.) - lists: list[StrictStr] | None = ["*"] + lists: list[StrictStr] = ["*"] # Process all room subscriptions defined in the Room Subscription API. (This is the default.) - rooms: list[StrictStr] | None = ["*"] + rooms: list[StrictStr] = ["*"] class TypingExtension(RequestBodyModel): - """The Typing Notification extension (MSC3961) + """The Typing Notification extension (MSC4508, previously MSC3961) Attributes: enabled @@ -499,11 +387,11 @@ class SlidingSyncBody(RequestBodyModel): extension to. """ - enabled: StrictBool | None = False + enabled: StrictBool = False # Process all lists defined in the Sliding Window API. (This is the default.) - lists: list[StrictStr] | None = ["*"] + lists: list[StrictStr] = ["*"] # Process all room subscriptions defined in the Room Subscription API. (This is the default.) - rooms: list[StrictStr] | None = ["*"] + rooms: list[StrictStr] = ["*"] class ThreadSubscriptionsExtension(RequestBodyModel): """The Thread Subscriptions extension (MSC4308) @@ -513,21 +401,28 @@ class SlidingSyncBody(RequestBodyModel): limit: maximum number of subscription changes to return (default 100) """ - enabled: StrictBool | None = False + enabled: StrictBool = False limit: StrictInt = 100 class StickyEventsExtension(RequestBodyModel): - """The Sticky Events extension (MSC4354) + """The Sticky Events extension (MSC4354 / MSC4480) Attributes: enabled - limit: maximum number of sticky events to return in the extension (default 100) - since: either a string with the Sticky Events since token or absent + limit: maximum number of sticky events to return in the extension + (default 100). The server may cap this to a lower number. + lists: List of list keys (from the Sliding Window API) to apply this + extension to. + rooms: List of room IDs (from the Room Subscription API) to apply this + extension to. """ enabled: StrictBool = False limit: NonNegativeStrictInt = 100 - since: SlidingSyncStickyEventsToken | AbsentType = Absent + # Process all lists defined in the Sliding Window API. (This is the default.) + lists: list[StrictStr] = ["*"] + # Process all room subscriptions defined in the Room Subscription API. (This is the default.) + rooms: list[StrictStr] = ["*"] class ProfilesExtension(RequestBodyModel): """The Profile Updates extension (MSC4262) diff --git a/tests/replication/test_sharded_sticky_events.py b/tests/replication/test_sharded_sticky_events.py index b3a3b2d3ac..9de3cc9cc3 100644 --- a/tests/replication/test_sharded_sticky_events.py +++ b/tests/replication/test_sharded_sticky_events.py @@ -22,7 +22,7 @@ import synapse.rest.admin from synapse.api.constants import EventTypes from synapse.rest.client import login, room, sync from synapse.server import HomeServer -from synapse.types import JsonDict +from synapse.types import JsonDict, SlidingSyncStreamToken from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -246,21 +246,18 @@ class ShardedStickyEventsTestCase(BaseMultiWorkerStreamTestCase): sticky_extension = channel.json_body["extensions"][ "org.matrix.msc4354.sticky_events" ] - next_batch = sticky_extension["next_batch"] self.assertEqual( sticky_extension, - { - "rooms": {self.room_id1: {"events": []}}, - "next_batch": next_batch, - }, + {"rooms": {self.room_id1: {"events": []}}}, sticky_extension, ) - # The token should carry the per-writer positions, as worker2 is stalled. - self.assertTrue(next_batch.startswith("sticky_m"), next_batch) - sync_body["extensions"]["org.matrix.msc4354.sticky_events"]["since"] = ( - next_batch - ) + # The sticky events position in `pos` should carry the per-writer + # positions, with worker1 ahead of the stalled worker2. + sticky_events_key = self.get_success( + SlidingSyncStreamToken.from_string(self.store, pos) + ).stream_token.sticky_events_key + self.assertIn("worker1", sticky_events_key.instance_map, sticky_events_key) # worker2 catches up, so an unsharded sticky events position is now free to # advance past worker1's write. diff --git a/tests/rest/client/sliding_sync/test_extension_sticky_events.py b/tests/rest/client/sliding_sync/test_extension_sticky_events.py index ddf3f980a0..b19c8db481 100644 --- a/tests/rest/client/sliding_sync/test_extension_sticky_events.py +++ b/tests/rest/client/sliding_sync/test_extension_sticky_events.py @@ -34,28 +34,28 @@ logger = logging.getLogger(__name__) DUMMY_LISTS = { "main": { - # Don't include any rooms in the top-N window - "ranges": [[0, 0]], + # Include (up to) the first 11 rooms in the window, but don't return any + # timeline events for them. + "ranges": [[0, 10]], "required_state": [], "timeline_limit": 0, } } """ Subscription lists that can be used in the Sliding Sync request `lists` field, -which sets up a subscription that is interested in all rooms but does not let any rooms into the window, -thus does not return any timelines. +which sets up a list whose window covers all of the (few) rooms used in these +tests, but does not return any timelines. -Sufficient to get sticky event updates as per MSC4354: - -> The server MUST include sticky events across all rooms that would be matched by at least one subscription list -> (i.e. all rooms that the client is interested in), even if the room does not appear in top-N window for that -> subscription list at this time. -> Rooms that would not be matched by a list are not included, as this means the client is not interested -> in those rooms. -> -> — https://github.com/matrix-org/matrix-spec-proposals/pull/4354/changes#diff-d76bc1a1d612c6da37d024f5b57f7b8352939b8db8a7ee9c6b71c1a848359afdR213-R217 +As per the common per-room extension semantics of MSC4508 (which MSC4480 follows), +only rooms within the range of a list (or room subscriptions) are in scope for +the sticky events extension. """ +STICKY_EVENTS_EXT = "org.matrix.msc4354.sticky_events" + +ROOM_SUBSCRIPTION = {"required_state": [], "timeline_limit": 0} +"""A room subscription that doesn't return any timeline events.""" + class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): """Tests for the sticky events sliding sync extension""" @@ -90,7 +90,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): self, response_body: JsonDict, expected_events_by_room: dict[str, list[str]] | None, - ) -> str | None: + ) -> None: """Assert the sliding sync response was successful and has the expected sticky events. @@ -99,20 +99,16 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): expected_events_by_room: map of room ID to list of event IDs to expect (in the order we expect them), or None if we expect an empty sticky events extension response - - Returns the next_batch token from the sticky events section, - unless we're expecting an empty response. """ extensions = response_body["extensions"] - sticky_events = extensions.get("org.matrix.msc4354.sticky_events") + sticky_events = extensions.get(STICKY_EVENTS_EXT) # If there are no expected events, we shouldn't get anything in the response if expected_events_by_room is None: self.assertIsNone(sticky_events) - return None + return self.assertIsNotNone(sticky_events) - self.assertIsInstance(sticky_events["next_batch"], str) actual_rooms = sticky_events["rooms"] # Check that we have the expected rooms @@ -131,8 +127,23 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): ttl = actual_event["unsigned"][EventUnsignedContentFields.STICKY_TTL] self.assertIsInstance(ttl, int) - self.assertIn("next_batch", sticky_events) - return sticky_events["next_batch"] + def _send_sticky_event( + self, + room_id: str, + tok: str, + *, + duration: Duration = Duration(minutes=5), + body: str = "sticky message", + event_type: str = EventTypes.Message, + ) -> str: + """Send a sticky event and return its event ID.""" + return self.helper.send_sticky_event( + room_id, + event_type, + duration=duration, + content={"body": body, "msgtype": "m.text"}, + tok=tok, + )["event_id"] def test_empty_sync(self) -> None: """Test that enabling sticky events extension works on initial and incremental sync, @@ -144,7 +155,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -160,8 +171,8 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): self._assert_sticky_events_response(response_body, None) def test_initial_sync(self) -> None: - """Test that we get sticky events when we don't specify a since token - (initial sync). + """Test that we get all unexpired sticky events on an initial sync (no + `pos`), and only new ones on subsequent incremental syncs. """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -185,30 +196,18 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body: JsonDict = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) - - # Assert the response and then get the next_batch for the next sliding sync request - next_batch = self._assert_sticky_events_response( - response_body, {room_id: [sticky_event_id]} - ) - assert next_batch is not None + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) # Do an incremental sync immediately again - sync_body = { - "lists": DUMMY_LISTS, - "extensions": { - "org.matrix.msc4354.sticky_events": { - "enabled": True, - "since": next_batch, - } - }, - } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) # Check we don't get that event again self._assert_sticky_events_response(response_body, None) @@ -223,7 +222,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): )["event_id"] # Now the incremental sync should give us that event - response_body, _ = self.do_sync(sync_body, tok=user1_tok) + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) self._assert_sticky_events_response( response_body, {room_id: [sticky_event_id2]} ) @@ -252,7 +251,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -286,7 +285,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -343,7 +342,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -428,7 +427,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): } }, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -512,7 +511,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): } }, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, } }, @@ -562,15 +561,13 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): # Initial sync sync_body = { "lists": DUMMY_LISTS, - "extensions": { - "org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 2} - }, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}}, } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) # We expect to see the first 2 sticky events by stream order # and they should be in that stream order - next_batch = self._assert_sticky_events_response( + self._assert_sticky_events_response( response_body, {room_id: sticky_event_ids[0:2]} ) @@ -578,30 +575,37 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): sync_body = { "lists": DUMMY_LISTS, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, - # This makes it incremental - "since": next_batch, "limit": 3, } }, } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) # Should get remaining events, in stream order again self._assert_sticky_events_response( response_body, {room_id: sticky_event_ids[2:4]} ) + # And nothing more after that + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # A new sticky event is delivered as normal afterwards + sticky_event_id = self._send_sticky_event(room_id, user2_tok) + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + def test_sticky_event_limit_zero_does_not_advance(self) -> None: """ Tests that sending a limit of 0 in the Sticky Events request extension returns no - sticky events and omits the extension from the response, rather than erroring. - - 'Regression' test for a bug that _almost_ existed. - (The `next_batch` token would be wrongly advanced internally, - but because no events were returned, the entire extension response was not serialised - and so the dodgy `next_batch` never actually got sent to a client.) + sticky events and omits the extension from the response, rather than erroring, + and that doing so doesn't cause any sticky events to be skipped. """ user1_id = self.register_user("user1", "pass") user1_tok = self.login(user1_id, "pass") @@ -613,47 +617,42 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): self.helper.join(room_id, user1_id, tok=user1_tok) # Send 2 sticky events (more than our limit of 1) - sticky_event_ids: list[str] = [] - for i in range(2): - event_id = self.helper.send_sticky_event( - room_id, - EventTypes.Message, - duration=Duration(minutes=5), - content={"body": f"sticky message {i}", "msgtype": "m.text"}, - tok=user2_tok, - )["event_id"] - sticky_event_ids.append(event_id) + sticky_event_ids = [ + self._send_sticky_event(room_id, user2_tok, body=f"sticky message {i}") + for i in range(2) + ] # Initial sync sync_body = { "lists": DUMMY_LISTS, - "extensions": { - "org.matrix.msc4354.sticky_events": {"enabled": True, "limit": 1} - }, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 1}}, } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) # We expect to see the first sticky event - next_batch = self._assert_sticky_events_response( + self._assert_sticky_events_response( response_body, {room_id: sticky_event_ids[0:1]} ) # Incremental sync with limit=0 sync_body = { "lists": DUMMY_LISTS, - "extensions": { - "org.matrix.msc4354.sticky_events": { - "enabled": True, - # This makes it incremental - "since": next_batch, - "limit": 0, - } - }, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 0}}, } - response_body, _ = self.do_sync(sync_body, tok=user1_tok) - # You could imagine a buggy system returns a `next_batch` here with no events, - # but Synapse just omits the response extension altogether. - next_batch = self._assert_sticky_events_response(response_body, None) + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # The second sticky event is still delivered once the limit allows it + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 1}}, + } + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response( + response_body, {room_id: sticky_event_ids[1:2]} + ) def test_deduplication_with_timeline(self) -> None: """ @@ -710,7 +709,7 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): } }, "extensions": { - "org.matrix.msc4354.sticky_events": { + STICKY_EVENTS_EXT: { "enabled": True, "limit": 2, } @@ -718,9 +717,9 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): } response_body, _ = self.do_sync(sync_body, tok=user1_tok) - events_in_sticky_section = response_body["extensions"][ - "org.matrix.msc4354.sticky_events" - ]["rooms"][room_id]["events"] + events_in_sticky_section = response_body["extensions"][STICKY_EVENTS_EXT][ + "rooms" + ][room_id]["events"] event_ids_in_sticky_section = [e["event_id"] for e in events_in_sticky_section] events_in_timeline_section = response_body["rooms"][room_id]["timeline"] @@ -735,3 +734,573 @@ class SlidingSyncStickyEventsExtensionTestCase(SlidingSyncBase): self.assertEqual( event_ids_in_timeline_section, [sticky_event_ids[1], non_sticky_event_id] ) + + def test_retry_with_same_pos_redelivers(self) -> None: + """ + Test that retrying a request with the same `pos` (e.g. because the + response was lost) delivers the same sticky events again. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True}}, + } + _, from_token = self.do_sync(sync_body, tok=user1_tok) + + sticky_event_id = self._send_sticky_event(room_id, user2_tok) + + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # Retry with the same `pos` + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + def test_enable_extension_partway_through_connection(self) -> None: + """ + Test that enabling the extension for the first time partway through a + connection returns all unexpired sticky events for the rooms in scope, + including ones sent before the request's `pos`. + + This allows clients to defer enabling the extension past their initial + request, as required by MSC4508. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + sticky_event_id = self._send_sticky_event(room_id, user2_tok) + + # Initial sync and an incremental sync, both without the extension + sync_body: JsonDict = {"lists": DUMMY_LISTS} + _, from_token = self.do_sync(sync_body, tok=user1_tok) + _, from_token = self.do_sync(sync_body, since=from_token, tok=user1_tok) + + # Now enable the extension: we should get the sticky event sent before + # the connection even started. + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True}}, + } + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # But not again on the next request + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + def test_reenable_extension_after_disabling(self) -> None: + """ + Test that re-enabling the extension after having disabled it (or omitted + it) on the same connection returns the unexpired sticky events that were + sent while it was disabled, as well as those sent after the request's + `pos`. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + enabled_sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True}}, + } + disabled_sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": False}}, + } + omitted_sync_body: JsonDict = {"lists": DUMMY_LISTS} + + _, from_token = self.do_sync(enabled_sync_body, tok=user1_tok) + + # Disable the extension and send a sticky event while it is disabled + _, from_token = self.do_sync( + disabled_sync_body, since=from_token, tok=user1_tok + ) + sticky_event_id1 = self._send_sticky_event( + room_id, user2_tok, body="while disabled" + ) + response_body, from_token = self.do_sync( + disabled_sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Omit the extension altogether and send another sticky event, plus one + # that will have expired by the time the extension is re-enabled. + self._send_sticky_event( + room_id, user2_tok, duration=Duration(seconds=2), body="expires soon" + ) + sticky_event_id2 = self._send_sticky_event( + room_id, user2_tok, body="while omitted" + ) + response_body, from_token = self.do_sync( + omitted_sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + self.reactor.advance(3) + + # Re-enable: the unexpired sticky events sent in the meantime are replayed + response_body, from_token = self.do_sync( + enabled_sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id1, sticky_event_id2]} + ) + + # But not again afterwards, and new sticky events are delivered as normal + response_body, from_token = self.do_sync( + enabled_sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + sticky_event_id3 = self._send_sticky_event(room_id, user2_tok) + response_body, _ = self.do_sync( + enabled_sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id3]} + ) + + def test_newly_joined_room_gets_all_unexpired_sticky_events(self) -> None: + """ + Test that when the user joins a room on an incremental sync, all unexpired + sticky events in the room are returned, including ones sent before `pos`. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + sticky_event_id1 = self._send_sticky_event(room_id, user2_tok, body="one") + + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True}}, + } + # user1 isn't in the room yet, so sees nothing + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + # user1 joins and sees the historical sticky event + self.helper.join(room_id, user1_id, tok=user1_tok) + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id1]} + ) + + # user1 leaves, another sticky event is sent, and user1 re-joins: they + # should get all unexpired sticky events in the room again. + self.helper.leave(room_id, user1_id, tok=user1_tok) + _, from_token = self.do_sync(sync_body, since=from_token, tok=user1_tok) + sticky_event_id2 = self._send_sticky_event(room_id, user2_tok, body="two") + self.helper.join(room_id, user1_id, tok=user1_tok) + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_id1, sticky_event_id2]} + ) + + def test_scoping_by_lists_and_rooms(self) -> None: + """ + Test that the common per-room extension fields `lists` and `rooms` + (MSC4508) control which rooms sticky events are returned for. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + sticky_event_id = self._send_sticky_event(room_id, user2_tok) + + # No lists in scope, no room subscriptions: nothing is in scope + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + STICKY_EVENTS_EXT: {"enabled": True, "lists": [], "rooms": []} + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + # A list key that doesn't exist in the request is ignored + sync_body = { + "lists": DUMMY_LISTS, + "extensions": { + STICKY_EVENTS_EXT: {"enabled": True, "lists": ["other"], "rooms": []} + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + # Naming the list brings the room into scope + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "lists": ["main"]}}, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # A room subscription brings the room into scope, even with no lists + sync_body = { + "room_subscriptions": {room_id: ROOM_SUBSCRIPTION}, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "lists": []}}, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, {room_id: [sticky_event_id]}) + + # ...unless the subscription isn't named in `rooms` + sync_body = { + "room_subscriptions": {room_id: ROOM_SUBSCRIPTION}, + "extensions": { + STICKY_EVENTS_EXT: {"enabled": True, "lists": [], "rooms": []} + }, + } + response_body, _ = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + def test_room_outside_list_range_is_not_in_scope(self) -> None: + """ + Test that a room that matches a list but is outside the list's range is + not in scope for the extension, and that it gets all of its unexpired + sticky events when it comes into range. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_a, user1_id, tok=user1_tok) + room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_b, user1_id, tok=user1_tok) + + # A sticky event in room B, and then a regular message in room A so that + # room A is the most recent room (and therefore the only one in a range + # of [0, 0]). + sticky_event_id = self._send_sticky_event(room_id_b, user2_tok) + self.helper.send(room_id_a, body="bump A", tok=user2_tok) + + sync_body = { + "lists": { + "main": { + "ranges": [[0, 0]], + "required_state": [], + "timeline_limit": 0, + } + }, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True}}, + } + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) + self.assertEqual( + response_body["lists"]["main"]["ops"][0]["room_ids"], [room_id_a] + ) + # Room B is out of range, so its sticky event is not returned + self._assert_sticky_events_response(response_body, None) + + # Bump room B into range: we now get its (old) sticky event + self.helper.send(room_id_b, body="bump B", tok=user2_tok) + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self.assertEqual( + response_body["lists"]["main"]["ops"][0]["room_ids"], [room_id_b] + ) + self._assert_sticky_events_response( + response_body, {room_id_b: [sticky_event_id]} + ) + + # And not again afterwards + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + def test_room_leaving_and_reentering_scope(self) -> None: + """ + Test that when a room drops out of scope and later comes back into scope, + the sticky events sent in the meantime are returned (unless they have + expired by then). + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_a, user1_id, tok=user1_tok) + room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_b, user1_id, tok=user1_tok) + + def sync_body_with_scope(room_ids: list[str]) -> JsonDict: + return { + "room_subscriptions": { + room_id_a: ROOM_SUBSCRIPTION, + room_id_b: ROOM_SUBSCRIPTION, + }, + "extensions": { + STICKY_EVENTS_EXT: { + "enabled": True, + "lists": [], + "rooms": room_ids, + } + }, + } + + # Room A is in scope to begin with + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_a]), tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Room A drops out of scope + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Sticky events are sent in room A while it is out of scope + expiring_event_id = self._send_sticky_event( + room_id_a, + user2_tok, + duration=Duration(seconds=2), + body="expires soon", + ) + sticky_event_id = self._send_sticky_event( + room_id_a, user2_tok, body="still sticky" + ) + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Let the short-lived sticky event expire + self.reactor.advance(3) + + # Room A comes back into scope: we get the sticky event sent while it was + # out of scope, but not the one that has expired since. + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id_a: [sticky_event_id]} + ) + self.assertNotIn( + expiring_event_id, + [ + e["event_id"] + for e in response_body["extensions"][STICKY_EVENTS_EXT]["rooms"][ + room_id_a + ]["events"] + ], + ) + + # And not again afterwards + response_body, _ = self.do_sync( + sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + def test_pagination_across_rooms_entering_scope(self) -> None: + """ + Test that the `limit` bounds the response even when a room with many + unexpired sticky events comes into scope, with the remainder following on + subsequent requests in stream order, and that this doesn't lose track of + sticky events in other rooms that are out of scope in the meantime. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id_a = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_a, user1_id, tok=user1_tok) + room_id_b = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id_b, user1_id, tok=user1_tok) + + def sync_body_with_scope(room_ids: list[str]) -> JsonDict: + return { + "room_subscriptions": { + room_id_a: ROOM_SUBSCRIPTION, + room_id_b: ROOM_SUBSCRIPTION, + }, + "extensions": { + STICKY_EVENTS_EXT: { + "enabled": True, + "limit": 2, + "lists": [], + "rooms": room_ids, + } + }, + } + + # 3 sticky events in room B, which is out of scope to begin with + sticky_event_ids_b = [ + self._send_sticky_event(room_id_b, user2_tok, body=f"B {i}") + for i in range(3) + ] + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_a]), tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Room B comes into scope and room A drops out of scope, and a new sticky + # event arrives in room A (which must not be returned while out of scope). + sticky_event_id_a = self._send_sticky_event(room_id_a, user2_tok, body="A") + + # First page: the oldest 2 of room B's sticky events + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id_b: sticky_event_ids_b[0:2]} + ) + + # Second page: the remaining one + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id_b: sticky_event_ids_b[2:3]} + ) + + # Nothing more after that + response_body, from_token = self.do_sync( + sync_body_with_scope([room_id_b]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, None) + + # Room A comes back into scope and we get the sticky event that was sent + # while it was out of scope. + response_body, _ = self.do_sync( + sync_body_with_scope([room_id_a]), since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id_a: [sticky_event_id_a]} + ) + + def test_live_room_pagination(self) -> None: + """ + Test that `limit` also bounds the sticky events of rooms that are already + live on the connection, with the remainder following on subsequent + requests in stream order and nothing being lost or duplicated. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}}, + } + # The room is now live on the connection + response_body, from_token = self.do_sync(sync_body, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + sticky_event_ids = [ + self._send_sticky_event(room_id, user2_tok, body=f"sticky {i}") + for i in range(5) + ] + + # The sticky events arrive in pages of 2, in stream order + for page in (sticky_event_ids[0:2], sticky_event_ids[2:4]): + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response(response_body, {room_id: page}) + + # A new sticky event sent in the meantime is picked up after the backlog + sticky_event_id = self._send_sticky_event(room_id, user2_tok, body="newest") + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self._assert_sticky_events_response( + response_body, {room_id: [sticky_event_ids[4], sticky_event_id]} + ) + + # And then nothing more + response_body, _ = self.do_sync(sync_body, since=from_token, tok=user1_tok) + self._assert_sticky_events_response(response_body, None) + + def test_per_room_state_is_persisted(self) -> None: + """ + Test that the per-room sticky events position is persisted in the + per-connection state in the database (rather than only held in memory), + so that it survives restarts and is shared between workers. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + user2_id = self.register_user("user2", "pass") + user2_tok = self.login(user2_id, "pass") + + room_id = self.helper.create_room_as(user2_id, tok=user2_tok) + self.helper.join(room_id, user1_id, tok=user1_tok) + + def get_persisted_sticky_events_state(pos: str) -> list[tuple[str, str, str]]: + """Returns (room_id, room_status, last_token) rows for the given `pos`.""" + connection_position = int(pos.split("/", 1)[0]) + return self.get_success( + self.store.db_pool.simple_select_list( + table="sliding_sync_connection_streams", + keyvalues={ + "connection_position": connection_position, + "stream": "sticky_events", + }, + retcols=("room_id", "room_status", "last_token"), + desc="get_persisted_sticky_events_state", + ) + ) + + for _ in range(3): + self._send_sticky_event(room_id, user2_tok) + + sync_body = { + "lists": DUMMY_LISTS, + "extensions": {STICKY_EVENTS_EXT: {"enabled": True, "limit": 2}}, + } + + # The first page is cut short, so the room is recorded as partially sent, + # along with the position we got up to. + _, from_token = self.do_sync(sync_body, tok=user1_tok) + rows = get_persisted_sticky_events_state(from_token) + self.assertEqual(len(rows), 1, rows) + persisted_room_id, room_status, last_token = rows[0] + self.assertEqual(persisted_room_id, room_id) + self.assertEqual(room_status, "previously") + self.assertIsNotNone(last_token) + + # After the remainder is sent, the room is recorded as live (and the + # persisted position was correctly read back to continue from). + response_body, from_token = self.do_sync( + sync_body, since=from_token, tok=user1_tok + ) + self.assertEqual( + len( + response_body["extensions"][STICKY_EVENTS_EXT]["rooms"][room_id][ + "events" + ] + ), + 1, + ) + rows = get_persisted_sticky_events_state(from_token) + self.assertEqual(rows, [(room_id, "live", None)]) diff --git a/tests/storage/test_sticky_events.py b/tests/storage/test_sticky_events.py index e81ce129d4..9c8a7937b4 100644 --- a/tests/storage/test_sticky_events.py +++ b/tests/storage/test_sticky_events.py @@ -299,6 +299,74 @@ class StickyEventsTestCase(unittest.HomeserverTestCase): ), ) + def test_get_rooms_with_sticky_events_between(self) -> None: + """ + Tests `get_rooms_with_sticky_events_between` returns only the rooms with + sticky events in the given stream range. + """ + other_room_id = self.helper.create_room_as(self.user_id, tok=self.token) + room_ids = [self.room_id, other_room_id] + + start_token = self.store.get_sticky_events_stream_token() + + self.helper.send_sticky_event( + self.room_id, + EventTypes.Message, + duration=Duration(minutes=1), + content={"body": "message 1", "msgtype": "m.text"}, + tok=self.token, + ) + mid_token = self.store.get_sticky_events_stream_token() + + self.helper.send_sticky_event( + other_room_id, + EventTypes.Message, + duration=Duration(minutes=1), + content={"body": "message 2", "msgtype": "m.text"}, + tok=self.token, + ) + end_token = self.store.get_sticky_events_stream_token() + + # Both rooms changed over the whole range + self.assertEqual( + self.get_success( + self.store.get_rooms_with_sticky_events_between( + room_ids, from_token=start_token, to_token=end_token + ) + ), + {self.room_id, other_room_id}, + ) + + # Only the other room changed in the second half of the range + self.assertEqual( + self.get_success( + self.store.get_rooms_with_sticky_events_between( + room_ids, from_token=mid_token, to_token=end_token + ) + ), + {other_room_id}, + ) + + # Nothing changed in an empty range + self.assertEqual( + self.get_success( + self.store.get_rooms_with_sticky_events_between( + room_ids, from_token=end_token, to_token=end_token + ) + ), + set(), + ) + + # Rooms we didn't ask about are not returned + self.assertEqual( + self.get_success( + self.store.get_rooms_with_sticky_events_between( + [self.room_id], from_token=start_token, to_token=end_token + ) + ), + {self.room_id}, + ) + def test_outlier_events_not_in_table(self) -> None: """ Tests the behaviour of outliered and then de-outliered events in the