Optimize backfill receiving to have less missing prev_event thrashing (scratch)

changelog.d/13863.bugfix

@@ -0,0 +1 @@
+Fix `have_seen_event` cache not being invalidated after we persist an event which causes inefficiency effects like extra `/state` federation calls.

docker/complement/conf/workers-shared-extra.yaml.j2

@@ -31,7 +31,7 @@ federation_ip_range_blacklist: []
 # Disable server rate-limiting
 rc_federation:
   window_size: 1000
-  sleep_limit: 10
+  sleep_limit: 99999
   sleep_delay: 500
   reject_limit: 99999
   concurrent: 3

synapse/handlers/federation.py

@@ -38,7 +38,7 @@
 from unpaddedbase64 import decode_base64
 
 from synapse import event_auth
-from synapse.api.constants import EventContentFields, EventTypes, Membership
+from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
 from synapse.api.errors import (
     AuthError,
     CodeMessageException,
@@ -60,13 +60,7 @@
 from synapse.federation.federation_client import InvalidResponseError
 from synapse.http.servlet import assert_params_in_dict
 from synapse.logging.context import nested_logging_context
-from synapse.logging.tracing import (
-    SynapseTags,
-    set_attribute,
-    start_active_span,
-    tag_args,
-    trace,
-)
+from synapse.logging.tracing import SynapseTags, set_attribute, tag_args, trace
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.module_api import NOT_SPAM
 from synapse.replication.http.federation import (
@@ -125,6 +119,7 @@ class _BackfillPoint:
 
     event_id: str
     depth: int
+    stream_ordering: int
     type: _BackfillPointType
 
 
@@ -231,16 +226,24 @@ async def _maybe_backfill_inner(
                 processing. Only used for timing.
         """
         backwards_extremities = [
-            _BackfillPoint(event_id, depth, _BackfillPointType.BACKWARDS_EXTREMITY)
-            for event_id, depth in await self.store.get_backfill_points_in_room(room_id)
+            _BackfillPoint(
+                event_id, depth, stream_ordering, _BackfillPointType.BACKWARDS_EXTREMITY
+            )
+            for event_id, depth, stream_ordering in await self.store.get_backfill_points_in_room(
+                room_id=room_id,
+                current_depth=current_depth,
+            )
         ]
 
         insertion_events_to_be_backfilled: List[_BackfillPoint] = []
         if self.hs.config.experimental.msc2716_enabled:
             insertion_events_to_be_backfilled = [
-                _BackfillPoint(event_id, depth, _BackfillPointType.INSERTION_PONT)
-                for event_id, depth in await self.store.get_insertion_event_backward_extremities_in_room(
-                    room_id
+                _BackfillPoint(
+                    event_id, depth, stream_ordering, _BackfillPointType.INSERTION_PONT
+                )
+                for event_id, depth, stream_ordering in await self.store.get_insertion_event_backward_extremities_in_room(
+                    room_id=room_id,
+                    current_depth=current_depth,
                 )
             ]
         logger.debug(
@@ -249,18 +252,14 @@ async def _maybe_backfill_inner(
             insertion_events_to_be_backfilled,
         )
 
-        if not backwards_extremities and not insertion_events_to_be_backfilled:
-            logger.debug("Not backfilling as no extremeties found.")
-            return False
-
         # we now have a list of potential places to backpaginate from. We prefer to
         # start with the most recent (ie, max depth), so let's sort the list.
         sorted_backfill_points: List[_BackfillPoint] = sorted(
             itertools.chain(
                 backwards_extremities,
                 insertion_events_to_be_backfilled,
             ),
-            key=lambda e: -int(e.depth),
+            key=lambda e: (-e.depth, -e.stream_ordering, e.event_id),
         )
 
         logger.debug(
@@ -273,6 +272,29 @@ async def _maybe_backfill_inner(
             sorted_backfill_points,
         )
 
+        # If we have no backfill points lower than the `current_depth` then
+        # either we can a) bail or b) still attempt to backfill. We opt to try
+        # backfilling anyway just in case we do get relevant events.
+        if not sorted_backfill_points and current_depth != MAX_DEPTH:
+            logger.debug(
+                "_maybe_backfill_inner: all backfill points are *after* current depth. Backfilling anyway."
+            )
+            return await self._maybe_backfill_inner(
+                room_id=room_id,
+                # We use `MAX_DEPTH` so that we find all backfill points next
+                # time (all events are below the `MAX_DEPTH`)
+                current_depth=MAX_DEPTH,
+                limit=limit,
+                processing_start_time=processing_start_time,
+            )
+        elif not sorted_backfill_points and current_depth == MAX_DEPTH:
+            # Even after trying again with `MAX_DEPTH`, we didn't find any
+            # backward extremities to backfill from.
+            logger.debug(
+                "_maybe_backfill_inner: Not backfilling as no backward extremeties found."
+            )
+            return False
+
         # If we're approaching an extremity we trigger a backfill, otherwise we
         # no-op.
         #
@@ -286,43 +308,16 @@ async def _maybe_backfill_inner(
         # XXX: shouldn't we do this *after* the filter by depth below? Again, we don't
         # care about events that have happened after our current position.
         #
-        max_depth = sorted_backfill_points[0].depth
-        if current_depth - 2 * limit > max_depth:
+        max_depth_of_backfill_points = sorted_backfill_points[0].depth
+        if current_depth - 2 * limit > max_depth_of_backfill_points:
             logger.debug(
                 "Not backfilling as we don't need to. %d < %d - 2 * %d",
-                max_depth,
+                max_depth_of_backfill_points,
                 current_depth,
                 limit,
             )
             return False
 
-        # We ignore extremities that have a greater depth than our current depth
-        # as:
-        #    1. we don't really care about getting events that have happened
-        #       after our current position; and
-        #    2. we have likely previously tried and failed to backfill from that
-        #       extremity, so to avoid getting "stuck" requesting the same
-        #       backfill repeatedly we drop those extremities.
-        #
-        # However, we need to check that the filtered extremities are non-empty.
-        # If they are empty then either we can a) bail or b) still attempt to
-        # backfill. We opt to try backfilling anyway just in case we do get
-        # relevant events.
-        #
-        filtered_sorted_backfill_points = [
-            t for t in sorted_backfill_points if t.depth <= current_depth
-        ]
-        if filtered_sorted_backfill_points:
-            logger.debug(
-                "_maybe_backfill_inner: backfill points before current depth: %s",
-                filtered_sorted_backfill_points,
-            )
-            sorted_backfill_points = filtered_sorted_backfill_points
-        else:
-            logger.debug(
-                "_maybe_backfill_inner: all backfill points are *after* current depth. Backfilling anyway."
-            )
-
         # For performance's sake, we only want to paginate from a particular extremity
         # if we can actually see the events we'll get. Otherwise, we'd just spend a lot
         # of resources to get redacted events. We check each extremity in turn and
@@ -344,14 +339,10 @@ async def _maybe_backfill_inner(
         # attempting to paginate before backfill reached the visible history.
 
         extremities_to_request: List[str] = []
-        for i, bp in enumerate(sorted_backfill_points):
+        for bp in sorted_backfill_points:
             if len(extremities_to_request) >= 5:
                 break
 
-            set_attribute(
-                SynapseTags.RESULT_PREFIX + "backfill_point" + str(i), str(bp)
-            )
-
             # For regular backwards extremities, we don't have the extremity events
             # themselves, so we need to actually check the events that reference them -
             # their "successor" events.
@@ -396,7 +387,7 @@ async def _maybe_backfill_inner(
             )
             return False
 
-        logger.debug(
+        logger.info(
             "_maybe_backfill_inner: extremities_to_request %s", extremities_to_request
         )
         set_attribute(
@@ -408,13 +399,12 @@ async def _maybe_backfill_inner(
             str(len(extremities_to_request)),
         )
 
-        with start_active_span("getting likely_domains"):
-            # Now we need to decide which hosts to hit first.
-            # First we try hosts that are already in the room.
-            # TODO: HEURISTIC ALERT.
-            likely_domains = (
-                await self._storage_controllers.state.get_current_hosts_in_room(room_id)
-            )
+        # Now we need to decide which hosts to hit first.
+        # First we try hosts that are already in the room.
+        # TODO: HEURISTIC ALERT.
+        likely_domains = (
+            await self._storage_controllers.state.get_current_hosts_in_room(room_id)
+        )
 
         async def try_backfill(domains: Collection[str]) -> bool:
             # TODO: Should we try multiple of these at a time?