From 0933943dd67b8b2f28b48341f4e6b2188bf5062d Mon Sep 17 00:00:00 2001 From: Jason Volk Date: Thu, 26 Feb 2026 13:06:36 +0000 Subject: [PATCH] Remove bad_event_ratelimiter entries after expiration. Signed-off-by: Jason Volk --- src/service/rooms/event_handler/fetch_auth.rs | 9 +++++++-- .../event_handler/handle_incoming_pdu.rs | 3 +++ src/service/rooms/event_handler/mod.rs | 19 +++++++++++++++++-- 3 files changed, 27 insertions(+), 4 deletions(-) diff --git a/src/service/rooms/event_handler/fetch_auth.rs b/src/service/rooms/event_handler/fetch_auth.rs index 298f75f4..7a442e5a 100644 --- a/src/service/rooms/event_handler/fetch_auth.rs +++ b/src/service/rooms/event_handler/fetch_auth.rs @@ -4,7 +4,7 @@ use std::{ time::Duration, }; -use futures::{FutureExt, StreamExt}; +use futures::{FutureExt, StreamExt, TryFutureExt}; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, RoomId, RoomVersionId, ServerName, api::federation::event::get_event, @@ -89,6 +89,7 @@ where if next_id == id { pdus.push((pdu, Some(json))); } + self.cancel_back_off(&next_id); } else { self.back_off(&next_id); } @@ -152,8 +153,11 @@ async fn fetch_auth_chain( .services .federation .execute(origin, get_event::v1::Request { event_id: next_id.clone() }) + .inspect_err(|e| debug_error!(?next_id, "Failed to fetch event: {e}")) + .inspect_ok(|_| { + self.cancel_back_off(&next_id); + }) .await - .inspect_err(|e| debug_error!("Failed to fetch event {next_id}: {e}")) else { debug_warn!("Backing off from {next_id}"); self.back_off(&next_id); @@ -168,6 +172,7 @@ async fn fetch_auth_chain( continue; }; + self.cancel_back_off(&next_id); if calculated_event_id != next_id { warn!( "Server didn't return event id we requested: requested: {next_id}, we got \ diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index a1fcf6bb..99cc930a 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -170,6 +170,9 @@ pub async fn handle_incoming_pdu<'a>( warn!("Prev {prev_id} failed: {e}"); self.back_off(prev_id); }) + .inspect_ok(|()| { + self.cancel_back_off(prev_id); + }) .map(|_| self.services.server.check_running()) }) .boxed() diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index b34300b7..ae77eae5 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/src/service/rooms/event_handler/mod.rs @@ -76,7 +76,16 @@ impl crate::Service for Service { } #[implement(Service)] -fn back_off(&self, event_id: &EventId) { +fn cancel_back_off(&self, event_id: &EventId) -> bool { + self.bad_event_ratelimiter + .write() + .expect("locked") + .remove(event_id) + .is_some() +} + +#[implement(Service)] +fn back_off(&self, event_id: &EventId) -> bool { use hash_map::Entry::{Occupied, Vacant}; match self @@ -87,9 +96,11 @@ fn back_off(&self, event_id: &EventId) { { | Vacant(e) => { e.insert((Instant::now(), 1)); + true }, | Occupied(mut e) => { *e.get_mut() = (Instant::now(), e.get().1.saturating_add(1)); + false }, } } @@ -106,7 +117,11 @@ fn is_backed_off(&self, event_id: &EventId, range: Range) -> bool { return false; }; - continue_exponential_backoff(range.start, range.end, time.elapsed(), tries) + if !continue_exponential_backoff(range.start, range.end, time.elapsed(), tries) { + return false; + } + + true } #[implement(Service)]