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/federation/federation_server.py

@@ -196,8 +196,12 @@ async def on_backfill_request(
                 origin, room_id, versions, limit
             )
 
+            logger.info("on_backfill_request pdus(%d)=%s", len(pdus), pdus)
+
             res = self._transaction_dict_from_pdus(pdus)
 
+            logger.info("on_backfill_request res=%s", res)
+
         return 200, res
 
     async def on_incoming_transaction(

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,12 +249,13 @@ async def _maybe_backfill_inner(
             t for t in sorted_extremeties_tuple if int(t[1]) <= current_depth
         ]
 
-        logger.debug(
-            "room_id: %s, backfill: current_depth: %s, limit: %s, max_depth: %s, extrems: %s filtered_sorted_extremeties_tuple: %s",
+        logger.info(
+            "room_id: %s, backfill: current_depth: %s, limit: %s, max_depth: %s, extrems (%d): %s filtered_sorted_extremeties_tuple: %s",
             room_id,
             current_depth,
             limit,
             max_depth,
+            len(sorted_extremeties_tuple),
             sorted_extremeties_tuple,
             filtered_sorted_extremeties_tuple,
         )
@@ -270,6 +271,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
@@ -1058,6 +1061,7 @@ async def on_backfill_request(
         events = await self.store.get_backfill_events(room_id, pdu_list, limit)
 
         events = await filter_events_for_server(self.storage, origin, events)
+        logger.info("on_backfill_request resultant events(%d)=%s", len(events), events)
 
         return events
 

synapse/handlers/federation_event.py

@@ -416,6 +416,19 @@ 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(%d)=%s",
+            len(events),
+            [
+                {
+                    "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 +442,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 +1274,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/message.py

@@ -955,8 +955,10 @@ async def create_new_client_event(
         # event and then try to auth it (which fails with a somewhat confusing "No
         # create event in auth events")
         assert (
-            builder.type == EventTypes.Create or len(prev_event_ids) > 0
-        ), "Attempting to create an event with no prev_events"
+            builder.type == EventTypes.Create
+            or len(prev_event_ids) > 0
+            or len(auth_event_ids) > 0
+        ), "Attempting to create an event with no prev_events or auth_event_ids"
 
         event = await builder.build(
             prev_event_ids=prev_event_ids,

synapse/handlers/room_batch.py

@@ -184,7 +184,7 @@ async def persist_state_events_at_start(
 
         # Make the state events float off on their own so we don't have a
         # bunch of `@mxid joined the room` noise between each batch
-        prev_event_id_for_state_chain = generate_fake_event_id()
+        prev_event_ids_for_state_chain = []  # generate_fake_event_id()
 
         for state_event in state_events_at_start:
             assert_params_in_dict(
@@ -221,12 +221,21 @@ async def persist_state_events_at_start(
                     action=membership,
                     content=event_dict["content"],
                     outlier=True,
-                    prev_event_ids=[prev_event_id_for_state_chain],
+                    prev_event_ids=prev_event_ids_for_state_chain,
                     # Make sure to use a copy of this list because we modify it
                     # later in the loop here. Otherwise it will be the same
                     # reference and also update in the event when we append later.
                     auth_event_ids=auth_event_ids.copy(),
                 )
+
+                mem_event = await self.store.get_event(event_id)
+                logger.info(
+                    "room_batch mem_event_id=%s depth=%s stream_ordering=%s prev_event_ids=%s",
+                    mem_event.event_id,
+                    mem_event.depth,
+                    mem_event.internal_metadata.stream_ordering,
+                    mem_event.prev_event_ids(),
+                )
             else:
                 # TODO: Add some complement tests that adds state that is not member joins
                 # and will use this code path. Maybe we only want to support join state events
@@ -240,7 +249,7 @@ async def persist_state_events_at_start(
                     ),
                     event_dict,
                     outlier=True,
-                    prev_event_ids=[prev_event_id_for_state_chain],
+                    prev_event_ids=prev_event_ids_for_state_chain,
                     # Make sure to use a copy of this list because we modify it
                     # later in the loop here. Otherwise it will be the same
                     # reference and also update in the event when we append later.
@@ -251,7 +260,7 @@ async def persist_state_events_at_start(
             state_event_ids_at_start.append(event_id)
             auth_event_ids.append(event_id)
             # Connect all the state in a floating chain
-            prev_event_id_for_state_chain = event_id
+            prev_event_ids_for_state_chain = [event_id]
 
         return state_event_ids_at_start
 
@@ -353,13 +362,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/handlers/room_member.py

@@ -644,7 +644,7 @@ async def update_membership_locked(
             if block_invite:
                 raise SynapseError(403, "Invites have been disabled on this server")
 
-        if prev_event_ids:
+        if prev_event_ids is not None:
             return await self._local_membership_update(
                 requester=requester,
                 target=target,

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