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 fa362609c..bff76a382 100644 --- a/src/service/rooms/event_handler/fetch_and_handle_outliers.rs +++ b/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::{ @@ -132,6 +135,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 @@ -145,6 +149,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() @@ -158,7 +163,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); @@ -166,11 +171,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| { @@ -184,13 +189,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}" )))) @@ -198,6 +204,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" ); @@ -207,17 +214,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; @@ -226,10 +233,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)); @@ -249,7 +256,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, @@ -262,9 +269,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? @@ -272,7 +281,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 } @@ -445,6 +454,7 @@ impl super::Service { via: &ServerName, min_depth: UInt, ) -> conduwuit::Result> { + let start = Instant::now(); #[cfg(debug_assertions)] { let missing_count = head @@ -456,9 +466,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), @@ -486,7 +498,7 @@ impl super::Service { let mut discovered = HashMap::with_capacity(head.prev_events.len()); let mut latest_events = vec![head.event_id().to_owned()]; - debug!( + debug!(elapsed=?start.elapsed(), %room_id, event_id=%head.event_id(), %iteration_limit, @@ -497,7 +509,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, @@ -522,10 +534,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 { @@ -536,7 +548,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 ); @@ -567,7 +579,7 @@ impl super::Service { } if latest_events.is_empty() { - debug!( + debug!(elapsed=?start.elapsed(), %limit, %via, %iteration, @@ -582,7 +594,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, @@ -593,7 +605,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, @@ -607,6 +619,7 @@ impl super::Service { } } + trace!(elapsed=?start.elapsed(), "Finished get_missing_events"); Ok(discovered) } } diff --git a/src/service/rooms/event_handler/fetch_prev.rs b/src/service/rooms/event_handler/fetch_prev.rs index 9cacf2f40..729339e6e 100644 --- a/src/service/rooms/event_handler/fetch_prev.rs +++ b/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::>() .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(()) } } diff --git a/src/service/rooms/event_handler/fetch_state.rs b/src/service/rooms/event_handler/fetch_state.rs index ebfac081c..f399e1696 100644 --- a/src/service/rooms/event_handler/fetch_state.rs +++ b/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>> { + 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 = HashMap::with_capacity(res.pdu_ids.len()); let to_fetch: Vec = 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 = 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> { + 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) } }