Stop getting missing prev_events after we already know their signature is invalid

changelog.d/13815.feature

@@ -0,0 +1 @@
+Keep track when an event pulled over federation fails its signature check so we can intelligently back-off in the future.

synapse/federation/federation_base.py

@@ -13,7 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Awaitable, Callable, Optional
 
 from synapse.api.constants import MAX_DEPTH, EventContentFields, EventTypes, Membership
 from synapse.api.errors import Codes, SynapseError
@@ -58,7 +58,12 @@ def __init__(self, hs: "HomeServer"):
 
     @trace
     async def _check_sigs_and_hash(
-        self, room_version: RoomVersion, pdu: EventBase
+        self,
+        room_version: RoomVersion,
+        pdu: EventBase,
+        record_failure_callback: Optional[
+            Callable[[EventBase, str], Awaitable[None]]
+        ] = None,
     ) -> EventBase:
         """Checks that event is correctly signed by the sending server.
 
@@ -70,6 +75,11 @@ async def _check_sigs_and_hash(
         Args:
             room_version: The room version of the PDU
             pdu: the event to be checked
+            record_failure_callback: A callback to run whenever the given event
+                fails signature or hash checks. This includes exceptions
+                that would be normally be thrown/raised but also things like
+                checking for event tampering where we just return the redacted
+                event.
 
         Returns:
               * the original event if the checks pass
@@ -80,7 +90,12 @@ async def _check_sigs_and_hash(
           InvalidEventSignatureError if the signature check failed. Nothing
              will be logged in this case.
         """
-        await _check_sigs_on_pdu(self.keyring, room_version, pdu)
+        try:
+            await _check_sigs_on_pdu(self.keyring, room_version, pdu)
+        except InvalidEventSignatureError as exc:
+            if record_failure_callback:
+                await record_failure_callback(pdu, str(exc))
+            raise exc
 
         if not check_event_content_hash(pdu):
             # let's try to distinguish between failures because the event was
@@ -116,6 +131,10 @@ async def _check_sigs_and_hash(
                         "event_id": pdu.event_id,
                     }
                 )
+                if record_failure_callback:
+                    await record_failure_callback(
+                        pdu, "Event content has been tampered with"
+                    )
             return redacted_event
 
         spam_check = await self.spam_checker.check_event_for_spam(pdu)

synapse/federation/federation_client.py

@@ -278,7 +278,7 @@ async def backfill(
         pdus = [event_from_pdu_json(p, room_version) for p in transaction_data_pdus]
 
         # Check signatures and hash of pdus, removing any from the list that fail checks
-        pdus[:] = await self._check_sigs_and_hash_and_fetch(
+        pdus[:] = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
             dest, pdus, room_version=room_version
         )
 
@@ -328,7 +328,17 @@ async def get_pdu_from_destination_raw(
 
             # Check signatures are correct.
             try:
-                signed_pdu = await self._check_sigs_and_hash(room_version, pdu)
+
+                async def _record_failure_callback(
+                    event: EventBase, cause: str
+                ) -> None:
+                    await self.store.record_event_failed_pull_attempt(
+                        event.room_id, event.event_id, cause
+                    )
+
+                signed_pdu = await self._check_sigs_and_hash(
+                    room_version, pdu, _record_failure_callback
+                )
             except InvalidEventSignatureError as e:
                 errmsg = f"event id {pdu.event_id}: {e}"
                 logger.warning("%s", errmsg)
@@ -547,24 +557,28 @@ async def get_room_state(
             len(auth_event_map),
         )
 
-        valid_auth_events = await self._check_sigs_and_hash_and_fetch(
+        valid_auth_events = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
             destination, auth_event_map.values(), room_version
         )
 
-        valid_state_events = await self._check_sigs_and_hash_and_fetch(
-            destination, state_event_map.values(), room_version
+        valid_state_events = (
+            await self._check_sigs_and_hash_for_pulled_events_and_fetch(
+                destination, state_event_map.values(), room_version
+            )
         )
 
         return valid_state_events, valid_auth_events
 
     @trace
-    async def _check_sigs_and_hash_and_fetch(
+    async def _check_sigs_and_hash_for_pulled_events_and_fetch(
         self,
         origin: str,
         pdus: Collection[EventBase],
         room_version: RoomVersion,
     ) -> List[EventBase]:
-        """Checks the signatures and hashes of a list of events.
+        """
+        Checks the signatures and hashes of a list of pulled events we got from
+        federation and records any signature failures as failed pull attempts.
 
         If a PDU fails its signature check then we check if we have it in
         the database, and if not then request it from the sender's server (if that
@@ -597,11 +611,17 @@ async def _check_sigs_and_hash_and_fetch(
 
         valid_pdus: List[EventBase] = []
 
+        async def _record_failure_callback(event: EventBase, cause: str) -> None:
+            await self.store.record_event_failed_pull_attempt(
+                event.room_id, event.event_id, cause
+            )
+
         async def _execute(pdu: EventBase) -> None:
             valid_pdu = await self._check_sigs_and_hash_and_fetch_one(
                 pdu=pdu,
                 origin=origin,
                 room_version=room_version,
+                record_failure_callback=_record_failure_callback,
             )
 
             if valid_pdu:
@@ -618,6 +638,9 @@ async def _check_sigs_and_hash_and_fetch_one(
         pdu: EventBase,
         origin: str,
         room_version: RoomVersion,
+        record_failure_callback: Optional[
+            Callable[[EventBase, str], Awaitable[None]]
+        ] = None,
     ) -> Optional[EventBase]:
         """Takes a PDU and checks its signatures and hashes.
 
@@ -634,14 +657,21 @@ async def _check_sigs_and_hash_and_fetch_one(
             origin
             pdu
             room_version
+            record_failure_callback: A callback to run whenever the given event
+                fails signature or hash checks. This includes exceptions
+                that would be normally be thrown/raised but also things like
+                checking for event tampering where we just return the redacted
+                event.
 
         Returns:
             The PDU (possibly redacted) if it has valid signatures and hashes.
             None if no valid copy could be found.
         """
 
         try:
-            return await self._check_sigs_and_hash(room_version, pdu)
+            return await self._check_sigs_and_hash(
+                room_version, pdu, record_failure_callback
+            )
         except InvalidEventSignatureError as e:
             logger.warning(
                 "Signature on retrieved event %s was invalid (%s). "
@@ -694,7 +724,7 @@ async def get_event_auth(
 
         auth_chain = [event_from_pdu_json(p, room_version) for p in res["auth_chain"]]
 
-        signed_auth = await self._check_sigs_and_hash_and_fetch(
+        signed_auth = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
             destination, auth_chain, room_version=room_version
         )
 
@@ -1401,7 +1431,7 @@ async def get_missing_events(
                 event_from_pdu_json(e, room_version) for e in content.get("events", [])
             ]
 
-            signed_events = await self._check_sigs_and_hash_and_fetch(
+            signed_events = await self._check_sigs_and_hash_for_pulled_events_and_fetch(
                 destination, events, room_version=room_version
             )
         except HttpResponseException as e:

synapse/handlers/federation_event.py

@@ -965,6 +965,12 @@ async def _compute_event_context_with_maybe_missing_prevs(
         seen = await self._store.have_events_in_timeline(prevs)
         missing_prevs = prevs - seen
 
+        # Filter out events we've tried to pull recently
+        prevs_to_ignore = await self.store.filter_events_with_pull_attempt_backoff(
+            room_id, missing_prevs
+        )
+        missing_prevs = missing_prevs - prevs_to_ignore
+
         if not missing_prevs:
             return await self._state_handler.compute_event_context(event)
 

synapse/storage/databases/main/event_federation.py

@@ -22,6 +22,7 @@
     Iterable,
     List,
     Optional,
+    Sequence,
     Set,
     Tuple,
     cast,
@@ -73,14 +74,13 @@
 
 logger = logging.getLogger(__name__)
 
-BACKFILL_EVENT_BACKOFF_UPPER_BOUND_SECONDS: int = int(
+PULLED_EVENT_BACKOFF_UPPER_BOUND_SECONDS: int = int(
     datetime.timedelta(days=7).total_seconds()
 )
-BACKFILL_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS: int = int(
+PULLED_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)
 class BackfillQueueNavigationItem:
@@ -851,8 +851,8 @@ def get_backfill_points_in_room_txn(
                     False,
                     current_depth,
                     self._clock.time_msec(),
-                    1000 * BACKFILL_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
-                    1000 * BACKFILL_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
+                    1000 * PULLED_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
+                    1000 * PULLED_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
                     limit,
                 ),
             )
@@ -967,8 +967,8 @@ def get_insertion_event_backward_extremities_in_room_txn(
                     room_id,
                     current_depth,
                     self._clock.time_msec(),
-                    1000 * BACKFILL_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
-                    1000 * BACKFILL_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
+                    1000 * PULLED_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
+                    1000 * PULLED_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
                     limit,
                 ),
             )
@@ -1506,6 +1506,78 @@ def _record_event_failed_pull_attempt_upsert_txn(
 
         txn.execute(sql, (room_id, event_id, 1, self._clock.time_msec(), cause))
 
+    @trace
+    async def filter_events_with_pull_attempt_backoff(
+        self,
+        room_id: str,
+        event_ids: Sequence[str],
+    ) -> List[str]:
+        """
+        Filter out events that we've failed to pull before
+        recently. Uses exponential backoff.
+
+        Args:
+            room_id: The room that the events belong to
+            event_ids: A list of events to filter down
+
+        Returns:
+            List of event_ids that can be attempted to be pulled
+        """
+        return await self.db_pool.runInteraction(
+            "filter_events_with_pull_attempt_backoff",
+            self._filter_events_with_pull_attempt_backoff_txn,
+            room_id,
+            event_ids,
+        )
+
+    def _filter_events_with_pull_attempt_backoff_txn(
+        self,
+        txn: LoggingTransaction,
+        room_id: str,
+        event_ids: Sequence[str],
+    ) -> None:
+        where_event_ids_match_clause, values = make_in_list_sql_clause(
+            txn.database_engine, "event_id", event_ids
+        )
+
+        sql = """
+            SELECT event_id FROM event_failed_pull_attempts
+            WHERE
+                room_id = ?
+                %s /* where_event_ids_match_clause */
+                /**
+                * Exponential back-off (up to the upper bound) so we don't try to
+                * pull the same event over and over. ex. 2hr, 4hr, 8hr, 16hr, etc.
+                *
+                * We use `1 << n` as a power of 2 equivalent for compatibility
+                * with older SQLites. The left shift equivalent only works with
+                * powers of 2 because left shift is a binary operation (base-2).
+                * Otherwise, we would use `power(2, n)` or the power operator, `2^n`.
+                */
+                AND (
+                    event_id IS NULL
+                    OR ? /* current_time */ >= last_attempt_ts + /*least*/%s((1 << num_attempts) * ? /* step */, ? /* upper bound */)
+                )
+        """
+
+        if isinstance(self.database_engine, PostgresEngine):
+            least_function = "least"
+        elif isinstance(self.database_engine, Sqlite3Engine):
+            least_function = "min"
+        else:
+            raise RuntimeError("Unknown database engine")
+
+        txn.execute(
+            sql % (where_event_ids_match_clause, least_function),
+            (
+                room_id,
+                *values,
+                self._clock.time_msec(),
+                1000 * PULLED_EVENT_EXPONENTIAL_BACKOFF_STEP_SECONDS,
+                1000 * PULLED_EVENT_BACKOFF_UPPER_BOUND_SECONDS,
+            ),
+        )
+
     async def get_missing_events(
         self,
         room_id: str,