From 4d7826b006371ca2b55a1eec9976e22fd3408995 Mon Sep 17 00:00:00 2001 From: Devon Hudson Date: Wed, 8 Oct 2025 17:01:40 -0600 Subject: [PATCH] Filter events from extension if in timeline --- synapse/handlers/sliding_sync/extensions.py | 40 ++++- .../sliding_sync/test_extension_threads.py | 149 +++++++++++++++--- 2 files changed, 170 insertions(+), 19 deletions(-) diff --git a/synapse/handlers/sliding_sync/extensions.py b/synapse/handlers/sliding_sync/extensions.py index 5defd047aa..2c39838fa8 100644 --- a/synapse/handlers/sliding_sync/extensions.py +++ b/synapse/handlers/sliding_sync/extensions.py @@ -24,12 +24,13 @@ from typing import ( Optional, Sequence, Set, + Tuple, cast, ) from typing_extensions import TypeAlias, assert_never -from synapse.api.constants import AccountDataTypes, EduTypes +from synapse.api.constants import AccountDataTypes, EduTypes, RelationTypes from synapse.handlers.receipts import ReceiptEventSource from synapse.logging.opentracing import trace from synapse.storage.databases.main.receipts import ReceiptInRoom @@ -185,6 +186,7 @@ class SlidingSyncExtensionHandler: threads_coro = self.get_threads_extension_response( sync_config=sync_config, threads_request=sync_config.extensions.threads, + actual_room_response_map=actual_room_response_map, to_token=to_token, from_token=from_token, ) @@ -990,6 +992,7 @@ class SlidingSyncExtensionHandler: self, sync_config: SlidingSyncConfig, threads_request: SlidingSyncConfig.Extensions.ThreadsExtension, + actual_room_response_map: Mapping[str, SlidingSyncResult.RoomResult], to_token: StreamToken, from_token: Optional[SlidingSyncStreamToken], ) -> Optional[SlidingSyncResult.Extensions.ThreadsExtension]: @@ -998,6 +1001,9 @@ class SlidingSyncExtensionHandler: Args: sync_config: Sync configuration. threads_request: The threads extension from the request. + 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. to_token: The point in the stream to sync up to. from_token: The point in the stream to sync from. @@ -1024,6 +1030,29 @@ class SlidingSyncExtensionHandler: if len(all_thread_updates) == 0: return None + # 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. + # If include_roots=True, we include all threads regardless, because the client + # wants the thread root events. + threads_in_timeline: Set[Tuple[str, str]] = set() # (room_id, thread_id) + if not threads_request.include_roots: + for room_id, room_result in actual_room_response_map.items(): + if room_result.timeline_events: + for event in room_result.timeline_events: + # Check if this event is part of a thread + relates_to = event.content.get("m.relates_to") + if not isinstance(relates_to, dict): + continue + + rel_type = relates_to.get("rel_type") + + # If this is a thread reply, track the thread + if rel_type == RelationTypes.THREAD: + thread_id = relates_to.get("event_id") + if thread_id: + threads_in_timeline.add((room_id, thread_id)) + # 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 = [ @@ -1040,6 +1069,11 @@ class SlidingSyncExtensionHandler: thread_updates: Dict[str, Dict[str, _ThreadUpdate]] = {} for update in all_thread_updates: + # 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.room_id, update.thread_id) 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) @@ -1055,6 +1089,10 @@ class SlidingSyncExtensionHandler: ) ) + # If after filtering we have no thread updates, return None to omit the extension + if not thread_updates: + return None + return SlidingSyncResult.Extensions.ThreadsExtension( updates=thread_updates, prev_batch=prev_batch_token, diff --git a/tests/rest/client/sliding_sync/test_extension_threads.py b/tests/rest/client/sliding_sync/test_extension_threads.py index 50e34993a7..aeace02da7 100644 --- a/tests/rest/client/sliding_sync/test_extension_threads.py +++ b/tests/rest/client/sliding_sync/test_extension_threads.py @@ -587,16 +587,9 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos) # Assert: Thread update SHOULD have prev_batch (3 updates) - thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + prev_batch = response_body["extensions"][EXT_NAME]["updates"][room_id][ thread_root_id - ] - self.assertIn( - "prev_batch", - thread_update, - "Threads with multiple updates should have prev_batch", - ) - - prev_batch = thread_update["prev_batch"] + ]["prev_batch"] self.assertIsNotNone(prev_batch, "prev_batch should not be None") # Now use the prev_batch token with /relations endpoint to paginate backwards @@ -684,16 +677,9 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): response_body, _ = self.do_sync(sync_body, tok=user1_tok) # Assert: Thread update SHOULD have prev_batch on initial sync (2+ updates exist) - thread_update = response_body["extensions"][EXT_NAME]["updates"][room_id][ + prev_batch = response_body["extensions"][EXT_NAME]["updates"][room_id][ thread_root_id - ] - self.assertIn( - "prev_batch", - thread_update, - "Threads with multiple updates should have prev_batch even on initial sync", - ) - - prev_batch = thread_update["prev_batch"] + ]["prev_batch"] self.assertIsNotNone(prev_batch) # Use prev_batch with /relations to fetch the thread history @@ -721,3 +707,130 @@ class SlidingSyncThreadsExtensionTestCase(SlidingSyncBase): returned_event_ids, "Second reply (latest) should NOT be in relations response - already returned in sliding sync", ) + + def test_thread_in_timeline_omitted_without_include_roots(self) -> None: + """ + Test that threads with events in the room timeline are omitted from the + extension response when include_roots=False. When all threads are filtered out, + the entire extension should be omitted from the response. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + room_id = self.helper.create_room_as(user1_id, tok=user1_tok) + + # Create thread root + thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok) + thread_root_id = thread_root_resp["event_id"] + + # Initial sync to establish baseline + sync_body: JsonDict = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 5, + } + }, + "extensions": { + EXT_NAME: { + "enabled": True, + "include_roots": False, + } + }, + } + _, sync_pos = self.do_sync(sync_body, tok=user1_tok) + + # Send a reply to the thread + self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Reply 1", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + + # Incremental sync - the reply should be in the timeline + response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos) + + # Assert: Extension should be omitted entirely since the only thread with updates + # is already visible in the timeline (include_roots=False) + self.assertNotIn( + EXT_NAME, + response_body.get("extensions", {}), + "Extension should be omitted when all threads are filtered out (in timeline with include_roots=False)", + ) + + def test_thread_in_timeline_included_with_include_roots(self) -> None: + """ + Test that threads with events in the room timeline are still included in the + extension response when include_roots=True, because the client wants the root event. + """ + user1_id = self.register_user("user1", "pass") + user1_tok = self.login(user1_id, "pass") + room_id = self.helper.create_room_as(user1_id, tok=user1_tok) + + # Create thread root + thread_root_resp = self.helper.send(room_id, body="Thread root", tok=user1_tok) + thread_root_id = thread_root_resp["event_id"] + + # Initial sync to establish baseline + sync_body: JsonDict = { + "lists": { + "foo-list": { + "ranges": [[0, 1]], + "required_state": [], + "timeline_limit": 5, + } + }, + "extensions": { + EXT_NAME: { + "enabled": True, + "include_roots": True, + } + }, + } + _, sync_pos = self.do_sync(sync_body, tok=user1_tok) + + # Send a reply to the thread + reply_resp = self.helper.send_event( + room_id, + type="m.room.message", + content={ + "msgtype": "m.text", + "body": "Reply 1", + "m.relates_to": { + "rel_type": RelationTypes.THREAD, + "event_id": thread_root_id, + }, + }, + tok=user1_tok, + ) + reply_id = reply_resp["event_id"] + + # Incremental sync - the reply should be in the timeline + response_body, _ = self.do_sync(sync_body, tok=user1_tok, since=sync_pos) + + # Assert: The thread reply should be in the room timeline + room_response = response_body["rooms"][room_id] + timeline_event_ids = [event["event_id"] for event in room_response["timeline"]] + self.assertIn( + reply_id, + timeline_event_ids, + "Thread reply should be in the room timeline", + ) + + # Assert: Thread SHOULD be in extension (include_roots=True) + thread_updates = response_body["extensions"][EXT_NAME]["updates"][room_id] + self.assertIn( + thread_root_id, + thread_updates, + "Thread should be included in extension when include_roots=True, even if in timeline", + ) + # Verify the thread root event is present + self.assertIn("thread_root", thread_updates[thread_root_id])