diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index d8e10d3e8..a6d90c49f 100644 --- a/src/service/federation/execute.rs +++ b/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( | Ok(response) => self.handle_response::(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( parts, body.as_ref(), )) + .inspect(|_| self.mark_healthy(dest)) .map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}"))) } } diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index e15f7f960..1affa53aa 100644 --- a/src/service/federation/mod.rs +++ b/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, + )) + } + } }