chore(deps): update rust crate axum-extra to 0.12.0

Cargo.lock

@@ -445,13 +445,14 @@ dependencies = [
 
 [[package]]
 name = "axum-extra"
-version = "0.10.3"
+version = "0.12.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9963ff19f40c6102c76756ef0a46004c0d58957d87259fc9208ff8441c12ab96"
+checksum = "fef252edff26ddba56bbcdf2ee3307b8129acb86f5749b68990c168a6fcc9c76"
 dependencies = [
  "axum",
  "axum-core",
  "bytes",
+ "futures-core",
  "futures-util",
  "headers",
  "http",
@@ -459,8 +460,6 @@ dependencies = [
  "http-body-util",
  "mime",
  "pin-project-lite",
- "rustversion",
- "serde_core",
  "tower-layer",
  "tower-service",
  "tracing",

Cargo.toml

@@ -97,7 +97,7 @@ features = [
 ]
 
 [workspace.dependencies.axum-extra]
-version = "0.10.1"
+version = "0.12.0"
 default-features = false
 features = ["typed-header", "tracing"]
 

src/api/router/auth.rs

@@ -4,7 +4,7 @@
 	headers::{Authorization, authorization::Bearer},
 	typed_header::TypedHeaderRejectionReason,
 };
-use conduwuit::{Err, Error, Result, debug_error, debug_info, err, warn};
+use conduwuit::{Err, Error, Result, debug_error, err, warn};
 use futures::{
 	TryFutureExt,
 	future::{
@@ -329,14 +329,6 @@ async fn auth_server(
 		return Err!(Request(Forbidden("Failed to verify X-Matrix signatures.")));
 	}
 
-	if services.sending.server_is_offline(destination).await {
-		debug_info!(?destination, "server returned from being offline");
-		services
-			.sending
-			.mark_server_online(destination, false)
-			.await;
-	}
-
 	Ok(Auth {
 		origin: origin.to_owned().into(),
 		sender_user: None,

src/service/sending/mod.rs

@@ -5,7 +5,6 @@
 mod sender;
 
 use std::{
-	collections::HashSet,
 	fmt::Debug,
 	hash::{DefaultHasher, Hash, Hasher},
 	iter::once,
@@ -20,8 +19,8 @@
 	warn,
 };
 use futures::{FutureExt, Stream, StreamExt};
-use ruma::{OwnedServerName, RoomId, ServerName, UserId, api::OutgoingRequest};
-use tokio::{sync::RwLock, task, task::JoinSet};
+use ruma::{RoomId, ServerName, UserId, api::OutgoingRequest};
+use tokio::{task, task::JoinSet};
 
 use self::data::Data;
 pub use self::{
@@ -38,7 +37,6 @@ pub struct Service {
 	server: Arc<Server>,
 	services: Services,
 	channels: Vec<(loole::Sender<Msg>, loole::Receiver<Msg>)>,
-	pub offline_servers: RwLock<HashSet<OwnedServerName>>,
 }
 
 struct Services {
@@ -54,7 +52,6 @@ struct Services {
 	account_data: Dep<account_data::Service>,
 	appservice: Dep<crate::appservice::Service>,
 	pusher: Dep<pusher::Service>,
-	resolver: Dep<crate::resolver::Service>,
 	federation: Dep<federation::Service>,
 }
 
@@ -99,11 +96,9 @@ fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 				account_data: args.depend::<account_data::Service>("account_data"),
 				appservice: args.depend::<crate::appservice::Service>("appservice"),
 				pusher: args.depend::<pusher::Service>("pusher"),
-				resolver: args.depend::<crate::resolver::Service>("resolver"),
 				federation: args.depend::<federation::Service>("federation"),
 			},
 			channels: (0..num_senders).map(|_| loole::unbounded()).collect(),
-			offline_servers: RwLock::new(HashSet::new()),
 		}))
 	}
 
@@ -151,8 +146,6 @@ fn interrupt(&self) {
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 
 	fn unconstrained(&self) -> bool { true }
-
-	async fn clear_cache(&self) { self.offline_servers.write().await.clear(); }
 }
 
 impl Service {
@@ -386,39 +379,6 @@ pub(super) fn shard_id(&self, dest: &Destination) -> usize {
 		let chans = self.channels.len().max(1);
 		hash.overflowing_rem(chans).0
 	}
-
-	/// Marks a server as offline
-	pub async fn mark_server_offline(&self, server: OwnedServerName) {
-		self.offline_servers.write().await.insert(server);
-	}
-
-	/// Marks a server as online again and flushes the senders if it was
-	/// previously marked as offline
-	pub async fn mark_server_online(&self, server: &ServerName, skip_flush: bool) {
-		if self.offline_servers.write().await.remove(server) && !skip_flush {
-			// Flush the senders if this server was previously offline
-			self.services.resolver.cache.del_destination(server);
-			self.services.resolver.cache.del_override(server);
-			self.dispatch(Msg {
-				dest: Destination::Federation(server.to_owned()),
-				event: SendingEvent::Flush,
-				queue_id: Vec::<u8>::new(),
-			})
-			.inspect_err(|e| {
-				error!(
-					?server,
-					?e,
-					"failed to dispatch flush message for server coming back online"
-				);
-			})
-			.ok();
-		}
-	}
-
-	/// Checks if a server is currently marked as offline
-	pub async fn server_is_offline(&self, server: &ServerName) -> bool {
-		self.offline_servers.read().await.contains(server)
-	}
 }
 
 fn num_senders(args: &crate::Args<'_>) -> usize {

src/service/sending/sender.rs

@@ -9,7 +9,6 @@
 };
 
 use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
-use conduwuit::debug_warn;
 use conduwuit_core::{
 	Error, Event, Result, debug, err, error,
 	result::LogErr,
@@ -136,13 +135,7 @@ 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.clone(), statuses, &e);
-				if let Destination::Federation(server_name) = dest {
-					debug_warn!(?server_name, "marking server offline due to error: {e:?}");
-					self.mark_server_offline(server_name).await;
-				}
-			},
+			| Err((dest, e)) => Self::handle_response_err(dest, statuses, &e),
 		}
 	}
 
@@ -187,12 +180,6 @@ async fn handle_response_ok<'a>(
 		} else {
 			statuses.remove(dest);
 		}
-		if let Destination::Federation(server_name) = dest {
-			self.mark_server_online(server_name, true).await;
-			// We skip the flush here because we were already able to contact
-			// the server, and have queued any pending events, and the
-			// resolver cache will be fine.
-		}
 	}
 
 	#[allow(clippy::needless_pass_by_ref_mut)]
@@ -203,9 +190,6 @@ async fn handle_request<'a>(
 		futures: &mut SendingFutures<'a>,
 		statuses: &mut CurTransactionStatus,
 	) {
-		if msg.event == SendingEvent::Flush {
-			statuses.remove(&msg.dest);
-		}
 		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() {