diff --git a/src/admin/appservice/commands.rs b/src/admin/appservice/commands.rs index 9bccc643d..86fa51f27 100644 --- a/src/admin/appservice/commands.rs +++ b/src/admin/appservice/commands.rs @@ -1,4 +1,4 @@ -use conduwuit::{Err, Result, checked}; +use conduwuit::{Err, Result, checked, utils::ReadyExt}; use futures::{FutureExt, StreamExt, TryFutureExt}; impl crate::Context<'_> { @@ -49,7 +49,6 @@ impl crate::Context<'_> { .services .appservice .get_registration(&appservice_identifier) - .await { | None => return Err!("Appservice does not exist."), | Some(config) => { @@ -78,12 +77,11 @@ impl crate::Context<'_> { self.services .users .stream_local_users() - .filter_map(async |user_id| { - if self.services.appservice.is_user_id(&user_id).await { - Some(user_id) - } else { - None - } + .ready_filter_map(|user_id| { + self.services + .appservice + .is_user_id(&user_id) + .then_some(user_id) }) .for_each(async |user_id| { self.services diff --git a/src/admin/query/appservice.rs b/src/admin/query/appservice.rs index e2f04acf3..45dec01e5 100644 --- a/src/admin/query/appservice.rs +++ b/src/admin/query/appservice.rs @@ -24,7 +24,7 @@ pub(super) async fn process(subcommand: AppserviceCommand, context: &Context<'_> match subcommand { | AppserviceCommand::GetRegistration { appservice_id } => { let timer = tokio::time::Instant::now(); - let results = services.appservice.get_registration(&appservice_id).await; + let results = services.appservice.get_registration(&appservice_id); let query_time = timer.elapsed(); diff --git a/src/api/client/alias.rs b/src/api/client/alias.rs index 5d4dec081..f10e03083 100644 --- a/src/api/client/alias.rs +++ b/src/api/client/alias.rs @@ -20,8 +20,7 @@ pub(crate) async fn create_alias_route( services .rooms .alias - .appservice_checks(&body.room_alias, body.identity.appservice_info()) - .await?; + .appservice_checks(&body.room_alias, body.identity.appservice_info())?; // this isn't apart of alias_checks or delete alias route because we should // allow removing forbidden room aliases @@ -69,8 +68,7 @@ pub(crate) async fn delete_alias_route( services .rooms .alias - .appservice_checks(&body.room_alias, body.identity.appservice_info()) - .await?; + .appservice_checks(&body.room_alias, body.identity.appservice_info())?; services .rooms diff --git a/src/api/client/room/create.rs b/src/api/client/room/create.rs index 582169346..a2acf5ec1 100644 --- a/src/api/client/room/create.rs +++ b/src/api/client/room/create.rs @@ -758,11 +758,7 @@ async fn room_alias_check( if !info.aliases.is_match(full_room_alias.as_str()) { return Err!(Request(Exclusive("Room alias is not in namespace."))); } - } else if services - .appservice - .is_exclusive_alias(&full_room_alias) - .await - { + } else if services.appservice.is_exclusive_alias(&full_room_alias) { return Err!(Request(Exclusive("Room alias reserved by appservice.",))); } diff --git a/src/api/client/sync/v3/mod.rs b/src/api/client/sync/v3/mod.rs index 4987534cf..e6ac768f0 100644 --- a/src/api/client/sync/v3/mod.rs +++ b/src/api/client/sync/v3/mod.rs @@ -201,7 +201,7 @@ pub(crate) async fn sync_events_route( .update_device_last_seen(sender_user, Some(sender_device), client_ip) .await; - let mut wake_receiver = services.sync.subscribe_to_wake(sender_user).await; + let mut wake_receiver = services.sync.subscribe_to_wake(sender_user); let response = build_sync_events(&services, &body).await?; if body.body.full_state || !(response.rooms.is_empty() diff --git a/src/api/client/sync/v5.rs b/src/api/client/sync/v5.rs index 7f8997aec..4f21c1930 100644 --- a/src/api/client/sync/v5.rs +++ b/src/api/client/sync/v5.rs @@ -147,7 +147,7 @@ pub(crate) async fn sync_events_v5_route( .sync .update_snake_sync_request_with_cache(&snake_key, globalsince, &mut body); - let mut wake_receiver = services.sync.subscribe_to_wake(sender_user).await; + let mut wake_receiver = services.sync.subscribe_to_wake(sender_user); let mut collection = collect_sync_response( services, sender_user, diff --git a/src/api/client/unversioned.rs b/src/api/client/unversioned.rs index 814ee0d9e..207bcb620 100644 --- a/src/api/client/unversioned.rs +++ b/src/api/client/unversioned.rs @@ -64,7 +64,7 @@ pub(crate) async fn continuwuity_local_user_count( .stream_local_users() .filter_map(async |user_id| { if !services.users.status(&user_id).await.is_active() - || services.appservice.is_exclusive_user_id(&user_id).await + || services.appservice.is_exclusive_user_id(&user_id) { None } else { diff --git a/src/api/router/auth.rs b/src/api/router/auth.rs index cac270ee0..e731245df 100644 --- a/src/api/router/auth.rs +++ b/src/api/router/auth.rs @@ -346,7 +346,7 @@ async fn check_access_token( } Ok(ClientIdentity::User { sender_user, sender_device }) - } else if let Ok(appservice_info) = services.appservice.find_from_token(token).await { + } else if let Ok(appservice_info) = services.appservice.find_from_token(token) { let Ok(sender_user) = query.user_id.clone().map_or_else( || { UserId::parse_with_server_name( @@ -405,7 +405,7 @@ async fn check_access_token( } async fn check_appservice_token(services: &Services, token: &str) -> Result { - let Ok(appservice_info) = services.appservice.find_from_token(token).await else { + let Ok(appservice_info) = services.appservice.find_from_token(token) else { return Err!(Request(Unauthorized("Invalid appservice token."))); }; diff --git a/src/api/server/invite.rs b/src/api/server/invite.rs index 89069e050..06ee9f2b1 100644 --- a/src/api/server/invite.rs +++ b/src/api/server/invite.rs @@ -209,31 +209,35 @@ async fn create_invite( services.sync.wake(&recipient_user).await; - for appservice in services.appservice.read().await.values() { - if appservice.is_user_match(&recipient_user) { - let transaction_id = general_purpose::URL_SAFE_NO_PAD - .encode(sha256::hash(pdu.event_id.as_bytes())) - .into(); + let matching_appservices: Vec<_> = services + .appservice + .read() + .values() + .filter(|appservice| appservice.is_user_match(&recipient_user)) + .cloned() + .collect(); - let request = ruma::api::appservice::event::push_events::v1::Request::new( - transaction_id, - vec![pdu.to_format()], - ); + for appservice in matching_appservices { + let transaction_id = general_purpose::URL_SAFE_NO_PAD + .encode(sha256::hash(pdu.event_id.as_bytes())) + .into(); - services - .sending - .send_appservice_request(appservice.registration.clone(), request) - .await - .map_err(|e| { - error!( - "failed to notify appservice {} about incoming invite: {e}", - appservice.registration.id - ); - err!(BadServerResponse( - "Failed to notify appservice about incoming invite." - )) - })?; - } + let request = ruma::api::appservice::event::push_events::v1::Request::new( + transaction_id, + vec![pdu.to_format()], + ); + + services + .sending + .send_appservice_request(appservice.registration.clone(), request) + .await + .map_err(|e| { + error!( + "failed to notify appservice {} about incoming invite: {e}", + appservice.registration.id + ); + err!(BadServerResponse("Failed to notify appservice about incoming invite.")) + })?; } } diff --git a/src/core/mod.rs b/src/core/mod.rs index 20d49fc50..f176593c4 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -36,7 +36,10 @@ pub use config::Config; pub use error::Error; pub use info::version::*; pub use matrix::{Event, EventTypeExt, Pdu, PduCount, PduEvent, PduId, pdu, state_res}; -pub use parking_lot::{Mutex as SyncMutex, RwLock as SyncRwLock}; +pub use parking_lot::{ + Mutex as SyncMutex, RwLock as SyncRwLock, RwLockReadGuard as SyncRwLockReadGuard, + RwLockWriteGuard as SyncRwLockWriteGuard, +}; pub use server::Server; pub use utils::{result, result::Result}; diff --git a/src/service/appservice/mod.rs b/src/service/appservice/mod.rs index 9a4d6f772..8a69921a3 100644 --- a/src/service/appservice/mod.rs +++ b/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, + registration_info: SyncRwLock, services: Services, db: Data, } @@ -35,7 +34,7 @@ type Registrations = BTreeMap; impl crate::Service for Service { fn build(args: crate::Args<'_>) -> Result> { Ok(Arc::new(Self { - registration_info: RwLock::new(BTreeMap::new()), + registration_info: SyncRwLock::new(BTreeMap::new()), services: Services { globals: args.depend::("globals"), sending: args.depend::("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(®istration.as_token).await { + if let Ok(existing) = self.find_from_token(®istration.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 { + pub fn get_registration(&self, id: &str) -> Option { 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 { + pub fn find_from_token(&self, token: &str) -> Result { 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 + Send { - self.read() - .map(|info| info.keys().cloned().collect::>()) - .map(IntoIterator::into_iter) - .map(IterStream::stream) - .flatten_stream() + let ids: Vec = self.read().keys().cloned().collect(); + ids.stream() } pub fn iter_db_ids(&self) -> impl Stream> + Send { @@ -281,7 +267,5 @@ impl Service { }) } - pub fn read(&self) -> impl Future> + Send { - self.registration_info.read() - } + pub fn read(&self) -> SyncRwLockReadGuard<'_, Registrations> { self.registration_info.read() } } diff --git a/src/service/migrations.rs b/src/service/migrations.rs index 3ed2d28a4..81ff11dd8 100644 --- a/src/service/migrations.rs +++ b/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"); diff --git a/src/service/rooms/alias/mod.rs b/src/service/rooms/alias/mod.rs index 5056a02fc..bed0fd3c7 100644 --- a/src/service/rooms/alias/mod.rs +++ b/src/service/rooms/alias/mod.rs @@ -269,18 +269,26 @@ impl Service { ) -> Result> { 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."))); } diff --git a/src/service/rooms/timeline/append.rs b/src/service/rooms/timeline/append.rs index 5f09c4638..663f98543 100644 --- a/src/service/rooms/timeline/append.rs +++ b/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::>(); diff --git a/src/service/rooms/typing/mod.rs b/src/service/rooms/typing/mod.rs index 113b3cebf..0cd6385fa 100644 --- a/src/service/rooms/typing/mod.rs +++ b/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, services: Services, /// u64 is unix timestamp of timeout - pub typing: RwLock>>, + pub typing: SyncRwLock>>, /// timestamp of the last change to typing users - pub last_typing_update: RwLock>, + pub last_typing_update: SyncRwLock>, } struct Services { @@ -44,8 +43,8 @@ impl crate::Service for Service { sync: args.depend::("sync"), users: args.depend::("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> { - 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(); diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index 8cb1010cf..cc518f5a5 100644 --- a/src/service/sending/mod.rs +++ b/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() diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 308dcbd79..0967abb38 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -810,7 +810,7 @@ impl Service { id: String, events: Vec, ) -> 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")))), diff --git a/src/service/sync/mod.rs b/src/service/sync/mod.rs index 5b50b38de..e3a07c40d 100644 --- a/src/service/sync/mod.rs +++ b/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); type SnakeConnectionsVal = Arc>; #[derive(Default)] -struct Wakers(Mutex>>); +struct Wakers(SyncMutex>>); impl Wakers { - async fn subscribe(&self, user: &UserId) -> watch::Receiver { + fn subscribe(&self, user: &UserId) -> watch::Receiver { 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(&self, users: I) + fn wake_all(&self, users: I) where - I: IntoIterator + Send, + I: IntoIterator, { - 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 { - self.wakers.subscribe(user).await + pub fn subscribe_to_wake(&self, user: &UserId) -> watch::Receiver { + 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(); } diff --git a/src/service/users/account.rs b/src/service/users/account.rs index 39cadab32..9fc78d339 100644 --- a/src/service/users/account.rs +++ b/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."))); diff --git a/src/service/users/device.rs b/src/service/users/device.rs index f5f40c57c..6933ea715 100644 --- a/src/service/users/device.rs +++ b/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" )));