feat: Refuse to send requests to unhealthy servers

Also marks them as online if a request succeeds

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,12 @@ 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) => {
+				if error.is_connect() {
+					self.hit_unhealthy(dest.to_owned());
+				}
+				Err(handle_error(actual, &method, &url, error).expect_err("always returns error"))
+			},
 		}
 	}
 
@@ -237,6 +243,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

@@ -2,12 +2,17 @@
 
 use std::{collections::HashMap, sync::Arc, time::Duration};
 
+use assign::assign;
 use conduwuit::{
-	Result, Server, SyncRwLock,
+	Error, Result, Server, SyncRwLock,
 	utils::{millis_since_unix_epoch, time::exponential_backoff::min_exp_backoff_duration},
 };
 pub(crate) use execute::FederationPathBuilderInput;
-use ruma::{OwnedServerName, ServerName};
+use http::StatusCode;
+use ruma::{
+	OwnedServerName, ServerName,
+	api::error::{ErrorKind, LimitExceededErrorData, RetryAfter},
+};
 
 use crate::{Dep, client, moderation, server_keys};
 
@@ -106,4 +111,26 @@ pub fn mark_healthy(&self, server_name: &ServerName) {
 		let mut map = self.remote_health.write();
 		map.remove(server_name);
 	}
+
+	/// 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)",
+					retry_after.as_secs()
+				)
+				.into(),
+				StatusCode::TOO_MANY_REQUESTS,
+			))
+		}
+	}
 }