Merge branch 'performance' into 'next'

improvement: do not save typing edus in db

See merge request famedly/conduit!597

src/api/client_server/sync.rs

@@ -1055,10 +1055,17 @@ async fn load_joined_room(
         .map(|(_, _, v)| v)
         .collect();
 
-    if services().rooms.edus.typing.last_typing_update(room_id)? > since {
+    if services()
+        .rooms
+        .edus
+        .typing
+        .last_typing_update(room_id)
+        .await?
+        > since
+    {
         edus.push(
             serde_json::from_str(
-                &serde_json::to_string(&services().rooms.edus.typing.typings_all(room_id)?)
+                &serde_json::to_string(&services().rooms.edus.typing.typings_all(room_id).await?)
                     .expect("event is valid, we just created it"),
             )
             .expect("event is valid, we just created it"),

src/api/client_server/typing.rs

@@ -23,17 +23,23 @@ pub async fn create_typing_event_route(
     }
 
     if let Typing::Yes(duration) = body.state {
-        services().rooms.edus.typing.typing_add(
+        services()
+            .rooms
+            .edus
+            .typing
+            .typing_add(
                 sender_user,
                 &body.room_id,
                 duration.as_millis() as u64 + utils::millis_since_unix_epoch(),
-        )?;
+            )
+            .await?;
     } else {
         services()
             .rooms
             .edus
             .typing
-            .typing_remove(sender_user, &body.room_id)?;
+            .typing_remove(sender_user, &body.room_id)
+            .await?;
     }
 
     Ok(create_typing_event::v3::Response {})

src/api/server_server.rs

@@ -826,17 +826,23 @@ pub async fn send_transaction_message_route(
                     .is_joined(&typing.user_id, &typing.room_id)?
                 {
                     if typing.typing {
-                        services().rooms.edus.typing.typing_add(
+                        services()
+                            .rooms
+                            .edus
+                            .typing
+                            .typing_add(
                                 &typing.user_id,
                                 &typing.room_id,
                                 3000 + utils::millis_since_unix_epoch(),
-                        )?;
+                            )
+                            .await?;
                     } else {
                         services()
                             .rooms
                             .edus
                             .typing
-                            .typing_remove(&typing.user_id, &typing.room_id)?;
+                            .typing_remove(&typing.user_id, &typing.room_id)
+                            .await?;
                     }
                 }
             }

src/database/key_value/globals.rs

@@ -94,7 +94,9 @@ impl service::globals::Data for KeyValueDatabase {
             futures.push(self.pduid_pdu.watch_prefix(&short_roomid));
 
             // EDUs
-            futures.push(self.roomid_lasttypingupdate.watch_prefix(&roomid_bytes));
+            futures.push(Box::into_pin(Box::new(async move {
+                let _result = services().rooms.edus.typing.wait_for_update(&room_id).await;
+            })));
 
             futures.push(self.readreceiptid_readreceipt.watch_prefix(&roomid_prefix));
 

src/database/key_value/rooms/edus/mod.rs

@@ -1,6 +1,5 @@
 mod presence;
 mod read_receipt;
-mod typing;
 
 use crate::{database::KeyValueDatabase, service};
 

src/database/key_value/rooms/edus/typing.rs

@@ -1,127 +0,0 @@
-use std::{collections::HashSet, mem};
-
-use ruma::{OwnedUserId, RoomId, UserId};
-
-use crate::{database::KeyValueDatabase, service, services, utils, Error, Result};
-
-impl service::rooms::edus::typing::Data for KeyValueDatabase {
-    fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> {
-        let mut prefix = room_id.as_bytes().to_vec();
-        prefix.push(0xff);
-
-        let count = services().globals.next_count()?.to_be_bytes();
-
-        let mut room_typing_id = prefix;
-        room_typing_id.extend_from_slice(&timeout.to_be_bytes());
-        room_typing_id.push(0xff);
-        room_typing_id.extend_from_slice(&count);
-
-        self.typingid_userid
-            .insert(&room_typing_id, user_id.as_bytes())?;
-
-        self.roomid_lasttypingupdate
-            .insert(room_id.as_bytes(), &count)?;
-
-        Ok(())
-    }
-
-    fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
-        let mut prefix = room_id.as_bytes().to_vec();
-        prefix.push(0xff);
-
-        let user_id = user_id.to_string();
-
-        let mut found_outdated = false;
-
-        // Maybe there are multiple ones from calling roomtyping_add multiple times
-        for outdated_edu in self
-            .typingid_userid
-            .scan_prefix(prefix)
-            .filter(|(_, v)| &**v == user_id.as_bytes())
-        {
-            self.typingid_userid.remove(&outdated_edu.0)?;
-            found_outdated = true;
-        }
-
-        if found_outdated {
-            self.roomid_lasttypingupdate.insert(
-                room_id.as_bytes(),
-                &services().globals.next_count()?.to_be_bytes(),
-            )?;
-        }
-
-        Ok(())
-    }
-
-    fn typings_maintain(&self, room_id: &RoomId) -> Result<()> {
-        let mut prefix = room_id.as_bytes().to_vec();
-        prefix.push(0xff);
-
-        let current_timestamp = utils::millis_since_unix_epoch();
-
-        let mut found_outdated = false;
-
-        // Find all outdated edus before inserting a new one
-        for outdated_edu in self
-            .typingid_userid
-            .scan_prefix(prefix)
-            .map(|(key, _)| {
-                Ok::<_, Error>((
-                    key.clone(),
-                    utils::u64_from_bytes(
-                        &key.splitn(2, |&b| b == 0xff).nth(1).ok_or_else(|| {
-                            Error::bad_database("RoomTyping has invalid timestamp or delimiters.")
-                        })?[0..mem::size_of::<u64>()],
-                    )
-                    .map_err(|_| Error::bad_database("RoomTyping has invalid timestamp bytes."))?,
-                ))
-            })
-            .filter_map(|r| r.ok())
-            .take_while(|&(_, timestamp)| timestamp < current_timestamp)
-        {
-            // This is an outdated edu (time > timestamp)
-            self.typingid_userid.remove(&outdated_edu.0)?;
-            found_outdated = true;
-        }
-
-        if found_outdated {
-            self.roomid_lasttypingupdate.insert(
-                room_id.as_bytes(),
-                &services().globals.next_count()?.to_be_bytes(),
-            )?;
-        }
-
-        Ok(())
-    }
-
-    fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
-        Ok(self
-            .roomid_lasttypingupdate
-            .get(room_id.as_bytes())?
-            .map(|bytes| {
-                utils::u64_from_bytes(&bytes).map_err(|_| {
-                    Error::bad_database("Count in roomid_lastroomactiveupdate is invalid.")
-                })
-            })
-            .transpose()?
-            .unwrap_or(0))
-    }
-
-    fn typings_all(&self, room_id: &RoomId) -> Result<HashSet<OwnedUserId>> {
-        let mut prefix = room_id.as_bytes().to_vec();
-        prefix.push(0xff);
-
-        let mut user_ids = HashSet::new();
-
-        for (_, user_id) in self.typingid_userid.scan_prefix(prefix) {
-            let user_id = UserId::parse(utils::string_from_bytes(&user_id).map_err(|_| {
-                Error::bad_database("User ID in typingid_userid is invalid unicode.")
-            })?)
-            .map_err(|_| Error::bad_database("User ID in typingid_userid is invalid."))?;
-
-            user_ids.insert(user_id);
-        }
-
-        Ok(user_ids)
-    }
-}

src/database/mod.rs

@@ -70,8 +70,6 @@ pub struct KeyValueDatabase {
     pub(super) readreceiptid_readreceipt: Arc<dyn KvTree>, // ReadReceiptId = RoomId + Count + UserId
     pub(super) roomuserid_privateread: Arc<dyn KvTree>, // RoomUserId = Room + User, PrivateRead = Count
     pub(super) roomuserid_lastprivatereadupdate: Arc<dyn KvTree>, // LastPrivateReadUpdate = Count
-    pub(super) typingid_userid: Arc<dyn KvTree>,        // TypingId = RoomId + TimeoutTime + Count
-    pub(super) roomid_lasttypingupdate: Arc<dyn KvTree>, // LastRoomTypingUpdate = Count
     pub(super) presenceid_presence: Arc<dyn KvTree>,    // PresenceId = RoomId + Count + UserId
     pub(super) userid_lastpresenceupdate: Arc<dyn KvTree>, // LastPresenceUpdate = Count
 
@@ -301,8 +299,6 @@ impl KeyValueDatabase {
             roomuserid_privateread: builder.open_tree("roomuserid_privateread")?, // "Private" read receipt
             roomuserid_lastprivatereadupdate: builder
                 .open_tree("roomuserid_lastprivatereadupdate")?,
-            typingid_userid: builder.open_tree("typingid_userid")?,
-            roomid_lasttypingupdate: builder.open_tree("roomid_lasttypingupdate")?,
             presenceid_presence: builder.open_tree("presenceid_presence")?,
             userid_lastpresenceupdate: builder.open_tree("userid_lastpresenceupdate")?,
             pduid_pdu: builder.open_tree("pduid_pdu")?,

src/service/mod.rs

@@ -4,9 +4,10 @@ use std::{
 };
 
 use lru_cache::LruCache;
-use tokio::sync::Mutex;
+use tokio::sync::{broadcast, Mutex};
 
 use crate::{Config, Result};
+use tokio::sync::RwLock;
 
 pub mod account_data;
 pub mod admin;
@@ -65,7 +66,11 @@ impl Services {
                 edus: rooms::edus::Service {
                     presence: rooms::edus::presence::Service { db },
                     read_receipt: rooms::edus::read_receipt::Service { db },
-                    typing: rooms::edus::typing::Service { db },
+                    typing: rooms::edus::typing::Service {
+                        typing: RwLock::new(BTreeMap::new()),
+                        last_typing_update: RwLock::new(BTreeMap::new()),
+                        typing_update_sender: broadcast::channel(100).0,
+                    },
                 },
                 event_handler: rooms::event_handler::Service,
                 lazy_loading: rooms::lazy_loading::Service {

src/service/rooms/edus/mod.rs

@@ -2,7 +2,7 @@ pub mod presence;
 pub mod read_receipt;
 pub mod typing;
 
-pub trait Data: presence::Data + read_receipt::Data + typing::Data + 'static {}
+pub trait Data: presence::Data + read_receipt::Data + 'static {}
 
 pub struct Service {
     pub presence: presence::Service,

src/service/rooms/edus/typing/data.rs

@@ -1,21 +0,0 @@
-use crate::Result;
-use ruma::{OwnedUserId, RoomId, UserId};
-use std::collections::HashSet;
-
-pub trait Data: Send + Sync {
-    /// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
-    /// called.
-    fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()>;
-
-    /// Removes a user from typing before the timeout is reached.
-    fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()>;
-
-    /// Makes sure that typing events with old timestamps get removed.
-    fn typings_maintain(&self, room_id: &RoomId) -> Result<()>;
-
-    /// Returns the count of the last typing update in this room.
-    fn last_typing_update(&self, room_id: &RoomId) -> Result<u64>;
-
-    /// Returns all user ids currently typing.
-    fn typings_all(&self, room_id: &RoomId) -> Result<HashSet<OwnedUserId>>;
-}

src/service/rooms/edus/typing/mod.rs

@@ -1,48 +1,117 @@
-mod data;
+use ruma::{events::SyncEphemeralRoomEvent, OwnedRoomId, OwnedUserId, RoomId, UserId};
+use std::collections::BTreeMap;
+use tokio::sync::{broadcast, RwLock};
 
-pub use data::Data;
+use crate::{services, utils, Result};
-use ruma::{events::SyncEphemeralRoomEvent, RoomId, UserId};
-
-use crate::Result;
 
 pub struct Service {
-    pub db: &'static dyn Data,
+    pub typing: RwLock<BTreeMap<OwnedRoomId, BTreeMap<OwnedUserId, u64>>>, // u64 is unix timestamp of timeout
+    pub last_typing_update: RwLock<BTreeMap<OwnedRoomId, u64>>, // timestamp of the last change to typing users
+    pub typing_update_sender: broadcast::Sender<OwnedRoomId>,
 }
 
 impl Service {
     /// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
     /// called.
-    pub fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> {
+    pub async fn typing_add(&self, user_id: &UserId, room_id: &RoomId, timeout: u64) -> Result<()> {
-        self.db.typing_add(user_id, room_id, timeout)
+        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(), services().globals.next_count()?);
+        let _ = self.typing_update_sender.send(room_id.to_owned());
+        Ok(())
     }
 
     /// Removes a user from typing before the timeout is reached.
-    pub fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
+    pub async fn typing_remove(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
-        self.db.typing_remove(user_id, room_id)
+        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(), services().globals.next_count()?);
+        let _ = self.typing_update_sender.send(room_id.to_owned());
+        Ok(())
+    }
+
+    pub async fn wait_for_update(&self, room_id: &RoomId) -> Result<()> {
+        let mut receiver = self.typing_update_sender.subscribe();
+        while let Ok(next) = receiver.recv().await {
+            if next == room_id {
+                break;
+            }
+        }
+
+        Ok(())
     }
 
     /// Makes sure that typing events with old timestamps get removed.
-    fn typings_maintain(&self, room_id: &RoomId) -> Result<()> {
+    async fn typings_maintain(&self, room_id: &RoomId) -> Result<()> {
-        self.db.typings_maintain(room_id)
+        let current_timestamp = utils::millis_since_unix_epoch();
+        let mut removable = Vec::new();
+        {
+            let typing = self.typing.read().await;
+            let Some(room) = typing.get(room_id) else {
+                return Ok(());
+            };
+            for (user, timeout) in room {
+                if *timeout < current_timestamp {
+                    removable.push(user.clone());
+                }
+            }
+            drop(typing);
+        }
+        if !removable.is_empty() {
+            let typing = &mut self.typing.write().await;
+            let room = typing.entry(room_id.to_owned()).or_default();
+            for user in removable {
+                room.remove(&user);
+            }
+            self.last_typing_update
+                .write()
+                .await
+                .insert(room_id.to_owned(), services().globals.next_count()?);
+            let _ = self.typing_update_sender.send(room_id.to_owned());
+        }
+        Ok(())
     }
 
     /// Returns the count of the last typing update in this room.
-    pub fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
+    pub async fn last_typing_update(&self, room_id: &RoomId) -> Result<u64> {
-        self.typings_maintain(room_id)?;
+        self.typings_maintain(room_id).await?;
-
+        Ok(self
-        self.db.last_typing_update(room_id)
+            .last_typing_update
+            .read()
+            .await
+            .get(room_id)
+            .copied()
+            .unwrap_or(0))
     }
 
     /// Returns a new typing EDU.
-    pub fn typings_all(
+    pub async fn typings_all(
         &self,
         room_id: &RoomId,
     ) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
-        let user_ids = self.db.typings_all(room_id)?;
-
         Ok(SyncEphemeralRoomEvent {
             content: ruma::events::typing::TypingEventContent {
-                user_ids: user_ids.into_iter().collect(),
+                user_ids: self
+                    .typing
+                    .read()
+                    .await
+                    .get(room_id)
+                    .map(|m| m.keys().cloned().collect())
+                    .unwrap_or_default(),
             },
         })
     }