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

@@ -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)
+}

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.",))),

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<QueueItem>, // Events we want to send: event and full key
 		statuses: &mut CurTransactionStatus,
 	) -> Result<Option<Vec<SendingEvent>>> {
-		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<SendingEvent>) -> 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<Raw<...>>
-		// serde_json::from_str::<Raw<_>>(
-		//     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")
 	}