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

@@ -890,12 +890,50 @@ 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)
+
+        # We use an `OrderedDict` because even though we sort by depth when we process
+        # the list, it's still important that we maintain the order of the given events
+        # in case the depth of two events is the same. MSC2716 relies on events at the
+        # same depth and `/backfill`` gives a carefully crafted order that we should try
+        # to maintain.
+        new_event_dict = collections.OrderedDict(
+            (event.event_id, event) for event in new_events
+        )
+        # 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,
+            fresh_event_ids,
+        ) = await self._store.separate_event_ids_with_failed_pull_attempts(
+            new_event_dict.keys()
+        )
+
+        # 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 = [
+            new_event_dict[event_id] for 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 = [new_event_dict[event_id] for event_id in fresh_event_ids]
+        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,49 @@ def _record_event_failed_pull_attempt_upsert_txn(
 
         txn.execute(sql, (room_id, event_id, 1, self._clock.time_msec(), cause))
 
+    @trace
+    async def separate_event_ids_with_failed_pull_attempts(
+        self, event_ids: Collection[str]
+    ) -> Tuple[Collection[str], Collection[str]]:
+        """
+        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="separate_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
+        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
+        ]
+        fresh_event_ids = [
+            event_id
+            for event_id in event_ids
+            if event_id not in event_ids_with_failed_pull_attempts
+        ]
+
+        return (event_ids_with_failed_pull_attempts, fresh_event_ids)
+
     @trace
     async def get_event_ids_to_not_pull_from_backoff(
         self,

tests/storage/test_event_federation.py

@@ -1134,6 +1134,62 @@ 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_separate_event_ids_with_failed_pull_attempts(self) -> None:
+        """
+        Test to make sure we properly separate 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 `$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, fresh_event_ids) = self.get_success(
+            self.store.separate_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,
+            # We expect a 2^1 hour backoff after a single failed attempt.
+            ["$c_failed_event_id1", "$b_failed_event_id2", "$a_failed_event_id3"],
+        )
+
+        self.assertEqual(
+            fresh_event_ids,
+            # We expect a 2^1 hour backoff after a single failed attempt.
+            ["$c_fresh_event_id1", "$b_fresh_event_id2", "$a_fresh_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.