perf: Don't add trees we already have to latest boundary

src/service/rooms/event_handler/fetch_and_handle_outliers.rs

@@ -8,10 +8,10 @@ use conduwuit::{
 	Err, Event, PduEvent, debug, debug_error, debug_info, debug_warn, err, error,
 	state_res::lexicographical_topological_sort,
 	trace,
-	utils::{IterStream, math::Expected, stream::BroadbandExt},
+	utils::{BoolExt, IterStream, math::Expected},
 	warn,
 };
-use futures::{StreamExt, future::select_ok};
+use futures::{StreamExt, TryStreamExt, future::select_ok};
 use ruma::{
 	CanonicalJsonObject, CanonicalJsonValue, EventId, MilliSecondsSinceUnixEpoch, OwnedEventId,
 	OwnedRoomId, OwnedServerName, RoomId, ServerName, UInt,
@@ -458,23 +458,13 @@ impl super::Service {
 			let missing_count = head
 				.prev_events()
 				.stream()
-				.broad_filter_map(|event_id| async move {
-					match self
-						.services
-						.timeline
-						.get_non_outlier_pdu_json(event_id)
-						.await
-						.inspect(
-							|_| debug!(elapsed=?start.elapsed(),"Found prev_event {event_id} locally."),
-						)
-						.inspect_err(
-							|e| debug!(elapsed=?start.elapsed(),%e, "Could not find prev_event {event_id} locally."),
-						) {
-						| Ok(_) => None,
-						| Err(_) => Some(event_id),
+				.fold(0_u8, |i, event_id| async move {
+					if self.services.timeline.pdu_exists(event_id).await {
+						i.expected_add(1)
+					} else {
+						i
 					}
 				})
-				.count()
 				.await;
 			debug_assert_ne!(
 				missing_count, 0,
@@ -511,6 +501,7 @@ impl super::Service {
 				%limit,
 				%via,
 				%iteration,
+				%iteration_limit,
 				discovered=discovered.len(),
 				%min_depth,
 				"Attempting to gap fill missing events"
@@ -532,21 +523,30 @@ impl super::Service {
 				.await?;
 
 			if response.events.is_empty() {
-				debug_info!(elapsed=?start.elapsed(),%via, "Finished gap filling missing events (remote returned no more events).");
+				debug_info!(
+					elapsed=?start.elapsed(),
+					%via,
+					"Finished gap filling missing events (remote returned no more events)."
+				);
 				break;
 			}
-			debug_info!(elapsed=?start.elapsed(),"Got {} events back from remote", response.events.len());
+			debug_info!(
+				elapsed=?start.elapsed(),
+				"Got {} events back from remote",
+				response.events.len()
+			);
 
 			latest_events.clear();
 			for raw_event in response.events {
 				let (_, event_id, pdu_json) = self.parse_incoming_pdu(&raw_event).await?;
 				let pdu = PduEvent::from_id_val(&event_id, pdu_json).map_err(|e| {
-					err!(Request(BadJson("Failed to parse backfilled event {event_id}: {e}")))
+					err!(Request(BadJson("Failed to parse gapfilled event {event_id}: {e}")))
 				})?;
 
 				if pdu.depth < min_depth {
 					debug_warn!(
-						elapsed=?start.elapsed(),"Received PDU with depth {} below min_depth {}",
+						elapsed=?start.elapsed(),
+						"Received PDU with depth {} below min_depth {}",
 						pdu.depth,
 						min_depth
 					);
@@ -570,6 +570,27 @@ impl super::Service {
 						// immediately.
 						continue;
 					}
+					if let Ok(outlier) = self.services.timeline.get_pdu(prev_event_id).await {
+						// We already have this PDU as an outlier, don't ask for
+						// it. However, if we are missing any prev events for it, add it to the
+						// latest events anyway.
+						let outlier_missing_prevs = outlier
+							.prev_events()
+							.stream()
+							.fold(0_u8, |i, event_id| async move {
+								if self.services.timeline.pdu_exists(event_id).await {
+									i.expected_add(1)
+								} else {
+									i
+								}
+							})
+							.await;
+						if outlier_missing_prevs > 0 {
+							latest_events.push(prev_event_id.to_owned());
+						}
+						discovered.insert(prev_event_id.to_owned(), outlier);
+						continue;
+					}
 					latest_events.push(prev_event_id.to_owned());
 				}