diff --git a/synapse/handlers/sliding_sync/__init__.py b/synapse/handlers/sliding_sync/__init__.py index 255a041d0e..c15c9e1de6 100644 --- a/synapse/handlers/sliding_sync/__init__.py +++ b/synapse/handlers/sliding_sync/__init__.py @@ -305,6 +305,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, + room_membership_for_user_at_to_token_map=room_membership_for_user_map, 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 66663c5e68..44a1cf7058 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -12,6 +12,7 @@ # . # +from collections import defaultdict import itertools import logging from typing import ( @@ -19,6 +20,7 @@ from typing import ( AbstractSet, ChainMap, Dict, + List, Mapping, MutableMapping, Optional, @@ -37,14 +39,18 @@ from synapse.api.constants import ( RelationTypes, ) from synapse.handlers.receipts import ReceiptEventSource +from synapse.handlers.sliding_sync.room_lists import RoomsForUserType from synapse.logging.opentracing import trace from synapse.storage.databases.main.receipts import ReceiptInRoom +from synapse.storage.databases.main.relations import ThreadUpdateInfo from synapse.types import ( DeviceListUpdates, JsonMapping, MultiWriterStreamToken, + RoomStreamToken, SlidingSyncStreamToken, StrCollection, + StreamKeyType, StreamToken, ThreadSubscriptionsToken, ) @@ -60,6 +66,7 @@ from synapse.util.async_helpers import ( concurrently_execute, gather_optional_coroutines, ) +from synapse.visibility import filter_events_for_client _ThreadSubscription: TypeAlias = ( SlidingSyncResult.Extensions.ThreadSubscriptionsExtension.ThreadSubscription @@ -84,6 +91,7 @@ class SlidingSyncExtensionHandler: self.device_handler = hs.get_device_handler() self.push_rules_handler = hs.get_push_rules_handler() self.relations_handler = hs.get_relations_handler() + self._storage_controllers = hs.get_storage_controllers() self._enable_thread_subscriptions = hs.config.experimental.msc4306_enabled self._enable_threads_ext = hs.config.experimental.msc4360_enabled @@ -96,6 +104,7 @@ class SlidingSyncExtensionHandler: actual_lists: Mapping[str, SlidingSyncResult.SlidingWindowList], actual_room_ids: Set[str], actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], + room_membership_for_user_at_to_token_map: Mapping[str, RoomsForUserType], to_token: StreamToken, from_token: Optional[SlidingSyncStreamToken], ) -> SlidingSyncResult.Extensions: @@ -111,6 +120,8 @@ class SlidingSyncExtensionHandler: 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. + room_membership_for_user_at_to_token_map: A map of room ID to the membership + information for the user in the room at the time of `to_token`. to_token: The latest point in the stream to sync up to. from_token: The point in the stream to sync from. """ @@ -191,7 +202,9 @@ class SlidingSyncExtensionHandler: threads_coro = self.get_threads_extension_response( sync_config=sync_config, threads_request=sync_config.extensions.threads, + actual_room_ids=actual_room_ids, actual_room_response_map=actual_room_response_map, + room_membership_for_user_at_to_token_map=room_membership_for_user_at_to_token_map, to_token=to_token, from_token=from_token, ) @@ -997,7 +1010,9 @@ class SlidingSyncExtensionHandler: self, sync_config: SlidingSyncConfig, threads_request: SlidingSyncConfig.Extensions.ThreadsExtension, + actual_room_ids: Set[str], actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], + room_membership_for_user_at_to_token_map: Mapping[str, RoomsForUserType], to_token: StreamToken, from_token: Optional[SlidingSyncStreamToken], ) -> Optional[SlidingSyncResult.Extensions.ThreadsExtension]: @@ -1009,6 +1024,8 @@ class SlidingSyncExtensionHandler: actual_room_response_map: A map of room ID to room results in the sliding sync response. Used to determine which threads already have events in the room timeline. + room_membership_for_user_at_to_token_map: A map of room ID to the membership + information for the user in the room at the time of `to_token`. to_token: The point in the stream to sync up to. from_token: The point in the stream to sync from. @@ -1018,23 +1035,113 @@ class SlidingSyncExtensionHandler: if not threads_request.enabled: return None + + # if ( + # # No timeline for invite/knock rooms + # room_membership_for_user_at_to_token.membership + # not in (Membership.INVITE, Membership.KNOCK) + # ): + # limited = False + # # We want to start off using the `to_token` (vs `from_token`) because we look + # # backwards from the `to_token` up to the `timeline_limit` and we might not + # # reach the `from_token` before we hit the limit. We will update the room stream + # # position once we've fetched the events to point to the earliest event fetched. + # prev_batch_token = to_token + # + # # We're going to paginate backwards from the `to_token` + # to_bound = to_token.room_key + # # People shouldn't see past their leave/ban event + # if room_membership_for_user_at_to_token.membership in ( + # Membership.LEAVE, + # Membership.BAN, + # ): + # to_bound = room_membership_for_user_at_to_token.event_pos.to_room_stream_token() + # + # # For initial `/sync` (and other historical scenarios mentioned above), we + # # want to view a historical section of the timeline; to fetch events by + # # `topological_ordering` (best representation of the room DAG as others were + # # seeing it at the time). This also aligns with the order that `/messages` + # # returns events in. + # # + # # For incremental `/sync`, we want to get all updates for rooms since + # # the last `/sync` (regardless if those updates arrived late or happened + # # a while ago in the past); to fetch events by `stream_ordering` (in the + # # order they were received by the server). + # # + # # Relevant spec issue: https://github.com/matrix-org/matrix-spec/issues/1917 + # # + # # FIXME: Using workaround for mypy, + # # https://github.com/python/mypy/issues/10740#issuecomment-1997047277 and + # # https://github.com/python/mypy/issues/17479 + # paginate_room_events_by_topological_ordering: PaginateFunction = ( + # self.store.paginate_room_events_by_topological_ordering + # ) + # paginate_room_events_by_stream_ordering: PaginateFunction = ( + # self.store.paginate_room_events_by_stream_ordering + # ) + # pagination_method: PaginateFunction = ( + # # Use `topographical_ordering` for historical events + # paginate_room_events_by_topological_ordering + # if timeline_from_bound is None + # # Use `stream_ordering` for updates + # else paginate_room_events_by_stream_ordering + # ) + # timeline_events, new_room_key, limited = await pagination_method( + # room_id=room_id, + # # The bounds are reversed so we can paginate backwards + # # (from newer to older events) starting at to_bound. + # # This ensures we fill the `limit` with the newest events first, + # from_key=to_bound, + # to_key=timeline_from_bound, + # direction=Direction.BACKWARDS, + # limit=room_sync_config.timeline_limit, + # ) + # + # # Make sure we don't expose any events that the client shouldn't see + # timeline_events = await filter_events_for_client( + # self.storage_controllers, + # user.to_string(), + # timeline_events, + # is_peeking=room_membership_for_user_at_to_token.membership + # != Membership.JOIN, + # filter_send_to_client=True, + # ) + + # TODO: Improve by doing subqueries for rooms where user membership is changed + # Fetch thread updates globally across all joined rooms. # The database layer returns a StreamToken (exclusive) for prev_batch if there # are more results. ( all_thread_updates, prev_batch_token, - ) = await self.store.get_thread_updates_for_user( - user_id=sync_config.user.to_string(), + ) = await self.store.get_thread_updates_for_rooms( + room_ids=actual_room_ids, from_token=from_token.stream_token.room_key if from_token else None, to_token=to_token.room_key, limit=threads_request.limit, - include_thread_roots=threads_request.include_roots, ) if len(all_thread_updates) == 0: return None + all_event_ids = {update.event_id for updates in all_thread_updates.values() for update in updates} + all_events = await self.store.get_events_as_list(all_event_ids) + filtered_events = await filter_events_for_client(self._storage_controllers, sync_config.user.to_string(), all_events) + + filtered_updates: Dict[str, List[ThreadUpdateInfo]] = defaultdict(list) + for thread_id, updates in all_thread_updates.items(): + for update in updates: + for ev in filtered_events: + if update.event_id == ev.event_id: + filtered_updates[thread_id].append(update) + + # Optionally fetch thread root events + thread_root_event_map = {} + if threads_request.include_roots: + thread_root_events = await self.store.get_events_as_list(filtered_updates.keys()) + thread_root_event_map = {e.event_id: e for e in thread_root_events} + # Identify which threads already have events in the room timelines. # If include_roots=False, we'll omit these threads from the extension response # since the client already sees the thread activity in the timeline. @@ -1061,12 +1168,11 @@ class SlidingSyncExtensionHandler: # Collect thread root events and get bundled aggregations. # Only fetch bundled aggregations if we have thread root events to attach them to. thread_root_events = [ - update.thread_root_event - for update in all_thread_updates + root_event + for root_event in thread_root_event_map.values() # Don't fetch bundled aggregations for threads with events already in the # timeline response since they will get filtered out later anyway. - if update.thread_root_event - and update.thread_root_event.event_id not in threads_in_timeline + if root_event.event_id not in threads_in_timeline ] aggregations_map = {} if thread_root_events: @@ -1076,23 +1182,38 @@ class SlidingSyncExtensionHandler: ) thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {} - for update in all_thread_updates: + for thread_root, updates in filtered_updates.items(): # Skip this thread if it already has events in the room timeline # (unless include_roots=True, in which case we always include it) - if update.thread_id in threads_in_timeline: + if thread_root in threads_in_timeline: continue - # Only look up bundled aggregations if we have a thread root event - bundled_aggs = ( - aggregations_map.get(update.thread_id) - if update.thread_root_event - else None - ) + # We only care about the latest update for the thread. + # Since values were obtained from the db in DESC order, the first event in + # the list will be the latest event. + update = updates[0] - thread_updates.setdefault(update.room_id, {})[update.thread_id] = ( + # # Generate prev_batch token if this thread has more than one update. + per_thread_prev_batch = None + if len(updates) > 1: + # Create a token pointing to one position before the latest event's + # stream position. + # This makes it exclusive - /relations with dir=b won't return the + # latest event again. + # Use StreamToken.START as base (all other streams at 0) since only room + # position matters. + per_thread_prev_batch = StreamToken.START.copy_and_replace( + StreamKeyType.ROOM, RoomStreamToken(stream=update.stream_ordering - 1) + ) + + # Only look up bundled aggregations if we have a thread root event + bundled_aggs = (aggregations_map.get(thread_root)) + thread_root_event = thread_root_event_map.get(thread_root) + + thread_updates.setdefault(update.room_id, {})[thread_root] = ( _ThreadUpdate( - thread_root=update.thread_root_event, - prev_batch=update.prev_batch, + thread_root=thread_root_event, + prev_batch=per_thread_prev_batch, bundled_aggregations=bundled_aggs, ) ) diff --git a/synapse/storage/databases/main/relations.py b/synapse/storage/databases/main/relations.py index 5d412d38b3..96be8e7ffc 100644 --- a/synapse/storage/databases/main/relations.py +++ b/synapse/storage/databases/main/relations.py @@ -18,6 +18,7 @@ # # +from collections import defaultdict import logging from typing import ( TYPE_CHECKING, @@ -37,7 +38,7 @@ from typing import ( import attr -from synapse.api.constants import MAIN_TIMELINE, Direction, Membership, RelationTypes +from synapse.api.constants import MAIN_TIMELINE, Direction, RelationTypes from synapse.api.errors import SynapseError from synapse.events import EventBase from synapse.storage._base import SQLBaseStore @@ -117,10 +118,9 @@ class ThreadUpdateInfo: re-receiving the latest event that was already included in the sliding sync response. """ - thread_id: str + event_id: str room_id: str - thread_root_event: Optional[EventBase] - prev_batch: Optional[StreamToken] + stream_ordering: int class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): @@ -1150,15 +1150,14 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): "get_related_thread_id", _get_related_thread_id ) - async def get_thread_updates_for_user( + async def get_thread_updates_for_rooms( self, *, - user_id: str, + room_ids: Collection[str], from_token: Optional[RoomStreamToken] = None, to_token: Optional[RoomStreamToken] = None, limit: int = 5, - include_thread_roots: bool = False, - ) -> Tuple[Sequence[ThreadUpdateInfo], Optional[StreamToken]]: + ) -> Tuple[Dict[str, List[ThreadUpdateInfo]], Optional[StreamToken]]: """Get a list of updated threads, ordered by stream ordering of their latest reply, filtered to only include threads in rooms where the user is currently joined. @@ -1184,74 +1183,53 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): # Ensure bad limits aren't being passed in. assert limit > 0 - # Generate the pagination clause, if necessary. - # - # Find any threads where the latest reply is between the stream ordering bounds. - pagination_clause = "" - pagination_args: List[str] = [] - if from_token: - from_bound = from_token.stream - pagination_clause += " AND stream_ordering > ?" - pagination_args.append(str(from_bound)) - - if to_token: - to_bound = to_token.stream - pagination_clause += " AND stream_ordering <= ?" - pagination_args.append(str(to_bound)) - - # Build the update count clause - count events in the thread within the sync window - update_count_clause = "" - update_count_args: List[str] = [] - update_count_clause = f""" - (SELECT COUNT(*) - FROM event_relations AS er - INNER JOIN events AS e ON er.event_id = e.event_id - WHERE er.relates_to_id = threads.thread_id - AND er.relation_type = '{RelationTypes.THREAD}'""" - if from_token: - update_count_clause += " AND e.stream_ordering > ?" - update_count_args.append(str(from_token.stream)) - if to_token: - update_count_clause += " AND e.stream_ordering <= ?" - update_count_args.append(str(to_token.stream)) - update_count_clause += ")" - - # Filter threads to only those in rooms where the user is currently joined. - sql = f""" - SELECT thread_id, room_id, stream_ordering, {update_count_clause} AS update_count - FROM threads - WHERE EXISTS ( - SELECT 1 - FROM local_current_membership AS lcm - WHERE lcm.room_id = threads.room_id - AND lcm.user_id = ? - AND lcm.membership = ? - ) - {pagination_clause} - ORDER BY stream_ordering DESC - LIMIT ? - """ + if len(room_ids) == 0: + return ({}), None def _get_thread_updates_for_user_txn( txn: LoggingTransaction, - ) -> Tuple[List[Tuple[str, str, int, int]], Optional[int]]: - # Add 1 to the limit as a free way of determining if there are more results - # than the limit amount. If `limit + 1` results are returned, then there are - # more results. Otherwise we would need to do a separate query to determine - # if this was true when exactly `limit` results are returned. + ) -> Tuple[List[Tuple[str, str, str, int]], Optional[int]]: + room_clause, room_id_values = make_in_list_sql_clause(txn.database_engine, "e.room_id", room_ids) + + # Generate the pagination clause, if necessary. + pagination_clause = "" + pagination_args: List[str] = [] + if from_token: + from_bound = from_token.stream + pagination_clause += " AND stream_ordering > ?" + pagination_args.append(str(from_bound)) + + if to_token: + to_bound = to_token.stream + pagination_clause += " AND stream_ordering <= ?" + pagination_args.append(str(to_bound)) + + # TODO: improve the fact that multiple hits for the same thread means we + # won't get as many updates for the sss response + + # Find any thread events between the stream ordering bounds. + sql = f""" + SELECT e.event_id, er.relates_to_id, e.room_id, e.stream_ordering + FROM event_relations AS er + INNER JOIN events AS e ON er.event_id = e.event_id + WHERE er.relation_type = '{RelationTypes.THREAD}' + AND {room_clause} + {pagination_clause} + ORDER BY stream_ordering DESC + LIMIT ? + """ + txn.execute( sql, ( - *update_count_args, - user_id, - Membership.JOIN, + *room_id_values, *pagination_args, - limit + 1, + limit, ), ) - # SQL returns: thread_id, room_id, stream_ordering, update_count - rows = cast(List[Tuple[str, str, int, int]], txn.fetchall()) + # SQL returns: event_id, thread_id, room_id, stream_ordering + rows = cast(List[Tuple[str, str, str, int]], txn.fetchall()) # If there are more events, generate the next pagination key from the # last thread which will be returned. @@ -1261,7 +1239,7 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): # that will be the last row we return from this function. # This works as an exclusive bound that can be backpaginated from. # Use the stream_ordering field (index 2 in original rows) - next_token = rows[-2][2] + next_token = rows[-2][3] return rows[:limit], next_token @@ -1280,35 +1258,14 @@ class RelationsWorkerStore(EventsWorkerStore, SQLBaseStore): StreamKeyType.ROOM, RoomStreamToken(stream=next_token_int - 1) ) - # Optionally fetch thread root events - event_map = {} - if include_thread_roots and thread_infos: - thread_root_ids = [thread_id for thread_id, _, _, _ in thread_infos] - thread_root_events = await self.get_events_as_list(thread_root_ids) - event_map = {e.event_id: e for e in thread_root_events} - - # Build ThreadUpdateInfo objects with per-thread prev_batch tokens. - thread_update_infos = [] - for thread_id, room_id, stream_ordering, update_count in thread_infos: - # Generate prev_batch token if this thread has more than one update. - per_thread_prev_batch = None - if update_count > 1: - # Create a token pointing to one position before the latest event's - # stream position. - # This makes it exclusive - /relations with dir=b won't return the - # latest event again. - # Use StreamToken.START as base (all other streams at 0) since only room - # position matters. - per_thread_prev_batch = StreamToken.START.copy_and_replace( - StreamKeyType.ROOM, RoomStreamToken(stream=stream_ordering - 1) - ) - - thread_update_infos.append( + # Build ThreadUpdateInfo objects. + thread_update_infos: Dict[str, List[ThreadUpdateInfo]] = defaultdict(list) + for event_id, thread_id, room_id, stream_ordering in thread_infos: + thread_update_infos[thread_id].append( ThreadUpdateInfo( - thread_id=thread_id, + event_id=event_id, room_id=room_id, - thread_root_event=event_map.get(thread_id), - prev_batch=per_thread_prev_batch, + stream_ordering=stream_ordering, ) ) diff --git a/tests/rest/client/sliding_sync/test_extension_threads.py b/tests/rest/client/sliding_sync/test_extension_threads.py index 3855b03672..4a10313a93 100644 --- a/tests/rest/client/sliding_sync/test_extension_threads.py +++ b/tests/rest/client/sliding_sync/test_extension_threads.py @@ -134,6 +134,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # base = self.store.get_max_thread_subscriptions_stream_id() sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, # Set to 0, otherwise events will be in timeline, not extension + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -150,6 +157,7 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): {"updates": {room_id: {thread_root_id: {}}}}, ) + def test_threads_incremental_sync(self) -> None: """ Test new thread updates appear in incremental sync response. @@ -158,6 +166,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): user1_tok = self.login(user1_id, "pass") room_id = self.helper.create_room_as(user1_id, tok=user1_tok) sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -256,6 +271,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # User2 syncs with threads extension enabled sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -296,6 +318,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Initial sync for user2 sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -380,6 +409,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Sync with include_roots=True sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -434,6 +470,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Sync with include_roots=False (explicitly) sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -451,6 +494,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Also test with include_roots omitted (should behave the same) sync_body_no_param = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -478,6 +528,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Initial sync to establish baseline sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -529,6 +586,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Initial sync to establish baseline sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True, @@ -668,6 +732,13 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): # Initial sync (no from_token) sync_body = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 0, + } + }, "extensions": { EXT_NAME: { "enabled": True,