From c738c119f85bcdacc52666132746a8073ad6bac8 Mon Sep 17 00:00:00 2001 From: strawberry Date: Tue, 4 Jun 2024 01:13:43 -0400 Subject: [PATCH] delete unnecessary `real_users_cache`, fix overwriting push_target iter, add proper function for getting local active users in room this `real_users_cache` cache seems weird, and i have no idea what prompted its creation upstream. perhaps they did this because sqlite was very slow and their rocksdb setup is very poor, so a "solution" was to stick member counts in memory. slow iterators, scanning, etc do not apply to conduwuit where our rocksdb is extremely tuned, and i seriously doubt something like this would have any real world net-positive performance impact. also for some reason, there is suspicious logic where we overwrite the entire push target collection. both of these things could be a potential cause for receiving notifications in rooms we've left. Signed-off-by: strawberry --- src/database/kvdatabase.rs | 4 +-- src/service/globals/data.rs | 9 +---- src/service/rooms/state_cache/data.rs | 49 ++++++++------------------- src/service/rooms/state_cache/mod.rs | 14 ++++---- src/service/rooms/timeline/mod.rs | 8 ++--- src/service/sending/sender.rs | 4 +-- 6 files changed, 30 insertions(+), 58 deletions(-) diff --git a/src/database/kvdatabase.rs b/src/database/kvdatabase.rs index ddbb8e92..906712dc 100644 --- a/src/database/kvdatabase.rs +++ b/src/database/kvdatabase.rs @@ -1,5 +1,5 @@ use std::{ - collections::{BTreeMap, HashMap, HashSet}, + collections::{BTreeMap, HashMap}, path::Path, sync::{Arc, Mutex, RwLock}, }; @@ -150,7 +150,6 @@ pub struct KeyValueDatabase { pub senderkey_pusher: Arc, pub auth_chain_cache: Mutex, Arc<[u64]>>>, - pub our_real_users_cache: RwLock>>>, pub appservice_in_room_cache: RwLock>>, pub lasttimelinecount_cache: Mutex>, } @@ -265,7 +264,6 @@ impl KeyValueDatabase { auth_chain_cache: Mutex::new(LruCache::new( (f64::from(config.auth_chain_cache_capacity) * config.conduit_cache_capacity_modifier) as usize, )), - our_real_users_cache: RwLock::new(HashMap::new()), appservice_in_room_cache: RwLock::new(HashMap::new()), lasttimelinecount_cache: Mutex::new(HashMap::new()), }) diff --git a/src/service/globals/data.rs b/src/service/globals/data.rs index 5b26598b..354ff2f7 100644 --- a/src/service/globals/data.rs +++ b/src/service/globals/data.rs @@ -177,19 +177,16 @@ impl Data for KeyValueDatabase { fn memory_usage(&self) -> String { let auth_chain_cache = self.auth_chain_cache.lock().unwrap().len(); - let our_real_users_cache = self.our_real_users_cache.read().unwrap().len(); let appservice_in_room_cache = self.appservice_in_room_cache.read().unwrap().len(); let lasttimelinecount_cache = self.lasttimelinecount_cache.lock().unwrap().len(); let max_auth_chain_cache = self.auth_chain_cache.lock().unwrap().capacity(); - let max_our_real_users_cache = self.our_real_users_cache.read().unwrap().capacity(); let max_appservice_in_room_cache = self.appservice_in_room_cache.read().unwrap().capacity(); let max_lasttimelinecount_cache = self.lasttimelinecount_cache.lock().unwrap().capacity(); format!( "\ auth_chain_cache: {auth_chain_cache} / {max_auth_chain_cache} -our_real_users_cache: {our_real_users_cache} / {max_our_real_users_cache} appservice_in_room_cache: {appservice_in_room_cache} / {max_appservice_in_room_cache} lasttimelinecount_cache: {lasttimelinecount_cache} / {max_lasttimelinecount_cache}\n\n {}", @@ -203,14 +200,10 @@ lasttimelinecount_cache: {lasttimelinecount_cache} / {max_lasttimelinecount_cach *c = LruCache::new(c.capacity()); } if amount > 2 { - let c = &mut *self.our_real_users_cache.write().unwrap(); - *c = HashMap::new(); - } - if amount > 3 { let c = &mut *self.appservice_in_room_cache.write().unwrap(); *c = HashMap::new(); } - if amount > 4 { + if amount > 3 { let c = &mut *self.lasttimelinecount_cache.lock().unwrap(); *c = HashMap::new(); } diff --git a/src/service/rooms/state_cache/data.rs b/src/service/rooms/state_cache/data.rs index 08e93d89..5694b22c 100644 --- a/src/service/rooms/state_cache/data.rs +++ b/src/service/rooms/state_cache/data.rs @@ -1,4 +1,4 @@ -use std::{collections::HashSet, sync::Arc}; +use std::collections::HashSet; use itertools::Itertools; use ruma::{ @@ -29,8 +29,6 @@ pub trait Data: Send + Sync { fn update_joined_count(&self, room_id: &RoomId) -> Result<()>; - fn get_our_real_users(&self, room_id: &RoomId) -> Result>>; - fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> Result; /// Makes a user forget a room. @@ -48,6 +46,10 @@ pub trait Data: Send + Sync { /// Returns an iterator over all joined members of a room. fn room_members<'a>(&'a self, room_id: &RoomId) -> Box> + 'a>; + /// Returns a vec of all the users joined in a room who are active + /// (not guests, not deactivated users) + fn active_local_users_in_room(&self, room_id: &RoomId) -> Vec; + fn room_joined_count(&self, room_id: &RoomId) -> Result>; fn room_invited_count(&self, room_id: &RoomId) -> Result>; @@ -225,13 +227,9 @@ impl Data for KeyValueDatabase { let mut joinedcount = 0_u64; let mut invitedcount = 0_u64; let mut joined_servers = HashSet::new(); - let mut real_users = HashSet::new(); for joined in self.room_members(room_id).filter_map(Result::ok) { joined_servers.insert(joined.server_name().to_owned()); - if user_is_local(&joined) && !services().users.is_deactivated(&joined).unwrap_or(true) { - real_users.insert(joined); - } joinedcount = joinedcount.saturating_add(1); } @@ -245,11 +243,6 @@ impl Data for KeyValueDatabase { self.roomid_invitedcount .insert(room_id.as_bytes(), &invitedcount.to_be_bytes())?; - self.our_real_users_cache - .write() - .unwrap() - .insert(room_id.to_owned(), Arc::new(real_users)); - for old_joined_server in self.room_servers(room_id).filter_map(Result::ok) { if !joined_servers.remove(&old_joined_server) { // Server not in room anymore @@ -288,28 +281,6 @@ impl Data for KeyValueDatabase { Ok(()) } - #[tracing::instrument(skip(self, room_id))] - fn get_our_real_users(&self, room_id: &RoomId) -> Result>> { - let maybe = self - .our_real_users_cache - .read() - .unwrap() - .get(room_id) - .cloned(); - if let Some(users) = maybe { - Ok(users) - } else { - self.update_joined_count(room_id)?; - Ok(Arc::clone( - self.our_real_users_cache - .read() - .unwrap() - .get(room_id) - .unwrap(), - )) - } - } - #[tracing::instrument(skip(self, room_id, appservice))] fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> Result { let maybe = self @@ -429,6 +400,16 @@ impl Data for KeyValueDatabase { })) } + /// Returns a vec of all our local users joined in a room who are active + /// (not guests / not deactivated users) + #[tracing::instrument(skip(self))] + fn active_local_users_in_room(&self, room_id: &RoomId) -> Vec { + self.room_members(room_id) + .filter_map(Result::ok) + .filter(|user| user_is_local(user) && !services().users.is_deactivated(user).unwrap_or(true)) + .collect_vec() + } + /// Returns the number of users which are currently in a room #[tracing::instrument(skip(self))] fn room_joined_count(&self, room_id: &RoomId) -> Result> { diff --git a/src/service/rooms/state_cache/mod.rs b/src/service/rooms/state_cache/mod.rs index c9ac278c..ab59c0ad 100644 --- a/src/service/rooms/state_cache/mod.rs +++ b/src/service/rooms/state_cache/mod.rs @@ -1,4 +1,4 @@ -use std::{collections::HashSet, sync::Arc}; +use std::sync::Arc; use data::Data; use itertools::Itertools; @@ -212,11 +212,6 @@ impl Service { #[tracing::instrument(skip(self, room_id))] pub fn update_joined_count(&self, room_id: &RoomId) -> Result<()> { self.db.update_joined_count(room_id) } - #[tracing::instrument(skip(self, room_id))] - pub fn get_our_real_users(&self, room_id: &RoomId) -> Result>> { - self.db.get_our_real_users(room_id) - } - #[tracing::instrument(skip(self, room_id, appservice))] pub fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> Result { self.db.appservice_in_room(room_id, appservice) @@ -278,6 +273,13 @@ impl Service { #[tracing::instrument(skip(self))] pub fn room_joined_count(&self, room_id: &RoomId) -> Result> { self.db.room_joined_count(room_id) } + #[tracing::instrument(skip(self))] + /// Returns a vec of all the users joined in a room who are active + /// (not guests, not deactivated users) + pub fn active_local_users_in_room(&self, room_id: &RoomId) -> Vec { + self.db.active_local_users_in_room(room_id) + } + #[tracing::instrument(skip(self))] pub fn room_invited_count(&self, room_id: &RoomId) -> Result> { self.db.room_invited_count(room_id) } diff --git a/src/service/rooms/timeline/mod.rs b/src/service/rooms/timeline/mod.rs index b45b2a3b..a9482853 100644 --- a/src/service/rooms/timeline/mod.rs +++ b/src/service/rooms/timeline/mod.rs @@ -309,21 +309,19 @@ impl Service { let mut push_target = services() .rooms .state_cache - .get_our_real_users(&pdu.room_id)?; + .active_local_users_in_room(&pdu.room_id); if pdu.kind == TimelineEventType::RoomMember { if let Some(state_key) = &pdu.state_key { let target_user_id = UserId::parse(state_key.clone()).expect("This state_key was previously validated"); if !push_target.contains(&target_user_id) { - let mut target = push_target.as_ref().clone(); - target.insert(target_user_id); - push_target = Arc::new(target); + push_target.push(target_user_id); } } } - for user in push_target.iter() { + for user in &push_target { // Don't notify the user of their own events if user == &pdu.sender { continue; diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index c4de3e60..9192fdca 100644 --- a/src/service/sending/sender.rs +++ b/src/service/sending/sender.rs @@ -485,7 +485,7 @@ async fn send_events_dest_push( .ok_or_else(|| { ( dest.clone(), - Error::bad_database("[Push] Event in servernamevent_datas not found in db."), + Error::bad_database("[Push] Event in servernameevent_data not found in db."), ) })?, ); @@ -567,7 +567,7 @@ async fn send_events_dest_normal( ); ( dest.clone(), - Error::bad_database("[Normal] Event in servernamevent_datas not found in db."), + Error::bad_database("[Normal] Event in servernameevent_data not found in db."), ) })?, );