refactor(service): Use SyncRwLock and SyncMutex for in-memory services

Convert in-memory data structures in appservice, rooms::typing, and sync
wakers to use SyncRwLock and SyncMutex. Drop appservice locks before
making outbound network requests in room alias queries and federation
invites.

src/service/appservice/mod.rs

@@ -1,20 +1,19 @@
 mod namespace_regex;
 mod registration_info;
 
-use std::{collections::BTreeMap, iter::IntoIterator, sync::Arc};
+use std::{collections::BTreeMap, sync::Arc};
 
 use async_trait::async_trait;
-use conduwuit::{Err, Result, err, utils::stream::IterStream};
+use conduwuit::{Err, Result, SyncRwLock, SyncRwLockReadGuard, err, utils::stream::IterStream};
 use database::Map;
-use futures::{Future, FutureExt, Stream, TryStreamExt};
+use futures::{Stream, TryStreamExt};
 use ruma::{RoomAliasId, RoomId, UserId, api::appservice::Registration};
-use tokio::sync::{RwLock, RwLockReadGuard};
 
 pub use self::{namespace_regex::NamespaceRegex, registration_info::RegistrationInfo};
 use crate::{Dep, globals, sending, users};
 
 pub struct Service {
-	registration_info: RwLock<Registrations>,
+	registration_info: SyncRwLock<Registrations>,
 	services: Services,
 	db: Data,
 }
@@ -35,7 +34,7 @@ type Registrations = BTreeMap<String, RegistrationInfo>;
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
-			registration_info: RwLock::new(BTreeMap::new()),
+			registration_info: SyncRwLock::new(BTreeMap::new()),
 			services: Services {
 				globals: args.depend::<globals::Service>("globals"),
 				sending: args.depend::<sending::Service>("sending"),
@@ -123,7 +122,6 @@ impl Service {
 
 		self.registration_info
 			.write()
-			.await
 			.insert(id, registration.try_into()?);
 
 		Ok(())
@@ -139,7 +137,7 @@ impl Service {
 
 		// Check for token collision with other appservices (allow
 		// re-registration of same appservice)
-		if let Ok(existing) = self.find_from_token(&registration.as_token).await {
+		if let Ok(existing) = self.find_from_token(&registration.as_token) {
 			if existing.registration.id != registration.id {
 				return Err(err!(Request(InvalidParam(
 					"Cannot register appservice: Token is already used by appservice '{}'. \
@@ -182,7 +180,6 @@ impl Service {
 		// removes the appservice registration info
 		self.registration_info
 			.write()
-			.await
 			.remove(appservice_id)
 			.ok_or_else(|| err!("Appservice not found"))?;
 
@@ -197,19 +194,17 @@ impl Service {
 			.await
 	}
 
-	pub async fn get_registration(&self, id: &str) -> Option<Registration> {
+	pub fn get_registration(&self, id: &str) -> Option<Registration> {
 		self.registration_info
 			.read()
-			.await
 			.get(id)
 			.cloned()
 			.map(|info| info.registration)
 	}
 
 	/// Returns Result to match users::find_from_token for select_ok usage
-	pub async fn find_from_token(&self, token: &str) -> Result<RegistrationInfo> {
+	pub fn find_from_token(&self, token: &str) -> Result<RegistrationInfo> {
 		self.read()
-			.await
 			.values()
 			.find(|info| info.registration.as_token == token)
 			.cloned()
@@ -217,26 +212,21 @@ impl Service {
 	}
 
 	/// Checks if a given user id matches any exclusive appservice regex
-	pub async fn is_exclusive_user_id(&self, user_id: &UserId) -> bool {
+	pub fn is_exclusive_user_id(&self, user_id: &UserId) -> bool {
 		self.read()
-			.await
 			.values()
 			.any(|info| info.is_exclusive_user_match(user_id))
 	}
 
 	/// Checks if a given user id matches any appservice regex, exclusive or
 	/// otherwise
-	pub async fn is_user_id(&self, user_id: &UserId) -> bool {
-		self.read()
-			.await
-			.values()
-			.any(|info| info.is_user_match(user_id))
+	pub fn is_user_id(&self, user_id: &UserId) -> bool {
+		self.read().values().any(|info| info.is_user_match(user_id))
 	}
 
 	/// Checks if a given room alias matches any exclusive appservice regex
-	pub async fn is_exclusive_alias(&self, alias: &RoomAliasId) -> bool {
+	pub fn is_exclusive_alias(&self, alias: &RoomAliasId) -> bool {
 		self.read()
-			.await
 			.values()
 			.any(|info| info.aliases.is_exclusive_match(alias.as_str()))
 	}
@@ -245,19 +235,15 @@ impl Service {
 	///
 	/// TODO: use this?
 	#[allow(dead_code)]
-	pub async fn is_exclusive_room_id(&self, room_id: &RoomId) -> bool {
+	pub fn is_exclusive_room_id(&self, room_id: &RoomId) -> bool {
 		self.read()
-			.await
 			.values()
 			.any(|info| info.rooms.is_exclusive_match(room_id.as_str()))
 	}
 
 	pub fn iter_ids(&self) -> impl Stream<Item = String> + Send {
-		self.read()
-			.map(|info| info.keys().cloned().collect::<Vec<_>>())
-			.map(IntoIterator::into_iter)
-			.map(IterStream::stream)
-			.flatten_stream()
+		let ids: Vec<String> = self.read().keys().cloned().collect();
+		ids.stream()
 	}
 
 	pub fn iter_db_ids(&self) -> impl Stream<Item = Result<(String, Registration)>> + Send {
@@ -281,7 +267,5 @@ impl Service {
 			})
 	}
 
-	pub fn read(&self) -> impl Future<Output = RwLockReadGuard<'_, Registrations>> + Send {
-		self.registration_info.read()
-	}
+	pub fn read(&self) -> SyncRwLockReadGuard<'_, Registrations> { self.registration_info.read() }
 }

src/service/migrations.rs

@@ -853,7 +853,7 @@ async fn split_userid_password(services: &Services) -> Result {
 				userid_password.remove(&user_id);
 				remote_users = remote_users.saturating_add(1);
 			} else if hash.is_empty() {
-				if !(services.appservice.is_user_id(&user_id).await
+				if !(services.appservice.is_user_id(&user_id)
 					|| user_id == services.globals.server_user)
 				{
 					info!("Marking {user_id} as deactivated");

src/service/rooms/alias/mod.rs

@@ -269,18 +269,26 @@ impl Service {
 	) -> Result<Option<OwnedRoomId>> {
 		use ruma::api::appservice::query::query_room_alias;
 
-		for appservice in self.services.appservice.read().await.values() {
-			if appservice.aliases.is_match(room_alias.as_str())
-				&& matches!(
-					self.services
-						.sending
-						.send_appservice_request(
-							appservice.registration.clone(),
-							query_room_alias::v1::Request::new(room_alias.to_owned()),
-						)
-						.await,
-					Ok(Some(_opt_result))
-				) {
+		let matching_appservices: Vec<_> = self
+			.services
+			.appservice
+			.read()
+			.values()
+			.filter(|appservice| appservice.aliases.is_match(room_alias.as_str()))
+			.cloned()
+			.collect();
+
+		for appservice in matching_appservices {
+			if matches!(
+				self.services
+					.sending
+					.send_appservice_request(
+						appservice.registration.clone(),
+						query_room_alias::v1::Request::new(room_alias.to_owned()),
+					)
+					.await,
+				Ok(Some(_opt_result))
+			) {
 				return self
 					.resolve_local_alias(room_alias)
 					.await
@@ -292,7 +300,7 @@ impl Service {
 		Ok(None)
 	}
 
-	pub async fn appservice_checks(
+	pub fn appservice_checks(
 		&self,
 		room_alias: &RoomAliasId,
 		appservice_info: Option<&RegistrationInfo>,
@@ -309,12 +317,7 @@ impl Service {
 			if !info.aliases.is_match(room_alias.as_str()) {
 				return Err!(Request(Exclusive("Room alias is not in namespace.")));
 			}
-		} else if self
-			.services
-			.appservice
-			.is_exclusive_alias(room_alias)
-			.await
-		{
+		} else if self.services.appservice.is_exclusive_alias(room_alias) {
 			return Err!(Request(Exclusive("Room alias reserved by appservice.")));
 		}
 

src/service/rooms/timeline/append.rs

@@ -809,7 +809,6 @@ impl super::Service {
 			.services
 			.appservice
 			.read()
-			.await
 			.values()
 			.map(ToOwned::to_owned) // TODO: is this to_owned expensive?
 			.collect::<Vec<_>>();

src/service/rooms/typing/mod.rs

@@ -1,7 +1,7 @@
 use std::{collections::BTreeMap, sync::Arc};
 
 use conduwuit::{
-	Result, Server, debug_info,
+	Result, Server, SyncRwLock, debug_info,
 	result::LogErr,
 	utils::{self, IterStream},
 };
@@ -14,7 +14,6 @@ use ruma::{
 	},
 	events::{EphemeralRoomEvent, SyncEphemeralRoomEvent, typing::TypingEventContent},
 };
-use tokio::sync::RwLock;
 
 use crate::{Dep, globals, sending, sending::EduBuf, sync, users};
 
@@ -22,9 +21,9 @@ pub struct Service {
 	server: Arc<Server>,
 	services: Services,
 	/// u64 is unix timestamp of timeout
-	pub typing: RwLock<BTreeMap<OwnedRoomId, BTreeMap<OwnedUserId, u64>>>,
+	pub typing: SyncRwLock<BTreeMap<OwnedRoomId, BTreeMap<OwnedUserId, u64>>>,
 	/// timestamp of the last change to typing users
-	pub last_typing_update: RwLock<BTreeMap<OwnedRoomId, u64>>,
+	pub last_typing_update: SyncRwLock<BTreeMap<OwnedRoomId, u64>>,
 }
 
 struct Services {
@@ -44,8 +43,8 @@ impl crate::Service for Service {
 				sync: args.depend::<sync::Service>("sync"),
 				users: args.depend::<users::Service>("users"),
 			},
-			typing: RwLock::new(BTreeMap::new()),
-			last_typing_update: RwLock::new(BTreeMap::new()),
+			typing: SyncRwLock::new(BTreeMap::new()),
+			last_typing_update: SyncRwLock::new(BTreeMap::new()),
 		}))
 	}
 
@@ -65,14 +64,12 @@ impl Service {
 		// update clients
 		self.typing
 			.write()
-			.await
 			.entry(room_id.to_owned())
 			.or_default()
 			.insert(user_id.to_owned(), timeout);
 
 		self.last_typing_update
 			.write()
-			.await
 			.insert(room_id.to_owned(), self.services.globals.next_count()?);
 
 		self.services.sync.wake_all_joined(room_id).await;
@@ -94,14 +91,12 @@ impl Service {
 		// update clients
 		self.typing
 			.write()
-			.await
 			.entry(room_id.to_owned())
 			.or_default()
 			.remove(user_id);
 
 		self.last_typing_update
 			.write()
-			.await
 			.insert(room_id.to_owned(), self.services.globals.next_count()?);
 
 		self.services.sync.wake_all_joined(room_id).await;
@@ -123,7 +118,7 @@ impl Service {
 		let mut removable = Vec::new();
 
 		{
-			let typing = self.typing.read().await;
+			let typing = self.typing.read();
 			let Some(room) = typing.get(room_id) else {
 				return Ok(());
 			};
@@ -139,7 +134,7 @@ impl Service {
 			return Ok(());
 		}
 		{
-			let mut typing = self.typing.write().await;
+			let mut typing = self.typing.write();
 			let room = typing.entry(room_id.to_owned()).or_default();
 			for user in &removable {
 				debug_info!("typing timeout {user:?} in {room_id:?}");
@@ -150,7 +145,6 @@ impl Service {
 		// update clients
 		self.last_typing_update
 			.write()
-			.await
 			.insert(room_id.to_owned(), self.services.globals.next_count()?);
 
 		self.services.sync.wake_all_joined(room_id).await;
@@ -174,15 +168,14 @@ impl Service {
 		Ok(self
 			.last_typing_update
 			.read()
-			.await
 			.get(room_id)
 			.copied()
 			.unwrap_or(0))
 	}
 
 	/// Returns a new typing EDU's content.
-	pub async fn typings_content(&self, room_id: &RoomId) -> TypingEventContent {
-		let room_typing_indicators = self.typing.read().await.get(room_id).cloned();
+	pub fn typings_content(&self, room_id: &RoomId) -> TypingEventContent {
+		let room_typing_indicators = self.typing.read().get(room_id).cloned();
 
 		let Some(typing_indicators) = room_typing_indicators else {
 			return TypingEventContent::new(Vec::new());
@@ -203,7 +196,7 @@ impl Service {
 		room_id: &RoomId,
 		sender_user: &UserId,
 	) -> Result<Vec<OwnedUserId>> {
-		let room_typing_indicators = self.typing.read().await.get(room_id).cloned();
+		let room_typing_indicators = self.typing.read().get(room_id).cloned();
 
 		let Some(typing_indicators) = room_typing_indicators else {
 			return Ok(Vec::new());
@@ -265,7 +258,7 @@ impl Service {
 	async fn appservice_send(&self, room_id: &RoomId) -> Result<()> {
 		let edu = EphemeralData::Typing(EphemeralRoomEvent::new(
 			room_id.to_owned(),
-			self.typings_content(room_id).await,
+			self.typings_content(room_id),
 		));
 
 		let mut buf = EduBuf::new();

src/service/sending/mod.rs

@@ -270,7 +270,6 @@ impl Service {
 			.services
 			.appservice
 			.read()
-			.await
 			.values()
 			.filter(|appservice| appservice.registration.receive_ephemeral && filter(appservice))
 			.cloned()

src/service/sending/sender.rs

@@ -810,7 +810,7 @@ impl Service {
 		id: String,
 		events: Vec<SendingEvent>,
 	) -> SendingResult {
-		let Some(appservice) = self.services.appservice.get_registration(&id).await else {
+		let Some(appservice) = self.services.appservice.get_registration(&id) else {
 			return Err((
 				Destination::Appservice(id.clone()),
 				Box::new(err!(Database(warn!(?id, "Missing appservice registration")))),

src/service/sync/mod.rs

@@ -8,7 +8,7 @@ use futures::StreamExt;
 use ruma::{
 	OwnedDeviceId, OwnedRoomId, OwnedUserId, RoomId, UserId, api::client::sync::sync_events::v5,
 };
-use tokio::sync::{Mutex, watch};
+use tokio::sync::watch;
 
 use crate::{Dep, rooms};
 
@@ -71,28 +71,27 @@ pub type SnakeConnectionsKey = (OwnedUserId, OwnedDeviceId, Option<String>);
 type SnakeConnectionsVal = Arc<SyncMutex<SnakeSyncCache>>;
 
 #[derive(Default)]
-struct Wakers(Mutex<HashMap<OwnedUserId, watch::Sender<u64>>>);
+struct Wakers(SyncMutex<HashMap<OwnedUserId, watch::Sender<u64>>>);
 
 impl Wakers {
-	async fn subscribe(&self, user: &UserId) -> watch::Receiver<u64> {
+	fn subscribe(&self, user: &UserId) -> watch::Receiver<u64> {
 		self.0
 			.lock()
-			.await
 			.entry(user.to_owned())
 			.or_insert_with(|| watch::channel(0).0)
 			.subscribe()
 	}
 
-	async fn wake(&self, user: &UserId) {
-		let mut wakers = self.0.lock().await;
+	fn wake(&self, user: &UserId) {
+		let mut wakers = self.0.lock();
 		wake_waker(&mut wakers, user);
 	}
 
-	async fn wake_all<I>(&self, users: I)
+	fn wake_all<I>(&self, users: I)
 	where
-		I: IntoIterator<Item = OwnedUserId> + Send,
+		I: IntoIterator<Item = OwnedUserId>,
 	{
-		let mut wakers = self.0.lock().await;
+		let mut wakers = self.0.lock();
 		for user in users {
 			wake_waker(&mut wakers, &user);
 		}
@@ -129,8 +128,8 @@ impl crate::Service for Service {
 }
 
 impl Service {
-	pub async fn subscribe_to_wake(&self, user: &UserId) -> watch::Receiver<u64> {
-		self.wakers.subscribe(user).await
+	pub fn subscribe_to_wake(&self, user: &UserId) -> watch::Receiver<u64> {
+		self.wakers.subscribe(user)
 	}
 
 	/// Wake the target user's sync loop. Call this when something
@@ -143,7 +142,7 @@ impl Service {
 	pub async fn wake(&self, user: &UserId) {
 		trace!(?user, "Waking user's sync loops");
 
-		self.wakers.wake(user).await;
+		self.wakers.wake(user);
 	}
 
 	/// Wake all of our users who are joined to the specified room.
@@ -156,7 +155,7 @@ impl Service {
 			.collect()
 			.await;
 
-		self.wakers.wake_all(users).await;
+		self.wakers.wake_all(users);
 	}
 
 	pub fn snake_connection_cached(&self, key: &SnakeConnectionsKey) -> bool {
@@ -359,11 +358,11 @@ mod tests {
 	async fn wake_after_subscription_is_observable() {
 		let wakers = Wakers::default();
 		let user = UserId::parse("@alice:example.com").unwrap();
-		let mut receiver = wakers.subscribe(&user).await;
+		let mut receiver = wakers.subscribe(&user);
 		let generation = *receiver.borrow_and_update();
 
-		wakers.wake(&user).await;
-		wakers.wake(&user).await;
+		wakers.wake(&user);
+		wakers.wake(&user);
 
 		receiver.changed().await.unwrap();
 		assert_eq!(*receiver.borrow_and_update(), generation.wrapping_add(2));
@@ -373,10 +372,10 @@ mod tests {
 	async fn wake_notifies_all_subscribers() {
 		let wakers = Wakers::default();
 		let user = UserId::parse("@alice:example.com").unwrap();
-		let mut first = wakers.subscribe(&user).await;
-		let mut second = wakers.subscribe(&user).await;
+		let mut first = wakers.subscribe(&user);
+		let mut second = wakers.subscribe(&user);
 
-		wakers.wake(&user).await;
+		wakers.wake(&user);
 
 		first.changed().await.unwrap();
 		second.changed().await.unwrap();
@@ -387,24 +386,24 @@ mod tests {
 		let wakers = Wakers::default();
 		let user = UserId::parse("@alice:example.com").unwrap();
 
-		wakers.wake(&user).await;
-		assert!(wakers.0.lock().await.is_empty());
+		wakers.wake(&user);
+		assert!(wakers.0.lock().is_empty());
 
-		drop(wakers.subscribe(&user).await);
-		assert_eq!(wakers.0.lock().await.len(), 1);
+		drop(wakers.subscribe(&user));
+		assert_eq!(wakers.0.lock().len(), 1);
 
-		wakers.wake(&user).await;
-		assert!(wakers.0.lock().await.is_empty());
+		wakers.wake(&user);
+		assert!(wakers.0.lock().is_empty());
 	}
 
 	#[tokio::test]
 	async fn subscriber_still_woken_after_a_sibling_disconnects() {
 		let wakers = Wakers::default();
 		let user = UserId::parse("@alice:example.com").unwrap();
-		let mut receiver = wakers.subscribe(&user).await;
-		drop(wakers.subscribe(&user).await);
+		let mut receiver = wakers.subscribe(&user);
+		drop(wakers.subscribe(&user));
 
-		wakers.wake(&user).await;
+		wakers.wake(&user);
 
 		receiver.changed().await.unwrap();
 	}

src/service/users/account.rs

@@ -405,11 +405,7 @@ impl super::Service {
 						"Username is not in this appservice's namespace."
 					)));
 				}
-			} else if self
-				.services
-				.appservice
-				.is_exclusive_user_id(&user_id)
-				.await
+			} else if self.services.appservice.is_exclusive_user_id(&user_id)
 				&& !emergency_mode_enabled
 			{
 				return Err!(Request(Exclusive("Username is reserved by an appservice.")));

src/service/users/device.rs

@@ -175,13 +175,7 @@ impl super::Service {
 		}
 
 		// Check for token collision with appservices
-		if self
-			.services
-			.appservice
-			.find_from_token(&token)
-			.await
-			.is_ok()
-		{
+		if self.services.appservice.find_from_token(&token).is_ok() {
 			return Err!(Request(InvalidParam(
 				"Token conflicts with an existing appservice token"
 			)));