fix: Fall back to legacy behaviour when prev events are missed from get_missing_events

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<Pdu>(
 		&self,
-		origin: &'a ServerName,
+		origin: &ServerName,
 		events: Vec<OwnedEventId>,
-		create_event: &'a Pdu,
-		room_id: &'a RoomId,
+		create_event: &Pdu,
+		room_id: &RoomId,
 	) -> HashMap<OwnedEventId, PduEvent>
 	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::<Vec<_>>();
-				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<Item = &OwnedServerName>,
+		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::<Vec<_>>();
+		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<Pdu>(
+		&self,
+		origin: &ServerName,
+		events: Vec<OwnedEventId>,
+		create_event: &Pdu,
+		room_id: &RoomId,
+	) -> HashMap<OwnedEventId, PduEvent>
+	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<OwnedEventId> = 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::<Vec<_>>()
+				.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
+	}
 }

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::<Vec<_>>()
 			.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}",
 				),
+				| _ => {},
 			}
 		}
 

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);
 			}

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:?}");