feat: Server health tracking

conduwuit-example.toml

@@ -554,9 +554,28 @@
 #
 #sender_idle_timeout = 180
 
-# Federation sender transaction retry backoff limit (seconds).
+# Federation sender retry backoff base (seconds).
 #
-#sender_retry_backoff_limit = 86400
+# This period will be doubled for each failed federation request until
+# either the remote server becomes healthy, or the value is clamped to
+# `sender_retry_backoff_limit`.
+#
+#sender_retry_backoff_base = 60
+
+# Federation sender retry backoff limit (seconds).
+#
+# Defaults to one week. Requests will never stop being retried if their
+# backoff period exceeds this value, however the maximum amount of time
+# between each request will instead be clamped at this value.
+#
+# The backoff period is reset if a successful request is made, or
+# continuwuity receives a request from the server that is being backed off
+# from.
+#
+# It is not recommended to lower this value below 48 hours or above
+# 1 year.
+#
+#sender_retry_backoff_limit = 806400
 
 # Appservice URL request connection timeout. Defaults to 35 seconds as
 # generally appservices are hosted within the same network.

src/admin/debug/commands.rs

@@ -1237,4 +1237,60 @@ pub(super) async fn rooms_by_extremity_count(&self, page: Option<usize>) -> Resu
 		))
 		.await
 	}
+
+	pub(super) async fn servers_in_backoff(
+		&self,
+		room_id: Option<OwnedRoomId>,
+		name: Option<OwnedServerName>,
+	) -> Result {
+		let backoff_map = self.services.federation.remote_health();
+
+		if backoff_map.is_empty() {
+			return Err!("No servers in backoff.");
+		}
+
+		let stale_map = self.services.federation.stale_destinations();
+
+		if let Some(specific_server_name) = name {
+			let entry = backoff_map.get(&specific_server_name);
+			return match entry {
+				| Some((retries, next_retry)) =>
+					self.write_str(&format!(
+						"{specific_server_name}: {retries} retries, next retry at {next_retry}"
+					))
+					.await,
+				| None =>
+					self.write_str(&format!("Server {specific_server_name} is not in backoff."))
+						.await,
+			};
+		}
+
+		self.write_str("| Server Name | Retries | Next connection attempt | Is stale |\n")
+			.await?;
+		self.write_str("| ----------- | ------- | ----------------------- | -------- |\n")
+			.await?;
+
+		if let Some(specific_room) = room_id {
+			let mut servers_in_room =
+				self.services.rooms.state_cache.room_servers(&specific_room);
+			while let Some(server_name) = servers_in_room.next().await {
+				let Some((retries, retry_at)) = backoff_map.get(&server_name) else { continue };
+				let is_stale = if stale_map.contains(&server_name) { "Yes" } else { "No" };
+				self.write_str(&format!(
+					"| {server_name} | {retries} | {retry_at} | {is_stale} |\n"
+				))
+				.await?;
+			}
+
+			return Ok(());
+		}
+
+		for (server_name, (retries, retry_at)) in &backoff_map {
+			let is_stale = if stale_map.contains(server_name) { "Yes" } else { "No" };
+			self.write_str(&format!("| {server_name} | {retries} | {retry_at} | {is_stale} |\n"))
+				.await?;
+		}
+
+		Ok(())
+	}
 }

src/admin/debug/mod.rs

@@ -250,6 +250,13 @@ pub enum DebugCommand {
 		page: Option<usize>,
 	},
 
+	ServersInBackoff {
+		#[arg(short, long, alias = "room")]
+		room_id: Option<OwnedRoomId>,
+		#[arg(short, long)]
+		name: Option<OwnedServerName>,
+	},
+
 	/// Developer test stubs
 	#[command(subcommand)]
 	#[allow(non_snake_case)]

src/api/router/auth.rs

@@ -149,6 +149,9 @@ async fn verify<B: AsRef<[u8]> + Sync>(
 					)));
 				}
 
+				// Ping the server as healthy
+				services.federation.mark_healthy(&output.origin);
+
 				Ok(output.origin)
 			},
 			| Err(err) =>

src/core/config/mod.rs

@@ -694,9 +694,30 @@ pub struct Config {
 	#[serde(default = "default_sender_idle_timeout")]
 	pub sender_idle_timeout: u64,
 
-	/// Federation sender transaction retry backoff limit (seconds).
+	/// Federation sender retry backoff base (seconds).
 	///
-	/// default: 86400
+	/// This period will be doubled for each failed federation request until
+	/// either the remote server becomes healthy, or the value is clamped to
+	/// `sender_retry_backoff_limit`.
+	///
+	/// default: 60
+	#[serde(default = "default_sender_retry_backoff_base")]
+	pub sender_retry_backoff_base: u64,
+
+	/// Federation sender retry backoff limit (seconds).
+	///
+	/// Defaults to one week. Requests will never stop being retried if their
+	/// backoff period exceeds this value, however the maximum amount of time
+	/// between each request will instead be clamped at this value.
+	///
+	/// The backoff period is reset if a successful request is made, or
+	/// continuwuity receives a request from the server that is being backed off
+	/// from.
+	///
+	/// It is not recommended to lower this value below 48 hours or above
+	/// 1 year.
+	///
+	/// default: 806400
 	#[serde(default = "default_sender_retry_backoff_limit")]
 	pub sender_retry_backoff_limit: u64,
 
@@ -2980,6 +3001,8 @@ fn default_sender_timeout() -> u64 { 180 }
 
 fn default_sender_idle_timeout() -> u64 { 180 }
 
+fn default_sender_retry_backoff_base() -> u64 { 60 }
+
 fn default_sender_retry_backoff_limit() -> u64 { 86400 }
 
 fn default_appservice_timeout() -> u64 { 35 }

src/core/error/mod.rs

@@ -287,3 +287,7 @@ fn from(err: reqwest::Error) -> Self { Self(err) }
 impl From<reqwest::Error> for Error {
 	fn from(err: reqwest::Error) -> Self { Self::Reqwest(err.into()) }
 }
+
+impl From<FormattedReqwestError> for reqwest::Error {
+	fn from(val: FormattedReqwestError) -> Self { val.0 }
+}

src/core/utils/mod.rs

@@ -38,8 +38,8 @@
 	string::{str_from_bytes, string_from_bytes},
 	sys::compute::available_parallelism,
 	time::{
-		exponential_backoff::{continue_exponential_backoff, continue_exponential_backoff_secs},
-		now_millis as millis_since_unix_epoch, timepoint_ago, timepoint_from_now,
+		exponential_backoff::should_continue_backoff, now_millis as millis_since_unix_epoch,
+		timepoint_ago, timepoint_from_now,
 	},
 };
 

src/core/utils/time/exponential_backoff.rs

@@ -1,38 +1,23 @@
-use std::{cmp, time::Duration};
+use std::time::Duration;
 
-/// Returns false if the exponential backoff has expired based on the inputs
+/// Returns false if the backoff interval has expired based on the inputs,
+/// meaning the operation should be retried.
 #[inline]
 #[must_use]
-pub fn continue_exponential_backoff_secs(
-	min: u64,
-	max: u64,
-	elapsed: Duration,
-	tries: u32,
-) -> bool {
-	let min = Duration::from_secs(min);
-	let max = Duration::from_secs(max);
-	continue_exponential_backoff(min, max, elapsed, tries)
-}
-
-/// Returns false if the exponential backoff has expired based on the inputs
-#[inline]
-#[must_use]
-pub fn continue_exponential_backoff(
+pub fn should_continue_backoff(
 	min: Duration,
 	max: Duration,
 	elapsed: Duration,
 	tries: u32,
 ) -> bool {
-	let min = min.saturating_mul(tries).saturating_mul(tries);
-	let min = cmp::min(min, max);
-	elapsed < min
+	elapsed < next_interval(min, max, tries)
 }
 
-/// Determines the minimum number of backoff seconds
+/// Determines the interval that should be waited before retrying the operation
+/// using the algorithm: `(min * retries).min(max)`.
 #[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)
+#[inline]
+pub fn next_interval(min: Duration, max: Duration, retries: u32) -> Duration {
+	// TODO(nex): jitter?
+	min.saturating_mul(retries).min(max)
 }

src/service/federation/execute.rs

@@ -131,6 +131,8 @@ pub async fn execute_on<'i, T, PathBuilderInput>(
 			))));
 		}
 
+		self.ensure_remote_is_healthy(dest)?;
+
 		let actual = self
 			.services
 			.client
@@ -183,8 +185,25 @@ async fn perform<T>(
 			| Ok(response) =>
 				self.handle_response::<T>(dest, actual, &method, &url, response)
 					.await,
-			| Err(error) =>
-				Err(handle_error(actual, &method, &url, error).expect_err("always returns error")),
+			| Err(error) => {
+				// This awful wrapping hack is required because `reqwest::Error` does not
+				// implement `Clone`, but we need to convert it into a local error type to pass
+				// to should_mark_stale, but handle_error then itself expects the original
+				// reqwest error.
+				// handle_error *could* just take the wrapped error, but it'd have to unwrap it
+				// anyway.
+				let wrapped = Error::Reqwest(error.into());
+				if self.should_mark_stale(&wrapped) {
+					debug_info!("{dest} is unhealthy & stale due to a connect error");
+					self.mark_destination_stale(dest);
+					self.hit_unhealthy(dest.to_owned());
+				}
+				let Error::Reqwest(unwrapped) = wrapped else {
+					unreachable!("wrapped reqwest error must unwrap to a reqwest error");
+				};
+				Err(handle_error(actual, &method, &url, unwrapped.into())
+					.expect_err("always returns error"))
+			},
 		}
 	}
 
@@ -237,6 +256,7 @@ async fn handle_response<T>(
 			parts,
 			body.as_ref(),
 		))
+		.inspect(|_| self.mark_healthy(dest))
 		.map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}")))
 	}
 }

src/service/federation/mod.rs

@@ -1,13 +1,30 @@
 mod execute;
-use std::sync::Arc;
 
-use conduwuit::{Result, Server};
+use std::{
+	collections::{HashMap, HashSet},
+	sync::Arc,
+	time::Duration,
+};
+
+use assign::assign;
+use async_trait::async_trait;
+use conduwuit::{
+	Error, Result, Server, SyncRwLock, debug,
+	utils::{millis_since_unix_epoch, time::exponential_backoff::next_interval},
+};
 pub(crate) use execute::FederationPathBuilderInput;
+use http::StatusCode;
+use ruma::{
+	OwnedServerName, ServerName,
+	api::error::{ErrorKind, LimitExceededErrorData, RetryAfter},
+};
 
 use crate::{Dep, client, moderation, server_keys};
 
 pub struct Service {
 	services: Services,
+	pub remote_health: SyncRwLock<HashMap<OwnedServerName, (u32, u64)>>,
+	pub stale_destinations: SyncRwLock<HashSet<OwnedServerName>>,
 }
 
 struct Services {
@@ -17,6 +34,7 @@ struct Services {
 	moderation: Dep<moderation::Service>,
 }
 
+#[async_trait]
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
@@ -26,8 +44,164 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				server_keys: args.depend::<server_keys::Service>("server_keys"),
 				moderation: args.depend::<moderation::Service>("moderation"),
 			},
+			remote_health: SyncRwLock::new(HashMap::new()),
+			stale_destinations: SyncRwLock::new(HashSet::new()),
 		}))
 	}
 
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
+
+	async fn clear_cache(&self) {
+		let mut map = self.remote_health.write();
+		map.clear();
+	}
+}
+
+impl Service {
+	/// Checks if a remote is "healthy". "Healthy" is defined by either:
+	///
+	/// * The remote has not been marked as having a failed request, OR
+	/// * The next retry timestamp is in the past
+	pub fn is_healthy(&self, server_name: &ServerName) -> bool {
+		let map = self.remote_health.read();
+		let unix_now = millis_since_unix_epoch();
+		if let Some((_, next_retry)) = map.get(server_name) {
+			unix_now >= *next_retry
+		} else {
+			true
+		}
+	}
+
+	/// Returns how long the server should wait before attempting to contact the
+	/// remote again.
+	pub fn retry_after(&self, server_name: &ServerName) -> Option<Duration> {
+		let map = self.remote_health.read();
+		let unix_now = millis_since_unix_epoch();
+		map.get(server_name)
+			.map(|(_, next_retry)| Duration::from_millis((*next_retry).saturating_sub(unix_now)))
+	}
+
+	/// Marks or updates a remote's health status as unhealthy. If the remote is
+	/// not already marked as unhealthy, a new entry is created. Otherwise, the
+	/// retry count is incremented and
+	pub fn hit_unhealthy(&self, server_name: OwnedServerName) {
+		let unix_now = millis_since_unix_epoch();
+		let mut map = self.remote_health.write();
+		let sn2 = server_name.clone(); // for logging since map.entry() moves
+		let (retries, next_retry) = map.entry(server_name).or_default();
+		if *next_retry > unix_now {
+			// Don't update the retry marker if we are already in a backoff
+			// period. This prevents the backoff skyrocketing if multiple
+			// concurrent or closely-related requests fail and consequently try
+			// to mark as offline.
+			return;
+		}
+
+		let min = Duration::from_secs(self.services.server.config.sender_retry_backoff_base);
+		let max = Duration::from_secs(self.services.server.config.sender_retry_backoff_limit);
+
+		*retries = retries.saturating_add(1);
+		*next_retry = unix_now.saturating_add(
+			u64::try_from(next_interval(min, max, *retries).as_millis())
+				.expect("backoff milliseconds should not exceed u64::MAX"),
+		);
+		debug!(
+			"{} is (now) unhealthy ({} retries, blocked until: {})",
+			sn2, *retries, *next_retry
+		);
+	}
+
+	/// Marks a server as "healthy" by removing it from the health map.
+	///
+	/// TODO: flush senders too
+	pub fn mark_healthy(&self, server_name: &ServerName) {
+		// TODO: We need to make sure the sender flush DOESN'T trigger if this is called
+		// by the senders themselves.
+		let mut health_map = self.remote_health.write();
+		if health_map.remove(server_name).is_some() {
+			debug!("{} is now healthy", server_name);
+		}
+		// The lock for remote_health is deliberately retained until the end of the
+		// function to prevent parallel requests from marking the server as healthy
+		// and then immediately marking it as unhealthy again due to the stale cache
+		// we might be about to clear
+		let mut stale_destinations = self.stale_destinations.write();
+		if stale_destinations.remove(server_name) {
+			debug!(
+				"{server_name} is no longer unhealthy but was stale, clearing destination cache \
+				 entry"
+			);
+			self.services
+				.client
+				.matrix_resolver
+				.remove_cache_entry(server_name.as_str());
+		}
+	}
+
+	/// Returns a rate-limited error if the remote is unhealthy.
+	fn ensure_remote_is_healthy(&self, server_name: &ServerName) -> Result<()> {
+		if self.is_healthy(server_name) {
+			Ok(())
+		} else {
+			let retry_after = self
+				.retry_after(server_name)
+				.expect("remote is unhealthy and must have an accompanying retry timestamp");
+			Err(Error::Request(
+				ErrorKind::LimitExceeded(assign!(LimitExceededErrorData::new(), {
+					retry_after: Some(RetryAfter::Delay(retry_after)),
+				})),
+				format!(
+					"Remote server {} is currently unhealthy (not retrying for another {} \
+					 seconds)",
+					server_name,
+					retry_after.as_secs()
+				)
+				.into(),
+				StatusCode::TOO_MANY_REQUESTS,
+			))
+		}
+	}
+
+	/// Marks a destination as stale, which will cause the destination cache to
+	/// be invalidated next time we receive a request FROM that destination.
+	/// This does not inherently mark the remote as "unhealthy".
+	///
+	/// Typically, this should only be done if a connection error is
+	/// encountered, which might indicate that the address of the destination
+	/// is incorrect or has since moved.
+	pub fn mark_destination_stale(&self, server_name: &ServerName) {
+		self.stale_destinations
+			.write()
+			.insert(server_name.to_owned());
+	}
+
+	/// Determines whether a destination should be marked "stale" depending on
+	/// the returned error.
+	pub fn should_mark_stale(&self, error: &Error) -> bool {
+		if let Error::Reqwest(error) = error {
+			if error.is_connect() {
+				return true;
+			}
+		}
+
+		match error.status_code() {
+			// Some special servers account for this specifically
+			| StatusCode::MISDIRECTED_REQUEST
+			// Common error codes observed for misdirected requests
+			// | StatusCode::NOT_FOUND  This one can be encountered naturally
+			| StatusCode::METHOD_NOT_ALLOWED
+			| StatusCode::IM_A_TEAPOT => true,
+			_ => false,
+		}
+	}
+
+	/// Returns a clone of the internal remote health tracking map.
+	pub fn remote_health(&self) -> HashMap<OwnedServerName, (u32, u64)> {
+		self.remote_health.read().clone()
+	}
+
+	/// Returns a clone of the internal stale destinations set.
+	pub fn stale_destinations(&self) -> HashSet<OwnedServerName> {
+		self.stale_destinations.read().clone()
+	}
 }

src/service/sending/mod.rs

@@ -132,11 +132,11 @@ async fn worker(self: Arc<Self>) -> Result {
 
 		while let Some(ret) = senders.join_next_with_id().await {
 			match ret {
-				| Ok((id, _)) => {
-					debug!(?id, "sender worker finished");
+				| Ok((id, error)) => {
+					debug!(?id, ?error, "sender worker finished");
 				},
 				| Err(error) => {
-					error!(id = ?error.id(), ?error, "sender worker finished");
+					error!(id = ?error.id(), ?error, "sender worker failed");
 				},
 			}
 		}

src/service/sending/sender.rs

@@ -10,14 +10,15 @@
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
 use conduwuit::{
-	debug_info, debug_warn, info, utils::time::exponential_backoff::min_exp_backoff_duration,
+	debug_info, debug_warn, info, trace,
+	utils::{should_continue_backoff, time::exponential_backoff::next_interval},
 };
 use conduwuit_core::{
 	Error, Event, Result, at, debug, err, error,
 	matrix::pdu::sticky,
 	result::LogErr,
 	utils::{
-		ReadyExt, calculate_hash, continue_exponential_backoff_secs,
+		ReadyExt, calculate_hash,
 		future::TryExtExt,
 		stream::{BroadbandExt, IterStream, WidebandExt},
 	},
@@ -29,6 +30,7 @@
 	join, pin_mut,
 	stream::FuturesUnordered,
 };
+use http::StatusCode;
 use ruma::{
 	CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedRoomId, OwnedServerName, OwnedUserId,
 	RoomId, ServerName, UInt,
@@ -142,20 +144,41 @@ async fn handle_response<'a>(
 	) {
 		match response {
 			| Ok(dest) => self.handle_response_ok(&dest, futures, statuses).await,
-			| Err((dest, e)) => Self::handle_response_err(dest, statuses, &e),
+			| Err((dest, e)) => {
+				if let Destination::Federation(dest) = &dest {
+					if self.services.federation.should_mark_stale(&e)
+						|| e.status_code() == StatusCode::NOT_FOUND
+					{
+						debug!("{dest} is now unhealthy & stale due to a connect error: {e:?}");
+						self.services.federation.mark_destination_stale(dest);
+						self.services.federation.hit_unhealthy(dest.clone());
+					} else if e.status_code() != StatusCode::OK {
+						debug!("{dest} is now unhealthy due to error response: {e:?}");
+						self.services.federation.hit_unhealthy(dest.clone());
+					}
+				}
+				Self::handle_response_err(dest, statuses, &e);
+			},
 		}
 	}
 
 	/// Handles a response error, incrementing the backoff factor.
-	fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, e: &Error) {
-		debug!(dest = ?dest, "{e:?}");
+	fn handle_response_err(dest: Destination, statuses: &mut CurTransactionStatus, er: &Error) {
+		debug!(dest = ?dest, "{er:?}");
+		let dest2 = dest.clone();
 		statuses.entry(dest).and_modify(|e| {
 			*e = match e {
 				| TransactionStatus::Running => TransactionStatus::Failed(1, Instant::now()),
 				| &mut TransactionStatus::Retrying(ref n) =>
 					TransactionStatus::Failed(n.saturating_add(1), Instant::now()),
 				| TransactionStatus::Failed(..) => {
-					panic!("Request that was not even running failed?!")
+					panic!(
+						"{}",
+						format!(
+							"Request to {dest2:?} that was not even running ({e:?}) failed: \
+							 {er:?}"
+						)
+					)
 				},
 			}
 		});
@@ -171,6 +194,9 @@ async fn handle_response_ok<'a>(
 	) {
 		let _cork = self.db.db.cork();
 		self.db.delete_all_active_requests_for(dest).await;
+		if let Destination::Federation(server_name) = dest {
+			self.services.federation.mark_healthy(server_name);
+		}
 
 		// Find events that have been added since starting the last request
 		let new_events = self
@@ -203,12 +229,22 @@ async fn handle_request<'a>(
 		statuses: &mut CurTransactionStatus,
 	) {
 		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() {
-				futures.push(self.send_events(msg.dest, events));
-			} else {
-				statuses.remove(&msg.dest);
-			}
+		match self.select_events(&msg.dest, iv, statuses).await {
+			| Ok(Some(events)) => {
+				if events.is_empty() {
+					// Nothing more to send to this remote
+					statuses.remove(&msg.dest);
+				} else {
+					futures.push(self.send_events(msg.dest, events));
+				}
+			},
+			| Ok(None) => {
+				trace!(
+					?msg.dest,
+					"Ignoring request to send to destination (nothing to send/already busy)"
+				);
+			},
+			| Err(e) => error!(error=?e, ?msg.dest, "Failed to select events"),
 		}
 	}
 
@@ -383,17 +419,37 @@ fn should_attempt_send(
 		statuses: &mut CurTransactionStatus,
 	) -> Result<(bool, bool)> {
 		let (mut allow, mut retry) = (true, false);
+		if let Destination::Federation(server_name) = dest {
+			let entry = statuses.entry(dest.clone()).and_modify(|e| match e {
+				| TransactionStatus::Running | TransactionStatus::Retrying(_) => {
+					allow = false; // already running
+					trace!("Transaction is already running for {dest:?}");
+				},
+				| TransactionStatus::Failed(tries, _) => {
+					*e = TransactionStatus::Retrying(*tries);
+					retry = true;
+					trace!("Previous transaction failed for {dest:?} ({e:?}), will retry");
+				},
+			});
+			if let Some(retry_after) = self.services.federation.retry_after(server_name) {
+				allow = allow && retry_after.is_zero();
+			}
+			if allow {
+				trace!(current_status=?entry, "Inserting running status for {dest:?}");
+				entry.or_insert(TransactionStatus::Running);
+			}
+			return Ok((allow, retry));
+		}
 		statuses
 			.entry(dest.clone())
 			.and_modify(|e| match e {
 				| 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)
+					let min = Duration::from_secs(self.server.config.sender_retry_backoff_base);
+					let max = Duration::from_secs(self.server.config.sender_retry_backoff_limit);
+					if should_continue_backoff(min, max, time.elapsed(), *tries)
 						&& !matches!(dest, Destination::Appservice(_))
 					{
-						let retry_after = min_exp_backoff_duration(min, max, *tries);
+						let retry_after = next_interval(min, max, *tries);
 						debug_warn!("Not retrying destination for another {retry_after:?}");
 						allow = false;
 					} else {
@@ -446,6 +502,8 @@ async fn select_edus(&self, server_name: &ServerName) -> Result<(EduVec, u64)> {
 		events.extend(presence.into_iter().flatten());
 		events.extend(receipts.into_iter().flatten());
 
+		// TODO(nex): some EDUs like typing need flattening
+
 		Ok((events, max_edu_count.load(Ordering::Acquire)))
 	}
 
@@ -902,6 +960,7 @@ async fn send_events_dest_federation(
 			%txn_id,
 			pdus=pdu_count,
 			edus=edu_count,
+			%server,
 			"Sending transaction to remote"
 		);
 		let start = Instant::now();
@@ -915,6 +974,7 @@ async fn send_events_dest_federation(
 			pdus=pdu_count,
 			edus=edu_count,
 			elapsed=?start.elapsed(),
+			%server,
 			"Finished sending transaction"
 		);