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

scripts-dev/complement.sh

@@ -102,7 +102,7 @@ if [ -z "$skip_docker_build" ]; then
     # Build the unified Complement image (from the worker Synapse image we just built).
     echo_if_github "::group::Build Docker image: complement/Dockerfile"
     docker build -t complement-synapse \
-           -f "docker/complement/Dockerfile" "docker/complement"
+          -f "docker/complement/Dockerfile" "docker/complement"
     echo_if_github "::endgroup::"
 fi
 
@@ -161,4 +161,4 @@ fi
 echo "Images built; running complement"
 cd "$COMPLEMENT_DIR"
 
-go test -v -tags $test_tags -count=1 "${extra_test_args[@]}" "$@" ./tests/...
+go test -v -tags $test_tags -count=1 "${extra_test_args[@]}" "$@" ./tests/

synapse/federation/federation_client.py

@@ -308,8 +308,8 @@ async def get_pdu_from_destination_raw(
             destination, event_id, timeout=timeout
         )
 
-        logger.debug(
-            "retrieved event id %s from %s: %r",
+        logger.info(
+            "get_pdu_raw retrieved event id %s from %s: %r",
             event_id,
             destination,
             transaction_data,
@@ -319,6 +319,12 @@ async def get_pdu_from_destination_raw(
             event_from_pdu_json(p, room_version) for p in transaction_data["pdus"]
         ]
 
+        logger.info(
+            "get_pdu_raw pdu_list[0]=%s outlier=%s",
+            pdu_list[0],
+            pdu_list[0].internal_metadata.outlier,
+        )
+
         if pdu_list and pdu_list[0]:
             pdu = pdu_list[0]
 
@@ -361,6 +367,12 @@ 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)
+        logger.info(
+            "get_pdu from cache event_id=%s ev=%s outlier=%s",
+            event_id,
+            ev,
+            ev and ev.internal_metadata.outlier,
+        )
         if ev:
             return ev
 

synapse/handlers/federation_event.py

@@ -765,6 +765,11 @@ 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.
+        #
         # these should not be outliers.
         assert (
             not event.internal_metadata.is_outlier()
@@ -1315,6 +1320,42 @@ async def _handle_marker_event(self, origin: str, marker_event: EventBase) -> No
             marker_event,
         )
 
+    async def backfill_event(
+        self, destination: str, room_id: str, event_id: str
+    ) -> None:
+        room_version = await self._store.get_room_version(room_id)
+
+        logger.info("backfill_event event_id=%s", event_id)
+
+        eventAsdf = await self._federation_client.get_pdu(
+            [destination],
+            event_id,
+            room_version,
+        )
+        event = await self._federation_client.get_pdu_from_destination_raw(
+            destination,
+            event_id,
+            room_version,
+        )
+        logger.info(
+            "backfill_event get_pdu=%s get_pdu_from_destination_raw=%s",
+            eventAsdf,
+            event,
+        )
+        # # FIXME: Too sketchy?
+        # event.internal_metadata.outlier = False
+
+        logger.info(
+            "backfill_event event=%s outlier=%s", event, event.internal_metadata.outlier
+        )
+
+        await self._process_pulled_events(
+            destination,
+            [event],
+            # Prevent notifications going to clients
+            backfilled=True,
+        )
+
     async def _get_events_and_persist(
         self, destination: str, room_id: str, event_ids: Collection[str]
     ) -> None:

synapse/handlers/room.py

@@ -1333,6 +1333,7 @@ def __init__(self, hs: "HomeServer"):
         self.store = hs.get_datastores().main
         self.state_handler = hs.get_state_handler()
         self.federation_client = hs.get_federation_client()
+        self.federation_event_handler = hs.get_federation_event_handler()
         self._storage_controllers = hs.get_storage_controllers()
 
     async def get_event_for_timestamp(
@@ -1365,7 +1366,7 @@ async def get_event_for_timestamp(
         local_event_id = await self.store.get_event_id_for_timestamp(
             room_id, timestamp, direction
         )
-        logger.debug(
+        logger.info(
             "get_event_for_timestamp: locally, we found event_id=%s closest to timestamp=%s",
             local_event_id,
             timestamp,
@@ -1375,6 +1376,7 @@ async def get_event_for_timestamp(
         # the timestamp given and the event we were able to find locally
         is_event_next_to_backward_gap = False
         is_event_next_to_forward_gap = False
+        local_event = None
         if local_event_id:
             local_event = await self.store.get_event(
                 local_event_id, allow_none=False, allow_rejected=False
@@ -1400,7 +1402,7 @@ async def get_event_for_timestamp(
             or is_event_next_to_backward_gap
             or is_event_next_to_forward_gap
         ):
-            logger.debug(
+            logger.info(
                 "get_event_for_timestamp: locally, we found event_id=%s closest to timestamp=%s which is next to a gap in event history so we're asking other homeservers first",
                 local_event_id,
                 timestamp,
@@ -1421,31 +1423,48 @@ async def get_event_for_timestamp(
                     remote_response = await self.federation_client.timestamp_to_event(
                         domain, room_id, timestamp, direction
                     )
-                    logger.debug(
+                    logger.info(
                         "get_event_for_timestamp: response from domain(%s)=%s",
                         domain,
                         remote_response,
                     )
 
+                    remote_event_id = remote_response.event_id
+                    origin_server_ts = remote_response.origin_server_ts
+
                     # TODO: Do we want to persist this as an extremity?
                     # TODO: I think ideally, we would try to backfill from
                     # this event and run this whole
                     # `get_event_for_timestamp` function again to make sure
                     # they didn't give us an event from their gappy history.
-                    remote_event_id = remote_response.event_id
-                    origin_server_ts = remote_response.origin_server_ts
+                    # await self.federation_event_handler._get_events_and_persist(
+                    #     domain,
+                    #     room_id,
+                    #     [remote_event_id],
+                    # )
+                    await self.federation_event_handler.backfill_event(
+                        domain, room_id, remote_event_id
+                    )
+                    logger.info(
+                        "get_event_for_timestamp: asdf %s",
+                        remote_event_id,
+                    )
 
                     # Only return the remote event if it's closer than the local event
                     if not local_event or (
                         abs(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",
+                            remote_event_id,
+                        )
                         return remote_event_id, 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
-                    logger.debug(
-                        "Failed to fetch /timestamp_to_event from %s because of exception(%s) %s args=%s",
+                    logger.info(
+                        "get_event_for_timestamp: Failed to fetch /timestamp_to_event from %s because of exception(%s) %s args=%s",
                         domain,
                         type(ex).__name__,
                         ex,
@@ -1454,14 +1473,15 @@ async def get_event_for_timestamp(
                 except Exception as ex:
                     # But we do want to see some exceptions in our code
                     logger.warning(
-                        "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(%s) %s args=%s",
                         domain,
                         type(ex).__name__,
                         ex,
                         ex.args,
+                        exc_info=(type(ex), ex, ex.__traceback__),
                     )
 
-        if not local_event_id:
+        if not local_event_id or not local_event:
             raise SynapseError(
                 404,
                 "Unable to find event from %s in direction %s" % (timestamp, direction),