Keep track when we try and fail to process a pulled event

changelog.d/13589.feature

@@ -0,0 +1 @@
+Keep track when we attempt to backfill an event but fail so we can intelligently back-off in the future.

synapse/handlers/federation_event.py

@@ -862,6 +862,8 @@ async def _process_pulled_event(
             self._sanity_check_event(event)
         except SynapseError as err:
             logger.warning("Event %s failed sanity check: %s", event_id, err)
+            if backfilled:
+                await self._store.record_event_backfill_attempt(event_id)
             return
 
         try:
@@ -897,6 +899,9 @@ async def _process_pulled_event(
                     backfilled=backfilled,
                 )
         except FederationError as e:
+            if backfilled:
+                await self._store.record_event_backfill_attempt(event_id)
+
             if e.code == 403:
                 logger.warning("Pulled event %s failed history check.", event_id)
             else:

synapse/storage/databases/main/event_federation.py

@@ -1292,6 +1292,79 @@ def _get_backfill_events(
 
         return event_id_results
 
+    @trace
+    async def record_event_backfill_attempt(self, event_id: str) -> None:
+        if self.database_engine.can_native_upsert:
+            await self.db_pool.runInteraction(
+                "record_event_backfill_attempt",
+                self._record_event_backfill_attempt_upsert_native_txn,
+                event_id,
+                db_autocommit=True,  # Safe as its a single upsert
+            )
+        else:
+            await self.db_pool.runInteraction(
+                "record_event_backfill_attempt",
+                self._record_event_backfill_attempt_upsert_emulated_txn,
+                event_id,
+            )
+
+    def _record_event_backfill_attempt_upsert_native_txn(
+        self,
+        txn: LoggingTransaction,
+        event_id: str,
+    ) -> None:
+        assert self.database_engine.can_native_upsert
+
+        sql = """
+            INSERT INTO event_backfill_attempts (
+                event_id, num_attempts, last_attempt_ts
+            )
+                VALUES (?, ?, ?)
+            ON CONFLICT (event_id) DO UPDATE SET
+                event_id=EXCLUDED.event_id,
+                num_attempts=event_backfill_attempts.num_attempts + 1,
+                last_attempt_ts=EXCLUDED.last_attempt_ts;
+        """
+
+        txn.execute(sql, (event_id, 1, self._clock.time_msec()))
+
+    def _record_event_backfill_attempt_upsert_emulated_txn(
+        self,
+        txn: LoggingTransaction,
+        event_id: str,
+    ) -> None:
+        self.database_engine.lock_table(txn, "event_backfill_attempts")
+
+        prev_row = self.db_pool.simple_select_one_txn(
+            txn,
+            table="event_backfill_attempts",
+            keyvalues={"event_id": event_id},
+            retcols=("num_attempts"),
+            allow_none=True,
+        )
+
+        if not prev_row:
+            self.db_pool.simple_insert_txn(
+                txn,
+                table="event_backfill_attempts",
+                values={
+                    "event_id": event_id,
+                    "num_attempts": 1,
+                    "last_attempt_ts": self._clock.time_msec(),
+                },
+            )
+        else:
+            self.db_pool.simple_update_one_txn(
+                txn,
+                table="event_backfill_attempts",
+                keyvalues={"event_id": event_id},
+                updatevalues={
+                    "event_id": event_id,
+                    "num_attempts": prev_row["num_attempts"] + 1,
+                    "last_attempt_ts": self._clock.time_msec(),
+                },
+            )
+
     async def get_missing_events(
         self,
         room_id: str,

synapse/storage/databases/main/events.py

@@ -2435,17 +2435,34 @@ def _update_backward_extremeties(
             "DELETE FROM event_backward_extremities"
             " WHERE event_id = ? AND room_id = ?"
         )
+        backward_extremity_tuples_to_remove = [
+            (ev.event_id, ev.room_id)
+            for ev in events
+            if not ev.internal_metadata.is_outlier()
+            # If we encountered an event with no prev_events, then we might
+            # as well remove it now because it won't ever have anything else
+            # to backfill from.
+            or len(ev.prev_event_ids()) == 0
+        ]
         txn.execute_batch(
             query,
-            [
-                (ev.event_id, ev.room_id)
-                for ev in events
-                if not ev.internal_metadata.is_outlier()
-                # If we encountered an event with no prev_events, then we might
-                # as well remove it now because it won't ever have anything else
-                # to backfill from.
-                or len(ev.prev_event_ids()) == 0
-            ],
+            backward_extremity_tuples_to_remove,
+        )
+
+        # Since we no longer need these backward extremities, it also means that
+        # they won't be backfilled from again so we no longer need to store the
+        # backfill attempts around it.
+        query = """
+            DELETE FROM event_backfill_attempts
+            WHERE event_id = ?
+        """
+        backward_extremity_event_ids_to_remove = [
+            (extremity_tuple[0],)
+            for extremity_tuple in backward_extremity_tuples_to_remove
+        ]
+        txn.execute_batch(
+            query,
+            backward_extremity_event_ids_to_remove,
         )
 
 

synapse/storage/schema/main/delta/72/04event_backfill_access_time.sql

@@ -0,0 +1,27 @@
+/* Copyright 2022 The Matrix.org Foundation C.I.C
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+-- Add a table that keeps track of when we last tried to backfill an event. This
+-- allows us to be more intelligent when we decide to retry (we don't need to
+-- fail over and over) and we can process that event in the background so we
+-- don't block on it each time.
+CREATE TABLE IF NOT EXISTS event_backfill_attempts(
+    event_id TEXT NOT NULL,
+    num_attempts INT NOT NULL,
+    last_attempt_ts BIGINT NOT NULL
+);
+
+CREATE UNIQUE INDEX IF NOT EXISTS event_backfill_attempts_event_id ON event_backfill_attempts(event_id);

tests/handlers/test_federation_event.py

@@ -227,3 +227,223 @@ async def get_event(destination: str, event_id: str, timeout=None):
 
         if prev_exists_as_outlier:
             self.mock_federation_transport_client.get_event.assert_not_called()
+
+    def test_process_pulled_event_records_failed_backfill_attempts(
+        self,
+    ) -> None:
+        """
+        Test to make sure that failed backfill attempts for an event are
+        recorded in the `event_backfill_attempts` table.
+
+        In this test, we pretend we are processing a "pulled" event via
+        backfill. The pulled event has a fake `prev_event` which our server has
+        obviously never seen before so it attempts to request the state at that
+        `prev_event` which expectedly fails because it's a fake event. Because
+        the server can't fetch the state at the missing `prev_event`, the
+        "pulled" event fails the history check and is fails to process.
+
+        We check that we correctly record the number of failed backfill attempts
+        to the pulled event and as a sanity check, that the "pulled" event isn't
+        persisted. expect.
+        """
+        OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}"
+        main_store = self.hs.get_datastores().main
+
+        # Create the room
+        user_id = self.register_user("kermit", "test")
+        tok = self.login("kermit", "test")
+        room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
+        room_version = self.get_success(main_store.get_room_version(room_id))
+
+        # We expect an outbound request to /state_ids, so stub that out
+        self.mock_federation_transport_client.get_room_state_ids.return_value = make_awaitable(
+            {
+                # Mimic the other server not knowing about the state at all.
+                # We want to cause Synapse to throw an error (`Unable to get
+                # missing prev_event $fake_prev_event`) and fail to backfill
+                # the pulled event.
+                "pdu_ids": [],
+                "auth_chain_ids": [],
+            }
+        )
+        # We also expect an outbound request to /state
+        self.mock_federation_transport_client.get_room_state.return_value = make_awaitable(
+            StateRequestResponse(
+                # Mimic the other server not knowing about the state at all.
+                # We want to cause Synapse to throw an error (`Unable to get
+                # missing prev_event $fake_prev_event`) and fail to backfill
+                # the pulled event.
+                auth_events=[],
+                state=[],
+            )
+        )
+
+        pulled_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "type": "test_regular_type",
+                    "room_id": room_id,
+                    "sender": OTHER_USER,
+                    "prev_events": [
+                        # The fake prev event will make the pulled event fail
+                        # the history check (`Unable to get missing prev_event
+                        # $fake_prev_event`)
+                        "$fake_prev_event"
+                    ],
+                    "auth_events": [],
+                    "origin_server_ts": 1,
+                    "depth": 12,
+                    "content": {"body": "pulled"},
+                }
+            ),
+            room_version,
+        )
+
+        # The function under test: try to process the pulled event
+        with LoggingContext("test"):
+            self.get_success(
+                self.hs.get_federation_event_handler()._process_pulled_event(
+                    self.OTHER_SERVER_NAME, pulled_event, backfilled=True
+                )
+            )
+
+        # Make sure our backfill attempt was recorded
+        backfill_num_attempts = self.get_success(
+            main_store.db_pool.simple_select_one_onecol(
+                table="event_backfill_attempts",
+                keyvalues={"event_id": pulled_event.event_id},
+                retcol="num_attempts",
+            )
+        )
+        self.assertEqual(backfill_num_attempts, 1)
+
+        # The function under test: try to process the pulled event again
+        with LoggingContext("test"):
+            self.get_success(
+                self.hs.get_federation_event_handler()._process_pulled_event(
+                    self.OTHER_SERVER_NAME, pulled_event, backfilled=True
+                )
+            )
+
+        # Make sure our second backfill attempt was recorded (`num_attempts` was incremented)
+        backfill_num_attempts = self.get_success(
+            main_store.db_pool.simple_select_one_onecol(
+                table="event_backfill_attempts",
+                keyvalues={"event_id": pulled_event.event_id},
+                retcol="num_attempts",
+            )
+        )
+        self.assertEqual(backfill_num_attempts, 2)
+
+        # And as a sanity check, make sure the event was not persisted through all of this.
+        persisted = self.get_success(
+            main_store.get_event(pulled_event.event_id, allow_none=True)
+        )
+        self.assertIsNone(
+            persisted,
+            "pulled event that fails the history check should not be persisted at all",
+        )
+
+    def test_process_pulled_event_clears_backfill_attempts_after_being_successfully_persisted(
+        self,
+    ) -> None:
+        """
+        Test to make sure that failed backfill attempts
+        (`event_backfill_attempts` table) for an event are cleared after the
+        event is successfully persisted.
+
+        In this test, we pretend we are processing a "pulled" event via
+        backfill. The pulled event succesfully processes and the backward
+        extremeties are updated along with clearing out any backfill attempts
+        for those old extremities.
+
+        We check that we correctly cleared failed backfill attempts of the
+        pulled event.
+        """
+        OTHER_USER = f"@user:{self.OTHER_SERVER_NAME}"
+        main_store = self.hs.get_datastores().main
+
+        # Create the room
+        user_id = self.register_user("kermit", "test")
+        tok = self.login("kermit", "test")
+        room_id = self.helper.create_room_as(room_creator=user_id, tok=tok)
+        room_version = self.get_success(main_store.get_room_version(room_id))
+
+        # allow the remote user to send state events
+        self.helper.send_state(
+            room_id,
+            "m.room.power_levels",
+            {"events_default": 0, "state_default": 0},
+            tok=tok,
+        )
+
+        # add the remote user to the room
+        member_event = self.get_success(
+            event_injection.inject_member_event(self.hs, room_id, OTHER_USER, "join")
+        )
+
+        initial_state_map = self.get_success(
+            main_store.get_partial_current_state_ids(room_id)
+        )
+
+        auth_event_ids = [
+            initial_state_map[("m.room.create", "")],
+            initial_state_map[("m.room.power_levels", "")],
+            member_event.event_id,
+        ]
+
+        pulled_event = make_event_from_dict(
+            self.add_hashes_and_signatures_from_other_server(
+                {
+                    "type": "test_regular_type",
+                    "room_id": room_id,
+                    "sender": OTHER_USER,
+                    "prev_events": [member_event.event_id],
+                    "auth_events": auth_event_ids,
+                    "origin_server_ts": 1,
+                    "depth": 12,
+                    "content": {"body": "pulled"},
+                }
+            ),
+            room_version,
+        )
+
+        # Fake the "pulled" event failing to backfill once so we can test
+        # if it's cleared out later on.
+        self.get_success(
+            main_store.record_event_backfill_attempt(pulled_event.event_id)
+        )
+        # Make sure we have a backfill attempt recorded for the pulled event
+        backfill_num_attempts = self.get_success(
+            main_store.db_pool.simple_select_one_onecol(
+                table="event_backfill_attempts",
+                keyvalues={"event_id": pulled_event.event_id},
+                retcol="num_attempts",
+            )
+        )
+        self.assertEqual(backfill_num_attempts, 1)
+
+        # The function under test: try to process the pulled event
+        with LoggingContext("test"):
+            self.get_success(
+                self.hs.get_federation_event_handler()._process_pulled_event(
+                    self.OTHER_SERVER_NAME, pulled_event, backfilled=True
+                )
+            )
+
+        # Make sure the backfill attempt for the pulled event are cleared
+        backfill_num_attempts = self.get_success(
+            main_store.db_pool.simple_select_one_onecol(
+                table="event_backfill_attempts",
+                keyvalues={"event_id": pulled_event.event_id},
+                retcol="num_attempts",
+                allow_none=True,
+            )
+        )
+        self.assertIsNone(backfill_num_attempts)
+
+        # And as a sanity check, make sure the "pulled" event was persisted.
+        persisted = self.get_success(
+            main_store.get_event(pulled_event.event_id, allow_none=True)
+        )
+        self.assertIsNotNone(persisted, "pulled event was not persisted at all")