Process previously failed backfill events in the background

changelog.d/15585.feature

@@ -0,0 +1 @@
+Process previously failed backfill events in the background to avoid blocking requests for something that is bound to fail again.

synapse/handlers/federation_event.py

@@ -865,7 +865,7 @@ async def _process_pulled_events(
             [event.event_id for event in events]
         )
 
-        new_events = []
+        new_events: List[EventBase] = []
         for event in events:
             event_id = event.event_id
 
@@ -895,12 +895,63 @@ async def _process_pulled_events(
             str(len(new_events)),
         )
 
-        # We want to sort these by depth so we process them and
-        # tell clients about them in order.
-        sorted_events = sorted(new_events, key=lambda x: x.depth)
-        for ev in sorted_events:
-            with nested_logging_context(ev.event_id):
-                await self._process_pulled_event(origin, ev, backfilled=backfilled)
+        @trace
+        async def _process_new_pulled_events(new_events: Collection[EventBase]) -> None:
+            # We want to sort these by depth so we process them and
+            # tell clients about them in order.
+            sorted_events = sorted(new_events, key=lambda x: x.depth)
+            for ev in sorted_events:
+                with nested_logging_context(ev.event_id):
+                    await self._process_pulled_event(origin, ev, backfilled=backfilled)
+
+        # Check if we've already tried to process these events at some point in the
+        # past. We aren't concerned with the expontntial backoff here, just whether it
+        # has failed to be processed before.
+        event_ids_with_failed_pull_attempts = (
+            await self._store.get_event_ids_with_failed_pull_attempts(
+                [event.event_id for event in new_events]
+            )
+        )
+
+        # Process previously failed backfill events in the background to not waste
+        # time on something that is bound to fail again.
+        events_with_failed_pull_attempts = [
+            event
+            for event in new_events
+            if event.event_id in event_ids_with_failed_pull_attempts
+        ]
+        set_tag(
+            SynapseTags.FUNC_ARG_PREFIX + "events_with_failed_pull_attempts",
+            str(event_ids_with_failed_pull_attempts),
+        )
+        set_tag(
+            SynapseTags.RESULT_PREFIX + "events_with_failed_pull_attempts.length",
+            str(len(events_with_failed_pull_attempts)),
+        )
+        if len(events_with_failed_pull_attempts) > 0:
+            run_as_background_process(
+                "_process_new_pulled_events_with_failed_pull_attempts",
+                _process_new_pulled_events,
+                events_with_failed_pull_attempts,
+            )
+
+        # We can optimistically try to process and wait for the event to be fully
+        # persisted if we've never tried before.
+        fresh_events = [
+            event
+            for event in new_events
+            if event.event_id not in event_ids_with_failed_pull_attempts
+        ]
+        set_tag(
+            SynapseTags.FUNC_ARG_PREFIX + "fresh_events",
+            str([event.event_id for event in fresh_events]),
+        )
+        set_tag(
+            SynapseTags.RESULT_PREFIX + "fresh_events.length",
+            str(len(fresh_events)),
+        )
+        if len(fresh_events) > 0:
+            await _process_new_pulled_events(fresh_events)
 
     @trace
     @tag_args

synapse/storage/databases/main/event_federation.py

@@ -1583,6 +1583,43 @@ def _record_event_failed_pull_attempt_upsert_txn(
 
         txn.execute(sql, (room_id, event_id, 1, self._clock.time_msec(), cause))
 
+    @trace
+    async def get_event_ids_with_failed_pull_attempts(
+        self, event_ids: List[str]
+    ) -> List[str]:
+        """
+        Filter the given list of `event_ids` and return events which have any failed
+        pull attempts.
+
+        Args:
+            event_ids: A list of events to filter down.
+
+        Returns:
+            A filtered down list of `event_ids` that have previous failed pull attempts
+            (order is maintained).
+        """
+
+        rows = await self.db_pool.simple_select_many_batch(
+            table="event_failed_pull_attempts",
+            column="event_id",
+            iterable=event_ids,
+            keyvalues={},
+            retcols=("event_id",),
+            desc="get_event_ids_with_failed_pull_attempts",
+        )
+        event_ids_with_failed_pull_attempts_from_database = [
+            str(row["event_id"]) for row in rows
+        ]
+        # We want to maintain the order of the given `event_ids` so we re-construct the
+        # list since there is no gurantees from the database implementation/query.
+        event_ids_with_failed_pull_attempts = [
+            event_id
+            for event_id in event_ids
+            if event_id in event_ids_with_failed_pull_attempts_from_database
+        ]
+
+        return event_ids_with_failed_pull_attempts
+
     @trace
     async def get_event_ids_to_not_pull_from_backoff(
         self,

tests/storage/test_event_federation.py

@@ -1134,6 +1134,55 @@ def test_get_insertion_event_backward_extremities_in_room_attempted_event_retry_
         backfill_event_ids = [backfill_point[0] for backfill_point in backfill_points]
         self.assertEqual(backfill_event_ids, ["insertion_eventA"])
 
+    def test_get_event_ids_with_failed_pull_attempts(self) -> None:
+        """
+        Test to make sure we properly get event_ids based on whether they have any
+        failed pull attempts.
+        """
+        # Create the room
+        user_id = self.register_user("alice", "test")
+        tok = self.login("alice", "test")
+        room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
+
+        # We purposely record the failed pull attempt for `$c_failed_event_id3` first to
+        # make sure we return results in the order of the `event_ids` passed in instead
+        # of the order in which we find things in the database or the unordered
+        # collections we might accidentally use. They also purposely have reverse
+        # prefixed a-c in front to better test dubious sorting happening somewhere.
+        self.get_success(
+            self.store.record_event_failed_pull_attempt(
+                room_id, "$a_failed_event_id3", "fake cause"
+            )
+        )
+        self.get_success(
+            self.store.record_event_failed_pull_attempt(
+                room_id, "$c_failed_event_id1", "fake cause"
+            )
+        )
+        self.get_success(
+            self.store.record_event_failed_pull_attempt(
+                room_id, "$b_failed_event_id2", "fake cause"
+            )
+        )
+
+        event_ids_with_failed_pull_attempts = self.get_success(
+            self.store.get_event_ids_with_failed_pull_attempts(
+                event_ids=[
+                    "$c_failed_event_id1",
+                    "$c_fresh_event_id1",
+                    "$b_failed_event_id2",
+                    "$b_fresh_event_id2",
+                    "$a_failed_event_id3",
+                    "$a_fresh_event_id3",
+                ]
+            )
+        )
+
+        self.assertEqual(
+            event_ids_with_failed_pull_attempts,
+            ["$c_failed_event_id1", "$b_failed_event_id2", "$a_failed_event_id3"],
+        )
+
     def test_get_event_ids_to_not_pull_from_backoff(self) -> None:
         """
         Test to make sure only event IDs we should backoff from are returned.