Only use a prefetched summary for the queue entry it was started for

Regression test for a room linked from two spaces: the room is queued
once per edge (the queue is only deduplicated when popped), each entry
carrying the via and depth to summarise that edge with, and an entry
queued later is popped first. Keying the prefetches by room id alone
meant the popped entry could consume a summary computed for a different
edge - summarising with the wrong via (so a room reachable only via the
other parent is dropped from the hierarchy) or the wrong depth (so a
room at max_depth is returned with children_state).

Iterating the prefetch window in pop order, as the previous commit did,
is not sufficient on its own: the entry for the shallower edge can be
prefetched on an earlier iteration, before the deeper edge is queued at
all. Track the entry each prefetch was started for and recompute when it
is not the one popped.

The test fails without the entry check (the room is summarised via the
root's dead via and disappears from the hierarchy).

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":

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.