feat: Include timing information in debug logs

src/service/rooms/event_handler/fetch_and_handle_outliers.rs

@@ -1,4 +1,7 @@
-use std::collections::{HashMap, HashSet, VecDeque};
+use std::{
+	collections::{HashMap, HashSet, VecDeque},
+	time::Instant,
+};
 
 use assign::assign;
 use conduwuit::{
@@ -127,6 +130,7 @@ impl super::Service {
 		via: &ServerName,
 		min_depth: UInt,
 	) -> conduwuit::Result<HashMap<OwnedEventId, PduEvent>> {
+		let start = Instant::now();
 		#[cfg(debug_assertions)]
 		{
 			let missing_count = head
@@ -138,9 +142,11 @@ impl super::Service {
 						.timeline
 						.get_non_outlier_pdu_json(event_id)
 						.await
-						.inspect(|_| debug!("Found prev_event {event_id} locally."))
+						.inspect(
+							|_| debug!(elapsed=?start.elapsed(),"Found prev_event {event_id} locally."),
+						)
 						.inspect_err(
-							|e| debug!(%e, "Could not find prev_event {event_id} locally."),
+							|e| debug!(elapsed=?start.elapsed(),%e, "Could not find prev_event {event_id} locally."),
 						) {
 						| Ok(_) => None,
 						| Err(_) => Some(event_id),
@@ -168,7 +174,7 @@ impl super::Service {
 
 		let mut discovered = HashMap::with_capacity(head.prev_events.len());
 		let mut latest_events: Vec<OwnedEventId> = vec![head.event_id().to_owned()];
-		debug!(
+		debug!(elapsed=?start.elapsed(),
 			%room_id,
 			event_id=%head.event_id(),
 			%iteration_limit,
@@ -179,7 +185,7 @@ impl super::Service {
 				.expected_add(1)
 				.saturating_mul(10)
 				.min(GET_MISSING_EVENTS_MAX_BATCH_SIZE);
-			debug_info!(
+			debug_info!(elapsed=?start.elapsed(),
 				%limit,
 				%via,
 				%iteration,
@@ -204,10 +210,10 @@ impl super::Service {
 				.await?;
 
 			if response.events.is_empty() {
-				debug_info!(%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!("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 {
@@ -218,7 +224,7 @@ impl super::Service {
 
 				if pdu.depth < min_depth {
 					debug_warn!(
-						"Received PDU with depth {} below min_depth {}, ignoring",
+						elapsed=?start.elapsed(),"Received PDU with depth {} below min_depth {}, ignoring",
 						pdu.depth,
 						min_depth
 					);
@@ -249,7 +255,7 @@ impl super::Service {
 			}
 
 			if latest_events.is_empty() {
-				debug!(
+				debug!(elapsed=?start.elapsed(),
 					%limit,
 					%via,
 					%iteration,
@@ -264,7 +270,7 @@ impl super::Service {
 				// debug mode because this can be important, but typically not much can be done
 				// about it as a user.
 				#[cfg(debug_assertions)]
-				error!(
+				error!(elapsed=?start.elapsed(),
 					discovered=discovered.len(),
 					max_fetch_prev_events=self.services.server.config.max_fetch_prev_events,
 					%iteration,
@@ -275,7 +281,7 @@ impl super::Service {
 					"Encountered a gap too large to fill, giving up"
 				);
 				#[cfg(not(debug_assertions))]
-				warn!(
+				warn!(elapsed=?start.elapsed(),
 					discovered=discovered.len(),
 					max_fetch_prev_events=self.services.server.config.max_fetch_prev_events,
 					%iteration,
@@ -289,6 +295,7 @@ impl super::Service {
 			}
 		}
 
+		trace!(elapsed=?start.elapsed(), "Finished get_missing_events");
 		Ok(discovered)
 	}
 
@@ -333,6 +340,7 @@ impl super::Service {
 	where
 		Pdu: Event + Send + Sync,
 	{
+		let start = Instant::now();
 		let room_version_rules =
 			&get_room_version_rules(create_event).unwrap_or(RoomVersionRules::V1);
 		let mut candidates = self
@@ -346,6 +354,7 @@ impl super::Service {
 		let mut seeded_events =
 			HashMap::with_capacity(events.len().saturating_add(events.len().saturating_mul(3)));
 		trace!(
+			elapsed=?start.elapsed(),
 			"Fetching {} unknown PDUs on demand from {} candidates",
 			events.len(),
 			candidates.len()
@@ -359,7 +368,7 @@ impl super::Service {
 					continue;
 				}
 				if let Ok(local_pdu) = self.services.timeline.get_pdu(&next_id).await {
-					trace!("Found {next_id} in db");
+					trace!(elapsed=?start.elapsed(),"Found {next_id} in db");
 					let mut obj = local_pdu.into_canonical_object();
 					obj.remove("event_id");
 					seeded_events.insert(next_id.clone(), obj);
@@ -367,11 +376,11 @@ impl super::Service {
 				}
 				let attempts = seen.get(&*next_id).copied().unwrap_or_default();
 				if attempts >= 5 {
-					debug_error!(%attempts, %next_id, "Could not fetch missing event after 5 attempts, giving up");
+					debug_error!(elapsed=?start.elapsed(),%attempts, %next_id, "Could not fetch missing event after 5 attempts, giving up");
 					continue;
 				}
 
-				debug!("Fetching {next_id} over federation");
+				debug!(elapsed=?start.elapsed(),"Fetching {next_id} over federation");
 				let futures = candidates
 					.iter()
 					.map(|remote| {
@@ -385,13 +394,14 @@ impl super::Service {
 				let (event_id, value) = match select_ok(futures).await {
 					| Ok((x, _)) => x,
 					| Err(e) => {
-						warn!("failed to fetch missing event {next_id} from any candidate: {e}");
+						warn!(elapsed=?start.elapsed(),"failed to fetch missing event {next_id} from any candidate: {e}");
 						continue;
 					},
 				};
 				let auth_events =
 					match expect_event_id_array(&value, "auth_events").map_err(|e| {
 						err!(Request(BadJson(warn!(
+							elapsed=?start.elapsed(),
 							%event_id,
 							"Failed to parse event fetched from remote: {e}"
 						))))
@@ -399,6 +409,7 @@ impl super::Service {
 						| Ok(auth_events) => auth_events,
 						| Err(e) => {
 							warn!(
+								elapsed=?start.elapsed(),
 								?e,
 								"event {event_id} is malformed (bad auth_events), skipping"
 							);
@@ -408,17 +419,17 @@ impl super::Service {
 				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");
+						trace!(elapsed=?start.elapsed(),"Found auth event {next_id} in db");
 						let mut obj = local_pdu.into_canonical_object();
 						obj.remove("event_id");
 						seeded_events.insert(id.clone(), obj);
 						continue;
 					}
 					if seeded_events.contains_key(&auth_event_id) {
-						trace!(%auth_event_id, "Already found auth event");
+						trace!(elapsed=?start.elapsed(),%auth_event_id, "Already found auth event");
 						continue;
 					}
-					debug!("Missing auth event {auth_event_id} for event {next_id}");
+					debug!(elapsed=?start.elapsed(),"Missing auth event {auth_event_id} for event {next_id}");
 					seen.insert(auth_event_id.clone(), attempts.saturating_add(1));
 					todo.push_back(auth_event_id);
 					have_all_auth = false;
@@ -427,10 +438,10 @@ impl super::Service {
 				// all of its auth events have been fetched.
 				// TODO: This may result in infinite looping, needs a breaker
 				if have_all_auth {
-					debug!(%next_id, "Have all auth events");
+					debug!(elapsed=?start.elapsed(),%next_id, "Have all auth events");
 					seeded_events.insert(next_id, value);
 				} else {
-					debug_warn!(
+					debug_warn!(elapsed=?start.elapsed(),
 						"Fetched {next_id} but missing some auth events, will have to re-fetch."
 					);
 					seen.insert(next_id.clone(), attempts.saturating_add(1));
@@ -450,7 +461,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();
-			debug_info!("Handling missing event {id} as outlier");
+			debug_info!(elapsed=?start.elapsed(),"Handling missing event {id} as outlier");
 			assert_eq!(pdu_json.get("event_id"), None, "pdu_json had event_id");
 			match Box::pin(self.handle_outlier_pdu(
 				origin,
@@ -463,9 +474,11 @@ impl super::Service {
 			.await
 			{
 				| Ok((pdu, _)) => {
+					trace!(elapsed=?start.elapsed(), "Persisted {id}");
 					let _ = pdus.insert(id, pdu);
 				},
-				| Err(e) => warn!("Authentication of event {id} failed: {e:?}"),
+				| Err(e) =>
+					warn!(elapsed=?start.elapsed(),"Authentication of event {id} failed: {e:?}"),
 			}
 
 			// TODO: should this try to promote to timeline?
@@ -473,7 +486,7 @@ impl super::Service {
 			// now.
 		}
 
-		trace!("Fetched and handled {} missing PDUs", pdus.len());
+		trace!(elapsed=?start.elapsed(),"Finished fetch_and_handle_missing_events: fetched and handled {} missing PDUs", pdus.len());
 		pdus
 	}
 }

src/service/rooms/event_handler/fetch_prev.rs

@@ -1,7 +1,7 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, time::Instant};
 
 use conduwuit::{
-	Event, PduEvent, debug, debug_info, debug_warn,
+	Event, PduEvent, debug, debug_info, debug_warn, trace,
 	utils::{BoolExt, IterStream, stream::BroadbandExt},
 };
 use futures::StreamExt;
@@ -19,6 +19,7 @@ impl super::Service {
 		incoming_pdu: &PduEvent,
 		origin: &ServerName,
 	) -> conduwuit::Result<()> {
+		let start = Instant::now();
 		let missing = incoming_pdu
 			.prev_events()
 			.stream()
@@ -33,10 +34,10 @@ impl super::Service {
 			.collect::<Vec<_>>()
 			.await;
 		if missing.is_empty() {
-			debug!(event_id=%incoming_pdu.event_id(), "No missing prev events.");
+			debug!(elapsed=?start.elapsed(), event_id=%incoming_pdu.event_id(), "No missing prev events.");
 			return Ok(());
 		}
-		debug!(%room_id, event_id=%incoming_pdu.event_id(), ?missing, "Fetching previous events");
+		debug!(elapsed=?start.elapsed(), %room_id, event_id=%incoming_pdu.event_id(), ?missing, "Fetching previous events");
 		let tail = self
 			.services
 			.state
@@ -62,7 +63,7 @@ impl super::Service {
 					),
 			)
 			.await?;
-		debug_info!("Fetched {} missing events", gapfilled.len());
+		debug_info!(elapsed=?start.elapsed(), "Fetched {} missing events", gapfilled.len());
 
 		// Persist all fetched events
 		let mapped = gapfilled
@@ -80,9 +81,12 @@ impl super::Service {
 			build_local_dag(&mapped).await?
 		};
 
+		let job_start = Instant::now();
+		trace!("Starting to persist {} prev events", to_persist.len());
 		for event_id in to_persist {
 			debug_info!("Persisting fetched prev event {event_id}");
 			let obj = mapped.get(&event_id).cloned().unwrap();
+			let persist_start = Instant::now();
 			match self
 				.handle_outlier_pdu(origin, create_event, &event_id, room_id, obj, false)
 				.await
@@ -91,16 +95,21 @@ impl super::Service {
 					self.upgrade_outlier_to_timeline_pdu(pdu, val, create_event, origin, room_id)
 						.await
 						.inspect_err(|e| {
-							debug_warn!("Failed to upgrade prev event {event_id}: {e}");
+							debug_warn!(total_elapsed=?start.elapsed(), job_elapsed=?job_start.elapsed(), task_elapsed=?persist_start.elapsed(), "Failed to upgrade prev event {event_id}: {e}");
 						})
 						.ok();
 				},
-				| Err(e) => debug_warn!("Failed to persist prev event {event_id}: {e}"),
+				| Err(e) =>
+					debug_warn!(total_elapsed=?start.elapsed(), job_elapsed=?job_start.elapsed(), task_elapsed=?persist_start.elapsed(), "Failed to persist prev event {event_id}: {e}"),
 			}
 		}
 
 		// NOTE because i keep forgetting: the caller persists incoming_pdu.
 		// we only care about its prev events
+		trace!(
+			total_elapsed=?start.elapsed(),
+			persist_elapsed=?job_start.elapsed(),
+		);
 		Ok(())
 	}
 }

src/service/rooms/event_handler/fetch_state.rs

@@ -1,6 +1,7 @@
 use std::{
 	cmp::max,
 	collections::{HashMap, hash_map},
+	time::Instant,
 };
 
 use conduwuit::{
@@ -34,6 +35,7 @@ impl super::Service {
 		room_id: &RoomId,
 		event_id: &EventId,
 	) -> Result<Option<HashMap<u64, OwnedEventId>>> {
+		let start = Instant::now();
 		trace!(%origin, "Asking remote for state_ids");
 		let res: get_room_state_ids::v1::Response = match self
 			.services
@@ -43,8 +45,9 @@ impl super::Service {
 				get_room_state_ids::v1::Request::new(event_id.to_owned(), room_id.to_owned()),
 			)
 			.await
-			.inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))
-		{
+			.inspect_err(
+				|e| debug_warn!(elapsed=?start.elapsed(), "Fetching state for event failed: {e}"),
+			) {
 			| Ok(resp) => Ok(resp),
 			| Err(e) =>
 				if e.is_not_found() {
@@ -58,7 +61,7 @@ impl super::Service {
 				},
 		}?;
 
-		debug!(events = res.pdu_ids.len(), "Fetching state events");
+		debug!(elapsed=?start.elapsed(), events = res.pdu_ids.len(), "Fetching state events");
 		let mut state_events: HashMap<OwnedEventId, PduEvent> =
 			HashMap::with_capacity(res.pdu_ids.len());
 		let to_fetch: Vec<OwnedEventId> = res
@@ -76,7 +79,7 @@ impl super::Service {
 			.collect()
 			.await;
 		if to_fetch.is_empty() {
-			debug!("All required state events are already known.");
+			debug!(elapsed=?start.elapsed(), "All required state events are already known.");
 			state_events = res
 				.pdu_ids
 				.iter()
@@ -103,6 +106,7 @@ impl super::Service {
 				// Since this endpoint might fail in huge rooms, we fall back to atomic fetch
 				// anyway.
 				debug_warn!(
+					elapsed=?start.elapsed(),
 					%missing_count,
 					%total_count,
 					"Fetching full state from remote server for event"
@@ -114,6 +118,7 @@ impl super::Service {
 					| Ok(state) => state,
 					| Err(e) => {
 						error!(
+							elapsed=?start.elapsed(),
 							error=?e,
 							%origin,
 							"Failed to fetch full state from remote, falling back to atomic fetch"
@@ -130,6 +135,7 @@ impl super::Service {
 				state_events.extend(fetched_state);
 			} else {
 				debug!(
+					elapsed=?start.elapsed(),
 					to_fetch = to_fetch.len(),
 					"Fetching missing events for state from remote"
 				);
@@ -142,7 +148,7 @@ impl super::Service {
 
 		let mut state: HashMap<ShortStateKey, OwnedEventId> =
 			HashMap::with_capacity(state_events.len());
-		debug!(events = state_events.len(), "Processing state events");
+		debug!(elapsed=?start.elapsed(), events = state_events.len(), "Processing state events");
 		for (event_id, pdu) in state_events {
 			let state_key = pdu.state_key().ok_or_else(|| {
 				err!(Request(BadJson("Found non-state pdu in state events: {event_id}")))
@@ -169,7 +175,7 @@ impl super::Service {
 				},
 			}
 		}
-
+		trace!(elapsed=?start.elapsed(), "fetch_state finished");
 		Ok(Some(state))
 	}
 
@@ -213,6 +219,7 @@ impl super::Service {
 		room_id: &RoomId,
 		event_id: &EventId,
 	) -> Result<HashMap<OwnedEventId, PduEvent>> {
+		let start = Instant::now();
 		trace!("Fetching full state from remote server");
 		let res: get_room_state::v1::Response = self
 			.services
@@ -223,7 +230,7 @@ impl super::Service {
 			)
 			.await
 			.inspect_err(|e| debug_warn!("Fetching state for event failed: {e}"))?;
-		debug!(count = res.auth_chain.len(), "Handling incoming auth chain...");
+		debug!(elapsed=?start.elapsed(), count = res.auth_chain.len(), "Handling incoming auth chain...");
 		res.auth_chain
 			.iter()
 			.stream()
@@ -260,8 +267,8 @@ impl super::Service {
 				},
 			)
 			.await;
-		debug!(count = res.pdus.len(), "Handling incoming state PDUs...");
-		Ok(res
+		debug!(elapsed=?start.elapsed(), count = res.pdus.len(), "Handling incoming state PDUs...");
+		let r = res
 			.pdus
 			.iter()
 			.stream()
@@ -287,6 +294,7 @@ impl super::Service {
 					.await
 					.inspect_err(|e| {
 						warn!(
+							elapsed=?start.elapsed(),
 							%incoming_room_id,
 							%incoming_event_id,
 							?e,
@@ -300,6 +308,8 @@ impl super::Service {
 				acc.insert(event.event_id().to_owned(), event);
 				acc
 			})
-			.await)
+			.await;
+		trace!(elapsed=?start.elapsed(), "fetch_full_state finished");
+		Ok(r)
 	}
 }