Backfill remote event fetched by MSC3030 so we can paginate from it later

synapse/federation/federation_client.py

@@ -53,7 +53,7 @@
     RoomVersion,
     RoomVersions,
 )
-from synapse.events import EventBase, builder
+from synapse.events import EventBase, builder, make_event_from_dict
 from synapse.federation.federation_base import (
     FederationBase,
     InvalidEventSignatureError,
@@ -309,7 +309,7 @@ async def get_pdu_from_destination_raw(
         )
 
         logger.debug(
-            "get_pdu_raw: retrieved event id %s from %s: %r",
+            "get_pdu_from_destination_raw: retrieved event id %s from %s: %r",
             event_id,
             destination,
             transaction_data,
@@ -360,9 +360,25 @@ async def get_pdu(
 
         # TODO: Rate limit the number of times we try and get the same event.
 
-        ev = self._get_pdu_cache.get(event_id)
-        if ev:
-            return ev
+        event_from_cache = self._get_pdu_cache.get(event_id)
+        if event_from_cache:
+            assert not event_from_cache.internal_metadata.outlier, (
+                "Event from cache unexpectedly an `outlier` when it should be pristine and untouched without metadata set. "
+                "We are probably not be returning a copy of the event because downstream callers are modifying the event reference we have in the cache."
+            )
+
+            # Make sure to return a copy because downstream callers will use
+            # this event reference directly and change our original, pristine,
+            # untouched PDU. For example when people mark the event as an
+            # `outlier` (`event.internal_metadata.outlier = true`), we don't
+            # want that to propagate back into the cache.
+            event_copy = make_event_from_dict(
+                event_from_cache.get_pdu_json(),
+                event_from_cache.room_version,
+                internal_metadata_dict=None,
+            )
+
+            return event_copy
 
         pdu_attempts = self.pdu_destination_tried.setdefault(event_id, {})
 
@@ -405,7 +421,22 @@ async def get_pdu(
         if signed_pdu:
             self._get_pdu_cache[event_id] = signed_pdu
 
-        return signed_pdu
+            # Make sure to return a copy because downstream callers will use this
+            # event reference directly and change our original, pristine, untouched
+            # PDU. For example when people mark the event as an `outlier`
+            # (`event.internal_metadata.outlier = true`), we don't want that to
+            # propagate back into the cache.
+            #
+            # We could get away with only making a new copy of the event when
+            # pulling from cache but it's probably better to have good hygiene and
+            # not dirty the cache in the first place as well.
+            event_copy = make_event_from_dict(
+                signed_pdu.get_pdu_json(),
+                signed_pdu.room_version,
+                internal_metadata_dict=None,
+            )
+
+        return event_copy
 
     async def get_room_state_ids(
         self, destination: str, room_id: str, event_id: str

synapse/handlers/federation_event.py

@@ -55,7 +55,7 @@
     check_state_independent_auth_rules,
     validate_event_for_room_version,
 )
-from synapse.events import EventBase, make_event_from_dict
+from synapse.events import EventBase
 from synapse.events.snapshot import EventContext
 from synapse.federation.federation_client import InvalidResponseError
 from synapse.logging.context import nested_logging_context
@@ -766,15 +766,22 @@ async def _process_pulled_event(
         """
         logger.info("Processing pulled event %s", event)
 
-        # TODO: Why does this matter? The whole point of this function is to
-        # persist random PDU's from backfill. It shouldn't matter whether we saw
-        # them somewhere else first as an outlier, then during backfill. This
-        # function handles de-outliering anyway.
+        # This function should only be used to backfill events. If you're trying
+        # to persist an outlier, use another method. If you happen to run into a
+        # situation where the event you're trying to backfill is marked as an
+        # `outlier`, then you should update that spot to return an `EventBase`
+        # without the `outlier` flag set.
         #
-        # these should not be outliers.
-        assert (
-            not event.internal_metadata.is_outlier()
-        ), "pulled event unexpectedly flagged as outlier"
+        # `EventBase` is used to represent both an event we have not yet
+        # persisted, and one that we have persisted and now keep in the cache.
+        # In an ideal world this method would only be called with the first type
+        # of event, but it turns out that's not actually the case and for
+        # example, you could get an event from cache that is marked as an
+        # `outlier` (fix up that spot though).
+        assert not event.internal_metadata.is_outlier(), (
+            "This is a safe-guard to make sure you're not trying to persist an outlier using this function (use something else). "
+            "If you're trying to backfill an event, this is the right method but you need pass in an event copy that doesn't have `event.internal_metada.outlier = true`."
+        )
 
         event_id = event.event_id
 
@@ -924,12 +931,15 @@ async def _get_state_ids_after_missing_prev_event(
         event_id: str,
     ) -> StateMap[str]:
         """Requests all of the room state at a given event from a remote homeserver.
+
         Args:
             destination: The remote homeserver to query for the state.
             room_id: The id of the room we're interested in.
             event_id: The id of the event we want the state at.
+
         Returns:
             The event ids of the state *after* the given event.
+
         Raises:
             InvalidResponseError: if the remote homeserver's response contains fields
                 of the wrong type.
@@ -1314,19 +1324,22 @@ async def _handle_marker_event(self, origin: str, marker_event: EventBase) -> No
             marker_event,
         )
 
-    async def backfill_event(
+    async def backfill_event_id(
         self, destination: str, room_id: str, event_id: str
-    ) -> None:
+    ) -> EventBase:
         """Backfill a single event and persist it as a non-outlier which means
         we also pull in all of the state and auth events necessary for it.
 
         Args:
             destination: The homeserver to pull the given event_id from.
             room_id: The room where the event is from.
             event_id: The event ID to backfill.
+
+        Raises:
+            FederationError if we are unable to find the event from the destination
         """
         logger.info(
-            "backfill_event event_id=%s from destination=%s", event_id, destination
+            "backfill_event_id: event_id=%s from destination=%s", event_id, destination
         )
 
         room_version = await self._store.get_room_version(room_id)
@@ -1346,29 +1359,18 @@ async def backfill_event(
                 affected=event_id,
             )
 
-        # We want to make a non-outlier event so it plays well with
-        # `_process_pulled_events()` -> `_update_outliers_txn()` to create a
-        # `state_group` and mimics what would happen in a regular backfill.
-        # `get_pdu()` can potentially return an `outlier` depending on the cache
-        # which we don't want.
-        event_non_outlier = make_event_from_dict(
-            event_from_response.get_pdu_json(),
-            event_from_response.room_version,
-            internal_metadata_dict=None,
-        )
-        assert not event_non_outlier.internal_metadata.outlier
-
         # Persist the event we just fetched, including pulling all of the state
-        # and auth events to de-outlier it. This function is weird and
-        # de-outliers but only works on non-outlier events. This also sets up
-        # the necessary `state_groups` for the event.
+        # and auth events to de-outlier it. This also sets up the necessary
+        # `state_groups` for the event.
         await self._process_pulled_events(
             destination,
-            [event_non_outlier],
+            [event_from_response],
             # Prevent notifications going to clients
             backfilled=True,
         )
 
+        return event_from_response
+
     async def _get_events_and_persist(
         self, destination: str, room_id: str, event_ids: Collection[str]
     ) -> None:

synapse/handlers/room.py

@@ -1481,7 +1481,7 @@ async def get_event_for_timestamp(
                     )
 
                     remote_event_id = remote_response.event_id
-                    origin_server_ts = remote_response.origin_server_ts
+                    remote_origin_server_ts = remote_response.origin_server_ts
 
                     # Backfill this event so we can get a pagination token for
                     # it with `/context` and paginate `/messages` from this
@@ -1492,20 +1492,37 @@ async def get_event_for_timestamp(
                     # sure they didn't give us an event from their gappy
                     # history. Also need a heuristic for when to stop recursing
                     # if they keep giving us gappy results.
-                    await self.federation_event_handler.backfill_event(
-                        domain, room_id, remote_event_id
+                    remote_event = (
+                        await self.federation_event_handler.backfill_event_id(
+                            domain, room_id, remote_event_id
+                        )
                     )
 
+                    # XXX: When we see that the remote server is not trustworthy,
+                    # maybe we should not ask them first in the future.
+                    if remote_origin_server_ts != remote_event.origin_server_ts:
+                        logger.info(
+                            "get_event_for_timestamp: Remote server (%s) claimed that remote_event_id=%s occured at remote_origin_server_ts=%s but that isn't true (actually occured at %s). Their claims are dubious and we should consider not trusting them.",
+                            domain,
+                            remote_event_id,
+                            remote_origin_server_ts,
+                            remote_event.origin_server_ts,
+                        )
+
                     # Only return the remote event if it's closer than the local event
                     if not local_event or (
-                        abs(origin_server_ts - timestamp)
+                        abs(remote_event.origin_server_ts - timestamp)
                         < abs(local_event.origin_server_ts - timestamp)
                     ):
                         logger.info(
-                            "get_event_for_timestamp: returning remote_event_id=%s since it's closer",
+                            "get_event_for_timestamp: returning remote_event_id=%s (%s) since it's closer to timestamp=%s than local_event=%s (%s)",
                             remote_event_id,
+                            remote_event.origin_server_ts,
+                            timestamp,
+                            local_event.event_id if local_event else None,
+                            local_event.origin_server_ts if local_event else None,
                         )
-                        return remote_event_id, origin_server_ts
+                        return remote_event_id, remote_origin_server_ts
                 except (HttpResponseException, InvalidResponseError) as ex:
                     # Let's not put a high priority on some other homeserver
                     # failing to respond or giving a random response
@@ -1516,15 +1533,12 @@ async def get_event_for_timestamp(
                         ex,
                         ex.args,
                     )
-                except Exception as ex:
+                except Exception:
                     # But we do want to see some exceptions in our code
                     logger.warning(
-                        "get_event_for_timestamp: Failed to fetch /timestamp_to_event from %s because of exception(%s) %s args=%s",
+                        "get_event_for_timestamp: Failed to fetch /timestamp_to_event from %s because of exception",
                         domain,
-                        type(ex).__name__,
-                        ex,
-                        ex.args,
-                        exc_info=(type(ex), ex, ex.__traceback__),
+                        exc_info=True,
                     )
 
         # To appease mypy, we have to add both of these conditions to check for