From e4922814b9be9834d18bd1a78e1aedbc96c324a9 Mon Sep 17 00:00:00 2001 From: timedout Date: Fri, 26 Jun 2026 08:48:43 +0100 Subject: [PATCH] refactor: Add better documentation to sender service, add additional logging traces This is a surprise tool that will help us later! --- src/core/utils/time/exponential_backoff.rs | 9 ++ src/service/sending/antispam.rs | 26 +++-- src/service/sending/sender.rs | 116 +++++++++++++-------- 3 files changed, 99 insertions(+), 52 deletions(-) diff --git a/src/core/utils/time/exponential_backoff.rs b/src/core/utils/time/exponential_backoff.rs index 682c2592e..76f0be2b2 100644 --- a/src/core/utils/time/exponential_backoff.rs +++ b/src/core/utils/time/exponential_backoff.rs @@ -27,3 +27,12 @@ pub fn continue_exponential_backoff( let min = cmp::min(min, max); elapsed < min } + +/// Determines the minimum number of backoff seconds +#[must_use] +pub fn min_exp_backoff_duration(min: u64, max: u64, retries: u32) -> Duration { + let min = Duration::from_secs(min) + .saturating_mul(retries) + .saturating_mul(retries); + Duration::from_secs(max).min(min) +} diff --git a/src/service/sending/antispam.rs b/src/service/sending/antispam.rs index 9cc1b52c3..7aa1b248f 100644 --- a/src/service/sending/antispam.rs +++ b/src/service/sending/antispam.rs @@ -1,7 +1,7 @@ -use std::{borrow::Cow, fmt::Debug, mem}; +use std::{borrow::Cow, fmt::Debug, mem, time::Instant}; use bytes::BytesMut; -use conduwuit::{Err, Result, debug_error, err, utils, utils::response::LimitReadExt, warn}; +use conduwuit::{Err, Result, debug_info, err, utils, utils::response::LimitReadExt, warn}; use reqwest::Client; use ruma::api::{ IncomingResponse, OutgoingRequest, @@ -32,10 +32,20 @@ where .map(BytesMut::freeze); let reqwest_request = reqwest::Request::try_from(http_request)?; + let method = reqwest_request.method().clone(); + let url = reqwest_request.url().clone(); + debug_info!("Sending request to appservice: {} {}", method, url); + let start = Instant::now(); let mut response = client.execute(reqwest_request).await.map_err(|e| { - warn!("Could not send request to antispam: {e:?}"); - e + err!(BadServerResponse(error!(?e, "Failed to contact antispam service."))) })?; + debug_info!( + "Received response (HTTP {}) from antispam service in {:?}: {} {}", + response.status(), + start.elapsed().as_millis(), + method, + url, + ); // reqwest::Response -> http::Response conversion let status = response.status(); @@ -49,10 +59,14 @@ where .expect("http::response::Builder is usable"), ); - let body = response.limit_read(65535).await?; // TODO: handle timeout + let body = response.limit_read(65535).await.map_err(|e| { + err!(BadServerResponse(error!( + ?e, + "Failed to read response body from antispam service." + ))) + })?; // TODO: handle timeout if !status.is_success() { - debug_error!("Antispam response bytes: {:?}", utils::string_from_bytes(&body)); return match status { | http::StatusCode::FORBIDDEN => Err!(Request(Forbidden("Request was rejected by antispam service.",))), diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 21d5c5547..0c44d66d9 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -1,6 +1,7 @@ use std::{ collections::{BTreeMap, HashMap, HashSet}, fmt::Debug, + future::ready, sync::{ Arc, atomic::{AtomicU64, AtomicUsize, Ordering}, @@ -9,7 +10,10 @@ use std::{ }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; -use conduwuit::info; +use conduwuit::{ + debug_error, debug_info, debug_warn, info, + utils::time::exponential_backoff::min_exp_backoff_duration, +}; use conduwuit_core::{ Error, Event, Result, at, debug, err, error, result::LogErr, @@ -80,9 +84,11 @@ impl Service { let mut statuses: CurTransactionStatus = CurTransactionStatus::new(); let mut futures: SendingFutures<'_> = FuturesUnordered::new(); + info!(shard = id, "Running startup netburst"); self.startup_netburst(id, &mut futures, &mut statuses) .boxed() .await; + info!(shard = id, "Finished startup netburst"); self.work_loop(id, &mut futures, &mut statuses).await; @@ -127,6 +133,8 @@ impl Service { } } + /// Handles a response, calling the appropriate sub-handler depending on if + /// the request was successful or not. #[tracing::instrument(name = "response", level = "debug", skip_all)] async fn handle_response<'a>( &'a self, @@ -140,6 +148,7 @@ impl Service { } } + /// Handles a response error, incrementing the backoff factor. fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) { debug!(dest = ?dest, "{e:?}"); statuses.entry(dest).and_modify(|e| { @@ -154,11 +163,11 @@ impl Service { }); } - #[allow(clippy::needless_pass_by_ref_mut)] + /// Handles a successful response, queueing up more sends if necessary. async fn handle_response_ok<'a>( &'a self, dest: &Destination, - futures: &mut SendingFutures<'a>, + futures: &SendingFutures<'a>, statuses: &mut CurTransactionStatus, ) { let _cork = self.db.db.cork(); @@ -183,12 +192,14 @@ impl Service { } } - #[allow(clippy::needless_pass_by_ref_mut)] + /// Handles sending a new request by queuing up any new events. If there are + /// no new events to send the destination, it is removed from the status map + /// instead. #[tracing::instrument(name = "request", level = "debug", skip_all)] async fn handle_request<'a>( &'a self, msg: Msg, - futures: &mut SendingFutures<'a>, + futures: &SendingFutures<'a>, statuses: &mut CurTransactionStatus, ) { let iv = vec![(msg.queue_id, msg.event)]; @@ -201,6 +212,8 @@ impl Service { } } + /// Cleans up dangling futures by waiting for them to complete, aborting + /// early if the shutdown timeout is reached. #[tracing::instrument( name = "finish", level = "info", @@ -218,7 +231,7 @@ impl Service { let now = Instant::now(); let deadline = now.checked_add(timeout).unwrap_or(now); loop { - trace!("Waiting for {} requests to complete...", futures.len()); + info!("Waiting for {} requests to complete...", futures.len()); select! { () = sleep_until(deadline) => return, response = futures.next() => match response { @@ -230,6 +243,8 @@ impl Service { } } + /// Performs startup netburst, which sends any previously queued up but not + /// yet sent requests immediately. #[tracing::instrument( name = "netburst", level = "debug", @@ -255,7 +270,12 @@ impl Service { let entry = txns.entry(dest.clone()).or_default(); if self.server.config.startup_netburst_keep >= 0 && entry.len() >= keep { - warn!("Dropping unsent event {dest:?} {:?}", String::from_utf8_lossy(&key)); + warn!( + startup_netburst_keep = self.server.config.startup_netburst_keep, + queue_size = entry.len(), + "Dropping unsent event to {dest:?}: {:?}", + String::from_utf8_lossy(&key) + ); self.db.delete_active_request(&key); } else { entry.push(event); @@ -270,8 +290,9 @@ impl Service { } } + /// Selects any new events to send to the given destination. #[tracing::instrument( - name = "select",, + name = "select", level = "debug", skip_all, fields( @@ -285,7 +306,7 @@ impl Service { new_events: Vec, // Events we want to send: event and full key statuses: &mut CurTransactionStatus, ) -> Result>> { - let (allow, retry) = self.select_events_current(dest, statuses)?; + let (allow, retry) = self.should_attempt_send(dest, statuses)?; // Nothing can be done for this remote, bail out. if !allow { @@ -328,29 +349,34 @@ impl Service { Ok(Some(events)) } - fn select_events_current( + /// Determines if the destination should be retried, honouring exponential + /// backoff. + fn should_attempt_send( &self, dest: &Destination, statuses: &mut CurTransactionStatus, ) -> Result<(bool, bool)> { let (mut allow, mut retry) = (true, false); statuses - .entry(dest.clone()) // TODO: can we avoid cloning? + .entry(dest.clone()) .and_modify(|e| match e { - TransactionStatus::Failed(tries, time) => { + | TransactionStatus::Failed(tries, time) => { // Fail if a request has failed recently (exponential backoff) let min = self.server.config.sender_timeout; let max = self.server.config.sender_retry_backoff_limit; if continue_exponential_backoff_secs(min, max, time.elapsed(), *tries) && !matches!(dest, Destination::Appservice(_)) { + let retry_after = min_exp_backoff_duration(min, max, *tries); + debug_warn!("Not retrying destination for another {retry_after:?}"); allow = false; } else { retry = true; + debug_info!("Will retry destination ({} tries)", *tries); *e = TransactionStatus::Retrying(*tries); } }, - TransactionStatus::Running | TransactionStatus::Retrying(_) => { + | TransactionStatus::Running | TransactionStatus::Retrying(_) => { allow = false; // already running }, }) @@ -359,11 +385,8 @@ impl Service { Ok((allow, retry)) } - #[tracing::instrument( - name = "edus",, - level = "debug", - skip_all, - )] + /// 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)> { // selection window let since = self.db.get_latest_educount(server_name).await; @@ -640,8 +663,14 @@ impl Service { Some(buf) } + /// Actually sends fetched events to the destination. + /// If the events vec is empty, a stub response is returned, and a debug + /// warning is logged. fn send_events(&self, dest: Destination, events: Vec) -> SendingFuture<'_> { - debug_assert!(!events.is_empty(), "sending empty transaction"); + if events.is_empty() { + debug_warn!(?dest, "Attempted to send empty transaction!"); + return std::future::ready(Ok(dest)).boxed(); + } match dest { | Destination::Federation(server) => self.send_events_dest_federation(server, events).boxed(), @@ -708,9 +737,6 @@ impl Service { let txn_id = &*URL_SAFE_NO_PAD.encode(txn_hash); - //debug_assert!(pdu_jsons.len() + edu_jsons.len() > 0, "sending empty - // transaction"); - let mut request = ruma::api::appservice::event::push_events::v1::Request::new(txn_id.into(), pdu_jsons); request.ephemeral = edu_jsons; @@ -843,12 +869,28 @@ impl Service { ); request.pdus = pdus; request.edus = edus; + let pdu_count = request.pdus.len(); + let edu_count = request.edus.len(); + debug_info!( + %txn_id, + pdus=pdu_count, + edus=edu_count, + "Sending transaction to remote" + ); + let start = Instant::now(); let result = self .services .federation .execute_signed(&self.services.client.sender, &server, request) .await; + debug_info!( + %txn_id, + pdus=pdu_count, + edus=edu_count, + elapsed=?start.elapsed(), + "Finished sending transaction" + ); for (event_id, result) in result.iter().flat_map(|resp| resp.pdus.iter()) { if let Err(e) = result { @@ -868,7 +910,8 @@ impl Service { } } - /// This does not return a full `Pdu` it is only to satisfy ruma's types. + /// Converts and sanitises an outgoing PDU object for federation + /// transmission. pub async fn convert_to_outgoing_federation_event( &self, mut pdu_json: CanonicalJsonObject, @@ -877,31 +920,12 @@ impl Service { .get_mut("unsigned") .and_then(|val| val.as_object_mut()) { + // TODO: remove all unsigned data over federation. unsigned.remove("transaction_id"); } - - // room v3 and above removed the "event_id" field from remote PDU format - if let Some(room_id) = pdu_json - .get("room_id") - .and_then(|val| RoomId::parse(val.as_str()?).ok()) - { - match self.services.state.get_room_version(&room_id).await { - | Ok(room_version_id) => match room_version_id { - | RoomVersionId::V1 | RoomVersionId::V2 => {}, - | _ => _ = pdu_json.remove("event_id"), - }, - | Err(_) => _ = pdu_json.remove("event_id"), - } - } else { - pdu_json.remove("event_id"); - } - - // TODO: another option would be to convert it to a canonical string to validate - // size and return a Result> - // serde_json::from_str::>( - // ruma::serde::to_canonical_json_string(pdu_json).expect("CanonicalJson is - // valid serde_json::Value"), ) - // .expect("Raw::from_value always works") + // We don't support any room versions that have the event_id in the PDU JSON, + // so we can safely remove it here. + pdu_json.remove("event_id"); to_raw_value(&pdu_json).expect("CanonicalJson is valid serde_json::Value") }