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 ab91c870c..446cecdcc 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/src/service/rooms/event_handler/fetch_and_handle_outliers.rs @@ -10,7 +10,7 @@ Err, Event, PduEvent, debug, debug_error, debug_info, debug_warn, err, state_res::lexicographical_topological_sort, trace, - utils::{IterStream, math::Expected}, + utils::{IterStream, math::Expected, stream::BroadbandExt}, warn, }; use futures::{StreamExt, future::select_ok}; @@ -101,7 +101,7 @@ impl super::Service { /// returned the correct event. /// Allows `fetch_and_handle_missing_events` to atomically fetch events from /// multiple remotes in parallel. - async fn fetch_and_handle_missing_event_via( + async fn fetch_event_via( &self, remote: OwnedServerName, event_id: OwnedEventId, @@ -132,13 +132,17 @@ async fn fetch_and_handle_missing_event_via( /// known as "atomic fetch". Should only be used for fetching missing auth /// events or resolving missing events from state_ids. For all other uses, /// use get_missing_events. - #[tracing::instrument(name = "get_missing_events_atomic", skip_all)] - pub(super) async fn fetch_and_handle_missing_events<'a, Pdu>( + /// + /// This function manually walks auth_events trees in a breadth-first + /// search, and persists all fetched events as outliers when all the + /// backwards extremities have been resolved. + #[tracing::instrument(name = "get_missing_auth_events_atomic", skip_all)] + pub(super) async fn fetch_and_handle_auth_events( &self, - origin: &'a ServerName, + origin: &ServerName, events: Vec, - create_event: &'a Pdu, - room_id: &'a RoomId, + create_event: &Pdu, + room_id: &RoomId, ) -> HashMap where Pdu: Event + Send + Sync, @@ -188,18 +192,11 @@ pub(super) async fn fetch_and_handle_missing_events<'a, Pdu>( } debug!(elapsed=?start.elapsed(),"Fetching {next_id} over federation"); - let futures = candidates - .iter() - .map(|remote| { - Box::pin(self.fetch_and_handle_missing_event_via( - remote.clone(), - next_id.clone(), - room_version_rules, - )) - }) - .collect::>(); - let (event_id, value) = match select_ok(futures).await { - | Ok((x, _)) => x, + let (event_id, value) = match self + .fetch_event_vias(candidates.iter(), &next_id, room_version_rules) + .await + { + | Ok(x) => x, | Err(e) => { warn!(elapsed=?start.elapsed(),"failed to fetch missing event {next_id} from any candidate: {e}"); continue; @@ -680,4 +677,106 @@ pub async fn get_missing_events( trace!(elapsed=?start.elapsed(), "Finished get_missing_events"); Ok(discovered) } + + async fn fetch_event_vias( + &self, + candidates: impl Iterator, + event_id: &EventId, + room_version_rules: &RoomVersionRules, + ) -> conduwuit::Result<(OwnedEventId, CanonicalJsonObject)> { + if let Ok(pdu_json) = self.services.timeline.get_pdu_json(event_id).await { + return Ok((event_id.to_owned(), pdu_json)); + } + let futures = candidates + .map(|remote| { + Box::pin(self.fetch_event_via( + remote.to_owned(), + event_id.to_owned(), + room_version_rules, + )) + }) + .collect::>(); + select_ok(futures).await.map(|(res, _)| res) + } + + /// Similar to `fetch_and_handle_missing_events`, but simply walks the + /// prev events tree instead of the auth events tree. Additionally, it does + /// not *handle* fetched PDUs in any capacity. + #[tracing::instrument(name = "get_missing_prev_events_atomic", skip_all)] + pub(super) async fn fetch_prev_events( + &self, + origin: &ServerName, + events: Vec, + create_event: &Pdu, + room_id: &RoomId, + ) -> HashMap + where + Pdu: Event + Send + Sync, + { + let room_version_rules = + &get_room_version_rules(create_event).unwrap_or(RoomVersionRules::V1); + let mut candidates = self + .services + .timeline + .candidate_backfill_servers(room_id) + .await; + candidates.insert(origin.to_owned()); + + let mut todo: VecDeque = VecDeque::from(events); + let mut discovered_events = HashMap::new(); + while let Some(next_id) = todo.pop_front() { + if discovered_events.len() >= self.services.server.config.max_fetch_prev_events.into() + { + debug_warn!( + "Encountered a gap too large to fill, giving up (fetched {} events)", + discovered_events.len() + ); + break; + } + let pdu = match self + .fetch_event_vias(candidates.iter(), &next_id, room_version_rules) + .await + { + | Ok((_, data)) => data, + | Err(e) => { + warn!("Failed to fetch prev event {next_id} from any candidate: {e}"); + continue; + }, + }; + + let prev_events = match expect_event_id_array(&pdu, "prev_events").map_err(|e| { + err!(Request(BadJson(warn!( + event_id=%next_id, + "Failed to parse event fetched from remote: {e}" + )))) + }) { + | Ok(auth_events) => auth_events, + | Err(e) => { + warn!(?e, "event {next_id} is malformed (bad prev_events), skipping"); + continue; + }, + }; + let missing_prev = prev_events + .iter() + .stream() + .broad_filter_map(|event_id| async { + if discovered_events.contains_key(event_id) + || self.services.timeline.pdu_exists(event_id).await + { + None + } else { + Some(event_id.to_owned()) + } + }) + .collect::>() + .await; + todo.extend(missing_prev); + discovered_events.insert( + next_id.clone(), + PduEvent::from_id_val(&next_id, pdu).expect("fetched PDU was already validated"), + ); + } + + discovered_events + } } diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index c36a0f3b5..2e95f39ab 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/src/service/rooms/event_handler/fetch_prev.rs @@ -5,7 +5,7 @@ utils::{BoolExt, IterStream, stream::BroadbandExt}, }; use futures::StreamExt; -use ruma::{CanonicalJsonObject, OwnedEventId, RoomId, ServerName}; +use ruma::{CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedEventId, RoomId, ServerName}; use crate::rooms::event_handler::build_local_dag; @@ -18,9 +18,10 @@ pub(super) async fn fetch_prevs( create_event: &PduEvent, incoming_pdu: &PduEvent, origin: &ServerName, + first_ts_in_room: MilliSecondsSinceUnixEpoch, ) -> conduwuit::Result<()> { let start = Instant::now(); - let missing = incoming_pdu + let mut missing = incoming_pdu .prev_events() .stream() .broad_filter_map(|event_id| async move { @@ -45,7 +46,7 @@ pub(super) async fn fetch_prevs( .collect::>() .await; - let gapfilled = self + let mut gapfilled = self .get_missing_events( room_id, incoming_pdu, @@ -64,6 +65,14 @@ pub(super) async fn fetch_prevs( ) .await?; debug_info!(elapsed=?start.elapsed(), "Fetched {} missing events", gapfilled.len()); + missing.retain(|eid| !gapfilled.contains_key(eid)); + if !missing.is_empty() { + debug_warn!(elapsed=?start.elapsed(), "Still missing {} events, falling back to atomic fetch.", missing.len()); + gapfilled.extend( + self.fetch_prev_events(origin, missing, create_event, room_id) + .await, + ); + } // Persist all fetched events let mapped = gapfilled @@ -98,7 +107,7 @@ pub(super) async fn fetch_prevs( .handle_outlier_pdu(origin, create_event, event_id, room_id, obj, false) .await { - | Ok((pdu, val)) => { + | Ok((pdu, val)) if pdu.origin_server_ts() >= first_ts_in_room => { self.upgrade_outlier_to_timeline_pdu(pdu, val, create_event, origin, room_id) .await .inspect_err(|e| { @@ -125,6 +134,7 @@ pub(super) async fn fetch_prevs( task_elapsed=?persist_start.elapsed(), "Failed to persist prev event {event_id}: {e}", ), + | _ => {}, } } diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index 414c3cbf9..ae306014f 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/src/service/rooms/event_handler/fetch_state.rs @@ -153,7 +153,7 @@ pub(super) async fn fetch_state( %origin, "Failed to fetch full state from remote, falling back to atomic fetch" ); - self.fetch_and_handle_missing_events( + self.fetch_and_handle_auth_events( origin, res.pdu_ids.clone(), create_event, @@ -168,7 +168,7 @@ pub(super) async fn fetch_state( %origin, "Remote did not return room state in an acceptable timeframe, falling back to atomic fetch" ); - self.fetch_and_handle_missing_events( + self.fetch_and_handle_auth_events( origin, res.pdu_ids.clone(), create_event, @@ -210,7 +210,7 @@ pub(super) async fn fetch_state( "Fetching missing events for state from remote" ); let fetched_state = self - .fetch_and_handle_missing_events(origin, to_fetch, create_event, room_id) + .fetch_and_handle_auth_events(origin, to_fetch, create_event, room_id) .await; state_events.extend(fetched_state); } diff --git a/src/service/rooms/event_handler/handle_incoming_pdu.rs b/src/service/rooms/event_handler/handle_incoming_pdu.rs index ca3c9da2f..d769c419e 100644 --- a/src/service/rooms/event_handler/handle_incoming_pdu.rs +++ b/src/service/rooms/event_handler/handle_incoming_pdu.rs @@ -252,7 +252,7 @@ pub async fn handle_incoming_pdu<'a>( // These are timeline events debug!("Fetching and persisting any missing prev events"); - self.fetch_prevs(room_id, create_event, &incoming_pdu, origin) + self.fetch_prevs(room_id, create_event, &incoming_pdu, origin, first_ts_in_room) .await .debug_inspect_err(|e| { error!("Failed to fetch and persist incoming event's prev_events: {e:?}");