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: Collection[EventBase] = []
         for event in events:
             event_id = event.event_id
 
@@ -890,12 +890,47 @@ async def _process_pulled_events(
             # Continue on with the events that are new to us.
             new_events.append(event)
 
-        # 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
+        ]
+        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
+        ]
+        if len(fresh_events) > 0:
+            await _process_new_pulled_events(fresh_events)
 
     @trace
     @tag_args

synapse/storage/databases/main/event_federation.py

@@ -46,7 +46,7 @@
 from synapse.storage.databases.main.events_worker import EventsWorkerStore
 from synapse.storage.databases.main.signatures import SignatureWorkerStore
 from synapse.storage.engines import PostgresEngine, Sqlite3Engine
-from synapse.types import JsonDict
+from synapse.types import JsonDict, StrCollection
 from synapse.util import json_encoder
 from synapse.util.caches.descriptors import cached
 from synapse.util.caches.lrucache import LruCache
@@ -1583,6 +1583,45 @@ 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: StrCollection
+    ) -> StrCollection:
+        """
+        Separate the given list of events into two lists based on whether they have any
+        failed pull attempts or not.
+
+        Args:
+            event_ids: A list of events to separate
+
+        Returns:
+            A tuple with two lists where the given event_ids are separated based on
+            whether they have any failed pull attempts or not
+            (event_ids_with_failed_pull_attempts, fresh_event_ids). Lists are ordered
+            the same as the given event_ids.
+        """
+
+        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 re-construct things
+        # 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.