Only try to backfill event if we haven't tried before recently (exponential backoff)

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.py

@@ -258,7 +258,14 @@ async def _maybe_backfill_inner(
         backwards_extremities = [
             _BackfillPoint(event_id, depth, _BackfillPointType.BACKWARDS_EXTREMITY)
             for event_id, depth in await self.store.get_oldest_event_ids_with_depth_in_room(
-                room_id
+                room_id=room_id,
+                # We don't want events that come after-in-time from our current
+                # position when we're backfilling looking backwards.
+                #
+                #                           current_depth (ignore events that come after this, ignore 2-4)
+                #                           |
+                # <oldest-in-time> [0]<--[1]▼<--[2]<--[3]<--[4] <newest-in-time>
+                current_depth=current_depth,
             )
         ]
 

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

@@ -11,6 +11,7 @@
 # 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.
+import datetime
 import itertools
 import logging
 from queue import Empty, PriorityQueue
@@ -71,6 +72,12 @@
 
 logger = logging.getLogger(__name__)
 
+BACKFILL_EVENT_BACKOFF_UPPER_BOUND_SECONDS: int = int(
+    datetime.timedelta(days=7).total_seconds()
+)
+BACKFILL_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS: int = int(
+    datetime.timedelta(hours=1).total_seconds()
+)
 
 # All the info we need while iterating the DAG while backfilling
 @attr.s(frozen=True, slots=True, auto_attribs=True)
@@ -715,7 +722,9 @@ def _get_auth_chain_difference_txn(
     @trace
     @tag_args
     async def get_oldest_event_ids_with_depth_in_room(
-        self, room_id: str
+        self,
+        room_id: str,
+        current_depth: int,
     ) -> List[Tuple[str, int]]:
         """Gets the oldest events(backwards extremities) in the room along with the
         aproximate depth.
@@ -735,34 +744,69 @@ async def get_oldest_event_ids_with_depth_in_room(
         def get_oldest_event_ids_with_depth_in_room_txn(
             txn: LoggingTransaction, room_id: str
         ) -> List[Tuple[str, int]]:
-            # Assemble a dictionary with event_id -> depth for the oldest events
+            # Assemble a tuple lookup of event_id -> depth for the oldest events
             # we know of in the room. Backwards extremeties are the oldest
             # events we know of in the room but we only know of them because
-            # some other event referenced them by prev_event and aren't peristed
-            # in our database yet (meaning we don't know their depth
+            # some other event referenced them by prev_event and aren't
+            # persisted in our database yet (meaning we don't know their depth
             # specifically). So we need to look for the aproximate depth from
             # the events connected to the current backwards extremeties.
             sql = """
-                SELECT b.event_id, MAX(e.depth) FROM events as e
+                SELECT backward_extrem.event_id, MAX(event.depth) FROM events as event
                 /**
                  * Get the edge connections from the event_edges table
                  * so we can see whether this event's prev_events points
                  * to a backward extremity in the next join.
                  */
-                INNER JOIN event_edges as g
-                ON g.event_id = e.event_id
+                INNER JOIN event_edges as edge
+                ON edge.event_id = event.event_id
                 /**
                  * We find the "oldest" events in the room by looking for
                  * events connected to backwards extremeties (oldest events
                  * in the room that we know of so far).
                  */
-                INNER JOIN event_backward_extremities as b
-                ON g.prev_event_id = b.event_id
-                WHERE b.room_id = ? AND g.is_state is ?
-                GROUP BY b.event_id
+                INNER JOIN event_backward_extremities as backward_extrem
+                ON edge.prev_event_id = backward_extrem.event_id
+                /**
+                 * We use this info to make sure we don't retry to use a backfill point
+                 * if we've already attempted to backfill from it recently.
+                 */
+                INNER JOIN event_backfill_attempts as backfill_attempt_info
+                ON backfill_attempt_info.event_id = backward_extrem.event_id
+                WHERE
+                    backward_extrem.room_id = ?
+                    /* We only care about normal events because TODO: why? */
+                    AND edge.is_state is ? /* False */
+                    /**
+                     * We only want backwards extremities that are older than or at
+                     * the same position of the given `current_depth` (where older
+                     * means less than a given depth).
+                     */
+                    AND MAX(event.depth) <= current_depth
+                    /**
+                     * Exponential back-off (up to the upper bound) so we don't retry the 
+                     * same backfill point over and over. ex. 2hr, 4hr, 8hr, 16hr, etc
+                     */
+                    AND ? /* current_time */ >= backfill_attempt_info.last_attempt_ts + least(2^backfill_attempt_info.num_attempts * ?, ? /* upper bound */)
+                /* TODO: Why? */
+                GROUP BY backward_extrem.event_id
+                /**
+                 * Sort from highest (closest to the `max_depth`) to the lowest depth
+                 * because the closest are most relevant to backfill from first.
+                 */
+                ORDER BY MAX(event.depth) DESC
             """
 
-            txn.execute(sql, (room_id, False))
+            txn.execute(
+                sql,
+                (
+                    room_id,
+                    False,
+                    self._clock.time_msec(),
+                    1000 * BACKFILL_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
+                    1000 * BACKFILL_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
+                ),
+            )
 
             return cast(List[Tuple[str, int]], txn.fetchall())
 
@@ -1292,6 +1336,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);