From 614f8adb48fd057bae5357298ad155afe08650ea Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 10:10:37 +0100 Subject: [PATCH] feat: Add more timing info Shots in the dark <3 --- src/service/sending/sender.rs | 140 +++++++++++++++++++++++++++------- 1 file changed, 112 insertions(+), 28 deletions(-) diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index be5abf7d5..d783cd21c 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -171,8 +171,21 @@ impl Service { futures: &mut SendingFutures<'a>, statuses: &mut CurTransactionStatus, ) { + let monitor = self.new_progress_monitor( + format!("Handle OK response from {dest:?}"), + Duration::from_secs(70), + ); + defer! {{ monitor.notify_one(); }} + let start = Instant::now(); let _cork = self.db.db.cork(); self.db.delete_all_active_requests_for(dest).await; + if start.elapsed().as_secs() > 1 { + warn!( + elapsed=?start.elapsed(), + "Deleting all active requests for {dest:?} took a long time" + ); + } + let start = Instant::now(); // Find events that have been added since starting the last request let new_events = self @@ -181,6 +194,12 @@ impl Service { .take(DEQUEUE_LIMIT) .collect::>() .await; + if start.elapsed().as_secs() > 1 { + warn!( + elapsed=?start.elapsed(), + "Fetching queued requests for {dest:?} took a long time" + ); + } // Insert any pdus we found if !new_events.is_empty() { @@ -204,6 +223,11 @@ impl Service { futures: &mut SendingFutures<'a>, statuses: &mut CurTransactionStatus, ) { + let monitor = self.new_progress_monitor( + format!("Handle send request {:?}", msg.queue_id), + Duration::from_secs(70), + ); + defer! {{ monitor.notify_one(); }} let iv = vec![(msg.queue_id, msg.event)]; if let Ok(Some(events)) = self.select_events(&msg.dest, iv, statuses).await { if !events.is_empty() { @@ -260,7 +284,10 @@ impl Service { futures: &mut SendingFutures<'a>, statuses: &mut CurTransactionStatus, ) { - let monitor = self.new_progress_monitor(format!("Startup netburst on shard {id}")); + let monitor = self.new_progress_monitor( + format!("Startup netburst on shard {id}"), + Duration::from_mins(2), + ); let keep = usize::try_from(self.server.config.startup_netburst_keep).unwrap_or(usize::MAX); let mut txns = HashMap::>::new(); @@ -322,11 +349,16 @@ impl Service { // Must retry any previous transaction for this remote. if retry { + let active_requests_task = self.new_progress_monitor( + format!("Fetch previous transactions for {dest:?}"), + Duration::from_secs(1), + ); self.db .active_requests_for(dest) .ready_for_each(|(_, e)| events.push(e)) .await; - + active_requests_task.notify_one(); + info!(?dest, "Attempting to retry destination with {} previous events", events.len()); return Ok(Some(events)); } @@ -392,6 +424,11 @@ impl Service { /// Selects a number of EDUs to send to the target. #[tracing::instrument(name = "edus", level = "debug", skip_all)] async fn select_edus(&self, server_name: &ServerName) -> Result<(EduVec, u64)> { + let monitor = self.new_progress_monitor( + format!("Select EDUs for {server_name}"), + Duration::from_secs(1), + ); + defer! {{ monitor.notify_one(); }} // selection window let since = self.db.get_latest_educount(server_name).await; let since_upper = self.services.globals.current_count()?; @@ -440,6 +477,11 @@ impl Service { max_edu_count: &AtomicU64, events_len: &AtomicUsize, ) -> EduVec { + let monitor = self.new_progress_monitor( + format!("Select device list update EDUs for {server_name}"), + Duration::from_secs(1), + ); + defer! {{ monitor.notify_one(); }} let mut events = EduVec::new(); let server_rooms = self.services.state_cache.server_rooms(server_name); @@ -497,6 +539,11 @@ impl Service { since: (u64, u64), max_edu_count: &AtomicU64, ) -> Option { + let monitor = self.new_progress_monitor( + format!("Select read receipt EDUs for {server_name}"), + Duration::from_secs(1), + ); + defer! {{ monitor.notify_one(); }} let mut num = 0; let receipts: BTreeMap = self .services @@ -603,6 +650,11 @@ impl Service { since: (u64, u64), max_edu_count: &AtomicU64, ) -> Option { + let monitor = self.new_progress_monitor( + format!("Select presence EDUs for {server_name}"), + Duration::from_secs(1), + ); + defer! {{ monitor.notify_one(); }} let presence_since = self.services.presence.presence_since(since.0); pin_mut!(presence_since); @@ -686,11 +738,15 @@ impl Service { /// Creates a new progress monitor with the given name, returning a marker /// that can be used to mark the progress monitor as finished. - fn new_progress_monitor(&self, task_name: String) -> Arc { + fn new_progress_monitor( + &self, + task_name: String, + tolerance: Duration, + ) -> Arc { let marker = Arc::new(tokio::sync::Notify::new()); let monitor_marker = marker.clone(); self.server.runtime().spawn(async move { - progress_monitor(&task_name, monitor_marker).await; + progress_monitor(&task_name, tolerance, monitor_marker).await; }); marker } @@ -708,11 +764,10 @@ impl Service { id: String, events: Vec, ) -> SendingResult { - let monitor = self.new_progress_monitor(format!( - "Send {} events to appservice {}", - events.len(), - id - )); + let monitor = self.new_progress_monitor( + format!("Send {} events to appservice {}", events.len(), id), + Duration::from_secs(5), + ); let Some(appservice) = self.services.appservice.get_registration(&id).await else { monitor.notify_one(); return Err(( @@ -786,10 +841,10 @@ impl Service { pushkey: String, events: Vec, ) -> SendingResult { - let monitor = self.new_progress_monitor(format!( - "Send {} events to pusher {pushkey} for {user_id}", - events.len(), - )); + let monitor = self.new_progress_monitor( + format!("Send {} events to pusher {pushkey} for {user_id}", events.len()), + Duration::from_secs(5), + ); let Ok(pusher) = self.services.pusher.get_pusher(&user_id, &pushkey).await else { monitor.notify_one(); return Err(( @@ -859,16 +914,18 @@ impl Service { server: OwnedServerName, events: Vec, ) -> SendingResult { - let master_monitor = self.new_progress_monitor(format!( - "Dispatch {} events to {server} (overall)", - events.len() - )); + let master_monitor = self.new_progress_monitor( + format!("Dispatch {} events to {server} (overall)", events.len()), + Duration::from_secs(70), + ); defer! {{ master_monitor.notify_one(); }} - let monitor = - self.new_progress_monitor(format!("Get {} events to send to {server}", events.len())); + let monitor = self.new_progress_monitor( + format!("Get {} events to send to {server}", events.len()), + Duration::from_secs(5), + ); let pdus: Vec<_> = events .iter() .filter_map(|pdu| match pdu { @@ -920,12 +977,15 @@ impl Service { "Sending transaction to remote" ); let start = Instant::now(); - let monitor = self.new_progress_monitor(format!( - "Send {} events ({} PDUs, {} EDUs) to {server}", - events.len(), - request.pdus.len(), - request.edus.len(), - )); + let monitor = self.new_progress_monitor( + format!( + "Send {} events ({} PDUs, {} EDUs) to {server}", + events.len(), + request.pdus.len(), + request.edus.len(), + ), + Duration::from_mins(1), + ); let result = self .services .federation @@ -983,7 +1043,11 @@ impl Service { /// seconds if the progress monitor is not marked as finished via the /// marker. /// Progress monitoring starts immediately. -async fn progress_monitor(task_name: &str, marker: Arc) { +async fn progress_monitor( + task_name: &str, + tolerance: Duration, + marker: Arc, +) { // TODO: this might be useful in more places than the sender service. // Only temporary for diagnostic purposes rn let start = Instant::now(); @@ -992,21 +1056,41 @@ async fn progress_monitor(task_name: &str, marker: Arc) { select! { _ = interval.tick() => { let elapsed = start.elapsed(); - if elapsed.as_secs() > 60 { + if elapsed > tolerance { warn!( %task_name, ?elapsed, + ?tolerance, "Task has been running for a long time (stuck?)" ); } else { trace!( %task_name, ?elapsed, + ?tolerance, "Task is still running" ); } }, - () = marker.notified() => return + () = marker.notified() => { + let elapsed = start.elapsed(); + if elapsed > tolerance { + warn!( + %task_name, + ?elapsed, + ?tolerance, + "Task took a long time (but didn't get stuck)" + ); + } else { + trace!( + %task_name, + ?elapsed, + ?tolerance, + "Task finished" + ); + } + break; + } } } }