fix: Use a timeout instead of deadline

Also parallelised federation calls in related functions

.pre-commit-config.yaml

@@ -23,7 +23,7 @@ repos:
         - id: check-added-large-files
 
   - repo: https://github.com/crate-ci/typos
-    rev: v1.40.0
+    rev: v1.41.0
     hooks:
       - id: typos
       - id: typos
@@ -31,7 +31,7 @@ repos:
         stages: [commit-msg]
 
   - repo: https://github.com/crate-ci/committed
-    rev: v1.1.8
+    rev: v1.1.9
     hooks:
     - id: committed
 

changelog.d/1257.bugfix

@@ -0,0 +1 @@
+Fixed unreliable room summary fetching and improved error messages. Contributed by @nex.

changelog.d/1261.bugfix

@@ -0,0 +1,2 @@
+Client requested timeout parameter is now applied to e2ee key lookups and claims. Related federation requests are now
+also concurrent. Contributed by @nex.

src/admin/user/commands.rs

@@ -465,7 +465,7 @@ pub(super) async fn force_join_list_of_local_users(
 
 	if server_admins.is_empty() {
 		return Err!("There are no admins set for this server.");
-	};
+	}
 
 	let (room_id, servers) = self
 		.services
@@ -580,7 +580,7 @@ pub(super) async fn force_join_all_local_users(
 
 	if server_admins.is_empty() {
 		return Err!("There are no admins set for this server.");
-	};
+	}
 
 	let (room_id, servers) = self
 		.services

src/api/client/keys.rs

@@ -1,7 +1,15 @@
-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::{
+	collections::{BTreeMap, HashMap, HashSet},
+	time::Duration,
+};
 
 use axum::extract::State;
-use conduwuit::{Err, Error, Result, debug, debug_warn, err, result::NotFound, utils};
+use conduwuit::{
+	Err, Error, Result, debug, debug_warn, err,
+	result::NotFound,
+	utils,
+	utils::{IterStream, stream::WidebandExt},
+};
 use conduwuit_service::{Services, users::parse_master_key};
 use futures::{StreamExt, stream::FuturesUnordered};
 use ruma::{
@@ -134,6 +142,7 @@ pub(crate) async fn get_keys_route(
 		&body.device_keys,
 		|u| u == sender_user,
 		true, // Always allow local users to see device names of other local users
+		body.timeout.unwrap_or(Duration::from_secs(10)),
 	)
 	.await
 }
@@ -145,7 +154,12 @@ pub(crate) async fn claim_keys_route(
 	State(services): State<crate::State>,
 	body: Ruma<claim_keys::v3::Request>,
 ) -> Result<claim_keys::v3::Response> {
-	claim_keys_helper(&services, &body.one_time_keys).await
+	claim_keys_helper(
+		&services,
+		&body.one_time_keys,
+		body.timeout.unwrap_or(Duration::from_secs(10)),
+	)
+	.await
 }
 
 /// # `POST /_matrix/client/r0/keys/device_signing/upload`
@@ -421,6 +435,7 @@ pub(crate) async fn get_keys_helper<F>(
 	device_keys_input: &BTreeMap<OwnedUserId, Vec<OwnedDeviceId>>,
 	allowed_signatures: F,
 	include_display_names: bool,
+	timeout: Duration,
 ) -> Result<get_keys::v3::Response>
 where
 	F: Fn(&UserId) -> bool + Send + Sync,
@@ -512,9 +527,10 @@ where
 
 	let mut failures = BTreeMap::new();
 
-	let mut futures: FuturesUnordered<_> = get_over_federation
+	let futures = get_over_federation
 		.into_iter()
-		.map(|(server, vec)| async move {
+		.stream()
+		.wide_filter_map(|(server, vec)| async move {
 			let mut device_keys_input_fed = BTreeMap::new();
 			for (user_id, keys) in vec {
 				device_keys_input_fed.insert(user_id.to_owned(), keys.clone());
@@ -522,17 +538,22 @@ where
 
 			let request =
 				federation::keys::get_keys::v1::Request { device_keys: device_keys_input_fed };
+			let response = tokio::time::timeout(
+				timeout,
+				services.sending.send_federation_request(server, request),
+			)
+			.await
+			// Need to flatten the Result<Result<V, E>, E> into Result<V, E>
+			.map_err(|_| err!(Request(Unknown("Timeout when getting keys over federation."))))
+			.and_then(|res| res);
 
-			let response = services
+			Some((server, response))
-				.sending
-				.send_federation_request(server, request)
-				.await;
-
-			(server, response)
 		})
-		.collect();
+		.collect::<FuturesUnordered<_>>()
+		.await
+		.into_iter();
 
-	while let Some((server, response)) = futures.next().await {
+	for (server, response) in futures {
 		match response {
 			| Ok(response) => {
 				for (user, master_key) in response.master_keys {
@@ -564,8 +585,8 @@ where
 				self_signing_keys.extend(response.self_signing_keys);
 				device_keys.extend(response.device_keys);
 			},
-			| _ => {
+			| Err(e) => {
-				failures.insert(server.to_string(), json!({}));
+				failures.insert(server.to_string(), json!({ "error": e.to_string() }));
 			},
 		}
 	}
@@ -608,6 +629,7 @@ fn add_unsigned_device_display_name(
 pub(crate) async fn claim_keys_helper(
 	services: &Services,
 	one_time_keys_input: &BTreeMap<OwnedUserId, BTreeMap<OwnedDeviceId, OneTimeKeyAlgorithm>>,
+	timeout: Duration,
 ) -> Result<claim_keys::v3::Response> {
 	let mut one_time_keys = BTreeMap::new();
 
@@ -638,32 +660,39 @@ pub(crate) async fn claim_keys_helper(
 
 	let mut failures = BTreeMap::new();
 
-	let mut futures: FuturesUnordered<_> = get_over_federation
+	let futures = get_over_federation
 		.into_iter()
-		.map(|(server, vec)| async move {
+		.stream()
+		.wide_filter_map(|(server, vec)| async move {
 			let mut one_time_keys_input_fed = BTreeMap::new();
 			for (user_id, keys) in vec {
 				one_time_keys_input_fed.insert(user_id.clone(), keys.clone());
 			}
-			(
+			let response = tokio::time::timeout(
-				server,
+				timeout,
-				services
+				services.sending.send_federation_request(
-					.sending
+					server,
-					.send_federation_request(server, federation::keys::claim_keys::v1::Request {
+					federation::keys::claim_keys::v1::Request {
 						one_time_keys: one_time_keys_input_fed,
-					})
+					},
-					.await,
+				),
 			)
+			.await
+			.map_err(|_| err!(Request(Unknown("Timeout when claiming keys over federation."))))
+			.and_then(|res| res);
+			Some((server, response))
 		})
-		.collect();
+		.collect::<FuturesUnordered<_>>()
+		.await
+		.into_iter();
 
-	while let Some((server, response)) = futures.next().await {
+	for (server, response) in futures {
 		match response {
 			| Ok(keys) => {
 				one_time_keys.extend(keys.one_time_keys);
 			},
-			| Err(_e) => {
+			| Err(e) => {
-				failures.insert(server.to_string(), json!({}));
+				failures.insert(server.to_string(), json!({"error": e.to_string()}));
 			},
 		}
 	}

src/api/client/room/summary.rs

@@ -1,11 +1,11 @@
 use axum::extract::State;
 use axum_client_ip::InsecureClientIp;
 use conduwuit::{
-	Err, Result, debug_warn, trace,
+	Err, Result, debug, debug_warn, info, trace,
 	utils::{IterStream, future::TryExtExt},
 };
 use futures::{
-	FutureExt, StreamExt,
+	FutureExt, StreamExt, TryFutureExt,
 	future::{OptionFuture, join3},
 	stream::FuturesUnordered,
 };
@@ -79,9 +79,15 @@ async fn room_summary_response(
 		.server_in_room(services.globals.server_name(), room_id)
 		.await
 	{
-		return local_room_summary_response(services, room_id, sender_user)
+		match local_room_summary_response(services, room_id, sender_user)
 			.boxed()
-			.await;
+			.await
+		{
+			| Ok(response) => return Ok(response),
+			| Err(e) => {
+				debug_warn!("Failed to get local room summary: {e:?}, falling back to remote");
+			},
+		}
 	}
 
 	let room =
@@ -111,26 +117,27 @@ async fn local_room_summary_response(
 	sender_user: Option<&UserId>,
 ) -> Result<get_summary::msc3266::Response> {
 	trace!(?sender_user, "Sending local room summary response for {room_id:?}");
-	let join_rule = services.rooms.state_accessor.get_join_rules(room_id);
+	let (join_rule, world_readable, guest_can_join) = join3(
-
+		services.rooms.state_accessor.get_join_rules(room_id),
-	let world_readable = services.rooms.state_accessor.is_world_readable(room_id);
+		services.rooms.state_accessor.is_world_readable(room_id),
-
+		services.rooms.state_accessor.guest_can_join(room_id),
-	let guest_can_join = services.rooms.state_accessor.guest_can_join(room_id);
-
-	let (join_rule, world_readable, guest_can_join) =
-		join3(join_rule, world_readable, guest_can_join).await;
-
-	trace!("{join_rule:?}, {world_readable:?}, {guest_can_join:?}");
-	user_can_see_summary(
-		services,
-		room_id,
-		&join_rule.clone().into(),
-		guest_can_join,
-		world_readable,
-		join_rule.allowed_rooms(),
-		sender_user,
 	)
-	.await?;
+	.await;
+
+	// Synapse allows server admins to bypass visibility checks.
+	// That seems neat so we'll copy that behaviour.
+	if sender_user.is_none() || !services.users.is_admin(sender_user.unwrap()).await {
+		user_can_see_summary(
+			services,
+			room_id,
+			&join_rule.clone().into(),
+			guest_can_join,
+			world_readable,
+			join_rule.allowed_rooms(),
+			sender_user,
+		)
+		.await?;
+	}
 
 	let canonical_alias = services
 		.rooms
@@ -231,15 +238,27 @@ async fn remote_room_summary_hierarchy_response(
 			"Federaton of room {room_id} is currently disabled on this server."
 		)));
 	}
+	if servers.is_empty() {
+		return Err!(Request(MissingParam(
+			"No servers were provided to fetch the room over federation"
+		)));
+	}
 
 	let request = get_hierarchy::v1::Request::new(room_id.to_owned());
 
 	let mut requests: FuturesUnordered<_> = servers
 		.iter()
 		.map(|server| {
+			info!("Fetching room summary for {room_id} from server {server}");
 			services
 				.sending
 				.send_federation_request(server, request.clone())
+				.inspect_ok(move |v| {
+					debug!("Fetched room summary for {room_id} from server {server}: {v:?}");
+				})
+				.inspect_err(move |e| {
+					info!("Failed to fetch room summary for {room_id} from server {server}: {e}");
+				})
 		})
 		.collect();
 
@@ -255,23 +274,23 @@ async fn remote_room_summary_hierarchy_response(
 			continue;
 		}
 
-		return user_can_see_summary(
+		if sender_user.is_none() || !services.users.is_admin(sender_user.unwrap()).await {
-			services,
+			return user_can_see_summary(
-			room_id,
+				services,
-			&room.join_rule,
+				room_id,
-			room.guest_can_join,
+				&room.join_rule,
-			room.world_readable,
+				room.guest_can_join,
-			room.allowed_room_ids.iter().map(AsRef::as_ref),
+				room.world_readable,
-			sender_user,
+				room.allowed_room_ids.iter().map(AsRef::as_ref),
-		)
+				sender_user,
-		.await
+			)
-		.map(|()| room);
+			.await
+			.map(|()| room);
+		}
+		return Ok(room);
 	}
 
-	Err!(Request(NotFound(
+	Err!(Request(NotFound("Room not found or is not accessible")))
-		"Room is unknown to this server and was unable to fetch over federation with the \
-		 provided servers available"
-	)))
 }
 
 async fn user_can_see_summary<'a, I>(
@@ -311,21 +330,14 @@ where
 				return Ok(());
 			}
 
-			Err!(Request(Forbidden(
+			Err!(Request(Forbidden("Room is not accessible")))
-				"Room is not world readable, not publicly accessible/joinable, restricted room \
-				 conditions not met, and guest access is forbidden. Not allowed to see details \
-				 of this room."
-			)))
 		},
 		| None => {
 			if is_public_room || world_readable {
 				return Ok(());
 			}
 
-			Err!(Request(Forbidden(
+			Err!(Request(Forbidden("Room is not accessible")))
-				"Room is not world readable or publicly accessible/joinable, authentication is \
-				 required"
-			)))
 		},
 	}
 }

src/api/client/well_known.rs

@@ -72,7 +72,7 @@ pub(crate) async fn well_known_support(
 	if contacts.is_empty() {
 		let admin_users = services.admin.get_admins().await;
 
-		for user_id in admin_users.iter() {
+		for user_id in &admin_users {
 			if *user_id == services.globals.server_user {
 				continue;
 			}

src/api/server/user.rs

@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use axum::extract::State;
 use conduwuit::{Error, Result};
 use futures::{FutureExt, StreamExt, TryFutureExt};
@@ -96,6 +98,7 @@ pub(crate) async fn get_keys_route(
 		&body.device_keys,
 		|u| Some(u.server_name()) == body.origin.as_deref(),
 		services.globals.allow_device_name_federation(),
+		Duration::from_secs(0),
 	)
 	.await?;
 
@@ -124,7 +127,8 @@ pub(crate) async fn claim_keys_route(
 		));
 	}
 
-	let result = claim_keys_helper(&services, &body.one_time_keys).await?;
+	let result =
+		claim_keys_helper(&services, &body.one_time_keys, Duration::from_secs(0)).await?;
 
 	Ok(claim_keys::v1::Response { one_time_keys: result.one_time_keys })
 }

src/service/admin/grant.rs

@@ -188,7 +188,7 @@ pub async fn revoke_admin(&self, user_id: &UserId) -> Result {
 		warn!(
 			"Revoking the admin status of {user_id} will not work correctly as they are within \
 			 the admins_list config."
-		)
+		);
 	}
 
 	let Ok(room_id) = self.get_admin_room().await else {

src/service/admin/mod.rs

@@ -406,16 +406,23 @@ impl Service {
 
 	/// Checks whether a given user is an admin of this server
 	pub async fn user_is_admin(&self, user_id: &UserId) -> bool {
-		if self.services.server.config.admins_list.contains(user_id) {
+		if self
+			.services
+			.server
+			.config
+			.admins_list
+			.contains(&user_id.to_owned())
+		{
 			return true;
 		}
 
 		if self.services.server.config.admins_from_room {
 			if let Ok(admin_room) = self.get_admin_room().await {
-				self.services
+				return self
+					.services
 					.state_cache
 					.is_joined(user_id, &admin_room)
-					.await
+					.await;
 			}
 		}
 

src/service/pusher/mod.rs

@@ -427,20 +427,20 @@ impl Service {
 				}
 
 				let d = vec![device];
-				let mut notifi = Notification::new(d);
+				let mut notify = Notification::new(d);
 
-				notifi.event_id = Some(event.event_id().to_owned());
+				notify.event_id = Some(event.event_id().to_owned());
-				notifi.room_id = Some(event.room_id().unwrap().to_owned());
+				notify.room_id = Some(event.room_id().unwrap().to_owned());
 				if http
 					.data
 					.get("org.matrix.msc4076.disable_badge_count")
 					.is_none() && http.data.get("disable_badge_count").is_none()
 				{
-					notifi.counts = NotificationCounts::new(unread, uint!(0));
+					notify.counts = NotificationCounts::new(unread, uint!(0));
 				} else {
 					// counts will not be serialised if it's the default (0, 0)
 					// skip_serializing_if = "NotificationCounts::is_default"
-					notifi.counts = NotificationCounts::default();
+					notify.counts = NotificationCounts::default();
 				}
 
 				if !event_id_only {
@@ -449,30 +449,30 @@ impl Service {
 							.iter()
 							.any(|t| matches!(t, Tweak::Highlight(true) | Tweak::Sound(_)))
 					{
-						notifi.prio = NotificationPriority::High;
+						notify.prio = NotificationPriority::High;
 					} else {
-						notifi.prio = NotificationPriority::Low;
+						notify.prio = NotificationPriority::Low;
 					}
-					notifi.sender = Some(event.sender().to_owned());
+					notify.sender = Some(event.sender().to_owned());
-					notifi.event_type = Some(event.kind().to_owned());
+					notify.event_type = Some(event.kind().to_owned());
-					notifi.content = serde_json::value::to_raw_value(event.content()).ok();
+					notify.content = serde_json::value::to_raw_value(event.content()).ok();
 
 					if *event.kind() == TimelineEventType::RoomMember {
-						notifi.user_is_target =
+						notify.user_is_target =
 							event.state_key() == Some(event.sender().as_str());
 					}
 
-					notifi.sender_display_name =
+					notify.sender_display_name =
 						self.services.users.displayname(event.sender()).await.ok();
 
-					notifi.room_name = self
+					notify.room_name = self
 						.services
 						.state_accessor
 						.get_name(event.room_id().unwrap())
 						.await
 						.ok();
 
-					notifi.room_alias = self
+					notify.room_alias = self
 						.services
 						.state_accessor
 						.get_canonical_alias(event.room_id().unwrap())
@@ -480,7 +480,7 @@ impl Service {
 						.ok();
 				}
 
-				self.send_request(&http.url, send_event_notification::v1::Request::new(notifi))
+				self.send_request(&http.url, send_event_notification::v1::Request::new(notify))
 					.await?;
 
 				Ok(())