From e4db5e27cf84ca6064c0163471603558adee4418 Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Wed, 5 Aug 2026 21:01:02 +0200 Subject: [PATCH 1/6] chore(deps): update rust crate livekit-api to 0.6.0 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 2c37c3e..b0fb5f7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,7 @@ log = "0.4" dotenv = "0.15.0" strum = "0.28.0" strum_macros = "0.28.0" -livekit-api = { version = "0.5.6", features = ["rustls-tls-native-roots"] } +livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] } livekit-protocol = "0.7.10" thiserror = "2.0.19" trust-dns-resolver = "0.23.2" From 257ebefe0c57e2bd40a034851dd6fba937c0b371 Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Wed, 5 Aug 2026 22:01:18 +0200 Subject: [PATCH 2/6] chore(deps): update rust crate livekit-protocol to v0.7.12 --- Cargo.lock | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bf277ff..e8bc4dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -616,7 +616,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -1452,9 +1452,9 @@ dependencies = [ [[package]] name = "livekit-protocol" -version = "0.7.11" +version = "0.7.12" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "735a237deeff60124edd7831fcabce32b4b9f667d5c4c5688c9d18a2808422c4" +checksum = "526f22bddf409e5f15449d55cf341647d7c16a1f43df9db9b05be106e4208e1c" dependencies = [ "pbjson", "pbjson-types", @@ -2327,7 +2327,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -2581,7 +2581,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -2640,7 +2640,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -3032,10 +3032,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] @@ -3596,7 +3596,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.52.0", ] [[package]] From 4eb2f23013007a3b57341343a84869ba3d10a6c5 Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Wed, 5 Aug 2026 22:01:22 +0200 Subject: [PATCH 3/6] chore(deps): update rust crate rustls to v0.23.43 --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bf277ff..21c1ac6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2586,9 +2586,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.42" +version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ "aws-lc-rs", "log", From da5a5a5dfff424fda297ebc271786e4ce2421379 Mon Sep 17 00:00:00 2001 From: Alex Date: Fri, 7 Aug 2026 23:54:28 +0200 Subject: [PATCH 4/6] [Fix] User States --- mtp-type-maps | 2 +- .../anonymous_client_connection.rs | 27 +- src/config.rs | 20 +- src/data/user.rs | 45 +- src/omega/capabilities.rs | 145 ++++ src/omega/mod.rs | 1 + src/omega/omega_connection.rs | 667 ++++++++++++++++-- src/rho/client_connection.rs | 202 ++++-- src/rho/connection.rs | 60 +- src/rho/iota_connection.rs | 178 ++++- src/rho/rho_connection.rs | 54 +- src/rho/rho_manager.rs | 13 + 12 files changed, 1169 insertions(+), 245 deletions(-) create mode 100644 src/omega/capabilities.rs diff --git a/mtp-type-maps b/mtp-type-maps index f753cfd..486541b 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit f753cfd406fc0050a8ccd76a90897205efceacd8 +Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index a333260..4712af0 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -8,7 +8,6 @@ use uuid::Uuid; use crate::anonymous_clients::anonymous_manager::{self, generate_username}; use crate::app_state::AppState; use crate::calls::call_group::call_invite_secret_from_cv; -use crate::data::user::UserStatus; use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender}; use crate::util::data_type_id; use crate::util::logger::PrintType; @@ -222,9 +221,13 @@ impl AnonymousClientConnection { .await; } - // Handle client status changes - if cv.is_type(CommunicationType::ClientChanged) { - self.handle_client_changed(cv).await; + // Presence is account-scoped and anonymous sessions have no + // persisted account preference to change. + if cv.is_type(CommunicationType::SetUserState) + || cv.is_type(CommunicationType::ClientChanged) + { + self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId) + .await; return; } @@ -321,22 +324,6 @@ impl AnonymousClientConnection { }); } - /// Handle client status change - async fn handle_client_changed(self: Arc, cv: CommunicationValue) { - if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { - let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); - self.state - .omega - .client_changed( - self.user_id as i64, - self.user_id as i64, - self.user_id as i64, - user_status, - ) - .await; - } - } - /// Handle call invite async fn handle_call_invite(self: Arc, cv: CommunicationValue) { let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64; diff --git a/src/config.rs b/src/config.rs index 7c10773..484da67 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,10 +1,12 @@ -use std::env; +use std::{env, time::Duration}; use thiserror::Error; const DEFAULT_RHO_PORT: u16 = 443; const DEFAULT_OMEGA_HOST: &str = "tensamin.net"; const DEFAULT_OMEGA_PORT: u16 = 9187; +const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20; +const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3; #[derive(Clone, Debug, Eq, PartialEq)] pub struct LiveKitConfig { @@ -19,6 +21,11 @@ pub struct Config { pub omega_host: String, pub omega_port: u16, pub omikron_id: u64, + /// Maximum duration of one route or subscription synchronization request. + pub omega_sync_timeout: Duration, + /// Number of synchronization requests before the transport is closed and + /// the normal reconnect loop starts. + pub omega_sync_retries: u32, pub livekit: Option, } @@ -38,6 +45,15 @@ impl Config { let rho_port = parse_or_default("RHO_PORT", DEFAULT_RHO_PORT)?; let omega_port = parse_or_default("OMEGA_PORT", DEFAULT_OMEGA_PORT)?; let omikron_id = parse_or_default("ID", 0_u64)?; + // Each synchronization request is bounded by this timeout. After + // omega_sync_retries attempts, Omikron closes the authenticated + // transport so the normal reconnect loop can establish a clean state. + let omega_sync_timeout = Duration::from_secs(parse_or_default( + "OMEGA_SYNC_TIMEOUT_SECONDS", + DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS, + )?); + let omega_sync_retries = + parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1); let omega_host = env::var("OMEGA_HOST") .unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string()) .trim() @@ -55,6 +71,8 @@ impl Config { omega_host, omega_port, omikron_id, + omega_sync_timeout, + omega_sync_retries, livekit: livekit_from_environment()?, }) } diff --git a/src/data/user.rs b/src/data/user.rs index 8587ffe..1eda0ba 100644 --- a/src/data/user.rs +++ b/src/data/user.rs @@ -1,7 +1,4 @@ -use strum::IntoEnumIterator; -use strum_macros::EnumIter; - -#[derive(Debug, Clone, PartialEq, EnumIter, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] #[allow(unused, non_camel_case_types)] pub enum UserStatus { user_offline, @@ -20,12 +17,40 @@ impl UserStatus { pub fn to_string(&self) -> String { format!("{:?}", self) } - pub fn from_str(s: &str) -> Option { - for sel in UserStatus::iter() { - if &sel.to_string() == s { - return Some(sel); - } + pub fn from_client_preference(s: &str) -> Option { + match s { + "user_online" => Some(Self::user_online), + "user_idle" => Some(Self::user_idle), + "user_dnd" => Some(Self::user_dnd), + "user_wc" => Some(Self::user_wc), + "user_invisible" => Some(Self::user_invisible), + _ => None, } - None + } + + pub fn public_value(&self) -> Self { + match self { + Self::user_invisible => Self::user_offline, + value => value.clone(), + } + } + + pub fn from_str(s: &str) -> Option { + Self::from_client_preference(s) + } +} + +#[cfg(test)] +mod tests { + use super::UserStatus; + + #[test] + fn network_parser_rejects_derived_states() { + assert_eq!(UserStatus::from_str("user_offline"), None); + assert_eq!(UserStatus::from_str("iota_offline"), None); + assert_eq!( + UserStatus::from_str("user_online"), + Some(UserStatus::user_online) + ); } } diff --git a/src/omega/capabilities.rs b/src/omega/capabilities.rs new file mode 100644 index 0000000..b658068 --- /dev/null +++ b/src/omega/capabilities.rs @@ -0,0 +1,145 @@ +use std::collections::BTreeSet; + +const OMIKRON_PREFIX: &str = "omikron;caps="; +const OMEGA_PREFIX: &str = "omega;caps="; +const SET_USER_STATE: &str = "set_user_state_v1"; +const STATE_SUBSCRIBE: &str = "state_subscribe_v1"; +const SESSION_SNAPSHOT: &str = "session_snapshot_v1"; +const CLIENT_STATE_PUSH: &str = "client_state_push_v1"; + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct PeerCapabilities { + pub set_user_state_v1: bool, + pub state_subscribe_v1: bool, + pub session_snapshot_v1: bool, + pub client_state_push_v1: bool, +} + +impl PeerCapabilities { + pub fn current() -> Self { + Self { + set_user_state_v1: true, + state_subscribe_v1: true, + session_snapshot_v1: true, + client_state_push_v1: true, + } + } + + pub fn identification_description(&self) -> String { + let mut names = Vec::new(); + if self.set_user_state_v1 { + names.push(SET_USER_STATE); + } + if self.state_subscribe_v1 { + names.push(STATE_SUBSCRIBE); + } + if self.session_snapshot_v1 { + names.push(SESSION_SNAPSHOT); + } + if self.client_state_push_v1 { + names.push(CLIENT_STATE_PUSH); + } + format!("{OMIKRON_PREFIX}{}", names.join(",")) + } + + pub fn from_identification_description(description: Option<&str>) -> Result { + parse_capabilities(description, OMIKRON_PREFIX) + } + + pub fn from_omega_identification_description(description: Option<&str>) -> Result { + // A missing Omega descriptor selects the documented legacy wire + // contracts: tuple route snapshots, GetStates-only subscriptions, + // and ClientChanged pushes. + parse_capabilities(description, OMEGA_PREFIX) + } +} + +fn parse_capabilities(description: Option<&str>, prefix: &str) -> Result { + let Some(description) = description else { + return Ok(PeerCapabilities::default()); + }; + if description == "omega" { + return Ok(PeerCapabilities::default()); + } + let Some(capabilities) = description.strip_prefix(prefix) else { + return Err(()); + }; + let mut seen = BTreeSet::new(); + for capability in capabilities.split(',') { + if capability.is_empty() || !seen.insert(capability) { + return Err(()); + } + } + if seen.iter().any(|capability| { + !matches!( + *capability, + SET_USER_STATE | STATE_SUBSCRIBE | SESSION_SNAPSHOT | CLIENT_STATE_PUSH + ) + }) { + return Err(()); + } + Ok(PeerCapabilities { + set_user_state_v1: seen.contains(SET_USER_STATE), + state_subscribe_v1: seen.contains(STATE_SUBSCRIBE), + session_snapshot_v1: seen.contains(SESSION_SNAPSHOT), + client_state_push_v1: seen.contains(CLIENT_STATE_PUSH), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn capability_description_round_trips() { + let capabilities = PeerCapabilities::current(); + assert_eq!( + PeerCapabilities::from_identification_description(Some( + &capabilities.identification_description() + )), + Ok(capabilities) + ); + } + + #[test] + fn malformed_capabilities_are_rejected() { + assert!( + PeerCapabilities::from_identification_description(Some("omikron;caps=unknown_v9")) + .is_err() + ); + assert!( + PeerCapabilities::from_identification_description(Some( + "omikron;caps=state_subscribe_v1,state_subscribe_v1" + )) + .is_err() + ); + assert!(PeerCapabilities::from_identification_description(Some("omikron;caps=")).is_err()); + } + + #[test] + fn missing_capabilities_mean_legacy_peer() { + assert_eq!( + PeerCapabilities::from_identification_description(None), + Ok(PeerCapabilities::default()) + ); + assert_eq!( + PeerCapabilities::from_omega_identification_description(Some("omega")), + Ok(PeerCapabilities::default()) + ); + } + + #[test] + fn omega_capabilities_use_the_omega_namespace() { + assert!( + PeerCapabilities::from_omega_identification_description(Some( + "omega;caps=set_user_state_v1" + )) + .unwrap() + .set_user_state_v1 + ); + assert!( + PeerCapabilities::from_identification_description(Some("omega;caps=set_user_state_v1")) + .is_err() + ); + } +} diff --git a/src/omega/mod.rs b/src/omega/mod.rs index cbe77d9..4108cc6 100644 --- a/src/omega/mod.rs +++ b/src/omega/mod.rs @@ -1 +1,2 @@ +pub mod capabilities; pub mod omega_connection; diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index 78a2d38..e66e62c 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,6 +1,9 @@ +use super::capabilities::PeerCapabilities; use crate::{ - config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out, - rho::rho_manager::RhoManager, util::logger::PrintType, + config::Config, + log_cv_in, log_cv_out, log_err, log_in, log_out, + rho::rho_manager::RhoManager, + util::{data_type_id, logger::PrintType}, }; use dashmap::DashMap; use mtp::client::{Client, MTPConnection, Sender}; @@ -13,7 +16,7 @@ use mtp::{ use mtp_transport::ConnectionHandle; use std::{sync::Arc, time::Duration}; use tokio::{ - sync::{Mutex, RwLock, mpsc, watch}, + sync::{Mutex, RwLock, mpsc, oneshot, watch}, task::JoinHandle, time::{Instant, sleep}, }; @@ -23,6 +26,7 @@ use uuid::Uuid; const RECONNECT_DELAY: Duration = Duration::from_secs(5); const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300); const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10); +const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1); const PING_INTERVAL: Duration = Duration::from_secs(5); const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); @@ -30,6 +34,37 @@ const MAX_CONCURRENT_REQUESTS: usize = 128; const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3; const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30); +fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> { + if !value.is_type(CommunicationType::ClientChanged) { + return None; + } + let receiver = i64::try_from(value.get_receiver()) + .ok() + .filter(|id| *id > 0)?; + let session_id = value + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + value + .get_data(DataType::UserId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + value.get_data(DataType::UserState).as_str()?; + Some((receiver, session_id)) +} + +fn parse_omega_capability_response(value: &CommunicationValue) -> Result { + if !value.is_type(CommunicationType::IdentificationResponse) { + return Err("not an identification response".to_string()); + } + PeerCapabilities::from_omega_identification_description( + value.get_data(DataType::Description).as_str(), + ) + .map_err(|_| "Omega advertised an invalid capability set".to_string()) +} + // ============================================================================ // Waiting Task System // ============================================================================ @@ -85,24 +120,33 @@ pub fn start_task_cleanup_loop(omega: Arc) { pub enum ConnectionState { Disconnected, Connecting, - Connected { identified: bool }, + Identifying, + SynchronizingRoutes, + SynchronizingSubscriptions, + Ready, } #[allow(unused_variables)] impl ConnectionState { pub fn is_connected(&self) -> bool { - match self { - ConnectionState::Connected { identified } => true, - _ => false, - } + !matches!( + self, + ConnectionState::Disconnected | ConnectionState::Connecting + ) } #[allow(dead_code)] pub fn is_identified(&self) -> bool { - match self { - ConnectionState::Connected { identified: true } => true, - _ => false, - } + matches!( + self, + ConnectionState::SynchronizingRoutes + | ConnectionState::SynchronizingSubscriptions + | ConnectionState::Ready + ) + } + + pub fn is_ready(&self) -> bool { + matches!(self, ConnectionState::Ready) } } @@ -117,6 +161,8 @@ pub struct OmegaConnection { host: String, port: u16, omikron_id: u64, + sync_timeout: Duration, + sync_retries: u32, pub connection_id: Uuid, shutdown_tx: Arc>>>, // Track if we should reconnect on close @@ -126,6 +172,14 @@ pub struct OmegaConnection { keyring: mtp::crypto::Keyring, rho: Arc, waiting_tasks: DashMap, + session_presence: DashMap<(i64, i64), SessionPresenceState>, + peer_capabilities: Arc>, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct SessionPresenceState { + pub subscribed_user_ids: Vec, + pub pending: bool, } impl OmegaConnection { @@ -143,6 +197,8 @@ impl OmegaConnection { host: config.omega_host.clone(), port: config.omega_port, omikron_id: config.omikron_id, + sync_timeout: config.omega_sync_timeout, + sync_retries: config.omega_sync_retries, connection_id: Uuid::new_v4(), shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), reconnect_on_close: Arc::new(RwLock::new(true)), @@ -154,6 +210,8 @@ impl OmegaConnection { keyring, rho, waiting_tasks: DashMap::new(), + session_presence: DashMap::new(), + peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())), } } @@ -267,6 +325,7 @@ impl OmegaConnection { let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port)) .with_client_id(self.omikron_id) + .with_description(PeerCapabilities::current().identification_description()) .with_policy( Policy::default() .with_send_mode(SendMode::SingleStreamPerMessage) @@ -302,18 +361,43 @@ impl OmegaConnection { // Store sender let sender_arc = Arc::new(connection.sender.clone()); *self.sender.write().await = Some(sender_arc.clone()); - *self.state.write().await = ConnectionState::Connected { identified: false }; + // A new authenticated transport gets a fresh request circuit. A + // previous transport can have failed while synchronizing and must not + // prevent this connection from making its initial requests. + self.circuit_breaker.lock().await.record_success(); + *self.state.write().await = ConnectionState::Identifying; // Get handle for close monitoring let sender_handle = sender_arc.handle().clone(); let connection = Arc::new(connection); + let (capability_tx, capability_rx) = oneshot::channel(); // Start read loop let read_self = self.clone(); let read_handle = tokio::spawn(async move { - read_self.read_loop(connection, sender_handle).await; + read_self + .read_loop(connection, sender_handle, Some(capability_tx)) + .await; }); + let negotiated_capabilities = + match tokio::time::timeout(CAPABILITY_NEGOTIATION_TIMEOUT, capability_rx).await { + Ok(Ok(Ok(capabilities))) => capabilities, + Ok(Ok(Err(error))) => { + self.fail_synchronization("capability negotiation", error) + .await; + let _ = read_handle.await; + return Err("Omega capability negotiation failed".to_string()); + } + Ok(Err(_)) | Err(_) => { + // An Omega from before capability negotiation sends no second + // identification response. Fall back to its documented + // legacy contracts rather than guessing modern support. + PeerCapabilities::default() + } + }; + *self.peer_capabilities.write().await = negotiated_capabilities; + // Tell omega our current state now that we're actually connected - // doing this after teardown (as before) sent into a sender that had // already been cleared, silently dropping the sync every time. @@ -354,8 +438,10 @@ impl OmegaConnection { } } - async fn sync_client_iota_status(self: Arc) { + async fn sync_client_iota_status(&self) { + *self.state.write().await = ConnectionState::SynchronizingRoutes; let mut connected_iota_ids: Vec = Vec::new(); + let mut connected_sessions: Vec = Vec::new(); let mut connected_user_ids: Vec = Vec::new(); let iota_ids = self.rho.iota_ids().await; @@ -367,21 +453,120 @@ impl OmegaConnection { for rho in rho_connections { for client_conn in rho.get_client_connections().await { - connected_user_ids.push(DataValue::SignedNumber( - client_conn.get_user_id().await.into(), - )); + let type_map = mtp::codec::TypeMap::latest(); + let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into()); + let session_id = DataValue::SignedNumber((client_conn.session_id as i128).into()); + let iota_id = DataValue::SignedNumber((rho.get_iota_id().await as i128).into()); + let peer_capabilities = self.peer_capabilities.read().await.clone(); + if peer_capabilities.session_snapshot_v1 { + connected_sessions.push(DataValue::Container(vec![ + (data_type_id(DataType::UserId, &type_map), user_id), + (data_type_id(DataType::SessionId, &type_map), session_id), + (data_type_id(DataType::IotaId, &type_map), iota_id), + ])); + } else { + connected_user_ids.push(user_id); + } } } - let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus) - .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) - .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) - .add_typed_default( - DataType::RhoConnections, - DataValue::SignedNumber(self.rho.connection_count().await as i128), - ); + let peer_capabilities = self.peer_capabilities.read().await.clone(); + let sync_msg = if peer_capabilities.session_snapshot_v1 { + CommunicationValue::new(CommunicationType::SyncClientIotaStatus) + .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) + .add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions)) + .add_typed_default( + DataType::RhoConnections, + DataValue::SignedNumber(self.rho.connection_count().await as i128), + ) + } else { + // Legacy Omega has no request/response contract for this message. + // Its exact snapshot is IotaIds + UserIds + RhoConnections. + CommunicationValue::new(CommunicationType::SyncClientIotaStatus) + .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids)) + .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids)) + .add_typed_default( + DataType::RhoConnections, + DataValue::SignedNumber(self.rho.connection_count().await as i128), + ) + }; - self.send_message(&sync_msg).await; + if !peer_capabilities.session_snapshot_v1 { + if let Err(error) = self.send_message_result(&sync_msg).await { + self.fail_synchronization("legacy route synchronization", error) + .await; + return; + } + *self.state.write().await = ConnectionState::SynchronizingSubscriptions; + if let Err(error) = self.restore_presence_subscriptions().await { + self.fail_synchronization("subscription restoration", error) + .await; + return; + } + *self.state.write().await = ConnectionState::Ready; + return; + } + + let mut response = Err("route synchronization did not start".to_string()); + for attempt in 0..self.sync_retries { + response = self + .await_response(&sync_msg, Some(self.sync_timeout)) + .await; + match &response { + Ok(value) if value.is_type(CommunicationType::Success) => break, + Ok(value) if value.is_type(CommunicationType::ErrorInvalidData) => break, + _ if attempt + 1 < self.sync_retries => { + sleep(Duration::from_millis(250 * 2u64.pow(attempt))).await + } + _ => break, + } + } + match response { + Ok(response) if response.is_type(CommunicationType::Success) => { + *self.state.write().await = ConnectionState::SynchronizingSubscriptions; + if let Err(error) = self.restore_presence_subscriptions().await { + log_err!( + 0, + PrintType::Omega, + "Presence subscription restoration failed: {}", + error + ); + self.fail_synchronization("subscription restoration", error) + .await; + return; + } + *self.state.write().await = ConnectionState::Ready; + } + Ok(response) => { + let error = format!( + "Omega rejected presence snapshot with {}", + response.get_type() + ); + self.fail_synchronization("route synchronization", error) + .await; + return; + } + Err(error) => { + self.fail_synchronization("route synchronization", error) + .await; + return; + } + } + } + + async fn fail_synchronization(&self, phase: &str, error: String) { + log_err!( + 0, + PrintType::Omega, + "Omega {} failed after {} attempts: {}; closing transport for reconnect", + phase, + self.sync_retries, + error + ); + *self.state.write().await = ConnectionState::Disconnected; + if let Some(sender) = self.sender.read().await.as_ref() { + sender.close().await; + } } // ------------------------------------------------------------------------- @@ -392,6 +577,7 @@ impl OmegaConnection { self: Arc, connection: Arc, sender_handle: Arc, + mut capability_tx: Option>>, ) { // Monitor both receiver and sender handle for close let mut close_rx = sender_handle.subscribe_close(); @@ -403,6 +589,14 @@ impl OmegaConnection { Ok(cv) => { log_cv_in!(PrintType::Omega, &cv); + if cv.is_type(CommunicationType::IdentificationResponse) { + if let Some(tx) = capability_tx.take() { + let result = parse_omega_capability_response(&cv); + let _ = tx.send(result); + } + continue; + } + let msg_id = cv.get_id(); if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) { if (task.task)(self.clone(), cv.clone()) { @@ -430,6 +624,20 @@ impl OmegaConnection { } } } + if cv.is_type(CommunicationType::ClientChanged) { + let Some((receiver, session_id)) = client_changed_target(&cv) + else { + continue; + }; + if let Some(client) = self + .rho + .get_client_connection(receiver, session_id) + .await + { + client.send_message(&cv).await; + } + continue; + } } Err(e) => { log_err!(0, PrintType::Omega, "Receive error: {}", e); @@ -488,6 +696,14 @@ impl OmegaConnection { let _ = self.try_send_message(cv).await; } + pub async fn send_message_result(&self, cv: &CommunicationValue) -> Result<(), String> { + self.try_send_message(cv).await + } + + pub async fn supports_set_user_state(&self) -> bool { + self.peer_capabilities.read().await.set_user_state_v1 + } + pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { if self.state.read().await.is_connected() { return Ok(()); @@ -607,72 +823,371 @@ impl OmegaConnection { self.state.read().await.is_identified() } + pub async fn is_ready(&self) -> bool { + self.state.read().await.is_ready() + } + #[allow(dead_code)] - pub async fn close_iota(&self, iota_id: i64) { + pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> { let cv = CommunicationValue::new(CommunicationType::IotaDisconnected) .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); - self.send_message(&cv).await; + let request_id = cv.get_id(); + let result = self.lifecycle_request(&cv).await; + if let Err(error) = &result { + self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error); + self.sync_client_iota_status().await; + } + result } - pub async fn client_changed( - &self, - _iota_id: i64, - user_id: i64, - session_id: i64, - state: UserStatus, - ) { - let cv = CommunicationValue::new(CommunicationType::ClientChanged) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())); - self.send_message(&cv).await; - } - - pub async fn client_disconnected(&self, user_id: i64, session_id: i64) { + pub async fn client_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> { + self.session_presence.remove(&(user_id, session_id)); let cv = CommunicationValue::new(CommunicationType::UserDisconnected) .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id.into()), ); - self.send_message(&cv).await; + let request_id = cv.get_id(); + let result = self.lifecycle_request(&cv).await; + if let Err(error) = &result { + self.log_lifecycle_failure( + "UserDisconnected", + user_id, + Some(session_id), + request_id, + error, + ); + self.sync_client_iota_status().await; + } + result } - pub async fn user_states(&self, user_id: i64, user_ids: Vec) { - let user_ids = user_ids - .iter() - .map(|v| DataValue::SignedNumber((*v).into())) - .collect(); - - let cv = CommunicationValue::new(CommunicationType::GetStates) - .add_typed_default(DataType::UserIds, DataValue::Array(user_ids)); - let msg_id = cv.get_id(); - - let rho_manager = self.rho.clone(); - self.waiting_tasks.insert( - msg_id, - WaitingTask { - task: Box::new( - move |_: Arc, response: CommunicationValue| { - let rho_manager = rho_manager.clone(); - tokio::spawn(async move { - let rho = rho_manager.get_for_user(user_id).await; - if let Some(rho) = rho { - for client in rho.get_client_connections_for_user(user_id).await { - client.send_message(&response).await; - } - } - }); - true - }, + pub async fn user_connected( + &self, + user_id: i64, + session_id: i64, + iota_id: i64, + ) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::UserConnected) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) + .add_typed_default( + DataType::UserState, + DataValue::Str("user_online".to_string()), + ) + .add_typed_default( + DataType::UpdatedAt, + DataValue::SignedNumber( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i128, ), - inserted_at: Instant::now(), + ); + let request_id = request.get_id(); + let result = self.lifecycle_request(&request).await; + if let Err(error) = &result { + self.log_lifecycle_failure( + "UserConnected", + user_id, + Some(session_id), + request_id, + error, + ); + self.sync_client_iota_status().await; + } + result + } + + pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::IotaConnected) + .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())); + let request_id = request.get_id(); + let result = self.lifecycle_request(&request).await; + if let Err(error) = &result { + self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error); + self.sync_client_iota_status().await; + } + result + } + + pub async fn reconcile_routes(&self) { + self.sync_client_iota_status().await; + } + + async fn lifecycle_request(&self, request: &CommunicationValue) -> Result<(), String> { + if !self.is_ready().await { + return Err("Omega is not synchronized".to_string()); + } + if !self.peer_capabilities.read().await.session_snapshot_v1 { + // Legacy Omega handles lifecycle messages as notifications and + // never returns Success. A transport-level send result is the + // only acknowledgement available under that contract. + return self.send_message_result(request).await; + } + let response = self + .await_response(request, Some(Duration::from_secs(20))) + .await?; + if response.is_type(CommunicationType::Success) { + Ok(()) + } else { + Err(format!("Omega rejected {}", response.get_type())) + } + } + + fn log_lifecycle_failure( + &self, + operation: &str, + entity_id: i64, + session_id: Option, + request_id: u32, + error: &str, + ) { + log_err!( + 0, + PrintType::Omega, + "Lifecycle operation={} entity_id={} session_id={:?} request_id={} failed: {}", + operation, + entity_id, + session_id, + request_id, + error + ); + } + + pub async fn replace_state_subscription( + &self, + user_id: i64, + session_id: i64, + mut user_ids: Vec, + ) -> Result<(), String> { + if user_id <= 0 || session_id <= 0 { + return Err("invalid user or session id".to_string()); + } + + user_ids.sort_unstable(); + user_ids.dedup(); + + let key = (user_id, session_id); + // Store the latest authoritative snapshot before checking readiness. + // A client can synchronize while the transport is reconnecting. + self.session_presence.insert( + key, + SessionPresenceState { + subscribed_user_ids: user_ids.clone(), + pending: true, }, ); - self.send_message(&cv).await; + if !self.is_ready().await { + return Err("Omega is not synchronized".to_string()); + } + + self.deliver_state_subscription(key, user_ids).await + } + + async fn deliver_state_subscription( + &self, + key: (i64, i64), + user_ids: Vec, + ) -> Result<(), String> { + let (user_id, session_id) = key; + + if !self.peer_capabilities.read().await.state_subscribe_v1 { + let delivered_ids = user_ids.clone(); + let result = self + .request_user_states(user_id, session_id, user_ids) + .await; + if result.is_ok() { + self.mark_presence_delivered(key, &delivered_ids).await; + } + return result; + } + + let subscription = CommunicationValue::new(CommunicationType::StateSubscribe) + .with_sender(user_id as u64) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default( + DataType::UserIds, + DataValue::Array( + user_ids + .iter() + .map(|user_id| DataValue::SignedNumber((*user_id).into())) + .collect(), + ), + ); + let response = self + .await_response(&subscription, Some(Duration::from_secs(20))) + .await?; + if !response.is_type(CommunicationType::Success) { + return Err(format!( + "StateSubscribe rejected with {}", + response.get_type() + )); + } + + let delivered_ids = user_ids.clone(); + self.request_user_states(user_id, session_id, user_ids) + .await?; + self.mark_presence_delivered(key, &delivered_ids).await; + Ok(()) + } + + async fn restore_presence_subscriptions(&self) -> Result<(), String> { + let mut active = Vec::new(); + for rho in self.rho.connections().await { + active.extend(rho.get_client_connections().await); + } + let active_keys: std::collections::HashSet<_> = active + .iter() + .map(|client| (client.user_id as i64, client.session_id as i64)) + .collect(); + + // Do not replay subscriptions for sessions that disappeared while the + // transport was down. + self.session_presence + .retain(|key, _| active_keys.contains(key)); + + for key in active_keys { + let Some(snapshot) = self.session_presence.get(&key).map(|entry| entry.clone()) else { + continue; + }; + if !snapshot.pending { + continue; + } + self.deliver_state_subscription(key, snapshot.subscribed_user_ids) + .await?; + } + Ok(()) + } + + async fn mark_presence_delivered(&self, key: (i64, i64), delivered_ids: &[i64]) { + if let Some(mut entry) = self.session_presence.get_mut(&key) + && entry.subscribed_user_ids == delivered_ids + { + entry.pending = false; + } + } + + async fn request_user_states( + &self, + user_id: i64, + session_id: i64, + user_ids: Vec, + ) -> Result<(), String> { + let request = CommunicationValue::new(CommunicationType::GetStates) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default( + DataType::UserIds, + DataValue::Array( + user_ids + .iter() + .map(|user_id| DataValue::SignedNumber((*user_id).into())) + .collect(), + ), + ); + let response = self + .await_response(&request, Some(Duration::from_secs(20))) + .await?; + let legacy_peer = !self.peer_capabilities.read().await.client_state_push_v1; + let response = + if legacy_peer && response.get_data(DataType::SessionId).as_number().is_none() { + response.add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(session_id.into()), + ) + } else { + response + }; + if !legacy_peer + && response.get_data(DataType::SessionId).as_number() != Some(session_id as i128) + { + return Err("GetStates response had the wrong session".to_string()); + } + + if let Some(client) = self.rho.get_client_connection(user_id, session_id).await { + client.send_message(&response).await; + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response}; + use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; + + fn notification() -> CommunicationValue { + CommunicationValue::new(CommunicationType::ClientChanged) + .with_receiver(7) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) + .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())) + } + + #[test] + fn client_changed_target_requires_all_routing_fields() { + assert_eq!(client_changed_target(¬ification()), Some((7, 11))); + + let mut missing_session = notification(); + missing_session.remove_data(DataType::SessionId); + assert_eq!(client_changed_target(&missing_session), None); + + let mut missing_state = notification(); + missing_state.remove_data(DataType::UserState); + assert_eq!(client_changed_target(&missing_state), None); + } + + #[test] + fn client_changed_target_rejects_non_positive_ids() { + let invalid = CommunicationValue::new(CommunicationType::ClientChanged) + .with_receiver(0) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(20)) + .add_typed_default(DataType::UserState, DataValue::Str("user_online".into())); + assert_eq!(client_changed_target(&invalid), None); + } + + #[test] + fn capability_response_negotiates_new_omega() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default( + DataType::Description, + DataValue::Str( + "omega;caps=set_user_state_v1,state_subscribe_v1,session_snapshot_v1,client_state_push_v1" + .into(), + ), + ); + let capabilities = parse_omega_capability_response(&response).unwrap(); + assert!(capabilities.state_subscribe_v1); + assert!(capabilities.client_state_push_v1); + } + + #[test] + fn missing_capability_response_is_legacy() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse); + let capabilities = parse_omega_capability_response(&response).unwrap(); + assert_eq!(capabilities, PeerCapabilities::default()); + } + + #[test] + fn unsupported_capability_response_is_rejected() { + let response = CommunicationValue::new(CommunicationType::IdentificationResponse) + .add_typed_default( + DataType::Description, + DataValue::Str("omega;caps=unsupported_v9".into()), + ); + assert!(parse_omega_capability_response(&response).is_err()); } } diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 64cb052..c5c7dee 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -24,7 +24,6 @@ pub struct ClientConnection { pub receiver: Arc, pub_key: Arc>>>, pub rho_connection: Arc>>>, - pub interested_users: Arc>>, is_open: Arc>, message_slots: Arc, } @@ -35,7 +34,6 @@ impl ClientConnection { state: general.state.clone(), pub_key: Arc::new(RwLock::new(None)), rho_connection: general.rho_connection.clone(), - interested_users: Arc::new(RwLock::new(Vec::new())), is_open: Arc::new(RwLock::new(true)), sender: general.sender.clone(), receiver: general.receiver.clone(), @@ -104,9 +102,26 @@ impl ClientConnection { let mut cv = cv; - // Handle client status changes - if cv.is_type(CommunicationType::ClientChanged) { - self.handle_client_changed(cv).await; + if cv.is_type(CommunicationType::SetUserState) { + self.handle_set_user_state(cv).await; + return; + } + + // Compatibility for clients predating SetUserState. The target + // user fields, if present, are deliberately ignored: an + // authenticated connection may only change its own state. + if cv.is_type(CommunicationType::ClientChanged) + && cv.get_data_opt(DataType::UserState).is_some() + { + self.handle_set_user_state( + CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .add_typed_default( + DataType::UserState, + cv.get_data(DataType::UserState).clone(), + ), + ) + .await; return; } @@ -191,6 +206,46 @@ impl ClientConnection { || cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::DeleteUser) { + if cv.is_type(CommunicationType::ChangeUserData) + && cv.get_data_opt(DataType::OnlineStatus).is_some() + { + let mut profile_request = cv.clone(); + let preference = profile_request.remove_data(DataType::OnlineStatus); + let state_request = CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .add_typed_default(DataType::UserState, preference.unwrap()); + let state_response = match self.request_set_user_state(state_request).await { + Ok(response) => response, + Err(error_type) => { + self.send_error_response(cv.get_id(), error_type).await; + return; + } + }; + if !state_response.is_type(CommunicationType::Success) { + self.send_message(&state_response).await; + return; + } + + if profile_request.data_len() == 0 { + self.send_message(&state_response).await; + return; + } + + match self + .await_omega_response(profile_request.with_sender(self.user_id)) + .await + { + Ok(response) if response.is_type(CommunicationType::Success) => { + self.send_message(&response).await; + } + Ok(response) => self.send_message(&response).await, + Err(_) => { + self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal) + .await; + } + } + return; + } let sender = self.get_user_id().await; self.handle_omega_forward(cv.with_sender(sender as u64)) .await; @@ -290,34 +345,91 @@ impl ClientConnection { }); } async fn handle_omega_forward(self: Arc, cv: CommunicationValue) { - let client_for_closure = self.clone(); - let response_cv = self - .state - .omega - .clone() - .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20))) - .await; - if let Ok(response_cv) = response_cv { - client_for_closure.send_message(&response_cv).await; + let request_id = cv.get_id(); + match self.await_omega_response(cv).await { + Ok(response_cv) => self.send_message(&response_cv).await, + Err(_) => { + self.send_error_response(request_id, CommunicationType::ErrorInternal) + .await; + } } } - /// Handle client status change - async fn handle_client_changed(self: Arc, cv: CommunicationValue) { - let user_id = self.get_user_id().await; - if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) { - let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online); - if let Some(rho_conn) = self.get_rho_connection().await { - self.state - .omega - .client_changed( - rho_conn.get_iota_id().await as i64, - user_id as i64, - self.session_id as i64, - user_status, - ) - .await; - } + async fn await_omega_response(&self, cv: CommunicationValue) -> Result { + self.state + .omega + .clone() + .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20))) + .await + .map_err(|_| ()) + } + + async fn request_set_user_state( + &self, + cv: CommunicationValue, + ) -> Result { + if !self.state.omega.is_ready().await { + return Err(CommunicationType::ErrorInternal); + } + let Some(state) = cv + .get_data(DataType::UserState) + .as_str() + .and_then(UserStatus::from_client_preference) + else { + return Err(CommunicationType::ErrorInvalidData); + }; + let Some(rho) = self.get_rho_connection().await else { + return Err(CommunicationType::ErrorNoIota); + }; + let request = if self.state.omega.supports_set_user_state().await { + CommunicationValue::new(CommunicationType::SetUserState) + .with_id(cv.get_id()) + .with_sender(self.user_id) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(self.user_id as i128), + ) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(self.session_id as i128), + ) + .add_typed_default( + DataType::IotaId, + DataValue::SignedNumber(rho.get_iota_id().await as i128), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) + } else { + // Legacy Omega accepts the original ClientChanged request. It is + // intentionally limited to the authenticated user's ID and a + // writable preference, with no derived connectivity state. + CommunicationValue::new(CommunicationType::ClientChanged) + .with_id(cv.get_id()) + .with_sender(self.user_id) + .add_typed_default( + DataType::UserId, + DataValue::SignedNumber(self.user_id as i128), + ) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())) + }; + if !self.state.omega.supports_set_user_state().await { + self.state + .omega + .send_message_result(&request) + .await + .map_err(|_| CommunicationType::ErrorInternal)?; + return Ok(CommunicationValue::new(CommunicationType::Success) + .with_id(cv.get_id()) + .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))); + } + self.await_omega_response(request) + .await + .map_err(|_| CommunicationType::ErrorInternal) + } + + async fn handle_set_user_state(self: Arc, cv: CommunicationValue) { + match self.request_set_user_state(cv.clone()).await { + Ok(response) => self.send_message(&response).await, + Err(error_type) => self.send_error_response(cv.get_id(), error_type).await, } } @@ -830,35 +942,6 @@ impl ClientConnection { let _ = self.sender.close(); } - /// Set interested users list - pub async fn set_interested_users(self: Arc, interested_ids: Vec) { - let mut interested_guard = self.interested_users.write().await; - *interested_guard = interested_ids; - } - #[allow(dead_code)] - pub async fn get_interested_users(self: Arc) -> Vec { - let interested_guard = self.interested_users.read().await; - interested_guard.clone() - } - - /// Check if interested in a user and send notification - #[allow(dead_code)] - pub async fn are_you_interested(self: Arc, user_id: i64, user_status: &str) { - let interested_guard = self.clone().get_interested_users().await; - if interested_guard.contains(&user_id) { - let status = if user_status == "user_invisible" { - "user_offline" - } else { - user_status - }; - let notification = CommunicationValue::new(CommunicationType::ClientChanged) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default(DataType::UserState, DataValue::Str(status.to_string())); - - self.send_message(¬ification).await; - } - } - /// Handle connection close pub async fn handle_close(self: Arc) { let user_id = self.get_user_id().await; @@ -880,7 +963,6 @@ impl Clone for ClientConnection { client_version: self.client_version.clone(), pub_key: Arc::clone(&self.pub_key), rho_connection: Arc::clone(&self.rho_connection), - interested_users: Arc::clone(&self.interested_users), is_open: Arc::clone(&self.is_open), message_slots: Arc::clone(&self.message_slots), } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index 137929b..cede813 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -136,10 +136,23 @@ impl GeneralConnection { if let Some(rho_conn) = rho { rho_conn.bind_user_id(user_id).await; rho_conn.add_client_connection(client.clone()).await; - self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) - .await; - self.send_initial_client_state_request(&rho_conn, user_id) - .await; + match self + .notify_user_connected(user_id, rho_conn.get_iota_id().await as i64) + .await + { + Ok(()) => { + self.send_initial_client_state_request(&rho_conn, user_id) + .await; + } + Err(error) => { + log_err!( + user_id, + PrintType::Client, + "UserConnected acknowledgement failed: {}", + error + ); + } + } } else { log_err!( user_id, @@ -155,29 +168,13 @@ impl GeneralConnection { true } - async fn notify_user_connected(&self, user_id: i64, iota_id: i64) { + async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> { let session_id = *self.session_id.read().await as i64; - let notify = CommunicationValue::new(CommunicationType::UserConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())) - .add_typed_default( - DataType::SessionId, - DataValue::SignedNumber(session_id.into()), - ) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into())) - .add_typed_default( - DataType::UserState, - DataValue::Str("user_online".to_string()), - ) - .add_typed_default( - DataType::UpdatedAt, - DataValue::SignedNumber( - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis() as i128, - ), - ); - self.state.omega.clone().send_message(¬ify).await; + self.state + .omega + .clone() + .user_connected(user_id, session_id, iota_id) + .await } async fn find_user_rho(&self, user_id: i64) -> Option> { @@ -227,9 +224,6 @@ impl GeneralConnection { async fn migrate_iota(self: &Arc) { let id = self.id; log_in!(id as i64, PrintType::Iota, "Iota {} connected", id); - let notify = CommunicationValue::new(CommunicationType::IotaConnected) - .add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into())); - self.state.omega.clone().send_message(¬ify).await; let iota = IotaConnection::from_general(self.clone(), id).await; let previous = self.state.rho.get_by_iota(id as i64).await; @@ -246,6 +240,14 @@ impl GeneralConnection { } } self.state.rho.add(rho).await; + if let Err(error) = self.state.omega.clone().iota_connected(id as i64).await { + log_err!( + id as i64, + PrintType::Iota, + "IotaConnected acknowledgement failed: {}", + error + ); + } self.load_iota_users(id).await; iota.start(); } diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 7f7086c..047f718 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -23,6 +23,47 @@ use tokio::sync::mpsc; use super::rho_connection::RhoConnection; +fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec)> { + if !value.is_type(CommunicationType::ClientStateSync) + && !value.is_type(CommunicationType::AddConversation) + { + return None; + } + + let user_id = i64::try_from(value.get_receiver()) + .ok() + .filter(|id| *id > 0)?; + let session_id = value + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0)?; + let DataValue::Array(values) = value.get_data(DataType::UserIds) else { + return None; + }; + + let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); + let mut contact_ids = Vec::with_capacity(values.len()); + for value in values { + let contact_id = match value { + DataValue::SignedNumber(id) => i64::try_from(*id).ok(), + DataValue::Container(entries) => entries.iter().find_map(|(key, value)| { + (*key == user_id_type) + .then(|| value.as_signed_number()) + .flatten() + .and_then(|id| i64::try_from(id).ok()) + }), + _ => None, + } + .filter(|id| *id > 0)?; + contact_ids.push(contact_id); + } + contact_ids.sort_unstable(); + contact_ids.dedup(); + + Some((user_id, session_id, contact_ids)) +} + #[allow(dead_code)] pub struct IotaConnection { pub state: Arc, @@ -172,6 +213,29 @@ impl IotaConnection { } } + async fn replace_state_subscription( + &self, + user_id: i64, + session_id: i64, + contact_ids: Vec, + ) -> Result<(), String> { + let Some(rho_connection) = self.get_rho_connection().await else { + return Err("Rho connection is unavailable".to_string()); + }; + if rho_connection + .get_client_connection(user_id, session_id) + .await + .is_none() + { + return Err("target client session is no longer connected".to_string()); + } + + self.state + .omega + .replace_state_subscription(user_id, session_id, contact_ids) + .await + } + /// Send a CommunicationValue to the Iota pub async fn send_message(&self, cv: &CommunicationValue) { log_cv_out!(PrintType::Iota, cv); @@ -206,6 +270,34 @@ impl IotaConnection { cv }; + // Iota's UserIds field is an authoritative replacement snapshot for + // the requesting client session, including contact mutations. + if let Some((user_id, session_id, contact_ids)) = contact_snapshot(&cv) { + if let Err(error) = self + .replace_state_subscription(user_id, session_id, contact_ids) + .await + { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "Failed to replace presence subscription for user {} session {}: {}", + user_id, + session_id, + error + ); + } + } + + if cv.is_type(CommunicationType::StateSubscribe) { + self.send_error_response( + cv.get_id(), + CommunicationType::ErrorInvalidData, + Some("StateSubscribe must come from an authoritative contact snapshot"), + ) + .await; + return; + } + // Handle GET_CHATS if cv.is_type(CommunicationType::GetChats) { self.handle_get_chats(cv).await; @@ -517,7 +609,20 @@ impl IotaConnection { } let mut interested_ids: Vec = Vec::new(); - let session_id = cv.get_data(DataType::SessionId).as_signed_number(); + let Some(session_id) = cv + .get_data(DataType::SessionId) + .as_signed_number() + .and_then(|id| i64::try_from(id).ok()) + .filter(|id| *id > 0) + else { + self.forward_to_client( + CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(cv.get_id()) + .with_receiver(user_id), + ) + .await; + return; + }; let tm = TypeMap::latest(); // Presence interest is the complete contact set, independent of @@ -658,20 +763,18 @@ impl IotaConnection { // ============================ // Notify Omega // ============================ - self.state - .omega - .user_states(user_id as i64, interested_ids.clone()) - .await; - - // ============================ - // Notify Rho - // ============================ - if let Some(rho_conn) = self.get_rho_connection().await { - if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) { - rho_conn - .set_interested(user_id as i64, session_id, interested_ids) - .await; - } + if let Err(error) = self + .replace_state_subscription(user_id as i64, session_id, interested_ids.clone()) + .await + { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "Failed to replace presence subscription for user {} session {}: {}", + user_id, + session_id, + error + ); } // ============================ @@ -779,3 +882,48 @@ impl std::fmt::Debug for IotaConnection { .finish() } } + +#[cfg(test)] +mod tests { + use super::contact_snapshot; + use crate::util::data_type_id; + use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap}; + + fn snapshot(user_ids: DataValue) -> CommunicationValue { + CommunicationValue::new(CommunicationType::ClientStateSync) + .with_receiver(7) + .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11)) + .add_typed_default(DataType::UserIds, user_ids) + } + + #[test] + fn contact_snapshot_deduplicates_flat_and_typed_ids() { + let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest()); + let parsed = contact_snapshot(&snapshot(DataValue::Array(vec![ + DataValue::SignedNumber(20), + DataValue::Container(vec![(user_id_type, DataValue::SignedNumber(21))]), + DataValue::SignedNumber(20), + ]))) + .unwrap(); + + assert_eq!(parsed, (7, 11, vec![20, 21])); + } + + #[test] + fn contact_snapshot_accepts_empty_contact_sets() { + assert_eq!( + contact_snapshot(&snapshot(DataValue::Array(Vec::new()))), + Some((7, 11, Vec::new())) + ); + } + + #[test] + fn contact_snapshot_rejects_malformed_contact_sets() { + assert_eq!( + contact_snapshot(&snapshot(DataValue::Array(vec![DataValue::Str( + "bad".into(), + )]))), + None + ); + } +} diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 46f0330..4c60560 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -1,6 +1,6 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection}; -use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; +use crate::{log_err, rho::app_connection::AppConnection}; use dashmap::DashMap; use mtp::codec::{CommunicationValue, DataType}; use std::sync::Arc; @@ -178,11 +178,21 @@ impl RhoConnection { return; } self.client_connections.remove(&key); - self.iota_connection + if let Err(error) = self + .iota_connection .state .omega .client_disconnected(target_user_id, target_session_id) - .await; + .await + { + log_err!( + target_user_id, + crate::util::logger::PrintType::Client, + "UserDisconnected acknowledgement failed for session {}: {}", + target_session_id, + error + ); + } let remaining_for_user = self .client_connections .iter() @@ -194,16 +204,6 @@ impl RhoConnection { .rho .remove_user_binding(target_user_id, self.get_iota_id().await as i64) .await; - self.iota_connection - .state - .omega - .client_changed( - self.get_iota_id().await as i64, - target_user_id, - target_session_id, - UserStatus::user_offline, - ) - .await; } } @@ -228,7 +228,14 @@ impl RhoConnection { } // Notify OmegaConnection - self.iota_connection.state.omega.close_iota(iota_id).await; + if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await { + log_err!( + iota_id, + crate::util::logger::PrintType::Iota, + "IotaDisconnected acknowledgement failed: {}", + error + ); + } } /// Send message from Iota to specific client @@ -255,25 +262,6 @@ impl RhoConnection { self.iota_connection.send_message(&cv).await; } - /// Set interested users for a specific client - pub async fn set_interested(&self, user_id: i64, session_id: i64, interested_ids: Vec) { - if let Some(connection) = self.get_client_connection(user_id, session_id).await { - connection.set_interested_users(interested_ids).await; - } - } - - /// Check if clients are interested in a user - #[allow(dead_code)] - pub async fn are_they_interested(&self, user_id: i64, user_status: &str) { - let connections = self.get_client_connections().await; - for connection in connections.iter() { - connection - .clone() - .are_you_interested(user_id, user_status) - .await; - } - } - /// Check if this RhoConnection contains a specific user ID #[allow(dead_code)] pub async fn contains_user(&self, user_id: &i64) -> bool { diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index 0345ae8..b86e205 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -135,4 +135,17 @@ impl RhoManager { .map(|entry| entry.value().clone()) .collect() } + + pub async fn get_client_connection( + &self, + user_id: i64, + session_id: i64, + ) -> Option> { + for rho in self.connections().await { + if let Some(client) = rho.get_client_connection(user_id, session_id).await { + return Some(client); + } + } + None + } } From 722eb9b025af5f8a89009e66a0e9910bfe0a3b5a Mon Sep 17 00:00:00 2001 From: Alex Date: Sun, 9 Aug 2026 02:51:11 +0200 Subject: [PATCH 5/6] [Fix] User deletion & migration --- Cargo.lock | 228 +++++++++--------- Cargo.toml | 2 +- .../anonymous_client_connection.rs | 1 - src/omega/omega_connection.rs | 16 ++ src/rho/client_connection.rs | 6 + src/rho/iota_connection.rs | 6 + src/rho/rho_connection.rs | 8 + src/rho/rho_manager.rs | 4 + 8 files changed, 151 insertions(+), 120 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bf277ff..7b23a80 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,18 +14,18 @@ dependencies = [ [[package]] name = "aho-corasick" -version = "1.1.4" +version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" dependencies = [ "memchr", ] [[package]] name = "android_system_properties" -version = "0.1.5" +version = "0.1.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc" dependencies = [ "libc", ] @@ -109,9 +109,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" [[package]] name = "aws-lc-rs" -version = "1.17.3" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1" +checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" dependencies = [ "aws-lc-sys", "untrusted 0.7.1", @@ -120,9 +120,9 @@ dependencies = [ [[package]] name = "aws-lc-sys" -version = "0.43.0" +version = "0.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c" +checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" dependencies = [ "cc", "cmake", @@ -143,6 +143,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64ct" version = "1.8.3" @@ -211,9 +217,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.4.0" +version = "1.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" +checksum = "9066c49992464636f92905fa096ec58baaa4d57ec19a5c096c68d3e25ef3d136" dependencies = [ "find-msvc-tools", "jobserver", @@ -319,12 +325,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "const-oid" -version = "0.9.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" - [[package]] name = "const-oid" version = "0.10.2" @@ -411,8 +411,23 @@ dependencies = [ "cfg-if", "cpufeatures 0.2.17", "curve25519-dalek-derive", - "digest 0.10.7", - "fiat-crypto", + "fiat-crypto 0.2.9", + "rustc_version", + "subtle", + "zeroize", +] + +[[package]] +name = "curve25519-dalek" +version = "5.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "curve25519-dalek-derive", + "digest 0.11.3", + "fiat-crypto 0.3.0", "rustc_version", "subtle", "zeroize", @@ -445,20 +460,9 @@ dependencies = [ [[package]] name = "data-encoding" -version = "2.11.0" +version = "2.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" - -[[package]] -name = "der" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" -dependencies = [ - "const-oid 0.9.6", - "pem-rfc7468", - "zeroize", -] +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" [[package]] name = "der" @@ -466,7 +470,8 @@ version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d" dependencies = [ - "const-oid 0.10.2", + "const-oid", + "pem-rfc7468", "zeroize", ] @@ -523,7 +528,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.1", - "const-oid 0.10.2", + "const-oid", "crypto-common 0.2.2", "ctutils", ] @@ -563,24 +568,25 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" [[package]] name = "ed25519" -version = "2.2.3" +version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" +checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a" dependencies = [ - "pkcs8 0.10.2", - "signature 2.2.0", + "pkcs8", + "signature 3.0.0", ] [[package]] name = "ed25519-dalek" -version = "2.2.0" +version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" +checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de" dependencies = [ - "curve25519-dalek", + "curve25519-dalek 5.0.0", "ed25519", "serde", - "sha2 0.10.9", + "sha2 0.11.0", + "signature 3.0.0", "subtle", "zeroize", ] @@ -616,7 +622,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -644,10 +650,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" [[package]] -name = "find-msvc-tools" -version = "0.1.9" +name = "fiat-crypto" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24" + +[[package]] +name = "find-msvc-tools" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de" [[package]] name = "fixedbitset" @@ -953,9 +965,9 @@ checksum = "1a9fcbcc408c5526c3ab80d534e5c86e7967c1fb7aa0a8c76abd1edc27deb877" [[package]] name = "http" -version = "1.4.2" +version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" dependencies = [ "bytes", "itoa", @@ -998,9 +1010,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" [[package]] name = "hybrid-array" -version = "0.4.13" +version = "0.4.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" dependencies = [ "ctutils", "typenum", @@ -1214,9 +1226,9 @@ dependencies = [ [[package]] name = "ipnet" -version = "2.12.0" +version = "2.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78" [[package]] name = "itertools" @@ -1363,9 +1375,9 @@ dependencies = [ [[package]] name = "keccak" -version = "0.2.0" +version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e24a010dd405bd7ed803e5253182815b41bf2e6a80cc3bfc066658e03a198aa" +checksum = "ffd9697dc4a9a62e2da93389f34400b77a28f0287711263cabb203b3ccb9c0e4" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -1452,9 +1464,9 @@ dependencies = [ [[package]] name = "livekit-protocol" -version = "0.7.11" +version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "735a237deeff60124edd7831fcabce32b4b9f667d5c4c5688c9d18a2808422c4" +checksum = "4d26880e94e2f9bab298445e7d86a3794453d211a12ddbd051bd9991a343f9ff" dependencies = [ "pbjson", "pbjson-types", @@ -1531,12 +1543,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "add6b9d92e496f16f4526d68ff29da1483aba4b119baeab8bed3b9e3544a6f3d" dependencies = [ - "const-oid 0.10.2", + "const-oid", "crypto-common 0.2.2", "ctutils", "hybrid-array", "module-lattice", - "pkcs8 0.11.0", + "pkcs8", "shake", "signature 3.0.0", ] @@ -1581,7 +1593,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "mtp-client", "mtp-codec", @@ -1597,7 +1609,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "mtp-codec", "mtp-common", @@ -1610,9 +1622,9 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "byteorder", "mtp-common", "mtp-crypto", @@ -1623,7 +1635,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "quinn", "rustls", @@ -1634,9 +1646,9 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ - "base64 0.22.1", + "base64 0.23.1", "chacha20poly1305", "ed25519-dalek", "getrandom 0.4.3", @@ -1648,7 +1660,7 @@ dependencies = [ "rustls", "serde", "sha2 0.11.0", - "thiserror 1.0.69", + "thiserror 2.0.19", "tokio", "zeroize", ] @@ -1656,24 +1668,24 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "mtp-crypto", "rand 0.10.2", - "thiserror 1.0.69", + "thiserror 2.0.19", "zeroize", ] [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "mtp-codec", "mtp-common", "mtp-crypto", "mtp-transport", - "rand 0.8.7", + "rand 0.10.2", "tokio", "tracing", "wtransport", @@ -1682,7 +1694,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "async-trait", "mtp-codec", @@ -1700,7 +1712,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "serde", "serde_yaml", @@ -1709,7 +1721,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e" +source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b" dependencies = [ "async-trait", "bytes", @@ -2102,9 +2114,9 @@ dependencies = [ [[package]] name = "pem-rfc7468" -version = "0.7.0" +version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" +checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9" dependencies = [ "base64ct", ] @@ -2131,24 +2143,14 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" -[[package]] -name = "pkcs8" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" -dependencies = [ - "der 0.7.10", - "spki 0.7.3", -] - [[package]] name = "pkcs8" version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7" dependencies = [ - "der 0.8.1", - "spki 0.8.0", + "der", + "spki", ] [[package]] @@ -2327,7 +2329,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2474,9 +2476,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.4.16" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" dependencies = [ "aho-corasick", "memchr", @@ -2581,14 +2583,14 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] name = "rustls" -version = "0.23.42" +version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ "aws-lc-rs", "log", @@ -2640,7 +2642,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2844,7 +2846,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09057cb2149ad4cbd2da1e26b351f9a4c354219421229c69c3063e6f61947c4a" dependencies = [ "digest 0.11.3", - "keccak 0.2.0", + "keccak 0.2.1", "sponge-cursor", ] @@ -2927,16 +2929,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "spki" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" -dependencies = [ - "base64ct", - "der 0.7.10", -] - [[package]] name = "spki" version = "0.8.0" @@ -2944,7 +2936,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f" dependencies = [ "base64ct", - "der 0.8.1", + "der", ] [[package]] @@ -3032,10 +3024,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3080,9 +3072,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.54" +version = "0.3.55" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244" +checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134" dependencies = [ "deranged", "num-conv", @@ -3152,13 +3144,13 @@ dependencies = [ [[package]] name = "tokio-macros" -version = "2.7.1" +version = "2.7.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] @@ -3596,7 +3588,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3851,7 +3843,7 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7e468321c81fb07fa7f4c636c3972b9100f0346e5b6a9f2bd0603a52f7ed277" dependencies = [ - "curve25519-dalek", + "curve25519-dalek 4.1.3", "rand_core 0.6.4", "serde", "zeroize", @@ -3911,18 +3903,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.55" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb" +checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.55" +version = "0.8.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb" +checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index 2c37c3e..ab2ba0d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,7 @@ dotenv = "0.15.0" strum = "0.28.0" strum_macros = "0.28.0" livekit-api = { version = "0.5.6", features = ["rustls-tls-native-roots"] } -livekit-protocol = "0.7.10" +livekit-protocol = "=0.7.10" thiserror = "2.0.19" trust-dns-resolver = "0.23.2" serde_json = "1.0.151" diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 4712af0..2ec1b78 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -302,7 +302,6 @@ impl AnonymousClientConnection { if cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::GetIotaData) - || cv.is_type(CommunicationType::DeleteUser) { self.handle_omega_forward(cv).await; return; diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index e66e62c..97f025b 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -624,6 +624,22 @@ impl OmegaConnection { } } } + if cv.is_type(CommunicationType::EraseHostedUserData) { + let Some(iota_id) = cv + .get_data(DataType::IotaId) + .as_number() + .and_then(|id| i64::try_from(id).ok()) + else { + log_err!(0, PrintType::Omega, "Discarded hosted-data erasure without IotaId"); + continue; + }; + if let Some(rho) = self.rho.get_by_iota(iota_id).await { + rho.get_iota_connection().send_message(&cv).await; + } else { + log_err!(iota_id, PrintType::Omega, "Hosted-data erasure is pending because the Iota is offline"); + } + continue; + } if cv.is_type(CommunicationType::ClientChanged) { let Some((receiver, session_id)) = client_changed_target(&cv) else { diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index c5c7dee..0a6f30f 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -77,6 +77,12 @@ impl ClientConnection { *self.rho_connection.write().await = Some(rho_connection); } + /// A membership snapshot can remove an account from an Iota while keeping + /// its authenticated Omega connection alive. + pub async fn clear_rho_connection(&self) { + *self.rho_connection.write().await = None; + } + /// Send a CommunicationValue to the client pub async fn send_message(self: Arc, cv: &CommunicationValue) { if !*self.is_open.read().await { diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 047f718..747a394 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -442,6 +442,12 @@ impl IotaConnection { || cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::GetIotaData) || cv.is_type(CommunicationType::DeleteIota) + || cv.is_type(CommunicationType::AttachUserBegin) + || cv.is_type(CommunicationType::AttachUserComplete) + || cv.is_type(CommunicationType::ReleaseUserFromIota) + || cv.is_type(CommunicationType::DeleteUserCredentialBegin) + || cv.is_type(CommunicationType::DeleteUserCredentialComplete) + || cv.is_type(CommunicationType::EraseHostedUserDataAck) { let sender = self.get_iota_id().await; diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 4c60560..9944572 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -86,6 +86,14 @@ impl RhoConnection { .collect() } + pub async fn detach_user_clients(&self, user_id: i64) { + let clients = self.get_client_connections_for_user(user_id).await; + for client in clients { + client.clear_rho_connection().await; + } + self.client_connections.retain(|(id, _), _| *id != user_id as u64); + } + #[allow(unused)] pub async fn get_app_connections( &self, diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index b86e205..5cb7ce3 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -66,7 +66,11 @@ impl RhoManager { ) .await; + let replacement = user_ids.iter().copied().collect::>(); for user_id in previous_users { + if !replacement.contains(&user_id) { + rho.detach_user_clients(user_id).await; + } if self .users .get(&user_id) From 1f3fec22eb0de70a0d9bcc2ac0865051ab6d72aa Mon Sep 17 00:00:00 2001 From: Rasensprenger Date: Tue, 11 Aug 2026 03:00:40 +0200 Subject: [PATCH 6/6] chore(deps): update rust crate livekit-api to 0.6.0 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index ab2ba0d..a9aa0d8 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,7 +29,7 @@ log = "0.4" dotenv = "0.15.0" strum = "0.28.0" strum_macros = "0.28.0" -livekit-api = { version = "0.5.6", features = ["rustls-tls-native-roots"] } +livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] } livekit-protocol = "=0.7.10" thiserror = "2.0.19" trust-dns-resolver = "0.23.2"