From 7f64de9727a0b0d90a32c737638e1811206ce390 Mon Sep 17 00:00:00 2001 From: timedout Date: Thu, 28 May 2026 17:42:01 +0100 Subject: [PATCH] fix: Calculate max iterations dynamically, and bump max prevs --- conduwuit-example.toml | 2 +- src/core/config/mod.rs | 4 +- .../fetch_and_handle_outliers.rs | 83 ++++++++++++++++--- 3 files changed, 73 insertions(+), 16 deletions(-) diff --git a/conduwuit-example.toml b/conduwuit-example.toml index a4d60bbc7..cc736826b 100644 --- a/conduwuit-example.toml +++ b/conduwuit-example.toml @@ -297,7 +297,7 @@ # This item is undocumented. Please contribute documentation for it. # -#max_fetch_prev_events = 256 +#max_fetch_prev_events = 1024 # How many incoming federation transactions the server is willing to be # processing at any given time before it becomes overloaded and starts diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 6ee251bf8..22c47e67d 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -375,7 +375,7 @@ pub struct Config { #[serde(default = "default_max_request_size")] pub max_request_size: usize, - /// default: 256 + /// default: 1024 #[serde(default = "default_max_fetch_prev_events")] pub max_fetch_prev_events: u16, @@ -2549,7 +2549,7 @@ fn default_pusher_timeout() -> u64 { 60 } fn default_pusher_idle_timeout() -> u64 { 15 } -fn default_max_fetch_prev_events() -> u16 { 256_u16 } +fn default_max_fetch_prev_events() -> u16 { 1024 } fn default_max_concurrent_inbound_transactions() -> usize { 150 } diff --git a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs index fec37ec28..3e01defd8 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -20,6 +20,8 @@ use super::get_room_version_rules; use crate::rooms::event_handler::parse_incoming_pdu::expect_event_id_array; +const GET_MISSING_EVENTS_MAX_BATCH_SIZE: u16 = 50; // matches src/server/get_missing_events.rs#LIMIT_MAX + /// Attempts to build a localised directed acyclic graph out of the given PDUs, /// returning them in a topologically sorted order. /// @@ -150,14 +152,39 @@ pub async fn get_missing_events( "event passed to get_missing_events is not missing any events (wasteful call)" ); }; + assert!(!tail.is_empty(), "empty tail"); + assert_ne!(via, self.services.globals.server_name(), "cannot ask ourselves for events"); - let mut discovered = HashMap::with_capacity(20); + // The iteration limit is in place to ensure that if the remote server leaves us + // in a state of infinite recursion (as old versions of continuwuity and + // predecessors would), we give up. However, get_missing_events doesn't return + // that many events per-request. Synapse returns 20, and conduwuit+ return 50. + // This means with a hard iteration limit, we might give up too early, before + // we get a chance to even come close to max_fetch_prev_events. As such, we'll + // calculate the max limit based on that config option based on these averages. + let max_fetch = self.services.server.config.max_fetch_prev_events; + let iteration_limit = max_fetch.saturating_div(20).max(10); + + let mut discovered = HashMap::with_capacity(head.prev_events.len()); let mut latest_events: Vec = vec![head.event_id().to_owned()]; - let mut iterations = 0_u8; - loop { - iterations = iterations.saturating_add(1); - let limit = iterations.saturating_mul(10).min(100); - debug_info!(%limit, %via, %iterations, discovered=discovered.len(), %min_depth, "Attempting to gap fill missing events"); + debug!( + %room_id, + event_id=%head.event_id(), + %iteration_limit, + "Fetching any missing events for head event", + ); + for iteration in 0..iteration_limit { + let limit = iteration + .saturating_mul(10) + .min(GET_MISSING_EVENTS_MAX_BATCH_SIZE); + debug_info!( + %limit, + %via, + %iteration, + discovered=discovered.len(), + %min_depth, + "Attempting to gap fill missing events" + ); let response: get_missing_events::v1::Response = self .services .sending @@ -193,11 +220,13 @@ pub async fn get_missing_events( pdu.depth, min_depth ); + discovered.insert(event_id.clone(), pdu); continue; } for prev_event_id in pdu.prev_events() { if discovered.contains_key(prev_event_id) { + // We already received this event. continue; } if self @@ -206,25 +235,53 @@ pub async fn get_missing_events( .non_outlier_pdu_exists(prev_event_id) .await { + // NOTE: we explicitly check for *non*-outlier events here, as if we end + // up discovering outlier events, we will be able to upgrade them + // immediately. continue; } latest_events.push(prev_event_id.to_owned()); - break; } discovered.insert(event_id.clone(), pdu); } if latest_events.is_empty() { + debug!( + %limit, + %via, + %iteration, + discovered=discovered.len(), + "No more events to fetch." + ); break; - } else if discovered.len() > self.services.server.config.max_fetch_prev_events.into() - || iterations >= 20 - { + } + if discovered.len() >= self.services.server.config.max_fetch_prev_events.into() { + // Stupid hack, debug_error!() drops the log to a DEBUG when not in debug mode, + // which is bad because this should at least produce a warning. It's an error in + // debug mode because this can be important, but typically not much can be done + // about it as a user. + #[cfg(debug_assertions)] error!( - filled=discovered.len(), + discovered=discovered.len(), max_fetch_prev_events=self.services.server.config.max_fetch_prev_events, - %iterations, - "Gap too large, giving up" + %iteration, + %iteration_limit, + %via, + event_id=%head.event_id(), + %room_id, + "Encountered a gap too large to fill, giving up" + ); + #[cfg(not(debug_assertions))] + warn!( + discovered=discovered.len(), + max_fetch_prev_events=self.services.server.config.max_fetch_prev_events, + %iteration, + %iteration_limit, + %via, + event_id=%head.event_id(), + %room_id, + "Encountered a gap too large to fill" ); break; }