From 98c1a466fde83c6c91bcb4f3a46d28c02fece04b Mon Sep 17 00:00:00 2001 From: timedout Date: Thu, 28 May 2026 18:38:27 +0100 Subject: [PATCH] feat: Enhance reliability by fetching full state when we're missing a lot of auth events --- .../fetch_and_handle_outliers.rs | 38 ++++- .../rooms/event_handler/fetch_state.rs | 159 +++++++++++++++--- .../event_handler/upgrade_outlier_pdu.rs | 9 +- 3 files changed, 175 insertions(+), 31 deletions(-) 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 3a881be52..0ecb81aca 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -139,8 +139,10 @@ impl super::Service { .candidate_backfill_servers(room_id) .await; candidates.insert(origin.to_owned()); + candidates.retain(|sn| self.services.globals.server_name() != sn); + assert_ne!(candidates.len(), 0, "no candidates to fetch missing events from"); let mut seeded_events = - HashMap::with_capacity(events.len() + (events.len().saturating_mul(3))); + HashMap::with_capacity(events.len().saturating_add(events.len().saturating_mul(3))); trace!( "Fetching {} unknown PDUs on demand from {} candidates", events.len(), @@ -193,8 +195,32 @@ impl super::Service { continue; }, }; - todo.extend(auth_events); - seeded_events.insert(event_id, value); + let mut have_all_auth = true; + for auth_event_id in auth_events { + if let Ok(local_pdu) = self.services.timeline.get_pdu(&next_id).await { + trace!("Found auth event {next_id} in db"); + seeded_events.insert(id.clone(), local_pdu.into_canonical_object()); + continue; + } + if seeded_events.contains_key(&auth_event_id) { + trace!(%auth_event_id, "Already found auth event"); + continue; + } + todo.push_back(auth_event_id); + have_all_auth = false; + } + // Insert this PDU back at the end of the queue so that it will be resolved once + // all of its auth events have been fetched. + // TODO: This may result in infinite looping, needs a breaker + if have_all_auth { + debug!(%event_id, "Have all auth events"); + seeded_events.insert(event_id, value); + } else { + debug_warn!( + "Fetched {event_id} but missing some auth events, will have to re-fetch." + ); + todo.push_back(event_id); + } } } @@ -209,7 +235,7 @@ impl super::Service { let mut pdus = HashMap::with_capacity(seeded_ordered.len()); for id in seeded_ordered { let pdu_json = seeded_events.remove(&id).unwrap(); - trace!("Handling missing event {id}"); + debug_info!("Handling missing event {id} as outlier"); match Box::pin(self.handle_outlier_pdu( origin, create_event, @@ -225,6 +251,10 @@ impl super::Service { }, | Err(e) => warn!("Authentication of event {id} failed: {e:?}"), } + + // TODO: should this try to promote to timeline? + // If we got here, we probably weren't able to promote it before + // now. } trace!("Fetched and handled {} missing PDUs", pdus.len()); diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index a97cd9b22..ea7ed3c3b 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/src/service/rooms/event_handler/fetch_state.rs @@ -1,33 +1,35 @@ use std::collections::{HashMap, hash_map}; -use conduwuit::{Err, Event, PduEvent, Result, debug, debug_warn, err, utils::IterStream}; +use conduwuit::{Err, Event, PduEvent, Result, debug, debug_warn, err, utils::IterStream, warn}; use futures::StreamExt; use ruma::{ - EventId, OwnedEventId, RoomId, ServerName, api::federation::event::get_room_state_ids, + EventId, OwnedEventId, RoomId, ServerName, + api::federation::event::{get_room_state, get_room_state_ids}, events::StateEventType, }; use crate::{conduwuit::utils::stream::BroadbandExt, rooms::short::ShortStateKey}; impl super::Service { - /// Call /state_ids to find out what the state at this pdu is. We trust the - /// server's response to some extent (sic), but we still do a lot of checks - /// on the events - #[tracing::instrument( - level = "debug", - skip_all, - fields(%origin), -)] - pub(super) async fn fetch_state( + /// Asks a remote server what the state at this event is. + /// It first attempts to call `GET /_matrix/federation/v1/state_ids` (fast). + /// If any events are missing, they are fetched from the remote, and + /// persisted as outliers, before being returned back to this function. If + /// we are missing a lot of events locally (>=50), this function falls back + /// to requesting the full state in PDU format from the remote (`GET + /// /_matrix/federation/v1/state, very slow in large rooms), and persists + /// them directly. + /// + /// The end result is a result containing a map of shortstatekeys to event + /// IDs. The underlying `Option` is always `Some`. + #[tracing::instrument(skip(self, create_event))] + pub(super) async fn fetch_state( &self, origin: &ServerName, - create_event: &Pdu, + create_event: &PduEvent, room_id: &RoomId, event_id: &EventId, - ) -> Result>> - where - Pdu: Event + Send + Sync, - { + ) -> Result>> { let res: get_room_state_ids::v1::Response = self .services .sending @@ -58,12 +60,25 @@ impl super::Service { if !to_fetch.is_empty() { if to_fetch.len() >= 100 { // That's a lot of events to fetch, just ask for the full state - // at that point. TODO: fetch /state + // at that point. + debug_warn!( + to_fetch = to_fetch.len(), + "Fetching full state from remote server for event" + ); + state_events.extend( + self.fetch_full_state(origin, create_event, room_id, event_id) + .await?, + ); + } else { + debug!( + to_fetch = to_fetch.len(), + "Fetching missing events for state from remote" + ); + state_events.extend( + self.fetch_and_handle_missing_events(origin, to_fetch, create_event, room_id) + .await, + ); } - state_events.extend( - self.fetch_and_handle_missing_events(origin, to_fetch, create_event, room_id) - .await, - ); } let mut state: HashMap = @@ -107,4 +122,106 @@ impl super::Service { Ok(Some(state)) } + + /// Fetches the full state via `GET /_matrix/federation/v1/state` from a + /// remote server, and persists all the incoming auth chain events and + /// state events as outliers, for use later. + /// + /// Any events that cannot be persisted are dropped with a warning. + /// TODO: make it noisy? + pub(super) async fn fetch_full_state( + &self, + origin: &ServerName, + create_event: &PduEvent, + room_id: &RoomId, + event_id: &EventId, + ) -> Result> { + let res: get_room_state::v1::Response = self + .services + .sending + .send_federation_request( + origin, + get_room_state::v1::Request::new(event_id.to_owned(), room_id.to_owned()), + ) + .await + .inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?; + debug!(count = res.auth_chain.len(), "Handling incoming auth chain..."); + res.auth_chain + .iter() + .stream() + .broad_filter_map(|raw_event_json| async { + if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok() + && parsed.0 == room_id + { + Some(parsed) + } else { + None + } + }) + .for_each_concurrent( + None, + |(incoming_room_id, incoming_event_id, incoming_event_json)| async move { + self.handle_outlier_pdu( + origin, + create_event, + &incoming_event_id, + &incoming_room_id, + incoming_event_json, + true, + ) + .await + .inspect_err(|e| { + warn!( + %incoming_room_id, + %incoming_event_id, + ?e, + "Failed to handle auth chain event from state fetch" + ); + }) + .ok(); + }, + ) + .await; + debug!(count = res.pdus.len(), "Handling incoming state PDUs..."); + Ok(res + .pdus + .iter() + .stream() + .broad_filter_map(|raw_event_json| async { + if let Some(parsed) = self.parse_incoming_pdu(raw_event_json).await.ok() + && parsed.0 == room_id + { + Some(parsed) + } else { + None + } + }) + .broad_filter_map( + |(incoming_room_id, incoming_event_id, incoming_event_json)| async move { + self.handle_outlier_pdu( + origin, + create_event, + &incoming_event_id, + &incoming_room_id, + incoming_event_json, + true, + ) + .await + .inspect_err(|e| { + warn!( + %incoming_room_id, + %incoming_event_id, + ?e, + "Failed to handle state event from state fetch" + ); + }) + .ok() + }, + ) + .fold(HashMap::new(), |mut acc, (event, _)| async move { + acc.insert(event.event_id().to_owned(), event); + acc + }) + .await) + } } diff --git a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs index 7a1865152..ac5f8ee14 100644 --- a/src/service/rooms/event_handler/upgrade_outlier_pdu.rs +++ b/src/service/rooms/event_handler/upgrade_outlier_pdu.rs @@ -23,17 +23,14 @@ use crate::rooms::{ }; impl super::Service { - pub(super) async fn upgrade_outlier_to_timeline_pdu( + pub(super) async fn upgrade_outlier_to_timeline_pdu( &self, incoming_pdu: PduEvent, mut val: CanonicalJsonObject, - create_event: &Pdu, + create_event: &PduEvent, origin: &ServerName, room_id: &RoomId, - ) -> Result> - where - Pdu: Event + Send + Sync, - { + ) -> Result> { // Skip the PDU if we already have it as a timeline event if let Ok(pduid) = self .services