diff --git a/common/client-core/src/client/topology_control/nym_api_provider.rs b/common/client-core/src/client/topology_control/nym_api_provider.rs index 291d10ebce..92b31890e3 100644 --- a/common/client-core/src/client/topology_control/nym_api_provider.rs +++ b/common/client-core/src/client/topology_control/nym_api_provider.rs @@ -19,7 +19,6 @@ pub(crate) struct NymApiTopologyProvider { impl NymApiTopologyProvider { pub(crate) fn new(mut nym_api_urls: Vec, client_version: String) -> Self { - println!("new"); nym_api_urls.shuffle(&mut thread_rng()); NymApiTopologyProvider { @@ -33,7 +32,6 @@ impl NymApiTopologyProvider { } fn use_next_nym_api(&mut self) { - println!("using next nym api"); if self.nym_api_urls.len() == 1 { warn!("There's only a single nym API available - it won't be possible to use a different one"); return; @@ -63,7 +61,6 @@ impl NymApiTopologyProvider { } async fn get_current_compatible_topology(&mut self) -> Option { - println!("getting current compatible topology"); let mixnodes = match self.validator_client.get_cached_active_mixnodes().await { Err(err) => { error!("failed to get network mixnodes - {err}"); @@ -112,7 +109,6 @@ impl NymApiTopologyProvider { #[async_trait] impl TopologyProvider for NymApiTopologyProvider { async fn get_new_topology(&mut self) -> Option { - println!("get_new_topology"); self.get_current_compatible_topology().await } } diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index caa6318a3a..18e5325739 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -81,10 +81,7 @@ impl fmt::Debug for GatewayClient { .field("shared_key", &self.shared_key) // .field("connection", &self.connection) .field("packet_router", &self.packet_router) - .field( - "response_timeout_duration", - &self.response_timeout_duration, - ) + .field("response_timeout_duration", &self.response_timeout_duration) // .field("bandwidth_controller", &self.bandwidth_controller) .field( "should_reconnect_on_failure", diff --git a/gateway/src/node/client_handling/active_clients.rs b/gateway/src/node/client_handling/active_clients.rs index 29330f5e19..e42f383e67 100644 --- a/gateway/src/node/client_handling/active_clients.rs +++ b/gateway/src/node/client_handling/active_clients.rs @@ -1,15 +1,16 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use crate::node::client_handling::websocket::message_receiver::MixMessageSender; use dashmap::DashMap; -use nym_sphinx::DestinationAddressBytes; use futures::channel::{mpsc, oneshot}; +use nym_sphinx::DestinationAddressBytes; use std::sync::Arc; +use super::websocket::message_receiver::{IsActiveRequestSender, MixMessageSender}; + #[derive(Clone)] pub(crate) struct ActiveClientsStore( - Arc>)>>, + Arc>, ); impl ActiveClientsStore { @@ -24,13 +25,16 @@ impl ActiveClientsStore { /// # Arguments /// /// * `client`: address of the client for which to obtain the handle. - pub(crate) fn get(&self, client: DestinationAddressBytes) -> Option { + pub(crate) fn get( + &self, + client: DestinationAddressBytes, + ) -> Option<(MixMessageSender, IsActiveRequestSender)> { let entry = self.0.get(&client)?; - let handle = &entry.value().0; + let handle = entry.value(); // if the entry is stale, remove it from the map // if handle.is_valid() { - if !handle.is_closed() { + if !handle.0.is_closed() { Some(handle.clone()) } else { // drop the reference to the map to prevent deadlocks @@ -40,54 +44,6 @@ impl ActiveClientsStore { } } - // pub(crate) fn get_is_recently_active(&self, client: DestinationAddressBytes) -> Option { - // let entry = self.0.get(&client)?; - // let is_recently_active = &entry.value().1; - // - // // if the entry is stale, remove it from the map - // // if handle.is_valid() { - // if !entry.value().0.is_closed() { - // Some(is_recently_active.clone()) - // } else { - // // drop the reference to the map to prevent deadlocks - // drop(entry); - // self.0.remove(&client); - // None - // } - // } - - pub(crate) fn get_is_active_sender(&self, client: DestinationAddressBytes) -> Option>> { - let entry = self.0.get(&client)?; - let is_active_sender = &entry.value().1; - - // if the entry is stale, remove it from the map - // if handle.is_valid() { - if !entry.value().0.is_closed() { - Some(is_active_sender.clone()) - } else { - // drop the reference to the map to prevent deadlocks - drop(entry); - self.0.remove(&client); - None - } - } - // - // pub(crate) fn register_activity(&self, client: DestinationAddressBytes) { - // if let Some(mut entry) = self.0.get_mut(&client) { - // entry.value_mut().1 = RecentlyActive::Yes; - // } - // } - - // pub(crate) fn reset_activity(&self, client: DestinationAddressBytes) { - // if let Some(mut entry) = self.0.get_mut(&client) { - // if entry.value_mut().1 == RecentlyActive::Yes { - // entry.value_mut().1 = RecentlyActive::YesButReset; - // } else if entry.value_mut().1 == RecentlyActive::YesButReset { - // entry.value_mut().1 = RecentlyActive::No; - // } - // } - // } - /// Indicates particular client has disconnected from the gateway and its handle should get removed. /// /// # Arguments @@ -103,7 +59,12 @@ impl ActiveClientsStore { /// /// * `client`: address of the client for which to insert the handle. /// * `handle`: the sender channel for all mix packets to be pushed back onto the websocket - pub(crate) fn insert(&self, client: DestinationAddressBytes, handle: MixMessageSender, is_active_sender: mpsc::UnboundedSender>) { + pub(crate) fn insert( + &self, + client: DestinationAddressBytes, + handle: MixMessageSender, + is_active_sender: mpsc::UnboundedSender>, + ) { self.0.insert(client, (handle, is_active_sender)); } @@ -111,5 +72,4 @@ impl ActiveClientsStore { pub(crate) fn size(&self) -> usize { self.0.len() } - } diff --git a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs index edeae24d44..ff99f91342 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/authenticated.rs @@ -2,10 +2,11 @@ // SPDX-License-Identifier: Apache-2.0 use crate::node::client_handling::websocket::connection_handler::{ClientDetails, FreshHandler}; -use crate::node::client_handling::websocket::message_receiver::MixMessageReceiver; +use crate::node::client_handling::websocket::message_receiver::{ + IsActiveRequestReceiver, MixMessageReceiver, +}; use crate::node::storage::error::StorageError; use crate::node::storage::Storage; -use futures::channel::{mpsc, oneshot}; use futures::StreamExt; use log::*; use nym_gateway_requests::iv::IVConversionError; @@ -98,7 +99,7 @@ pub(crate) struct AuthenticatedHandler { inner: FreshHandler, client: ClientDetails, mix_receiver: MixMessageReceiver, - is_active_request_receiver: mpsc::UnboundedReceiver>, + is_active_request_receiver: IsActiveRequestReceiver, } // explicitly remove handle from the global store upon being dropped @@ -129,7 +130,7 @@ where fresh: FreshHandler, client: ClientDetails, mix_receiver: MixMessageReceiver, - is_active_request_receiver: mpsc::UnboundedReceiver>, + is_active_request_receiver: IsActiveRequestReceiver, ) -> Self { AuthenticatedHandler { inner: fresh, @@ -139,12 +140,6 @@ where } } - // fn register_activity(&self) { - // self.inner - // .active_clients_store - // .register_activity(self.client.address) - // } - /// Explicitly removes handle from the global store. fn disconnect(self) { self.inner @@ -321,7 +316,6 @@ where /// /// * `bin_msg`: raw message to handle. async fn handle_binary(&self, bin_msg: Vec) -> Message { - println!("handle_binary"); // this function decrypts the request and checks the MAC match BinaryRequest::try_from_encrypted_tagged_bytes(bin_msg, &self.client.shared_keys) { Err(e) => { @@ -346,7 +340,6 @@ where /// /// * `raw_request`: raw message to handle. async fn handle_text(&mut self, raw_request: String) -> Message { - println!("handle_text"); match ClientControlRequest::try_from(raw_request) { Err(e) => RequestHandlingError::InvalidTextRequest(e).into_error_message(), Ok(request) => match request { @@ -363,6 +356,21 @@ where } } + /// Handles pong message received from the client. + /// If the client is still active, the handler that requested the ping will receive a reply. + async fn handle_pong(&mut self, msg: Vec) { + if let Ok(msg) = msg.try_into() { + let msg = u64::from_be_bytes(msg); + debug!("Received pong from client: {}", msg); + if let Some(tx) = self.inner.is_active_ping_pending_replies.remove(&msg) { + debug!("Reporting back"); + if let Err(err) = tx.send(true) { + warn!("Failed to send pong reply back to the requesting handler: {err}"); + } + } + } + } + /// Attempts to handle websocket message received from the connected client. /// /// # Arguments @@ -377,21 +385,7 @@ where Message::Binary(bin_msg) => Some(self.handle_binary(bin_msg).await), Message::Text(text_msg) => Some(self.handle_text(text_msg).await), Message::Pong(msg) => { - // println!("1 Received pong from client: {}", String::from_utf8_lossy(&s)); - // let ss = nym_sphinx::DestinationAddressBytes::try_from_byte_slice(&s).expect("JON"); - // let msg = msg.try_into(); - // let msg = u64::from_be_bytes(msg.try_into().expect("JON")); - if let Ok(msg) = msg.try_into() { - let msg = u64::from_be_bytes(msg); - // WIP(JON): for testing - if true { - println!("Received pong from client: {}", msg); - if let Some(tx) = self.inner.is_active_ping_pending_replies.remove(&msg) { - println!("Reporting back"); - tx.send(true).expect("JON"); - } - } - } + self.handle_pong(msg).await; None } _ => None, diff --git a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs index b064146a97..8061c941b4 100644 --- a/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs +++ b/gateway/src/node/client_handling/websocket/connection_handler/fresh.rs @@ -6,6 +6,7 @@ use crate::node::client_handling::websocket::connection_handler::coconut::Coconu use crate::node::client_handling::websocket::connection_handler::{ AuthenticatedHandler, ClientDetails, InitialAuthResult, SocketStream, }; +use crate::node::client_handling::websocket::message_receiver::IsActiveResultSender; use crate::node::storage::error::StorageError; use crate::node::storage::Storage; use futures::channel::oneshot; @@ -81,7 +82,7 @@ pub(crate) struct FreshHandler { // Occasionally the handler is requested to ping the connected client for confirm that it's // active, such as when a duplicate connection is detected. This hashmap stores the oneshot // senders that are used to return the result of the ping to the handler requesting the ping. - pub(crate) is_active_ping_pending_replies: HashMap>, + pub(crate) is_active_ping_pending_replies: HashMap, } impl FreshHandler @@ -426,14 +427,14 @@ where let encrypted_address = EncryptedAddressBytes::try_from_base58_string(enc_address)?; let iv = IV::try_from_base58_string(iv)?; - if let Some(mut tx) = self.active_clients_store.get_is_active_sender(address) { + if let Some((__, mut is_active_request_tx)) = self.active_clients_store.get(address) { log::warn!("Detected duplicate connection for client: {}", address); // Ask the other connection to ping if they are still active. // Use a oneshot channel to return the result to us let (ping_result_sender, ping_result_receiver) = oneshot::channel::(); log::debug!("Asking other connection to ping the connection client"); - tx.send(ping_result_sender).await.ok(); + is_active_request_tx.send(ping_result_sender).await.ok(); // Wait for the reply match tokio::time::timeout(Duration::from_millis(1000), ping_result_receiver).await { diff --git a/gateway/src/node/client_handling/websocket/message_receiver.rs b/gateway/src/node/client_handling/websocket/message_receiver.rs index 3ff265b2a7..b1759fae99 100644 --- a/gateway/src/node/client_handling/websocket/message_receiver.rs +++ b/gateway/src/node/client_handling/websocket/message_receiver.rs @@ -1,7 +1,13 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use futures::channel::mpsc; +use futures::channel::{mpsc, oneshot}; pub(crate) type MixMessageSender = mpsc::UnboundedSender>>; pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver>>; + +// Channels used for one handler to requester another handler to check that the client is still +// active. The result is then passed back to the requesting handler in the oneshot channel. +pub(crate) type IsActiveRequestSender = mpsc::UnboundedSender; +pub(crate) type IsActiveRequestReceiver = mpsc::UnboundedReceiver; +pub(crate) type IsActiveResultSender = oneshot::Sender; diff --git a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs index dd2bc5d402..afbdb421ee 100644 --- a/gateway/src/node/mixnet_handling/receiver/connection_handler.rs +++ b/gateway/src/node/mixnet_handling/receiver/connection_handler.rs @@ -70,7 +70,7 @@ impl ConnectionHandler { } fn update_clients_store_cache_entry(&mut self, client_address: DestinationAddressBytes) { - if let Some(client_sender) = self.active_clients_store.get(client_address) { + if let Some((client_sender, _)) = self.active_clients_store.get(client_address) { self.clients_store_cache .insert(client_address, client_sender); }