diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index 6fa9c1447..fb0f56c9f 100644 --- a/src/service/federation/execute.rs +++ b/src/service/federation/execute.rs @@ -187,7 +187,8 @@ impl super::Service { .await, | Err(error) => { if error.is_connect() { - debug_info!("{dest} is unhealthy due to a connect error"); + debug_info!("{dest} is unhealthy & stale due to a connect error"); + self.mark_destination_stale(dest); self.hit_unhealthy(dest.to_owned()); } Err(handle_error(actual, &method, &url, error).expect_err("always returns error")) diff --git a/src/service/federation/mod.rs b/src/service/federation/mod.rs index ccb6f4914..ce9285d78 100644 --- a/src/service/federation/mod.rs +++ b/src/service/federation/mod.rs @@ -1,6 +1,10 @@ mod execute; -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; use assign::assign; use async_trait::async_trait; @@ -20,6 +24,7 @@ use crate::{Dep, client, moderation, server_keys}; pub struct Service { services: Services, pub remote_health: SyncRwLock>, + stale_destinations: SyncRwLock>, } struct Services { @@ -40,6 +45,7 @@ impl crate::Service for Service { moderation: args.depend::("moderation"), }, remote_health: SyncRwLock::new(HashMap::new()), + stale_destinations: SyncRwLock::new(HashSet::new()), })) } @@ -111,10 +117,25 @@ impl Service { 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 map = self.remote_health.write(); - if map.remove(server_name).is_some() { + 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. @@ -138,4 +159,22 @@ impl Service { )) } } + + /// 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()); + } + + /// Returns true if the destination has been marked stale. + fn is_destination_stale(&self, server_name: &ServerName) -> bool { + self.stale_destinations.read().contains(server_name) + } } diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 6f285c24c..3711f2fcf 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -144,11 +144,17 @@ impl Service { match response { | Ok(dest) => self.handle_response_ok(&dest, futures, statuses).await, | Err((dest, e)) => { - if e.status_code().is_server_error() - && let Destination::Federation(dest) = &dest - { - debug!("{dest} is now unhealthy due to server error response: {e:?}"); - self.services.federation.hit_unhealthy(dest.clone()); + if let Destination::Federation(dest) = &dest { + if let Error::Reqwest(e) = &e + && e.is_connect() + { + 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().is_server_error() { + debug!("{dest} is now unhealthy due to server error response: {e:?}"); + self.services.federation.hit_unhealthy(dest.clone()); + } } Self::handle_response_err(dest, statuses, &e); },