diff --git a/conduwuit-example.toml b/conduwuit-example.toml index 8d476397e..b722280d7 100644 --- a/conduwuit-example.toml +++ b/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. diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index 225874520..2f1c00482 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -1237,4 +1237,60 @@ pub(super) async fn rooms_by_extremity_count(&self, page: Option) -> Resu )) .await } + + pub(super) async fn servers_in_backoff( + &self, + room_id: Option, + name: Option, + ) -> 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(()) + } } diff --git a/src/admin/debug/mod.rs b/src/admin/debug/mod.rs index ead072ccc..016713e04 100644 --- a/src/admin/debug/mod.rs +++ b/src/admin/debug/mod.rs @@ -250,6 +250,13 @@ pub enum DebugCommand { page: Option, }, + ServersInBackoff { + #[arg(short, long, alias = "room")] + room_id: Option, + #[arg(short, long)] + name: Option, + }, + /// Developer test stubs #[command(subcommand)] #[allow(non_snake_case)] diff --git a/src/api/router/auth.rs b/src/api/router/auth.rs index f1450419a..f8f72519c 100644 --- a/src/api/router/auth.rs +++ b/src/api/router/auth.rs @@ -149,6 +149,9 @@ async fn verify + Sync>( ))); } + // Ping the server as healthy + services.federation.mark_healthy(&output.origin); + Ok(output.origin) }, | Err(err) => diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 557ac5e65..3bb9b4326 100644 --- a/src/core/config/mod.rs +++ b/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 } diff --git a/src/core/error/mod.rs b/src/core/error/mod.rs index a4e8c40aa..ad3040fff 100644 --- a/src/core/error/mod.rs +++ b/src/core/error/mod.rs @@ -287,3 +287,7 @@ fn from(err: reqwest::Error) -> Self { Self(err) } impl From for Error { fn from(err: reqwest::Error) -> Self { Self::Reqwest(err.into()) } } + +impl From for reqwest::Error { + fn from(val: FormattedReqwestError) -> Self { val.0 } +} diff --git a/src/core/utils/mod.rs b/src/core/utils/mod.rs index 12baee8f8..e1e143497 100644 --- a/src/core/utils/mod.rs +++ b/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, }, }; diff --git a/src/core/utils/time/exponential_backoff.rs b/src/core/utils/time/exponential_backoff.rs index 76f0be2b2..aca2286d3 100644 --- a/src/core/utils/time/exponential_backoff.rs +++ b/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) } diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index d8e10d3e8..bd633195d 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,25 @@ 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) => { + // 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( 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 703f939b2..80a048f7f 100644 --- a/src/service/federation/mod.rs +++ b/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>, + pub stale_destinations: SyncRwLock>, } struct Services { @@ -17,6 +34,7 @@ struct Services { moderation: Dep, } +#[async_trait] impl crate::Service for Service { fn build(args: crate::Args<'_>) -> Result> { Ok(Arc::new(Self { @@ -26,8 +44,164 @@ fn build(args: crate::Args<'_>) -> Result> { server_keys: args.depend::("server_keys"), moderation: args.depend::("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 { + 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 { + self.remote_health.read().clone() + } + + /// Returns a clone of the internal stale destinations set. + pub fn stale_destinations(&self) -> HashSet { + self.stale_destinations.read().clone() + } } diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index ac9920f76..a6c971adf 100644 --- a/src/service/sending/mod.rs +++ b/src/service/sending/mod.rs @@ -132,11 +132,11 @@ async fn worker(self: Arc) -> 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"); }, } } diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index c4703ba43..442fed288 100644 --- a/src/service/sending/sender.rs +++ b/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" );