Fix historical messages backfilling in random order on remote homeservers (MSC2716)

changelog.d/11114.bugfix

@@ -0,0 +1 @@
+Fix [MSC2716](https://github.com/matrix-org/matrix-doc/pull/2716) historical messages backfilling in random order on remote homeservers.

scripts-dev/complement.sh

@@ -65,4 +65,4 @@ if [[ -n "$1" ]]; then
 fi
 
 # Run the tests!
-go test -v -tags synapse_blacklist,msc2946,msc3083,msc2403,msc2716 -count=1 $EXTRA_COMPLEMENT_ARGS ./tests/...
+go test -v -tags synapse_blacklist,msc2946,msc3083,msc2403,msc2716 -count=1 $EXTRA_COMPLEMENT_ARGS ./tests/main_test.go ./tests/msc2716_test.go

synapse/events/utils.py

@@ -283,13 +283,13 @@ def format_event_for_client_v1(d: JsonDict) -> JsonDict:
 
 def format_event_for_client_v2(d: JsonDict) -> JsonDict:
     drop_keys = (
-        "auth_events",
-        "prev_events",
-        "hashes",
-        "signatures",
-        "depth",
-        "origin",
-        "prev_state",
+        # "auth_events",
+        # "prev_events",
+        # "hashes",
+        # "signatures",
+        # "depth",
+        # "origin",
+        # "prev_state",
     )
     for key in drop_keys:
         d.pop(key, None)
@@ -340,6 +340,9 @@ def serialize_event(
 
     d["event_id"] = e.event_id
 
+    # TODO: Remove
+    d["stream_ordering"] = e.internal_metadata.stream_ordering
+
     if "age_ts" in d["unsigned"]:
         d["unsigned"]["age"] = time_now_ms - d["unsigned"]["age_ts"]
         del d["unsigned"]["age_ts"]

synapse/handlers/federation.py

@@ -148,14 +148,14 @@ async def _maybe_backfill_inner(
         insertion_events_to_be_backfilled = (
             await self.store.get_insertion_event_backwards_extremities_in_room(room_id)
         )
-        logger.debug(
+        logger.info(
             "_maybe_backfill_inner: extremities oldest_events_with_depth=%s insertion_events_to_be_backfilled=%s",
             oldest_events_with_depth,
             insertion_events_to_be_backfilled,
         )
 
         if not oldest_events_with_depth and not insertion_events_to_be_backfilled:
-            logger.debug("Not backfilling as no extremeties found.")
+            logger.info("Not backfilling as no extremeties found.")
             return False
 
         # We only want to paginate if we can actually see the events we'll get,
@@ -203,7 +203,7 @@ async def _maybe_backfill_inner(
             redact=False,
             check_history_visibility_only=True,
         )
-        logger.debug(
+        logger.info(
             "_maybe_backfill_inner: filtered_extremities %s", filtered_extremities
         )
 
@@ -230,7 +230,7 @@ async def _maybe_backfill_inner(
         # much larger factor will result in triggering a backfill request much
         # earlier than necessary.
         if current_depth - 2 * limit > max_depth:
-            logger.debug(
+            logger.info(
                 "Not backfilling as we don't need to. %d < %d - 2 * %d",
                 max_depth,
                 current_depth,
@@ -249,7 +249,7 @@ async def _maybe_backfill_inner(
             t for t in sorted_extremeties_tuple if int(t[1]) <= current_depth
         ]
 
-        logger.debug(
+        logger.info(
             "room_id: %s, backfill: current_depth: %s, limit: %s, max_depth: %s, extrems: %s filtered_sorted_extremeties_tuple: %s",
             room_id,
             current_depth,
@@ -270,6 +270,8 @@ async def _maybe_backfill_inner(
         # request URI to be too long.
         extremities = dict(sorted_extremeties_tuple[:5])
 
+        logger.info("backfill extremities=%s", extremities)
+
         # Now we need to decide which hosts to hit first.
 
         # First we try hosts that are already in the room

synapse/handlers/federation_event.py

@@ -416,6 +416,18 @@ async def backfill(
         events = await self._federation_client.backfill(
             dest, room_id, limit=limit, extremities=extremities
         )
+        logger.info(
+            "from remote server: got backfill response events=%s",
+            [
+                {
+                    "event_id": ev.event_id,
+                    "type": ev["type"],
+                    "depth": ev["depth"],
+                    "content": ev["content"].get("body", None),
+                }
+                for ev in events
+            ],
+        )
 
         if not events:
             return
@@ -429,7 +441,39 @@ async def backfill(
                     f"room {ev.room_id}, when we were backfilling in {room_id}"
                 )
 
-        await self._process_pulled_events(dest, events, backfilled=True)
+        await self._process_pulled_events(
+            dest,
+            # The /backfill response should start from `?v` and include the
+            # events that preceded it (so the list will be newest -> oldest). We
+            # reverse that order so the messages are oldest -> newest and we can
+            # persist the backfilled events without constantly have to go fetch
+            # missing prev_events which are probably included in the same
+            # backfill chunk.
+            # TODO: If we try to reverse this list, the stream_ordering will be backwards
+            # reversed(events),
+            events,
+            backfilled=True,
+        )
+
+        for ev in events:
+            event_after_persisted = await self._store.get_event(
+                ev.event_id, allow_none=True
+            )
+
+            if event_after_persisted:
+                logger.info(
+                    "from remote server: processed backfilled event_id=%s type=%s depth=%s stream_ordering=%s content=%s",
+                    ev.event_id,
+                    event_after_persisted["type"],
+                    event_after_persisted["depth"],
+                    event_after_persisted.internal_metadata.stream_ordering,
+                    event_after_persisted["content"].get("body", None),
+                )
+            else:
+                logger.info(
+                    "from remote server: processed backfilled event_id=%s failed to lookup",
+                    ev.event_id,
+                )
 
     async def _get_missing_events_for_pdu(
         self, origin: str, pdu: EventBase, prevs: Set[str], min_depth: int
@@ -1229,7 +1273,12 @@ def prep(event: EventBase) -> Optional[Tuple[EventBase, EventContext]]:
             return event, context
 
         events_to_persist = (x for x in (prep(event) for event in fetched_events) if x)
-        await self.persist_events_and_notify(room_id, tuple(events_to_persist))
+        await self.persist_events_and_notify(
+            room_id,
+            tuple(events_to_persist),
+            # TODO: Maybe this to get fetched missing events during backfill as backfilled also :/
+            backfilled=True,
+        )
 
     async def _check_event_auth(
         self,

synapse/handlers/room_batch.py

@@ -353,13 +353,20 @@ async def persist_historical_events(
         # Events are sorted by (topological_ordering, stream_ordering)
         # where topological_ordering is just depth.
         for (event, context) in reversed(events_to_persist):
-            await self.event_creation_handler.handle_new_client_event(
+            result_event = await self.event_creation_handler.handle_new_client_event(
                 await self.create_requester_for_user_id_from_app_service(
                     event["sender"], app_service_requester.app_service
                 ),
                 event=event,
                 context=context,
             )
+            logger.info(
+                "result_event depth=%s stream_ordering=%s event_id=%s body=%s",
+                result_event.depth,
+                result_event.internal_metadata.stream_ordering,
+                result_event.event_id,
+                result_event.content.get("body", None),
+            )
 
         return event_ids
 

synapse/rest/client/room.py

@@ -561,6 +561,7 @@ async def on_GET(
         pagination_config = await PaginationConfig.from_request(
             self.store, request, default_limit=10
         )
+        logger.info("/messages rest start pagination_config=%s", pagination_config)
         # Twisted will have processed the args by now.
         assert request.args is not None
         as_client_event = b"raw" not in request.args