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/mod.rs b/src/service/federation/mod.rs index 86ea9b662..c46d34a40 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -6,10 +6,7 @@ use async_trait::async_trait; use conduwuit::{ Error, Result, Server, SyncRwLock, debug, - utils::{ - math::Expected, millis_since_unix_epoch, - time::exponential_backoff::min_exp_backoff_duration, - }, + utils::{math::Expected, millis_since_unix_epoch, time::exponential_backoff::next_interval}, }; pub(crate) use execute::FederationPathBuilderInput; use http::StatusCode; @@ -94,12 +91,12 @@ pub fn hit_unhealthy(&self, server_name: OwnedServerName) { return; } - let min = self.services.server.config.sender_timeout; - let max = self.services.server.config.sender_retry_backoff_limit; + let min = Duration::from_secs(self.services.server.config.sender_timeout); + 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(min_exp_backoff_duration(min, max, *retries).as_millis()) + u64::try_from(next_interval(min, max, *retries).as_millis()) .expect("backoff milliseconds should not exceed u64::MAX"), ); debug!( diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index c3e886352..d64002e9c 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, + 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}, }, @@ -409,13 +410,12 @@ fn should_attempt_send( .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_timeout); + 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 {