diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 658f93735..d2283fc26 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -7,6 +7,7 @@ collections::{BTreeMap, BTreeSet, HashMap}, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}, path::PathBuf, + time::Duration, }; use conduwuit_macros::config_example_generator; @@ -2981,6 +2982,16 @@ pub fn new(server_name: OwnedServerName) -> Self { ]; impl Config { + /// Returns the capped linear retry delay for a federation send attempt. + #[must_use] + pub fn sender_retry_backoff(&self, attempt: u32) -> Duration { + crate::utils::linear_backoff( + Duration::from_secs(self.sender_retry_backoff_base), + Duration::from_secs(self.sender_retry_backoff_limit), + attempt, + ) + } + /// Get the base domain to use for user-facing URLs. #[must_use] pub fn get_client_domain(&self) -> Url { diff --git a/src/core/utils/mod.rs b/src/core/utils/mod.rs index 9e28210c1..4a5dc2bcf 100644 --- a/src/core/utils/mod.rs +++ b/src/core/utils/mod.rs @@ -37,7 +37,10 @@ stream::{IterStream, ReadyExt, Tools as StreamTools, TryReadyExt}, string::{str_from_bytes, string_from_bytes}, sys::compute::available_parallelism, - time::{now_millis as millis_since_unix_epoch, timepoint_ago, timepoint_from_now}, + time::{ + linear_backoff, millis_from_now, now_millis as millis_since_unix_epoch, timepoint_ago, + timepoint_from_now, + }, }; #[inline] diff --git a/src/core/utils/time.rs b/src/core/utils/time.rs index b731e6563..ff35601bb 100644 --- a/src/core/utils/time.rs +++ b/src/core/utils/time.rs @@ -16,6 +16,20 @@ pub fn now_millis() -> u64 { .as_millis() as u64 } +/// Returns a capped linear retry delay for a failed attempt. +#[must_use] +pub fn linear_backoff(base: Duration, limit: Duration, attempt: u32) -> Duration { + base.saturating_mul(attempt).min(limit) +} + +/// Returns a saturated Unix-millisecond deadline after `duration`. +#[must_use] +pub fn millis_from_now(duration: Duration) -> u64 { add_millis(now_millis(), duration) } + +fn add_millis(now: u64, duration: Duration) -> u64 { + now.saturating_add(u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)) +} + #[inline] pub fn parse_timepoint_ago(ago: &str) -> Result { timepoint_ago(parse_duration(ago)?) @@ -155,3 +169,29 @@ pub fn is_within_bounds( | TimeDirection::After => item_time >= time_boundary, } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::{add_millis, linear_backoff}; + + #[test] + fn linear_backoff_scales_and_caps() { + let base = Duration::from_secs(10); + let limit = Duration::from_secs(25); + + assert_eq!(linear_backoff(base, limit, 0), Duration::ZERO); + assert_eq!(linear_backoff(base, limit, 1), Duration::from_secs(10)); + assert_eq!(linear_backoff(base, limit, 2), Duration::from_secs(20)); + assert_eq!(linear_backoff(base, limit, 3), limit); + assert_eq!(linear_backoff(base, limit, u32::MAX), limit); + } + + #[test] + fn millis_deadline_saturates() { + assert_eq!(add_millis(10, Duration::from_millis(20)), 30); + assert_eq!(add_millis(u64::MAX, Duration::from_millis(1)), u64::MAX); + assert_eq!(add_millis(0, Duration::from_secs(u64::MAX)), u64::MAX); + } +} diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index ebbe7dc1c..76d0e124a 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -8,7 +8,10 @@ use assign::assign; use async_trait::async_trait; -use conduwuit::{Error, Result, Server, SyncRwLock, debug, utils::millis_since_unix_epoch}; +use conduwuit::{ + Error, Result, Server, SyncRwLock, debug, + utils::{millis_from_now, millis_since_unix_epoch}, +}; pub(crate) use execute::FederationPathBuilderInput; use http::StatusCode; use ruma::{ @@ -105,15 +108,9 @@ pub fn hit_unhealthy(&self, server_name: OwnedServerName) { 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); - let next_interval = min.saturating_mul(*retries).min(max); - *next_retry = unix_now.saturating_add( - u64::try_from(next_interval.as_millis()) - .expect("backoff milliseconds should not exceed u64::MAX"), - ); + let next_interval = self.services.server.config.sender_retry_backoff(*retries); + *next_retry = millis_from_now(next_interval); debug!( "{} is (now) unhealthy ({} retries, blocked until: {})", sn2, *retries, *next_retry diff --git a/src/service/pusher/webpush/mod.rs b/src/service/pusher/webpush/mod.rs index 6f66d6fdb..5fb8ad19b 100644 --- a/src/service/pusher/webpush/mod.rs +++ b/src/service/pusher/webpush/mod.rs @@ -132,7 +132,7 @@ pub(super) fn begin_webpush_activation(&self, sender: &UserId, pusher: &Pusher) let state = WebPushState { activated: false, ack_token: Some(token.clone()), - token_expires_at: utils::millis_since_unix_epoch().saturating_add(ACK_TOKEN_TTL_MS), + token_expires_at: utils::millis_from_now(Duration::from_millis(ACK_TOKEN_TTL_MS)), }; self.put_webpush_state(sender, pusher.ids.pushkey.as_str(), &state); @@ -301,7 +301,7 @@ async fn deliver_webpush( } let delay = retry_after - .unwrap_or_else(|| RETRY_BASE.saturating_mul(attempt)) + .unwrap_or_else(|| utils::linear_backoff(RETRY_BASE, RETRY_MAX, attempt)) .min(RETRY_MAX); sleep(delay).await; attempt = attempt.saturating_add(1); @@ -413,8 +413,7 @@ async fn back_off_webpush( |retry_after| retry_after.min(BACKOFF_MAX), ); - backoff.next_attempt_at = utils::millis_since_unix_epoch() - .saturating_add(delay.as_millis().try_into().unwrap_or(u64::MAX)); + backoff.next_attempt_at = utils::millis_from_now(delay); self.db.senderkey_webpushbackoff.put(key, Json(&backoff)); } diff --git a/src/service/rooms/event_handler/mod.rs b/src/service/rooms/event_handler/mod.rs index f20eb3037..75c12b73e 100644 --- a/src/service/rooms/event_handler/mod.rs +++ b/src/service/rooms/event_handler/mod.rs @@ -13,12 +13,7 @@ mod state_at_incoming; mod upgrade_outlier_pdu; -use std::{ - collections::HashMap, - fmt::Write, - sync::Arc, - time::{Duration, Instant}, -}; +use std::{collections::HashMap, fmt::Write, sync::Arc, time::Instant}; use assign::assign; use async_trait::async_trait; @@ -149,7 +144,11 @@ pub(super) fn ensure_can_pull_event(&self, event_id: &EventId) -> Result<()> { let Some((retry_count, last_retry)) = map.get(event_id) else { return Ok(()); }; - let next_retry = calc_next_retry(&self.services.server.config, *retry_count); + let next_retry = self + .services + .server + .config + .sender_retry_backoff(*retry_count); if last_retry.elapsed() >= next_retry { Ok(()) } else { @@ -169,7 +168,7 @@ pub(super) fn hit_failed_pdu_pull(&self, event_id: OwnedEventId) { let mut map = self.failed_pdu_pulls.write(); map.entry(event_id) .and_modify(|(retries, last_retry)| { - let next_retry = calc_next_retry(&self.services.server.config, *retries); + let next_retry = self.services.server.config.sender_retry_backoff(*retries); if next_retry > last_retry.elapsed() { // Don't increment the retry if a retry was illegal. // This can happen if an event was fetched multiple times @@ -201,9 +200,3 @@ fn get_room_version_rules(create_event: &Pdu) -> Result Duration { - let min = Duration::from_secs(config.sender_retry_backoff_base); - let max = Duration::from_secs(config.sender_retry_backoff_limit); - min.saturating_mul(retries).min(max) -} diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 690b90767..308dcbd79 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -450,9 +450,7 @@ fn should_attempt_send( .entry(dest.clone()) .and_modify(|e| match e { | TransactionStatus::Failed(tries, time) => { - let min = Duration::from_secs(self.server.config.sender_retry_backoff_base); - let max = Duration::from_secs(self.server.config.sender_retry_backoff_limit); - let retry_after = min.saturating_mul(*tries).min(max); + let retry_after = self.server.config.sender_retry_backoff(*tries); if time.elapsed() < retry_after { debug_warn!("Not retrying destination for another {retry_after:?}"); allow = false;