diff --git a/synapse/handlers/room_summary.py b/synapse/handlers/room_summary.py index 40c19ee699..47d1730941 100644 --- a/synapse/handlers/room_summary.py +++ b/synapse/handlers/room_summary.py @@ -312,7 +312,14 @@ class RoomSummaryHandler: # traversal order. Values are ("ok", result) / ("err", exception) so that a # prefetch which is never consumed (page limit hit first) cannot produce an # unhandled error. - prefetched: dict[str, "Deferred"] = {} + # + # The queue entry a summary was computed for is kept alongside it: the + # same room can be queued more than once (linked from two spaces; the + # queue is only deduplicated when popped), each entry carrying the + # via/depth to summarise that edge with, and an entry queued later can + # be popped first. A prefetch is therefore only usable by the very entry + # it was started for. + prefetched: dict[str, tuple[_RoomQueueEntry, "Deferred"]] = {} async def prefetch( entry: _RoomQueueEntry, @@ -341,16 +348,18 @@ class RoomSummaryHandler: # consume is wasted (and would be re-issued for the next page). window = min(PREFETCH_SUMMARIES, limit - len(rooms_result)) # NB reversed(): the queue is a stack, so the LAST entries are - # processed first. The same room can appear in the queue more than - # once (a room linked from two spaces; the queue is only deduped at - # pop time), and the entry carries the via/depth used to summarise - # it — so the entry registered here must be the one popped first. + # processed first, and prefetching them in that order means the + # entry that is popped next is the one whose summary is furthest + # along. for upcoming in reversed(room_queue[-window:]): if ( upcoming.room_id not in processed_rooms and upcoming.room_id not in prefetched ): - prefetched[upcoming.room_id] = run_in_background(prefetch, upcoming) + prefetched[upcoming.room_id] = ( + upcoming, + run_in_background(prefetch, upcoming), + ) queue_entry = room_queue.pop() room_id = queue_entry.room_id @@ -361,8 +370,12 @@ class RoomSummaryHandler: logger.debug("Processing room %s", room_id) - deferred = prefetched.pop(room_id, None) - if deferred is None: + # Only use a prefetch that was started for THIS entry: another + # entry for the same room carries a different via/depth, and would + # summarise a different edge. Any other prefetch is discarded (its + # errors are already captured in its result tuple). + prefetch_entry, deferred = prefetched.pop(room_id, (None, None)) + if prefetch_entry is not queue_entry or deferred is None: deferred = run_in_background(prefetch, queue_entry) status, value = await make_deferred_yieldable(deferred) if status == "err": diff --git a/tests/handlers/test_room_summary.py b/tests/handlers/test_room_summary.py index 1bb25f3b23..80ce57d026 100644 --- a/tests/handlers/test_room_summary.py +++ b/tests/handlers/test_room_summary.py @@ -33,7 +33,12 @@ from synapse.api.constants import ( RestrictedJoinRuleTypes, RoomTypes, ) -from synapse.api.errors import AuthError, NotFoundError, SynapseError +from synapse.api.errors import ( + AuthError, + HttpResponseException, + NotFoundError, + SynapseError, +) from synapse.api.room_versions import RoomVersions from synapse.federation.transport.client import TransportLayerClient from synapse.handlers.room_summary import _child_events_comparison_key, _RoomEntry @@ -1059,6 +1064,78 @@ class SpaceSummaryTestCase(unittest.HomeserverTestCase): ) self._assert_hierarchy(result, expected) + def test_fed_room_linked_from_two_spaces(self) -> None: + """ + A room linked from two spaces is summarised using the via of the edge + the traversal actually reaches it by. + + Summaries for upcoming rooms are fetched concurrently, ahead of the + strictly-ordered traversal; a room can be in the queue more than once + (it is only deduplicated when popped), and each queue entry carries the + via to summarise it with. The summary that gets used must therefore be + the one belonging to the entry that is popped, not any other entry for + the same room. + """ + root_via = "root." + self.hs.hostname + subspace_via = "subspace." + self.hs.hostname + fed_room = "#linked-twice:" + self.hs.hostname + "2" + + # A local subspace, ordered ahead of the federated room so it is + # traversed first — which pushes its own edge to the federated room onto + # the queue while the root's edge to it is still there. + subspace = self.helper.create_room_as( + self.user, + tok=self.token, + extra_content={ + "creation_content": {EventContentFields.ROOM_TYPE: RoomTypes.SPACE} + }, + ) + self._add_child(self.space, subspace, self.token, order="a") + self._add_child(self.space, fed_room, self.token, order="b", via=[root_via]) + self._add_child(subspace, fed_room, self.token, via=[subspace_via]) + + destinations: list[str] = [] + + async def get_room_hierarchy( + _self: TransportLayerClient, + destination: str, + room_id: str, + suggested_only: bool, + ) -> JsonDict: + destinations.append(destination) + # Only the subspace's via can actually serve the room, so summarising + # the wrong edge loses the room entirely. + if destination == root_via: + raise HttpResponseException(502, "Bad Gateway", b"{}") + return { + "room": {"room_id": fed_room, "world_readable": True}, + "children": [], + "inaccessible_children": [], + } + + with mock.patch( + "synapse.federation.transport.client.TransportLayerClient.get_room_hierarchy", + new=get_room_hierarchy, + ): + result = self.get_success( + self.handler.get_room_hierarchy(create_requester(self.user), self.space) + ) + + # The subspace's edge is the one the traversal pops, so that is the via + # the room must have been summarised with… + self.assertIn(subspace_via, destinations) + + # …and the room is returned (exactly once), listed under both parents. + self._assert_hierarchy( + result, + [ + (self.space, [subspace, fed_room, self.room]), + (subspace, [fed_room]), + (fed_room, ()), + (self.room, ()), + ], + ) + def test_fed_caching(self) -> None: """ Federation `/hierarchy` responses should be cached.