Include bundled aggregations for the latest event in a thread

changelog.d/12273.bugfix

@@ -0,0 +1 @@
+Fix a bug introduced in Synapse v1.48.0 where latest thread reply provided failed to include the proper bundled aggregations.

synapse/events/utils.py

@@ -425,13 +425,12 @@ def serialize_event(
 
         # Check if there are any bundled aggregations to include with the event.
         if bundle_aggregations:
-            event_aggregations = bundle_aggregations.get(event.event_id)
-            if event_aggregations:
+            if event.event_id in bundle_aggregations:
                 self._inject_bundled_aggregations(
                     event,
                     time_now,
                     config,
-                    event_aggregations,
+                    bundle_aggregations,
                     serialized_event,
                     apply_edits=apply_edits,
                 )
@@ -470,7 +469,7 @@ def _inject_bundled_aggregations(
         event: EventBase,
         time_now: int,
         config: SerializeEventConfig,
-        aggregations: "BundledAggregations",
+        bundled_aggregations: Dict[str, "BundledAggregations"],
         serialized_event: JsonDict,
         apply_edits: bool,
     ) -> None:
@@ -479,23 +478,33 @@ def _inject_bundled_aggregations(
         Args:
             event: The event being serialized.
             time_now: The current time in milliseconds
-            aggregations: The bundled aggregation to serialize.
-            serialized_event: The serialized event which may be modified.
             config: Event serialization config
+            bundled_aggregations: The bundled aggregation to serialize.
+            serialized_event: The serialized event which may be modified.
             apply_edits: Whether the content of the event should be modified to reflect
                any replacement in `aggregations.replace`.
         """
+
+        # We have already checked that aggregations exist for this event.
+        event_aggregations = bundled_aggregations[event.event_id]
+
+        # The JSON dictionary to be added under the unsigned property of the event
+        # being serialized.
         serialized_aggregations = {}
 
-        if aggregations.annotations:
-            serialized_aggregations[RelationTypes.ANNOTATION] = aggregations.annotations
+        if event_aggregations.annotations:
+            serialized_aggregations[
+                RelationTypes.ANNOTATION
+            ] = event_aggregations.annotations
 
-        if aggregations.references:
-            serialized_aggregations[RelationTypes.REFERENCE] = aggregations.references
+        if event_aggregations.references:
+            serialized_aggregations[
+                RelationTypes.REFERENCE
+            ] = event_aggregations.references
 
-        if aggregations.replace:
+        if event_aggregations.replace:
             # If there is an edit, optionally apply it to the event.
-            edit = aggregations.replace
+            edit = event_aggregations.replace
             if apply_edits:
                 self._apply_edit(event, serialized_event, edit)
 
@@ -506,19 +515,24 @@ def _inject_bundled_aggregations(
                 "sender": edit.sender,
             }
 
-        # If this event is the start of a thread, include a summary of the replies.
-        if aggregations.thread:
-            thread = aggregations.thread
+        # Include any threaded replies to this event.
+        #
+        # Note that is it not valid to start a thread on an event which already
+        # has a relation. Doing so would cause an infinite loop.
+        event_relation = event.content.get("m.relates_to")
+        include_thread = (
+            not isinstance(event_relation, dict)
+            or event_relation.get("rel_type") is None
+        )
+        if include_thread and event_aggregations.thread:
+            thread = event_aggregations.thread
 
-            # Don't bundle aggregations as this could recurse forever.
-            serialized_latest_event = serialize_event(
-                thread.latest_event, time_now, config=config
+            serialized_latest_event = self.serialize_event(
+                thread.latest_event,
+                time_now,
+                config=config,
+                bundle_aggregations=bundled_aggregations,
             )
-            # Manually apply an edit, if one exists.
-            if thread.latest_edit:
-                self._apply_edit(
-                    thread.latest_event, serialized_latest_event, thread.latest_edit
-                )
 
             thread_summary = {
                 "latest_event": serialized_latest_event,

synapse/handlers/relations.py

@@ -44,8 +44,6 @@
 class _ThreadAggregation:
     # The latest event in the thread.
     latest_event: EventBase
-    # The latest edit to the latest event in the thread.
-    latest_edit: Optional[EventBase]
     # The total number of events in the thread.
     count: int
     # True if the current user has sent an event to the thread.
@@ -256,10 +254,11 @@ async def get_annotations_for_event(
 
         return filtered_results
 
-    async def _get_bundled_aggregation_for_event(
+    async def _get_bundled_annotations_and_references_for_event(
         self, event: EventBase, ignored_users: FrozenSet[str]
-    ) -> Optional[BundledAggregations]:
-        """Generate bundled aggregations for an event.
+    ) -> Tuple[Optional[JsonDict], Optional[JsonDict]]:
+        """
+        Generate bundled aggregations for annotation and reference relations for an event.
 
         Note that this does not use a cache, but depends on cached methods.
 
@@ -268,8 +267,9 @@ async def _get_bundled_aggregation_for_event(
             ignored_users: The users ignored by the requesting user.
 
         Returns:
-            The bundled aggregations for an event, if bundled aggregations are
-            enabled and the event can have bundled aggregations.
+            A tuple of the bundled aggregations for annotation and reference relations.
+            Either or both entries in the tuple might be None if no relations
+            of that type exist.
         """
 
         # Do not bundle aggregations for an event which represents an edit or an
@@ -278,21 +278,17 @@ async def _get_bundled_aggregation_for_event(
         if isinstance(relates_to, (dict, frozendict)):
             relation_type = relates_to.get("rel_type")
             if relation_type in (RelationTypes.ANNOTATION, RelationTypes.REPLACE):
-                return None
+                return None, None
 
         event_id = event.event_id
         room_id = event.room_id
 
-        # The bundled aggregations to include, a mapping of relation type to a
-        # type-specific value. Some types include the direct return type here
-        # while others need more processing during serialization.
-        aggregations = BundledAggregations()
-
         annotations = await self.get_annotations_for_event(
             event_id, room_id, ignored_users=ignored_users
         )
+        serialized_annotations = None
         if annotations:
-            aggregations.annotations = {"chunk": annotations}
+            serialized_annotations = {"chunk": annotations}
 
         references, next_token = await self.get_relations_for_event(
             event_id,
@@ -301,18 +297,19 @@ async def _get_bundled_aggregation_for_event(
             RelationTypes.REFERENCE,
             ignored_users=ignored_users,
         )
+        serialized_references: Optional[JsonDict] = None
         if references:
-            aggregations.references = {
+            serialized_references = {
                 "chunk": [{"event_id": event.event_id} for event in references]
             }
 
             if next_token:
-                aggregations.references["next_batch"] = await next_token.to_string(
+                serialized_references["next_batch"] = await next_token.to_string(
                     self._main_store
                 )
 
         # Store the bundled aggregations in the event metadata for later use.
-        return aggregations
+        return serialized_annotations, serialized_references
 
     async def get_threads_for_events(
         self, event_ids: Collection[str], user_id: str, ignored_users: FrozenSet[str]
@@ -353,7 +350,7 @@ async def get_threads_for_events(
 
         for event_id, summary in summaries.items():
             if summary:
-                thread_count, latest_thread_event, edit = summary
+                thread_count, latest_thread_event = summary
 
                 # Subtract off the count of any ignored users.
                 for ignored_user in ignored_users:
@@ -398,7 +395,6 @@ async def get_threads_for_events(
 
                 results[event_id] = _ThreadAggregation(
                     latest_event=latest_thread_event,
-                    latest_edit=edit,
                     count=thread_count,
                     # If there's a thread summary it must also exist in the
                     # participated dictionary.
@@ -417,8 +413,13 @@ async def get_bundled_aggregations(
             user_id: The user requesting the bundled aggregations.
 
         Returns:
-            A map of event ID to the bundled aggregation for the event. Not all
-            events may have bundled aggregations in the results.
+            A map of event ID to the bundled aggregations for the event.
+
+            Not all requested events may exist in the results (if they don't have
+            bundled aggregations).
+
+            The results may include additional events which are related to the
+            requested events.
         """
         # De-duplicate events by ID to handle the same event requested multiple times.
         #
@@ -433,13 +434,38 @@ async def get_bundled_aggregations(
         # Fetch any ignored users of the requesting user.
         ignored_users = await self._main_store.ignored_users(user_id)
 
+        # Threads are special as the latest event of a thread might cause additional
+        # events to be fetched. Thus, we check those first!
+
+        # Fetch thread summaries (but only for the directly requested events).
+        #
+        # Note that you can't have threads off of other related events, but it is
+        # possible for a malicious homeserver to inject them anyway.
+        threads = await self.get_threads_for_events(
+            events_by_id.keys(), user_id, ignored_users
+        )
+        for event_id, thread in threads.items():
+            results.setdefault(event_id, BundledAggregations()).thread = thread
+
+            # If the latest event in a thread is not already being fetched,
+            # add it. This ensures that the bundled aggregations for the
+            # latest thread event is correct.
+            latest_thread_event = thread.latest_event
+            if latest_thread_event and latest_thread_event.event_id not in events_by_id:
+                events_by_id[latest_thread_event.event_id] = latest_thread_event
+
         # Fetch other relations per event.
         for event in events_by_id.values():
-            event_result = await self._get_bundled_aggregation_for_event(
+            (
+                annotations,
+                references,
+            ) = await self._get_bundled_annotations_and_references_for_event(
                 event, ignored_users
             )
-            if event_result:
-                results[event.event_id] = event_result
+            if annotations or references:
+                aggregations = results.setdefault(event.event_id, BundledAggregations())
+                aggregations.annotations = annotations
+                aggregations.references = references
 
         # Fetch any edits (but not for redacted events).
         #
@@ -455,10 +481,4 @@ async def get_bundled_aggregations(
         for event_id, edit in edits.items():
             results.setdefault(event_id, BundledAggregations()).replace = edit
 
-        threads = await self.get_threads_for_events(
-            events_by_id.keys(), user_id, ignored_users
-        )
-        for event_id, thread in threads.items():
-            results.setdefault(event_id, BundledAggregations()).thread = thread
-
         return results

synapse/storage/databases/main/relations.py

@@ -445,8 +445,8 @@ def get_thread_summary(self, event_id: str) -> Optional[Tuple[int, EventBase]]:
     @cachedList(cached_method_name="get_thread_summary", list_name="event_ids")
     async def get_thread_summaries(
         self, event_ids: Collection[str]
-    ) -> Dict[str, Optional[Tuple[int, EventBase, Optional[EventBase]]]]:
-        """Get the number of threaded replies, the latest reply (if any), and the latest edit for that reply for the given event.
+    ) -> Dict[str, Optional[Tuple[int, EventBase]]]:
+        """Get the number of threaded replies and the latest reply (if any) for the given events.
 
         Args:
             event_ids: Summarize the thread related to this event ID.
@@ -458,7 +458,6 @@ async def get_thread_summaries(
             Each summary is a tuple of:
                 The number of events in the thread.
                 The most recent event in the thread.
-                The most recent edit to the most recent event in the thread, if applicable.
         """
 
         def _get_thread_summaries_txn(
@@ -544,9 +543,6 @@ def _get_thread_summaries_txn(
 
         latest_events = await self.get_events(latest_event_ids.values())  # type: ignore[attr-defined]
 
-        # Check to see if any of those events are edited.
-        latest_edits = await self.get_applicable_edits(latest_event_ids.values())
-
         # Map to the event IDs to the thread summary.
         #
         # There might not be a summary due to there not being a thread or
@@ -557,8 +553,7 @@ def _get_thread_summaries_txn(
 
             summary = None
             if latest_event:
-                latest_edit = latest_edits.get(latest_event_id)
-                summary = (counts[parent_event_id], latest_event, latest_edit)
+                summary = (counts[parent_event_id], latest_event)
             summaries[parent_event_id] = summary
 
         return summaries