From 0d5e48ec8f4adf7adcbc21d8a0e7c9df052a1826 Mon Sep 17 00:00:00 2001 From: Alex Date: Mon, 27 Jul 2026 20:37:30 +0200 Subject: [PATCH] [Fix] Stability --- .cargo/config.toml | 2 +- .gitmodules | 3 + Cargo.lock | 93 ++++-- mtp-type-maps | 1 + .../anonymous_client_connection.rs | 7 +- src/omega/omega_connection.rs | 122 ++++++-- src/rho/client_connection.rs | 31 +- src/rho/connection.rs | 113 +++++++- src/rho/iota_connection.rs | 221 ++++++++++++-- src/rho/rho_connection.rs | 73 ++--- src/util/logger.rs | 7 +- type-maps.yaml | 269 ------------------ 12 files changed, 556 insertions(+), 386 deletions(-) create mode 100644 .gitmodules create mode 160000 mtp-type-maps delete mode 100644 type-maps.yaml diff --git a/.cargo/config.toml b/.cargo/config.toml index d363b83..46adaad 100644 --- a/.cargo/config.toml +++ b/.cargo/config.toml @@ -1,2 +1,2 @@ [env] -MTP_TYPE_MAPS = { value = "type-maps.yaml", relative = true } +MTP_TYPE_MAPS = { value = "mtp-type-maps/type-maps.yaml", relative = true } diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..3069632 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "mtp-type-maps"] + path = mtp-type-maps + url = ssh://git@git.methanium.net/tensamin/mtp-type-maps diff --git a/Cargo.lock b/Cargo.lock index f19e7c9..4a70dea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -92,7 +92,7 @@ checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -211,9 +211,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cc" -version = "1.3.0" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8" +checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9" dependencies = [ "find-msvc-tools", "jobserver", @@ -587,9 +587,9 @@ dependencies = [ [[package]] name = "either" -version = "1.16.0" +version = "1.17.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" +checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" [[package]] name = "enum-as-inner" @@ -655,6 +655,12 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + [[package]] name = "foldhash" version = "0.2.0" @@ -813,6 +819,25 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "h2" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "h3" version = "0.0.8" @@ -965,6 +990,12 @@ version = "1.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.13" @@ -985,9 +1016,11 @@ dependencies = [ "bytes", "futures-channel", "futures-core", + "h2", "http", "http-body", "httparse", + "httpdate", "itoa", "pin-project-lite", "smallvec", @@ -1346,9 +1379,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" [[package]] name = "libc" -version = "0.2.186" +version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" [[package]] name = "libm" @@ -1548,7 +1581,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-client", "mtp-codec", @@ -1564,7 +1597,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-codec", "mtp-common", @@ -1577,7 +1610,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "base64 0.22.1", "byteorder", @@ -1590,7 +1623,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "quinn", "rustls", @@ -1601,7 +1634,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "base64 0.22.1", "chacha20poly1305", @@ -1623,7 +1656,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -1634,7 +1667,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "mtp-codec", "mtp-common", @@ -1649,7 +1682,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "async-trait", "mtp-codec", @@ -1667,7 +1700,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "serde", "serde_yaml", @@ -1676,7 +1709,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#04760fd88d2bc3adf548a9ec532fa228227f0a49" +source = "git+https://git.methanium.net/Methanium/mtp.git#88ae866b91857c9b1650ea0ee2af2fdb70b92f10" dependencies = [ "async-trait", "bytes", @@ -1684,6 +1717,9 @@ dependencies = [ "h3-quinn", "h3-webtransport", "http", + "http-body-util", + "hyper", + "hyper-util", "mtp-codec", "mtp-common", "mtp-crypto", @@ -1694,6 +1730,8 @@ dependencies = [ "rustls", "thiserror 2.0.19", "tokio", + "tokio-rustls", + "tokio-stream", "tracing", ] @@ -2576,9 +2614,9 @@ dependencies = [ [[package]] name = "rustls-pki-types" -version = "1.15.0" +version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" dependencies = [ "web-time", "zeroize", @@ -2715,7 +2753,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -2958,9 +2996,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.2" +version = "3.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a207d6d6a2b7fc470b80443726053f18a2481b7e1eee970597051596567987a3" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" dependencies = [ "proc-macro2", "quote", @@ -3037,7 +3075,7 @@ checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" dependencies = [ "proc-macro2", "quote", - "syn 3.0.2", + "syn 3.0.3", ] [[package]] @@ -3135,9 +3173,9 @@ dependencies = [ [[package]] name = "tokio-stream" -version = "0.1.18" +version = "0.1.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" dependencies = [ "futures-core", "pin-project-lite", @@ -3162,13 +3200,14 @@ dependencies = [ [[package]] name = "tokio-util" -version = "0.7.18" +version = "0.7.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" dependencies = [ "bytes", "futures-core", "futures-sink", + "libc", "pin-project-lite", "tokio", ] diff --git a/mtp-type-maps b/mtp-type-maps new file mode 160000 index 0000000..594646a --- /dev/null +++ b/mtp-type-maps @@ -0,0 +1 @@ +Subproject commit 594646ac39d986f0787aa614a99d580035a67318 diff --git a/src/anonymous_clients/anonymous_client_connection.rs b/src/anonymous_clients/anonymous_client_connection.rs index 22f86e4..da2c650 100644 --- a/src/anonymous_clients/anonymous_client_connection.rs +++ b/src/anonymous_clients/anonymous_client_connection.rs @@ -356,7 +356,12 @@ impl AnonymousClientConnection { 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, user_status) + .client_changed( + self.user_id as i64, + self.user_id as i64, + self.user_id as i64, + user_status, + ) .await; } } diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index bb10ed2..45d0fb3 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -1,5 +1,5 @@ use crate::{ - config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, + config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out, rho::rho_manager::RhoManager, util::logger::PrintType, }; use dashmap::DashMap; @@ -24,6 +24,7 @@ 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 HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5); +const MAX_HEARTBEAT_FAILURES: usize = 3; const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); const MAX_CONCURRENT_REQUESTS: usize = 128; @@ -282,14 +283,14 @@ impl OmegaConnection { .with_send_mode(SendMode::SingleStreamPerMessage) .with_max_message_size(1_000_000_000) .with_timeouts( - Duration::from_millis(2_000), - Duration::from_millis(2_000), + Duration::from_millis(5_000), + Duration::from_millis(5_000), Duration::from_millis(30_000), ) .with_keep_alive(Some(Duration::from_secs(6))) .with_max_idle_timeout(Some(Duration::from_secs(30))) .with_receiver_queue_capacity(1000) - .with_max_concurrent_stream_tasks(10) + .with_max_concurrent_stream_tasks(64) .with_persistent_stream_retries(5, Duration::from_secs(5)), ); @@ -339,6 +340,21 @@ impl OmegaConnection { let result = read_handle.await; // Cleanup + let close_reason = sender_arc.close_reason(); + if let Some(reason) = close_reason { + log_err!( + 0, + PrintType::Omega, + "Omega transport ended with close reason: {:?}", + reason + ); + } else { + log_in!( + 0, + PrintType::Omega, + "Omega transport ended without a close reason" + ); + } *self.sender.write().await = None; *self.state.write().await = ConnectionState::Disconnected; @@ -463,6 +479,7 @@ impl OmegaConnection { } async fn heartbeat_loop(self: Arc) { + let mut consecutive_failures = 0; loop { sleep(HEARTBEAT_INTERVAL).await; @@ -481,11 +498,35 @@ impl OmegaConnection { break; } - self.send_ping().await; + match self.send_ping().await { + Ok(()) => consecutive_failures = 0, + Err(error) => { + consecutive_failures += 1; + log_err!( + 0, + PrintType::Omega, + "Heartbeat send failed ({}/{}): {}", + consecutive_failures, + MAX_HEARTBEAT_FAILURES, + error + ); + if consecutive_failures >= MAX_HEARTBEAT_FAILURES { + log_err!( + 0, + PrintType::Omega, + "Closing unhealthy Omega connection after repeated heartbeat failures" + ); + if let Some(sender) = self.sender.read().await.as_ref().cloned() { + sender.close().await; + } + break; + } + } + } } } - async fn send_ping(&self) { + async fn send_ping(&self) -> Result<(), String> { let ping = CommunicationValue::new(CommunicationType::Ping).add_typed_default( DataType::SendTime, DataValue::SignedNumber( @@ -495,7 +536,7 @@ impl OmegaConnection { .as_secs() as i128, ), ); - self.send_message(&ping).await; + self.try_send_message(&ping).await } async fn handle_pong(&self, cv: &CommunicationValue) { @@ -516,7 +557,7 @@ impl OmegaConnection { // Public API // ------------------------------------------------------------------------- - pub async fn send_message(&self, cv: &CommunicationValue) { + async fn try_send_message(&self, cv: &CommunicationValue) -> Result<(), String> { if !cv.is_type(CommunicationType::Pong) && !cv.is_type(CommunicationType::Ping) { log_cv_out!(PrintType::Omega, &cv); } @@ -531,7 +572,7 @@ impl OmegaConnection { if let Some(sender) = self.sender.write().await.take() { sender.close().await; } - return; + return Err("connection is closed".to_string()); } let sender_clone = Arc::clone(sender); @@ -539,10 +580,17 @@ impl OmegaConnection { if let Err(e) = sender_clone.send(cv).await { log_err!(0, PrintType::Omega, "Send failed: {}", e); + return Err(e.to_string()); } } else { log_err!(0, PrintType::Omega, "Cannot send: not connected"); + return Err("not connected".to_string()); } + Ok(()) + } + + pub async fn send_message(&self, cv: &CommunicationValue) { + let _ = self.try_send_message(cv).await; } pub async fn await_connection(&self, timeout_duration: Option) -> Result<(), String> { @@ -608,6 +656,17 @@ impl OmegaConnection { msg_id, WaitingTask { task: Box::new(move |_, response_cv| { + log_in!( + 0, + PrintType::Omega, + "Matched Omega response (request_id={}, response_id={}, type={})", + msg_id, + response_cv.get_id(), + response_cv + .get_comm_type_enum() + .map(|kind| kind.to_string()) + .unwrap_or_else(|| response_cv.get_type().to_string()) + ); let inner_tx = tx.clone(); tokio::spawn(async move { let _ = inner_tx.send(response_cv).await; @@ -618,7 +677,19 @@ impl OmegaConnection { }, ); - self.send_message(cv).await; + if let Err(error) = self.try_send_message(cv).await { + self.waiting_tasks.remove(&msg_id); + return Err(format!("failed to send request: {error}")); + } + log_out!( + 0, + PrintType::Omega, + "Sent Omega request (request_id={}, type={})", + msg_id, + cv.get_comm_type_enum() + .map(|kind| kind.to_string()) + .unwrap_or_else(|| cv.get_type().to_string()) + ); let timeout = timeout_duration.unwrap_or(Duration::from_secs(10)); @@ -648,20 +719,33 @@ impl OmegaConnection { self.send_message(&cv).await; } - pub async fn client_changed(&self, _iota_id: i64, user_id: i64, state: UserStatus) { - let msg_type = match state { - UserStatus::iota_offline => CommunicationType::UserDisconnected, - UserStatus::user_offline => CommunicationType::UserDisconnected, - UserStatus::user_invisible => CommunicationType::UserDisconnected, - _ => CommunicationType::UserConnected, - }; - - let cv = CommunicationValue::new(msg_type) + 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) { + 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; + } + pub async fn user_states(&self, user_id: i64, user_ids: Vec) { let user_ids = user_ids .iter() diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 615885d..0cd3e39 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -272,6 +272,28 @@ impl ClientConnection { } } + // Every Iota request is bound to the authenticated device. A + // caller may omit the field for compatibility, but can never + // select a different device session. + let expected_session_id = self.session_id as i128; + if let Some(session_id) = cv.get_data(DataType::SessionId).as_signed_number() { + if session_id != expected_session_id { + let response = CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(cv.get_id()) + .add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); + self.send_message(&response).await; + return; + } + } else { + cv = cv.add_typed_default( + DataType::SessionId, + DataValue::SignedNumber(expected_session_id), + ); + } + // Forward other messages to Iota self.forward_to_iota(cv).await; }); @@ -330,6 +352,7 @@ impl ClientConnection { .client_changed( rho_conn.get_iota_id().await as i64, user_id as i64, + self.session_id as i64, user_status, ) .await; @@ -868,7 +891,7 @@ impl ClientConnection { user_status }; let notification = CommunicationValue::new(CommunicationType::ClientChanged) - .add_typed_default(DataType::UserId, DataValue::Str(user_id.to_string())) + .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; @@ -876,12 +899,10 @@ impl ClientConnection { } /// Handle connection close - pub async fn handle_close(&self) { + pub async fn handle_close(self: Arc) { let user_id = self.get_user_id().await; if let Some(rho_conn) = self.state.rho.get_for_user(user_id as i64).await { - rho_conn - .close_client_connection(Arc::new(self.clone())) - .await; + rho_conn.close_client_connection(self.clone()).await; } } } diff --git a/src/rho/connection.rs b/src/rho/connection.rs index fb99436..a39219e 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -1,4 +1,5 @@ use std::{sync::Arc, time::Duration}; +use strum_macros::Display; use tokio::sync::RwLock; use uuid::Uuid; @@ -27,7 +28,7 @@ pub type MtpReceiver = WebMtpReceiver; * `Identification`/`Register` hello frames; it is kept only so * `app_connection.rs` still compiles. */ -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Display)] #[allow(dead_code)] pub enum ConnectionKind { Client, @@ -85,8 +86,20 @@ impl GeneralConnection { pub async fn handle(self: Arc) { log_in!(0, PrintType::General, "General connection handler started"); - self.migrate().await; - log_out!(0, PrintType::General, "General connection handler stopped"); + if self.migrate().await { + log_out!( + 0, + PrintType::General, + "General connection handler stopped, upgraded to {}", + self.connection_kind + ); + } else { + log_out!( + 0, + PrintType::General, + "General connection handler stopped without upgrade" + ); + } } async fn migrate(self: &Arc) -> bool { @@ -102,19 +115,67 @@ impl GeneralConnection { async fn migrate_client(self: &Arc) { let id = self.id; let user_id = id as i64; - self.notify_user_connected(user_id).await; + + let Ok(handshake) = self.receiver.receive().await else { + return; + }; + if !handshake.is_type(CommunicationType::ClientConnected) { + let error = CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(handshake.get_id()); + let _ = self.sender.send(&error).await; + return; + } + let Some(session_id) = handshake.get_data(DataType::SessionId).as_signed_number() else { + let _ = self + .sender + .send( + &CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(handshake.get_id()), + ) + .await; + return; + }; + if !(1..=i64::MAX as i128).contains(&session_id) { + let _ = self + .sender + .send( + &CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(handshake.get_id()), + ) + .await; + return; + } + let version = handshake + .get_data(DataType::VersionNumber) + .as_signed_number(); + if !matches!(version, Some(version) if version >= 0) { + let _ = self + .sender + .send( + &CommunicationValue::new(CommunicationType::ErrorInvalidData) + .with_id(handshake.get_id()), + ) + .await; + return; + } + *self.session_id.write().await = session_id as u64; let client = ClientConnection::from_general(self.clone(), id).await; let rho = self.find_user_rho(user_id).await; *self.rho_connection.write().await = rho.clone(); if let Some(rho_conn) = rho { - if let Some(response) = self.request_initial_client_state(&rho_conn, user_id).await { + 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; + if let Some(response) = self + .request_initial_client_state(&rho_conn, user_id, handshake) + .await + { let response = self.add_call_state(response, user_id).await; log_cv_out!(response); let _ = self.sender.send(&response).await; - rho_conn.bind_user_id(user_id).await; - rho_conn.add_client_connection(client.clone()).await; } } else { log_err!( @@ -128,9 +189,28 @@ impl GeneralConnection { client.start(); } - async fn notify_user_connected(&self, user_id: i64) { + async fn notify_user_connected(&self, user_id: i64, iota_id: i64) { + 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::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; } @@ -167,6 +247,7 @@ impl GeneralConnection { &self, rho: &Arc, user_id: i64, + handshake: CommunicationValue, ) -> Option { let session_id = *self.session_id.read().await as i64; let request = CommunicationValue::new(CommunicationType::ClientConnected) @@ -174,6 +255,18 @@ impl GeneralConnection { .add_typed_default( DataType::SessionId, DataValue::SignedNumber(session_id.into()), + ) + .add_typed_default( + DataType::VersionNumber, + handshake.get_data(DataType::VersionNumber).clone(), + ) + .add_typed_default( + DataType::CacheValid, + handshake.get_data(DataType::CacheValid).clone(), + ) + .add_typed_default( + DataType::CacheSchemaVersion, + handshake.get_data(DataType::CacheSchemaVersion).clone(), ); rho.get_iota_connection() .clone() @@ -187,7 +280,7 @@ impl GeneralConnection { response: CommunicationValue, user_id: i64, ) -> CommunicationValue { - let mut output = CommunicationValue::new(CommunicationType::IdentificationResponse); + let mut output = response.clone(); for (key, value) in response.iter_typed_data() { if key == Some(DataType::Contacts) { diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index d154bf7..5e26f80 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -225,22 +225,124 @@ impl IotaConnection { } if cv.is_type(CommunicationType::CompleteRegisterUser) { - let response_cv = self + // Registration carries the authenticated Iota ID separately so + // Omega can bind the allocation lease to this connection rather + // than trusting a client-supplied value. + let request = cv.clone().add_typed_default( + DataType::IotaId, + DataValue::SignedNumber(self.iota_id.into()), + ); + log_in!( + self.iota_id as i64, + PrintType::Omega, + "Forwarding CompleteRegisterUser to Omega (request_id={})", + request.get_id() + ); + let mut response_cv = self .state .omega .clone() - .await_response( - &cv.clone().with_sender(self.iota_id), - Some(Duration::from_secs(20)), - ) + .await_response(&request, Some(Duration::from_secs(8))) .await; - if let Ok(response_cv) = response_cv { - if response_cv.is_type(CommunicationType::Success) { - if let Some(user_id) = cv.get_data(DataType::UserId).as_number() { - self.add_user_id(user_id as u64).await; + if let Err(error) = &response_cv { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "CompleteRegisterUser request_id={} failed: {}; retrying once", + request.get_id(), + error + ); + response_cv = self + .state + .omega + .clone() + .await_response(&request, Some(Duration::from_secs(8))) + .await; + } + match response_cv { + Ok(response_cv) => { + log_in!( + self.iota_id as i64, + PrintType::Omega, + "Omega completed registration (request_id={}, response_id={}, type={})", + request.get_id(), + response_cv.get_id(), + response_cv + .get_comm_type_enum() + .map(|kind| kind.to_string()) + .unwrap_or_else(|| response_cv.get_type().to_string()) + ); + if response_cv.is_type(CommunicationType::Success) { + if let Some(user_id) = cv.get_data(DataType::UserId).as_number() { + self.add_user_id(user_id as u64).await; + } } + self.send_message(&response_cv).await; + } + Err(error) => { + // Omega may have committed the insert even when its + // Success response was lost in transit. Verify the exact + // generated user ID before reporting failure; GetUserData + // uses the proven request/response path and keeps this + // recovery idempotent. + let user_id = cv.get_data(DataType::UserId).as_number(); + if let Some(user_id) = user_id { + let verification = CommunicationValue::new(CommunicationType::GetUserData) + .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id)); + match self + .state + .omega + .clone() + .await_response(&verification, Some(Duration::from_secs(3))) + .await + { + Ok(verified) + if verified.get_data(DataType::UserId).as_number() + == Some(user_id) => + { + log_in!( + self.iota_id as i64, + PrintType::Omega, + "Registration response was lost; verified user {} remotely", + user_id + ); + self.add_user_id(user_id as u64).await; + self.send_message( + &CommunicationValue::new(CommunicationType::Success) + .with_id(cv.get_id()), + ) + .await; + return; + } + Ok(verified) => log_err!( + self.iota_id as i64, + PrintType::Omega, + "Registration verification returned an unexpected user (request_id={}, response_id={})", + verification.get_id(), + verified.get_id() + ), + Err(verify_error) => log_err!( + self.iota_id as i64, + PrintType::Omega, + "Registration verification failed after request_id={}: {}", + verification.get_id(), + verify_error + ), + } + } + log_err!( + self.iota_id as i64, + PrintType::Omega, + "CompleteRegisterUser forwarding failed: {}", + error + ); + self.send_error_response( + cv.get_id(), + CommunicationType::ErrorInternal, + Some(&format!("Omega forwarding failed: {error}")), + ) + .await; } - self.send_message(&response_cv).await; } return; } @@ -249,7 +351,6 @@ impl IotaConnection { || cv.is_type(CommunicationType::PushNotification) || cv.is_type(CommunicationType::GetUserData) || cv.is_type(CommunicationType::GetIotaData) - || cv.is_type(CommunicationType::GetRegister) || cv.is_type(CommunicationType::DeleteIota) { let sender = self.get_iota_id().await; @@ -258,12 +359,27 @@ impl IotaConnection { .await; return; } + + // Register allocation is scoped to this authenticated Iota. Keep the + // source ID in typed data so Omega can create a durable lease. + if cv.is_type(CommunicationType::GetRegister) { + self.handle_omega_forward_without_sender(cv).await; + return; + } self.forward_to_client(cv).await; } #[allow(dead_code)] - async fn send_error_response(&self, message_id: u32, error_type: CommunicationType) { - let error = CommunicationValue::new(error_type).with_id(message_id); + async fn send_error_response( + &self, + message_id: u32, + error_type: CommunicationType, + detail: Option<&str>, + ) { + let mut error = CommunicationValue::new(error_type).with_id(message_id); + if let Some(detail) = detail { + error = error.add_typed_default(DataType::ErrorType, DataValue::Str(detail.into())); + } self.send_message(&error).await; } @@ -284,6 +400,52 @@ impl IotaConnection { iota_for_closure.send_message(&response_cv).await; } } + + async fn handle_omega_forward_without_sender(self: Arc, cv: CommunicationValue) { + let iota_for_closure = self.clone(); + let request = cv.clone().add_typed_default( + DataType::IotaId, + DataValue::SignedNumber(self.iota_id.into()), + ); + let mut response_cv = self + .state + .omega + .clone() + .await_response(&request, Some(Duration::from_secs(8))) + .await; + if let Err(error) = &response_cv { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "GetRegister request_id={} failed: {}; retrying once", + request.get_id(), + error + ); + response_cv = self + .state + .omega + .clone() + .await_response(&request, Some(Duration::from_secs(8))) + .await; + } + match response_cv { + Ok(response_cv) => iota_for_closure.send_message(&response_cv).await, + Err(error) => { + log_err!( + self.iota_id as i64, + PrintType::Omega, + "GetRegister forwarding failed: {}", + error + ); + self.send_error_response( + cv.get_id(), + CommunicationType::ErrorInternal, + Some(&format!("Omega forwarding failed: {error}")), + ) + .await; + } + } + } /// Handle ping message async fn handle_ping(&self, cv: CommunicationValue) { if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { @@ -397,8 +559,31 @@ impl IotaConnection { } let mut interested_ids: Vec = Vec::new(); + let session_id = cv.get_data(DataType::SessionId).as_signed_number(); let tm = TypeMap::latest(); + // Presence interest is the complete contact set, independent of + // whether this account currently participates in a call. + if let DataValue::Array(users) = cv.get_data(DataType::UserIds) { + for user in users { + match user { + DataValue::SignedNumber(id) => interested_ids.push(*id as i64), + DataValue::Container(entries) => { + if let Some(DataValue::SignedNumber(id)) = + entries.iter().find_map(|(key, value)| { + (*key == data_type_id(DataType::UserId, &tm)).then_some(value) + }) + { + interested_ids.push(*id as i64); + } + } + _ => {} + } + } + interested_ids.sort_unstable(); + interested_ids.dedup(); + } + // ============================ // Load Calls // ============================ @@ -494,8 +679,6 @@ impl IotaConnection { if let Some(DataValue::SignedNumber(id)) = user_map.get(&data_type_id(DataType::UserId, &tm)) { - interested_ids.push(*id as i64); - if let Some(call_list) = invites.get(&(*id as i64)) && !call_list.is_empty() { @@ -526,9 +709,11 @@ impl IotaConnection { // Notify Rho // ============================ if let Some(rho_conn) = self.get_rho_connection().await { - rho_conn - .set_interested(user_id as i64, interested_ids) - .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; + } } // ============================ diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index c298b7a..c7e4f65 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -2,7 +2,7 @@ use super::{client_connection::ClientConnection, iota_connection::IotaConnection use crate::{data::user::UserStatus, rho::app_connection::AppConnection}; use dashmap::DashMap; -use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; +use mtp::codec::{CommunicationValue, DataType}; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; @@ -128,39 +128,51 @@ impl RhoConnection { let user_id = connection.user_id as i64; let session_id = connection.session_id as i64; - let notification = CommunicationValue::new(CommunicationType::ClientConnected) - .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into())); - - self.iota_connection.send_message(¬ification).await; - - let should_notify_online = !self + if let Some((_, old_connection)) = self .client_connections - .iter() - .any(|entry| entry.key().0 == user_id as u64); - self.client_connections - .remove(&(user_id as u64, session_id as u64)); + .remove(&(user_id as u64, session_id as u64)) + { + // A reconnect for a persistent device replaces the old transport; + // do this before publishing the new route. + old_connection.close().await; + } self.client_connections .insert((user_id as u64, session_id as u64), Arc::clone(&connection)); + } - if should_notify_online { - self.iota_connection - .state - .omega - .client_changed( - self.get_iota_id().await as i64, - user_id, - UserStatus::user_online, - ) - .await; + pub async fn get_client_connection( + &self, + user_id: i64, + session_id: i64, + ) -> Option> { + if user_id < 0 || session_id <= 0 { + return None; } + self.client_connections + .get(&(user_id as u64, session_id as u64)) + .map(|entry| entry.value().clone()) } /// Remove a client connection pub async fn close_client_connection(&self, connection: Arc) { let target_user_id = connection.user_id as i64; let target_session_id = connection.session_id as i64; - self.client_connections - .remove(&(target_user_id as u64, target_session_id as u64)); + // Do not let a stale transport tear down the replacement for the same + // persistent session. + let key = (target_user_id as u64, target_session_id as u64); + if self + .client_connections + .get(&key) + .is_some_and(|current| !Arc::ptr_eq(current.value(), &connection)) + { + return; + } + self.client_connections.remove(&key); + self.iota_connection + .state + .omega + .client_disconnected(target_user_id, target_session_id) + .await; let remaining_for_user = self .client_connections .iter() @@ -178,6 +190,7 @@ impl RhoConnection { .client_changed( self.get_iota_id().await as i64, target_user_id, + target_session_id, UserStatus::user_offline, ) .await; @@ -232,17 +245,9 @@ impl RhoConnection { } /// Set interested users for a specific client - pub async fn set_interested(&self, user_id: i64, interested_ids: Vec) { - let connections = self.get_client_connections().await; - for connection in connections.iter() { - let conn_user_id = connection.get_user_id().await; - if conn_user_id == user_id as u64 { - connection - .clone() - .set_interested_users(interested_ids.clone()) - .await; - break; - } + 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; } } diff --git a/src/util/logger.rs b/src/util/logger.rs index dfc26fa..97b17fa 100644 --- a/src/util/logger.rs +++ b/src/util/logger.rs @@ -173,8 +173,11 @@ pub fn format_cv(cv: &CommunicationValue) -> String { parts.push(format!("> {}", receiver)); } - let comm_type = cv.get_type().to_string(); - parts.push(format!("{}", comm_type)); + let comm_type = cv + .get_comm_type_enum() + .map(|kind| kind.to_string()) + .unwrap_or_else(|| cv.get_type().to_string()); + parts.push(format!("{} (id={})", comm_type, cv.get_id())); let data = cv.data(); diff --git a/type-maps.yaml b/type-maps.yaml deleted file mode 100644 index ce3d5e6..0000000 --- a/type-maps.yaml +++ /dev/null @@ -1,269 +0,0 @@ -protocol_version: "1.0" - -# Note that markers 0 to 31 are reserved for default use, manually working with them is not recommended - -# Fixed CommunicationType markers are: -# Error: 0 -# ErrorParsing: 1 -# ErrorBadVersion: 2 -# Disconnect: 3 -# Redirect: 4 -# Shutdown: 5 -# BadRequest: 6 -# Unauthorized: 7 -# Forbidden: 8 -# NotFound: 9 -# TooManyRequests: 10 -# InternalServerError: 11 -# BadGateway: 12 -# ServiceUnavailable: 13 -# GatewayTimeout: 14 -# Identification: 15 -# IdentificationResponse: 16 -# Register: 17 -# RegisterResponse: 18 -# Ping: 19 -# Pong: 20 - -# Fixed Data Type markers are: -# Error: 0 -# ErrorParsing: 1 -# ErrorMessage: 2 -# Version: 3 -# Description: 4 -# Timestamp: 5 -# Id: 6 -# ClientNonce: 7 -# ServerNonce: 8 -# PublicKeys: 9 -# Signature: 10 -# Connected: 11 - -type_maps: - "1.0": - CommunicationTypes: - ErrorProtocol: 33 - ErrorAnonymous: 34 - ErrorInternal: 35 - ErrorInvalidData: 36 - ErrorInvalidUserId: 37 - ErrorInvalidOmikronId: 38 - ErrorNotFound: 39 - ErrorNotAuthenticated: 40 - ErrorNoIota: 41 - ErrorInvalidChallenge: 42 - ErrorInvalidSecret: 43 - ErrorInvalidPrivateKey: 44 - ErrorInvalidPublicKey: 45 - ErrorNoUserId: 46 - ErrorNoCallId: 47 - ErrorInvalidCallId: 48 - Success: 49 - ShortenLink: 50 - SettingsSave: 51 - SettingsLoad: 52 - SettingsList: 53 - GlobalSettingsSave: 54 - GlobalSettingsLoad: 55 - Message: 56 - MessageState: 57 - MessageSend: 58 - MessageLive: 59 - MessageOtherIota: 60 - MessageChunk: 61 - MessageGet: 143 - MessagesGet: 62 - PushNotification: 63 - ReadNotification: 64 - GetNotifications: 65 - TauriIdentification: 66 - ChangeConfirm: 67 - ConfirmReceive: 68 - ConfirmRead: 69 - GetChats: 70 - GetStates: 71 - AddCommunity: 72 - RemoveCommunity: 73 - GetCommunities: 74 - RegisterIota: 81 - RegisterIotaSuccess: 82 - AddConversation: 85 - SendChat: 86 - ClientChanged: 87 - ClientConnected: 88 - ClientDisconnected: 89 - ClientClosed: 90 - PublicKey: 91 - PrivateKey: 92 - WebrtcSdp: 93 - WebrtcIce: 94 - StartStream: 95 - EndStream: 96 - WatchStream: 97 - CallToken: 98 - CallInvite: 99 - CallDisconnectUser: 100 - CallTimeoutUser: 101 - CallSetAnonymousJoining: 102 - CallData: 103 - EndCall: 104 - Function: 105 - Update: 106 - CreateUser: 107 - RhoUpdate: 108 - UserConnected: 109 - UserDisconnected: 110 - IotaConnected: 111 - IotaDisconnected: 112 - SyncClientIotaStatus: 113 - GetUserData: 114 - GetIotaData: 115 - IotaUserData: 116 - ChangeUserData: 117 - ChangeIotaData: 118 - GetRegister: 119 - CompleteRegisterUser: 120 - CompleteRegisterIota: 121 - DeleteUser: 122 - DeleteIota: 123 - StartRegister: 124 - CompleteRegister: 125 - GetApp: 126 - CreateApp: 127 - DeleteApp: 128 - SaveAppData: 129 - LoadAppData: 130 - AppIdentification: 131 - AppChallenge: 132 - AppChallengeResponse: 133 - AppIdentificationResponse: 134 - LoadTxtRecord: 135 - ErrorNotSet: 136 - SetChatSecret: 139 - GetChatSecret: 140 - ChatSecretResponse: 141 - ChatSecretForward: 142 - MessageEditLive: 144 - MessageEdit: 145 - MessageReactionAdd: 146 - MessageReactionRemove: 147 - MessageReactionLive: 148 - MessageDeleteLive: 150 - DataTypes: - ErrorType: 32 - ErrorProtocol: 33 - AcceptedIds: 34 - Uuid: 35 - RegisterId: 36 - Link: 37 - Settings: 38 - SettingsName: 39 - ChatPartnerId: 40 - ChatPartnerName: 41 - IotaId: 42 - UserId: 43 - UserIds: 44 - IotaIds: 45 - UserState: 46 - UserStates: 47 - UserPings: 48 - CallState: 49 - ScreenShare: 50 - PrivateKeyHash: 51 - # Accepted: 52 now part of default MTP - AcceptedProfiles: 53 - DeniedProfiles: 54 - Content: 55 - Messages: 56 - Notifications: 57 - SendTime: 58 - GetTime: 59 - GetVariant: 60 - SharedSecretOwn: 61 - SharedSecretOther: 62 - SharedSecretSign: 63 - SharedSecret: 64 - CallId: 65 - CallToken: 66 - CallSecret: 67 - Untill: 68 - Enabled: 69 - StartDate: 70 - EndDate: 71 - ReceiverId: 72 - SenderId: 73 - Signed: 75 - Message: 76 - MessageState: 77 - LastPing: 78 - PingIota: 79 - PingClients: 80 - Matches: 81 - Omikron: 82 - Offset: 83 - Amount: 84 - Position: 85 - Name: 86 - Path: 87 - Codec: 88 - Function: 89 - Payload: 90 - Result: 91 - Interactables: 92 - WantToWatch: 93 - Watcher: 94 - CreatedAt: 95 - Username: 96 - Display: 97 - Avatar: 98 - About: 99 - Status: 100 - PublicKey: 101 - SubLevel: 102 - SubEnd: 103 - CommunityAddress: 104 - CommunityTitle: 106 - Communities: 107 - RhoConnections: 108 - User: 109 - OnlineStatus: 110 - OmikronId: 111 - OmikronConnections: 112 - ResetToken: 113 - NewToken: 114 - CallInvited: 115 - CallMembers: 116 - Calls: 117 - Timeout: 118 - HasAdmin: 119 - LastMessageAt: 120 - Height: 121 - SentBySelf: 122 - SessionId: 123 - Contacts: 124 - LastMessage: 125 - AppIdentifier: 127 - AppPrivateKey: 128 - AppPublicKey: 129 - AppSession: 130 - AppData: 131 - TauriToken: 132 - Challenge: 133 - EncryptedPayload: 134 - SecurePayload: 135 - DeviceId: 136 - ClientId: 137 - SecretId: 142 - VersionNumber: 143 - EncryptedSecret: 144 - WrappingScheme: 146 - UpdatedAt: 147 - ChatId: 148 - KemCiphertext: 149 - SenderUserId: 152 - RecipientUserId: 153 - Recipients: 154 - Edited: 155 - Reactions: 156 - Reaction: 157 - ReplyId: 158