From f1a77e04045c46099aadafdb791747cebac0829d Mon Sep 17 00:00:00 2001 From: timedout Date: Wed, 19 Aug 2026 14:34:15 +0100 Subject: [PATCH] fix: Address review comments --- src/admin/debug/commands.rs | 2 +- src/admin/debug/mod.rs | 7 +++++++ src/api/router/auth.rs | 10 +++++++-- src/core/utils/mod.rs | 5 +---- src/core/utils/time.rs | 1 - src/core/utils/time/exponential_backoff.rs | 23 --------------------- src/service/federation/execute.rs | 11 ++++++++-- src/service/federation/mod.rs | 24 ++++++++++++---------- src/service/sending/sender.rs | 19 ++++++----------- 9 files changed, 45 insertions(+), 57 deletions(-) delete mode 100644 src/core/utils/time/exponential_backoff.rs diff --git a/src/admin/debug/commands.rs b/src/admin/debug/commands.rs index 2f1c00482..fc2892805 100644 --- a/src/admin/debug/commands.rs +++ b/src/admin/debug/commands.rs @@ -1246,7 +1246,7 @@ impl crate::Context<'_> { let backoff_map = self.services.federation.remote_health(); if backoff_map.is_empty() { - return Err!("No servers in backoff."); + return self.write_str("No servers are currently in backoff.").await; } let stale_map = self.services.federation.stale_destinations(); diff --git a/src/admin/debug/mod.rs b/src/admin/debug/mod.rs index 016713e04..c9d8d443d 100644 --- a/src/admin/debug/mod.rs +++ b/src/admin/debug/mod.rs @@ -250,6 +250,13 @@ pub enum DebugCommand { page: Option, }, + /// Lists remote servers that are currently in a "backoff" state. + /// + /// Servers in this list will not be contacted until the retry time elapses, + /// or they make contact first. + /// + /// "Stale" indicates that routing information may be out of date, and the + /// remote will be re-resolved immediately before the next contact attempt. ServersInBackoff { #[arg(short, long, alias = "room")] room_id: Option, diff --git a/src/api/router/auth.rs b/src/api/router/auth.rs index f8f72519c..f5fdc3e5c 100644 --- a/src/api/router/auth.rs +++ b/src/api/router/auth.rs @@ -1,6 +1,6 @@ use std::any::{Any, TypeId}; -use conduwuit::{Err, Error, Result, err}; +use conduwuit::{Err, Error, Result, err, utils::IterStream}; use http::StatusCode; use ruma::{ DeviceId, OwnedDeviceId, OwnedServerName, OwnedUserId, UserId, @@ -150,7 +150,13 @@ impl CheckAuth for ServerSignatures { } // Ping the server as healthy - services.federation.mark_healthy(&output.origin); + if services.federation.mark_healthy(&output.origin) { + services + .sending + .flush_servers([output.origin.clone()].stream()) + .await + .ok(); + } Ok(output.origin) }, diff --git a/src/core/utils/mod.rs b/src/core/utils/mod.rs index e1e143497..9e28210c1 100644 --- a/src/core/utils/mod.rs +++ b/src/core/utils/mod.rs @@ -37,10 +37,7 @@ pub use self::{ stream::{IterStream, ReadyExt, Tools as StreamTools, TryReadyExt}, string::{str_from_bytes, string_from_bytes}, sys::compute::available_parallelism, - time::{ - exponential_backoff::should_continue_backoff, now_millis as millis_since_unix_epoch, - timepoint_ago, timepoint_from_now, - }, + time::{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 fb03c1d88..b731e6563 100644 --- a/src/core/utils/time.rs +++ b/src/core/utils/time.rs @@ -1,4 +1,3 @@ -pub mod exponential_backoff; pub mod jitter; use std::time::{Duration, SystemTime, UNIX_EPOCH}; diff --git a/src/core/utils/time/exponential_backoff.rs b/src/core/utils/time/exponential_backoff.rs deleted file mode 100644 index aca2286d3..000000000 --- a/src/core/utils/time/exponential_backoff.rs +++ /dev/null @@ -1,23 +0,0 @@ -use std::time::Duration; - -/// Returns false if the backoff interval has expired based on the inputs, -/// meaning the operation should be retried. -#[inline] -#[must_use] -pub fn should_continue_backoff( - min: Duration, - max: Duration, - elapsed: Duration, - tries: u32, -) -> bool { - elapsed < next_interval(min, max, tries) -} - -/// Determines the interval that should be waited before retrying the operation -/// using the algorithm: `(min * retries).min(max)`. -#[must_use] -#[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 bd633195d..376188f83 100644 --- a/src/service/federation/execute.rs +++ b/src/service/federation/execute.rs @@ -256,8 +256,15 @@ impl super::Service { parts, body.as_ref(), )) - .inspect(|_| self.mark_healthy(dest)) - .map_err(|e| err!(BadServerResponse("Server returned bad 200 response: {e:?}"))) + .inspect(|_| { + self.mark_healthy(dest); + }) + .map_err(|e| { + // Bad 200 response is usually a sign of bad routing, so we'll mark as stale. + // We won't mark as offline though since we technically got a response. + self.mark_destination_stale(dest); + err!(BadServerResponse("Server returned bad 200 response: {e:?}")) + }) } } diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index 80a048f7f..afce8db95 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -8,10 +8,7 @@ use std::{ 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}, -}; +use conduwuit::{Error, Result, Server, SyncRwLock, debug, utils::millis_since_unix_epoch}; pub(crate) use execute::FederationPathBuilderInput; use http::StatusCode; use ruma::{ @@ -83,7 +80,9 @@ impl Service { /// 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 + /// retry count is incremented and next retry window is updated. + /// + /// Does not update the marker if the backoff period is already in effect. pub fn hit_unhealthy(&self, server_name: OwnedServerName) { let unix_now = millis_since_unix_epoch(); let mut map = self.remote_health.write(); @@ -101,8 +100,9 @@ impl Service { 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(min, max, *retries).as_millis()) + u64::try_from(next_interval.as_millis()) .expect("backoff milliseconds should not exceed u64::MAX"), ); debug!( @@ -113,12 +113,12 @@ impl Service { /// 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. + /// Returns true if the server was previously marked as unhealthy, false + /// otherwise. + pub fn mark_healthy(&self, server_name: &ServerName) -> bool { let mut health_map = self.remote_health.write(); - if health_map.remove(server_name).is_some() { + let was_unhealthy = health_map.remove(server_name).is_some(); + if was_unhealthy { debug!("{} is now healthy", server_name); } // The lock for remote_health is deliberately retained until the end of the @@ -136,6 +136,8 @@ impl Service { .matrix_resolver .remove_cache_entry(server_name.as_str()); } + + was_unhealthy } /// Returns a rate-limited error if the remote is unhealthy. diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 493dbdf25..3bf06f08c 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -9,10 +9,7 @@ use std::{ }; use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD}; -use conduwuit::{ - debug_info, debug_warn, info, trace, - utils::{should_continue_backoff, time::exponential_backoff::next_interval}, -}; +use conduwuit::{debug_info, debug_warn, info, trace}; use conduwuit_core::{ Error, Event, Result, at, debug, err, error, matrix::pdu::sticky, @@ -173,11 +170,7 @@ impl Service { TransactionStatus::Failed(n.saturating_add(1), Instant::now()), | TransactionStatus::Failed(..) => { panic!( - "{}", - format!( - "Request to {dest2:?} that was not even running ({e:?}) failed: \ - {er:?}" - ) + "Request to {dest2:?} that was not even running ({e:?}) failed: {er:?}" ) }, } @@ -446,16 +439,16 @@ impl Service { }); return Ok((allow, retry)); } + + // Legacy backoff tracking for non-server remotes statuses .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); - if should_continue_backoff(min, max, time.elapsed(), *tries) - && !matches!(dest, Destination::Appservice(_)) - { - let retry_after = next_interval(min, max, *tries); + let retry_after = min.saturating_mul(*tries).min(max); + if time.elapsed() < retry_after { debug_warn!("Not retrying destination for another {retry_after:?}"); allow = false; } else {