From 1de479ce8de29db8fc0720bfa13f42e8217cfdef Mon Sep 17 00:00:00 2001 From: Alois Date: Tue, 28 Jul 2026 21:05:25 +0200 Subject: [PATCH] (feat): update mtp --- Cargo.lock | 34 +++++----- client/src/client_connection.rs | 32 +-------- communities/src/community_connection.rs | 19 ------ iota-daemon-lib/src/command_router.rs | 19 ------ mtp-type-maps | 2 +- omikron-connector/src/lib.rs | 1 - omikron-connector/src/omikron_connection.rs | 75 +++++++-------------- omikron-connector/src/ping_pong_task.rs | 44 ------------ 8 files changed, 46 insertions(+), 180 deletions(-) delete mode 100644 omikron-connector/src/ping_pong_task.rs diff --git a/Cargo.lock b/Cargo.lock index 7d6d7af..32418ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1266,13 +1266,13 @@ dependencies = [ [[package]] name = "displaydoc" -version = "0.2.6" +version = "0.2.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 2.0.119", + "syn 3.0.3", ] [[package]] @@ -1987,7 +1987,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.5.10", + "socket2 0.6.5", "system-configuration", "tokio", "tower-service", @@ -2956,7 +2956,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "mtp-client", "mtp-codec", @@ -2972,7 +2972,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "mtp-codec", "mtp-common", @@ -2985,7 +2985,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "base64", "byteorder", @@ -2998,7 +2998,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "quinn", "rustls", @@ -3009,7 +3009,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "base64", "chacha20poly1305", @@ -3031,7 +3031,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -3042,7 +3042,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "mtp-codec", "mtp-common", @@ -3057,7 +3057,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "async-trait", "mtp-codec", @@ -3075,7 +3075,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "serde", "serde_yaml", @@ -3084,7 +3084,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#bcf8aee3716f1690f1284748ac1ff3cd22799fb0" +source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" dependencies = [ "async-trait", "bytes", @@ -3810,7 +3810,7 @@ dependencies = [ "quinn-udp", "rustc-hash", "rustls", - "socket2 0.5.10", + "socket2 0.6.5", "thiserror 2.0.19", "tokio", "tracing", @@ -3851,7 +3851,7 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.5.10", + "socket2 0.6.5", "tracing", "windows-sys 0.61.2", ] @@ -4835,7 +4835,7 @@ 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.61.2", diff --git a/client/src/client_connection.rs b/client/src/client_connection.rs index da728d4..cda5cf8 100644 --- a/client/src/client_connection.rs +++ b/client/src/client_connection.rs @@ -25,7 +25,6 @@ pub struct ClientConnection { sender: Arc>>>, receiver: Receiver, connection_loop_handle: Arc>>>, - pub ping: Arc>, pub connection_id: Uuid, shutdown_tx: Arc>>>, pub waiting_tasks: @@ -38,7 +37,6 @@ impl ClientConnection { sender: Arc>>>, receiver: Receiver, connection_loop_handle: Arc>>>, - ping: Arc>, connection_id: Uuid, shutdown_tx: Arc>>>, waiting_tasks: DashMap< @@ -51,7 +49,6 @@ impl ClientConnection { sender, receiver, connection_loop_handle, - ping, connection_id, shutdown_tx, waiting_tasks, @@ -96,34 +93,11 @@ impl ClientConnection { // ------------------------------------------------------------------------- // Message Handling // ------------------------------------------------------------------------- - async fn handle_ping(self: Arc, cv: CommunicationValue) { - // Update our ping if provided - if let DataValue::SignedNumber(last_ping) = cv.get_data(DataType::LastPing) { - let current = now_millis_i64(); - let mut ping_guard = self.ping.write().await; - *ping_guard = current - *last_ping as i64; - } - - // Send pong response - let response = CommunicationValue::new(CommunicationType::Pong) - .with_id(cv.get_id()) - .add_typed_default(DataType::PingIota, DataValue::SignedNumber(0)); - - self.send_message(&response).await; - } - pub async fn handle_message(self: Arc, cv: CommunicationValue) { - if !cv.is_type(CommunicationType::Ping) && !cv.is_type(CommunicationType::Pong) { - log_cv_in!(&cv); - } + log_cv_in!(&cv); let _msg_id = cv.get_id(); - if cv.is_type(CommunicationType::Ping) { - self.handle_ping(cv).await; - return; - } - if cv.is_type(CommunicationType::Challenge) { self.handle_challenge(&cv).await; return; @@ -566,9 +540,7 @@ impl ClientConnection { let sender_clone = Arc::clone(sender); drop(sender_guard); - if !cv.is_type(CommunicationType::Ping) && !cv.is_type(CommunicationType::Pong) { - log_cv_out!(&cv); - } + log_cv_out!(&cv); if let Err(e) = sender_clone.send(cv).await { return Err(e.to_string()); diff --git a/communities/src/community_connection.rs b/communities/src/community_connection.rs index 3ceca2b..36a23c0 100644 --- a/communities/src/community_connection.rs +++ b/communities/src/community_connection.rs @@ -30,7 +30,6 @@ pub struct CommunityConnection { challenged: Arc>, challenge: Arc>, auth: Arc>>, - pub ping: Arc>, } impl CommunityConnection { pub fn new( @@ -47,7 +46,6 @@ impl CommunityConnection { challenged: Arc::new(RwLock::new(false)), challenge: Arc::new(RwLock::new(String::new())), auth: Arc::new(RwLock::new(None)), - ping: Arc::new(RwLock::new(-1)), }) } pub async fn send_message(&self, message: &CommunicationValue) { @@ -84,11 +82,6 @@ impl CommunityConnection { return; } - if cv.is_type(CommunicationType::Ping) { - self.handle_ping(cv).await; - return; - } - if cv.is_type(CommunicationType::ClientChanged) { //self.handle_client_changed(cv).await; return; @@ -381,16 +374,4 @@ impl CommunityConnection { } } - async fn handle_ping(&self, cv: CommunicationValue) { - if let Some(last_ping) = cv.get_data(DataType::LastPing) { - if let Ok(ping_val) = last_ping.to_string().parse::() { - let mut ping_guard = self.ping.write().await; - *ping_guard = ping_val; - } - } - - let response = CommunicationValue::new(CommunicationType::Pong).with_id(cv.get_id()); - - self.send_message(&response).await; - } } diff --git a/iota-daemon-lib/src/command_router.rs b/iota-daemon-lib/src/command_router.rs index b47366a..58e627d 100644 --- a/iota-daemon-lib/src/command_router.rs +++ b/iota-daemon-lib/src/command_router.rs @@ -11,7 +11,6 @@ use iota_storage::users::user_manager; use iota_storage::util::config_util::{self}; use mtp::codec::{CommunicationType, CommunicationValue}; use std::sync::{Arc, Mutex}; -use std::time::Duration; use crate::daemon_state::{ShutdownReason, StartupPhase}; @@ -295,22 +294,4 @@ impl CommandRouter { } } } - - pub async fn ping(&self, seconds: u64) -> Result { - let response = self - .services - .omikron - .await_response( - &CommunicationValue::new(CommunicationType::Ping), - Duration::from_secs(seconds), - ) - .await; - match response { - Ok(value) => { - log!("{}", iota_logger::format_cv(&value)); - Ok("Ping response received".into()) - } - Err(error) => Err(format!("Ping error: {error:?}")), - } - } } diff --git a/mtp-type-maps b/mtp-type-maps index 594646a..f430cd3 160000 --- a/mtp-type-maps +++ b/mtp-type-maps @@ -1 +1 @@ -Subproject commit 594646ac39d986f0787aa614a99d580035a67318 +Subproject commit f430cd358b3d07a4cfd6982eb8917fec80c24a7d diff --git a/omikron-connector/src/lib.rs b/omikron-connector/src/lib.rs index cce748b..69044c0 100644 --- a/omikron-connector/src/lib.rs +++ b/omikron-connector/src/lib.rs @@ -1,7 +1,6 @@ pub mod client; pub mod omega_discovery; pub mod omikron_connection; -pub mod ping_pong_task; pub mod user_ops; pub use client::{OmikronClient, OmikronError, OmikronStartupError}; diff --git a/omikron-connector/src/omikron_connection.rs b/omikron-connector/src/omikron_connection.rs index e7e7b2c..e132b37 100755 --- a/omikron-connector/src/omikron_connection.rs +++ b/omikron-connector/src/omikron_connection.rs @@ -11,7 +11,6 @@ use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; use mtp::crypto::{Keyring, PublicKeyBundle}; use std::env; use std::path::{Path, PathBuf}; -use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; use tokio::sync::{Mutex, RwLock, Semaphore, oneshot, watch}; @@ -110,10 +109,10 @@ const OMIKRON_PUBLIC_KEY_PATH: &str = "omikron.mpkb"; 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 MAINTENANCE_INTERVAL: Duration = Duration::from_secs(5); const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60); const TASK_MAX_AGE: Duration = Duration::from_secs(60); -const MAX_MISSED_PONGS: u32 = 3; +const MAX_MISSED_PINGS: usize = 3; const MAX_CONCURRENT_HANDLERS: usize = 20; // ============================================================================ @@ -168,14 +167,13 @@ pub struct OmikronConnection { sender: Arc>>>, connection_loop_handle: Arc>>>, pub last_ping: Arc>, - heartbeat_handle: Arc>>>, + maintenance_handle: Arc>>>, pub connection_id: Uuid, shutdown_tx: Arc>>>, reconnect_on_close: Arc>, auth_failure: Arc>>, pub app_challenges: Arc>, pub app_sessions: Arc>, - pub(crate) missed_pongs: Arc, handler_semaphore: Arc, cancellation: CancellationToken, pub(crate) active_tasks: Arc>, @@ -201,14 +199,13 @@ impl OmikronConnection { sender: Arc::new(RwLock::new(None)), connection_loop_handle: Arc::new(Mutex::new(None)), last_ping: Arc::new(Mutex::new(-1)), - heartbeat_handle: Arc::new(Mutex::new(None)), + maintenance_handle: Arc::new(Mutex::new(None)), connection_id: Uuid::new_v4(), shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))), reconnect_on_close: Arc::new(RwLock::new(true)), auth_failure: Arc::new(RwLock::new(None)), app_challenges: Arc::new(DashMap::new()), app_sessions: Arc::new(DashMap::new()), - missed_pongs: Arc::new(AtomicU32::new(0)), handler_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_HANDLERS)), cancellation, active_tasks, @@ -267,7 +264,7 @@ impl OmikronConnection { handle.abort(); } - if let Some(handle) = self.heartbeat_handle.lock().await.take() { + if let Some(handle) = self.maintenance_handle.lock().await.take() { handle.abort(); } @@ -363,7 +360,9 @@ impl OmikronConnection { persistent_stream_max_retries: 5, persistent_stream_retry_backoff: Duration::from_secs(5), max_frames_per_stream: None, - }); + }) + .with_ping_interval(MAINTENANCE_INTERVAL) + .with_max_missed_pings(MAX_MISSED_PINGS); let connection = match Client::auth_connect_or_register( client_config, @@ -402,18 +401,18 @@ impl OmikronConnection { // Start read loop let connection = Arc::new(connection); let read_self = self.clone(); + let read_connection = connection.clone(); let read_handle = tokio::spawn(async move { - read_self.read_loop(connection).await; + read_self.read_loop(read_connection).await; }); log_t!("omikron_authenticated"); - // Start heartbeat - let heartbeat_self = self.clone(); - let heartbeat_handle = tokio::spawn(async move { - heartbeat_self.heartbeat_loop().await; + let maintenance_self = self.clone(); + let maintenance_handle = tokio::spawn(async move { + maintenance_self.maintenance_loop(connection).await; }); - *self.heartbeat_handle.lock().await = Some(heartbeat_handle); + *self.maintenance_handle.lock().await = Some(maintenance_handle); { self.active_tasks.insert("Omikron Listener".to_string()); @@ -427,7 +426,7 @@ impl OmikronConnection { self.active_tasks.remove("Omikron Listener"); } - if let Some(handle) = self.heartbeat_handle.lock().await.take() { + if let Some(handle) = self.maintenance_handle.lock().await.take() { handle.abort(); } @@ -589,7 +588,7 @@ impl OmikronConnection { } // ------------------------------------------------------------------------- - // Read Loop & Heartbeat + // Read Loop & Maintenance // ------------------------------------------------------------------------- async fn read_loop(self: Arc, connection: Arc) { @@ -603,11 +602,6 @@ impl OmikronConnection { continue; } } - if cv.is_type(CommunicationType::Pong) { - self.handle_pong(&cv).await; - continue; - } - let permit = self.handler_semaphore.clone().acquire_owned().await; let self_clone = self.clone(); tokio::spawn(async move { @@ -635,9 +629,9 @@ impl OmikronConnection { } } - async fn heartbeat_loop(self: Arc) { + async fn maintenance_loop(self: Arc, connection: Arc) { loop { - sleep(HEARTBEAT_INTERVAL).await; + sleep(MAINTENANCE_INTERVAL).await; if !self.state.read().await.is_connected() { break; @@ -651,19 +645,13 @@ impl OmikronConnection { break; } - if self.missed_pongs.load(Ordering::Relaxed) > MAX_MISSED_PONGS { - log!( - "Connection appears dead ({} consecutive missed pongs), closing sender", - self.missed_pongs.load(Ordering::Relaxed) - ); - if let Some(sender) = self.sender.read().await.as_ref() { - sender.close().await; - } - break; + if let Some(ping) = connection.get_ping() { + let ping_ms = ping.as_millis() as i64; + *self.last_ping.lock().await = ping_ms; + self.app.lock().unwrap().push_ping_val(ping_ms as f64); } self.flush_pending_chat_secret_forwards().await; - self.send_ping().await; } } @@ -854,9 +842,7 @@ impl OmikronConnection { // ------------------------------------------------------------------------- pub async fn handle_message(self: Arc, cv: CommunicationValue) { - if !cv.is_type(CommunicationType::Ping) && !cv.is_type(CommunicationType::Pong) { - log_cv_in!(&cv); - } + log_cv_in!(&cv); let msg_id = cv.get_id(); @@ -866,11 +852,6 @@ impl OmikronConnection { } } - if cv.is_type(CommunicationType::Pong) { - self.handle_pong(&cv).await; - return; - } - self.clone().handle_message_impl(cv).await; } @@ -1804,9 +1785,7 @@ impl OmikronConnection { let sender_clone = Arc::clone(sender); drop(sender_guard); - if !cv.is_type(CommunicationType::Ping) && !cv.is_type(CommunicationType::Pong) { - log_cv_out!(&cv); - } + log_cv_out!(&cv); if let Err(e) = sender_clone.send(cv).await { self.fail_all_waiting_tasks(format!( @@ -2117,14 +2096,13 @@ impl OmikronClient for OmikronConnection { sender: self.sender.clone(), connection_loop_handle: self.connection_loop_handle.clone(), last_ping: self.last_ping.clone(), - heartbeat_handle: self.heartbeat_handle.clone(), + maintenance_handle: self.maintenance_handle.clone(), connection_id: self.connection_id, shutdown_tx: self.shutdown_tx.clone(), reconnect_on_close: self.reconnect_on_close.clone(), auth_failure: self.auth_failure.clone(), app_challenges: self.app_challenges.clone(), app_sessions: self.app_sessions.clone(), - missed_pongs: self.missed_pongs.clone(), handler_semaphore: self.handler_semaphore.clone(), cancellation: self.cancellation.clone(), active_tasks: self.active_tasks.clone(), @@ -2141,14 +2119,13 @@ impl OmikronClient for OmikronConnection { sender: self.sender.clone(), connection_loop_handle: self.connection_loop_handle.clone(), last_ping: self.last_ping.clone(), - heartbeat_handle: self.heartbeat_handle.clone(), + maintenance_handle: self.maintenance_handle.clone(), connection_id: self.connection_id, shutdown_tx: self.shutdown_tx.clone(), reconnect_on_close: self.reconnect_on_close.clone(), auth_failure: self.auth_failure.clone(), app_challenges: self.app_challenges.clone(), app_sessions: self.app_sessions.clone(), - missed_pongs: self.missed_pongs.clone(), handler_semaphore: self.handler_semaphore.clone(), cancellation: self.cancellation.clone(), active_tasks: self.active_tasks.clone(), diff --git a/omikron-connector/src/ping_pong_task.rs b/omikron-connector/src/ping_pong_task.rs deleted file mode 100644 index d55cec7..0000000 --- a/omikron-connector/src/ping_pong_task.rs +++ /dev/null @@ -1,44 +0,0 @@ -use crate::omikron_connection::OmikronConnection; -use dashmap::DashMap; -use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue}; -use std::sync::LazyLock; -use std::sync::atomic::Ordering; -use std::time::Instant; -use tokio::time::Duration; - -static PING_TIMES: LazyLock> = LazyLock::new(|| DashMap::new()); - -impl OmikronConnection { - pub async fn send_ping(&self) { - let id = rand::random(); - - PING_TIMES.insert(id, Instant::now()); - - PING_TIMES.retain(|_, v| v.elapsed() < Duration::from_secs(30)); - - self.missed_pongs.fetch_add(1, Ordering::Relaxed); - - let ping_message = CommunicationValue::new(CommunicationType::Ping) - .with_id(id) - .add_typed_default( - DataType::LastPing, - DataValue::Array(vec![DataValue::SignedNumber( - *self.last_ping.lock().await as i128, - )]), - ); - - let _ = self.send_message(&ping_message).await; - } - - pub async fn handle_pong(&self, cv: &CommunicationValue) { - self.missed_pongs.store(0, Ordering::Relaxed); - - let id = cv.get_id(); - - if let Some((_, send_time)) = PING_TIMES.remove(&id) { - let ping_ms = Instant::now().duration_since(send_time).as_millis() as i64; - *self.last_ping.lock().await = ping_ms; - self.app.lock().unwrap().push_ping_val(ping_ms as f64); - } - } -}