Initial Glueball Refactors (#1506)

Author: PepperLola

glueball/Cargo.lock

@@ -16,16 +16,13 @@ colored = "3.1.1"
 crossterm = "0.28"
 dashmap = "6.2.1"
 directories = "6.0.0"
-futures-channel = "0.3.33"
 futures-util = "0.3.33"
 rand = "0.10.2"
-ratatui = "0.29"
+ratatui = "=0.30.0-alpha.1"
 rcgen = "0.14.8"
 rmp-serde = "1.3.1"
-rtrb = "0.3.4"
 rustls-pemfile = "2"
 serde = { version = "1.0.229", features = ["derive"] }
-serde_json = "1.0.151"
 tokio = { version = "1.53.1", features = ["rt-multi-thread", "net", "sync", "time", "macros"] }
 tokio-rustls = { version = "0.26", default-features = false, features = ["ring", "tls12", "logging"] }
 tokio-tungstenite = "0.30.0"

glueball/Cargo.toml

@@ -4,32 +4,27 @@ use crate::{
     state::State,
     util::server_sent_msg,
 };
-use rtrb::{Consumer, Producer, RingBuffer};
 use std::sync::Arc;
+use tokio::sync::mpsc::{self, Receiver, Sender};
 use tokio_tungstenite::tungstenite::Message;
 
-const MAX_PENDING_KICK_MESSAGES: usize = 12;
+const MAX_PENDING_ACTIONS: usize = 12;
 
 pub enum UserAction {
     Lock(RoomId),
     Kick(ClientId),
 }
 
-pub fn setup_user_action_system(state: &Arc<State>) -> Producer<UserAction> {
-    let (tx, rx) = RingBuffer::new(MAX_PENDING_KICK_MESSAGES);
+pub fn setup_user_action_system(state: &Arc<State>) -> Sender<UserAction> {
+    let (tx, rx) = mpsc::channel::<UserAction>(MAX_PENDING_ACTIONS);
     spawn_user_action_receiver(rx, state.clone());
 
     tx
 }
 
-fn spawn_user_action_receiver(mut rx: Consumer<UserAction>, state: Arc<State>) {
+fn spawn_user_action_receiver(mut rx: Receiver<UserAction>, state: Arc<State>) {
     tokio::spawn(async move {
-        loop {
-            let Ok(user_message) = rx.pop() else {
-                tokio::task::yield_now().await;
-                continue;
-            };
-
+        while let Some(user_message) = rx.recv().await {
             match user_message {
                 UserAction::Lock(room_id) => {
                     let _ = state.toggle_room_lock(&room_id);

glueball/src/action.rs

@@ -15,7 +15,7 @@ pub const DEFAULT_PORT: u16 = 2610;
 pub fn certification_directory() -> Result<PathBuf> {
     let dir = match ProjectDirs::from("com", "Autodesk", "synthesis-glueball") {
         Some(dirs) => dirs.data_dir().join("secrets"),
-        None => bail!("Cound not find certificate directory"),
+        None => bail!("Could not find certificate directory"),
     };
 
     Ok(dir)

glueball/src/config.rs

@@ -94,7 +94,7 @@ pub async fn handle_client_close(client_id: ClientId, state: &Arc<State>) -> Res
     });
 
     let Some(room) = state.get_room_of_client_mut(&client_id) else {
-        bail!("Client attempted to leave when they were not in a room ");
+        bail!("Client attempted to leave when they were not in a room");
     };
 
     let client_name = room.get_client_name(&client_id)?;

glueball/src/connection.rs

@@ -1,14 +1,13 @@
 use std::collections::{HashMap, VecDeque};
 
 use colored::{Color, ColoredString, Colorize};
-use ratatui::{
-    style::{Style, Stylize},
-    text::Line,
-};
+use ratatui::{style::Style, text::Line};
 use tokio::sync::mpsc::{self, Receiver};
 
 use crate::{LOG_TX, room::RoomId};
 
+/// Maximum number of log lines retained in each log (both per-room and system logs)
+/// Oldest lines are dropped once the buffer is full.
 pub const MAX_LOG_LINES: usize = 500;
 
 pub type RoomLogs = HashMap<RoomId, VecDeque<Event>>;
@@ -85,14 +84,14 @@ pub enum LogDestination {
     RemoveRoom(RoomId),
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum EventType {
     Info,
     Warning,
     Error,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Event {
     pub message: String,
     pub kind: EventType,
@@ -180,9 +179,19 @@ impl Logger {
         let _ = self.room_logs.remove(id);
     }
 
-    pub fn snapshot(&self) -> LogSnapshot {
+    pub fn new_snapshot(&self) -> LogSnapshot {
         (self.global_log.clone(), self.room_logs.clone())
     }
+
+    pub fn update_snapshot(&self, old_snapshot: &mut LogSnapshot) {
+        if old_snapshot.0 != self.global_log {
+            old_snapshot.0.clone_from(&self.global_log);
+        }
+
+        if old_snapshot.1 != self.room_logs {
+            old_snapshot.1.clone_from(&self.room_logs);
+        }
+    }
 }
 
 pub fn spawn_log_receiver<F>(mut logging_rx: Receiver<LogRequest>, handle_log: F)

glueball/src/logging.rs

@@ -40,10 +40,10 @@ static LOG_TX: OnceLock<Sender<LogRequest>> = OnceLock::new();
 async fn main() -> Result<()> {
     let _cleanup_trigger = Cleanup;
 
-    let config = retrieve_config()?;
-
     let logging_rx = create_logging_channel();
 
+    let config = retrieve_config()?;
+
     let state = Arc::new(State::new());
 
     if config.headless {
@@ -53,7 +53,7 @@ async fn main() -> Result<()> {
     }
 
     if let Some(ref room_id) = config.permanent_room {
-        state.new_permanent_room(&room_id);
+        state.new_permanent_room(room_id);
     }
 
     // `listener` will be used regardless of the security level specified
@@ -121,7 +121,7 @@ async fn run_secure_server(
 }
 
 fn setup_cli_logging(logging_rx: mpsc::Receiver<LogRequest>) {
-    // Read the logging channel and immediantly print result
+    // Read the logging channel and immediately print result
     let print_to_terminal =
         move |message: String, kind: EventType, log_destination: LogDestination| {
             match log_destination {

glueball/src/main.rs

@@ -20,6 +20,7 @@ use std::ops;
 use std::sync::Arc;
 
 type WsStream<S> = WebSocketStream<Prefixed<S>>;
+const MAX_CLIENT_NAME_LENGTH: usize = 100;
 
 /// Waits for and handles messages from the client that are intended for the server.
 ///
@@ -51,10 +52,13 @@ where
             // When they ask to initialize a connection, then we add them to a room
             // Or create a room for them
             Some(ClientToServerMessage::InitializeConnection { room_id, name }) => {
+                if name.len() > MAX_CLIENT_NAME_LENGTH {
+                    break None;
+                }
+
                 let info = state.initialize_client_in_room(tx, room_id, &name);
-                let (client_id, room_id) = match info {
-                    Some(info) => info,
-                    None => break None,
+                let Some((client_id, room_id)) = info else {
+                    break None;
                 };
 
                 let message = server_sent_msg(ServerToClientMessage::SendInfo {
@@ -90,10 +94,14 @@ where
     S: SynthesisStream,
 {
     let Some(Ok(Message::Binary(message_data))) = read.next().await else {
-        warn_global!("Client disconnected before handshake (probably a test)");
+        warn_global!("Client disconnected before handshake (probably connection test)");
         return None;
     };
 
+    if message_data.len() == 1 {
+        return None;
+    }
+
     let Ok(message) = deserialize_messagepack::<ClientToServerMessage>(&message_data[1..]) else {
         error_global!("{addr} sent an invalid initial message");
         return None;
@@ -134,8 +142,9 @@ pub async fn handle_client_message(
             // If we're here, that means the message has a client-client prefix
             // which we want anyway, so there's no need to prefix the message
             // we can just forward it!
-            let senders: Vec<ClientSender> = state.get_senders_from_user_room(client_id);
+            let senders: Vec<ClientSender> = state.get_senders_from_user_room(&client_id);
 
+            // TODO perf
             let tasks = senders.iter().map(|tx| tx.send(message.clone()));
             let _ = futures_util::future::join_all(tasks).await;
 

glueball/src/messaging.rs

@@ -9,10 +9,9 @@ use uuid::Uuid;
 pub type ClientMap = DashMap<ClientId, RoomId>;
 pub type RoomMap = DashMap<RoomId, Room>;
 
-/// Maximum number of log lines retained in each log (both per-room and system logs)
-/// Oldest lines are dropped once the buffer is full.
 const VALID_ROOM_ID_CHARACTERS: [char; 36] = valid_room_id_characters();
 const MAX_ROOM_COUNT: usize = 32;
+const MAX_MEMBER_COUNT_PER_ROOM: usize = 10;
 
 pub struct State {
     pub users: ClientMap,
@@ -34,7 +33,7 @@ impl State {
         name: &str,
     ) -> Option<(ClientId, RoomId)> {
         match room_id {
-            None if self.room_count() == MAX_ROOM_COUNT => None,
+            None if self.room_count() >= MAX_ROOM_COUNT => None,
             None => Some(self.add_room_and_host(name, tx)),
             Some(room_id) => self
                 .add_client_to_room(name, tx, &room_id)
@@ -66,13 +65,15 @@ impl State {
         (host_id, room_id)
     }
 
-    /// # Safety
+    /// # Deadlock
     /// Relinquish all locks on `self.room` before calling
     pub fn remove_client(&self, client_id: &ClientId) {
         let Some(room_id) = self.users.get(client_id).map(|a| a.value().clone()) else {
             return;
         };
 
+        self.users.remove(client_id);
+
         // Room lock held
         let Some(mut room) = self.rooms.get_mut(&room_id) else {
             return;
@@ -94,6 +95,7 @@ impl State {
         room_id: &RoomId,
     ) -> Option<ClientId> {
         let client_id = Uuid::new_v4();
+
         let Some(mut room) = self.rooms.get_mut(room_id) else {
             warn_global!("Attempted to add {client_id} into non-existant room {room_id}");
             return None;
@@ -103,6 +105,10 @@ impl State {
             return None;
         }
 
+        if room.members.len() >= MAX_MEMBER_COUNT_PER_ROOM {
+            return None;
+        }
+
         let client = Client::new(client_id, client_name.to_string(), client_tx);
         room.members.push(client);
         self.users.insert(client_id, room_id.clone());
@@ -125,13 +131,18 @@ impl State {
             return;
         }
 
+        if self.rooms.contains_key(room_id) {
+            error_global!("Attempted to create permanent room that already exists");
+            return;
+        }
+
         let room = Room {
             members: Vec::new(),
             host: None,
             locked: false,
             permanent: true,
         };
-        self.rooms.insert(room_id.to_string(), room);
+        self.rooms.insert(room_id.clone(), room);
     }
 
     pub fn get_room_of_client_mut(&self, client_id: &ClientId) -> Option<RefMut<'_, RoomId, Room>> {
@@ -149,12 +160,12 @@ impl State {
         self.rooms.get(&room_id)
     }
 
-    pub fn get_senders_from_user_room(&self, client_id: ClientId) -> Vec<ClientSender> {
-        let Some(room) = self.get_room_of_client(&client_id) else {
+    pub fn get_senders_from_user_room(&self, client_id: &ClientId) -> Vec<ClientSender> {
+        let Some(room) = self.get_room_of_client(client_id) else {
             return Vec::new();
         };
 
-        room.get_peer_senders(&client_id)
+        room.get_peer_senders(client_id)
     }
 
     /// Flips whether new clients can join `room_id`.
@@ -222,9 +233,14 @@ impl State {
 }
 
 pub fn is_valid_room_id(s: &str) -> bool {
-    s.trim().len() == 6
-        && s.chars()
-            .all(|c| c.is_ascii_digit() || c.is_ascii_uppercase())
+    let s = s.trim();
+
+    let correct_length = s.len() == 6;
+    let all_characters_valid = s
+        .chars()
+        .all(|c| c.is_ascii_digit() || c.is_ascii_uppercase());
+
+    correct_length && all_characters_valid
 }
 
 const fn valid_room_id_characters() -> [char; 36] {