From c3a07b31ec5b20fa1e48970586d7ddf8b28eb813 Mon Sep 17 00:00:00 2001 From: Alois Date: Tue, 28 Jul 2026 23:53:37 +0200 Subject: [PATCH] (fix): connections (again) --- Cargo.lock | 22 ++++++++++----------- src/omega/omega_connection.rs | 3 +-- src/rho/client_connection.rs | 4 ++++ src/rho/connection.rs | 6 ++++++ src/rho/iota_connection.rs | 9 ++++++++- src/rho/rho_connection.rs | 36 +++++++++++++++++++++++------------ src/rho/rho_manager.rs | 13 +++++++++++-- src/rho/server.rs | 2 +- 8 files changed, 66 insertions(+), 29 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f76f154..3f395d9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1581,7 +1581,7 @@ dependencies = [ [[package]] name = "mtp" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "mtp-client", "mtp-codec", @@ -1597,7 +1597,7 @@ dependencies = [ [[package]] name = "mtp-client" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "mtp-codec", "mtp-common", @@ -1610,7 +1610,7 @@ dependencies = [ [[package]] name = "mtp-codec" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "base64 0.22.1", "byteorder", @@ -1623,7 +1623,7 @@ dependencies = [ [[package]] name = "mtp-common" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "quinn", "rustls", @@ -1634,7 +1634,7 @@ dependencies = [ [[package]] name = "mtp-crypto" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "base64 0.22.1", "chacha20poly1305", @@ -1656,7 +1656,7 @@ dependencies = [ [[package]] name = "mtp-files" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "mtp-crypto", "rand 0.10.2", @@ -1667,7 +1667,7 @@ dependencies = [ [[package]] name = "mtp-host" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "mtp-codec", "mtp-common", @@ -1682,7 +1682,7 @@ dependencies = [ [[package]] name = "mtp-transport" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "async-trait", "mtp-codec", @@ -1700,7 +1700,7 @@ dependencies = [ [[package]] name = "mtp-type-map" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "serde", "serde_yaml", @@ -1709,7 +1709,7 @@ dependencies = [ [[package]] name = "mtp-webserver" version = "0.2.0" -source = "git+https://git.methanium.net/Methanium/mtp.git#590810ce59709efccd81ca04401c855be314b9c0" +source = "git+https://git.methanium.net/Methanium/mtp.git#6f673ba7f2834b95ce724babe131241df8a119c6" dependencies = [ "async-trait", "bytes", @@ -3032,7 +3032,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.4.3", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", diff --git a/src/omega/omega_connection.rs b/src/omega/omega_connection.rs index a9c9dbb..78a2d38 100644 --- a/src/omega/omega_connection.rs +++ b/src/omega/omega_connection.rs @@ -24,7 +24,6 @@ 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 PING_INTERVAL: Duration = Duration::from_secs(5); -const MAX_MISSED_PINGS: 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; @@ -284,7 +283,7 @@ impl OmegaConnection { .with_persistent_stream_retries(5, Duration::from_secs(5)), ) .with_ping_interval(PING_INTERVAL) - .with_max_missed_pings(MAX_MISSED_PINGS); + .with_max_missed_pings(0); let host_public_key = load_public_key_bundle("./omega.mpkb") .map_err(|e| format!("Failed to load omega.mpkb: {}", e))?; diff --git a/src/rho/client_connection.rs b/src/rho/client_connection.rs index 0a618d8..64cb052 100644 --- a/src/rho/client_connection.rs +++ b/src/rho/client_connection.rs @@ -75,6 +75,10 @@ impl ClientConnection { self.rho_connection.read().await.clone() } + pub async fn set_rho_connection(&self, rho_connection: Arc) { + *self.rho_connection.write().await = Some(rho_connection); + } + /// 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/connection.rs b/src/rho/connection.rs index 3a545f4..02957fd 100755 --- a/src/rho/connection.rs +++ b/src/rho/connection.rs @@ -234,6 +234,12 @@ impl GeneralConnection { let iota = IotaConnection::from_general(self.clone(), id).await; let rho = Arc::new(RhoConnection::new(iota.clone(), Vec::new()).await); iota.set_rho_connection(rho.clone()).await; + if let Some(previous) = self.state.rho.get_by_iota(id as i64).await { + for client in previous.drain_client_connections() { + client.set_rho_connection(rho.clone()).await; + rho.add_client_connection(client).await; + } + } self.state.rho.add(rho).await; self.load_iota_users(&iota, id).await; iota.start(); diff --git a/src/rho/iota_connection.rs b/src/rho/iota_connection.rs index 7d1b78e..7f7086c 100755 --- a/src/rho/iota_connection.rs +++ b/src/rho/iota_connection.rs @@ -66,7 +66,14 @@ impl IotaConnection { iota_for_closure.handle_message(cv).await; }); } - Err(_) => { + Err(error) => { + log_err!( + self_clone.iota_id as i64, + PrintType::Iota, + "Iota receive loop ended: {}; transport close reason: {:?}", + error, + self_clone.receiver.close_reason() + ); break; } } diff --git a/src/rho/rho_connection.rs b/src/rho/rho_connection.rs index 35e259d..46f0330 100644 --- a/src/rho/rho_connection.rs +++ b/src/rho/rho_connection.rs @@ -63,6 +63,17 @@ impl RhoConnection { .collect() } + pub fn drain_client_connections(&self) -> Vec> { + let keys: Vec<_> = self + .client_connections + .iter() + .map(|entry| *entry.key()) + .collect(); + keys.into_iter() + .filter_map(|key| self.client_connections.remove(&key).map(|(_, value)| value)) + .collect() + } + /// Get client connections for a specific user pub async fn get_client_connections_for_user( &self, @@ -198,25 +209,26 @@ impl RhoConnection { /// Close the Iota connection and all associated client connections pub async fn close_iota_connection(&self) { + let iota_id = self.get_iota_id().await as i64; + if self + .iota_connection + .state + .rho + .remove_if_current(iota_id, self) + .await + .is_none() + { + return; + } + // Close all client connections let connections = self.get_client_connections().await; for connection in connections { connection.close().await; } - // Remove from manager - self.iota_connection - .state - .rho - .remove(self.get_iota_id().await as i64) - .await; - // Notify OmegaConnection - self.iota_connection - .state - .omega - .close_iota(self.get_iota_id().await as i64) - .await; + self.iota_connection.state.omega.close_iota(iota_id).await; } /// Send message from Iota to specific client diff --git a/src/rho/rho_manager.rs b/src/rho/rho_manager.rs index da9acd8..0345ae8 100644 --- a/src/rho/rho_manager.rs +++ b/src/rho/rho_manager.rs @@ -89,8 +89,17 @@ impl RhoManager { } } - pub async fn remove(&self, iota_id: i64) -> Option> { - let rho = self.connections.remove(&iota_id).map(|(_, rho)| rho); + pub async fn remove_if_current( + &self, + iota_id: i64, + expected: &RhoConnection, + ) -> Option> { + let rho = self + .connections + .remove_if(&iota_id, |_, current| { + std::ptr::eq(current.as_ref(), expected) + }) + .map(|(_, rho)| rho); if let Some(rho) = rho.as_ref() { self.users.retain(|_, mapped| !Arc::ptr_eq(mapped, rho)); } diff --git a/src/rho/server.rs b/src/rho/server.rs index 78c75a4..ef3c85b 100644 --- a/src/rho/server.rs +++ b/src/rho/server.rs @@ -124,7 +124,7 @@ pub async fn start(state: Arc) -> Result<(), Box