diff --git a/nym-gateway-probe/src/lib.rs b/nym-gateway-probe/src/lib.rs index 2a321136c9..21c9fee778 100644 --- a/nym-gateway-probe/src/lib.rs +++ b/nym-gateway-probe/src/lib.rs @@ -7,8 +7,8 @@ use std::{ time::Duration, }; +use crate::socks5_test::{HttpsConnectivityResult, HttpsConnectivityTest}; use crate::types::Entry; -use crate::types::HttpsConnectivityTest; use anyhow::bail; use base64::{Engine as _, engine::general_purpose}; use bytes::BytesMut; @@ -50,7 +50,7 @@ use url::Url; use crate::{ icmp::{check_for_icmp_beacon_reply, icmp_identifier, send_ping_v4, send_ping_v6}, - types::{Exit, HttpsConnectivityResult, Socks5ProbeResults}, + types::{Exit, Socks5ProbeResults}, }; mod bandwidth_helpers; @@ -59,6 +59,7 @@ mod icmp; pub mod mode; mod netstack; pub mod nodes; +mod socks5_test; mod types; use crate::bandwidth_helpers::{acquire_bandwidth, import_bandwidth}; @@ -129,7 +130,10 @@ impl CredentialArgs { #[derive(Args)] pub struct Socks5Args { - #[arg(long, default_value_t = 45)] + #[arg(long, value_delimiter = ',')] + socks5_json_rpc_url_list: Vec, + + #[arg(long, default_value_t = 30)] mixnet_client_timeout_sec: u64, #[arg(long, default_value_t = 10)] @@ -327,28 +331,13 @@ impl Probe { .exit_gateway_nr(&exit_gateway)? .to_testable_node()?; - let socks5_outcome = { - if let Some(ref nr_details) = node_info.network_requester_details { - match do_socks5_connectivity_test( - &nr_details.address, - network_details, - Some(&directory), - self.socks5_args.mixnet_client_timeout_sec, - self.socks5_args.test_count, - ) - .await - { - Ok(results) => Some(results), - Err(e) => { - error!("SOCKS5 test failed: {}", e); - None - } - } - } else { - info!("No NR available, skipping SOCKS5 tests"); - None - } - }; + let socks5_outcome = self + .test_socks5_if_possible( + network_details, + &node_info.network_requester_details, + &directory, + ) + .await; let probe_result = ProbeResult { node: exit_gateway.to_base58_string(), @@ -375,6 +364,7 @@ impl Probe { only_lp_registration: bool, test_lp_wg: bool, min_mixnet_performance: Option, + network_details: NymNetworkDetails, ) -> anyhow::Result { let tickets_materials = self.credentials_args.decode_attached_ticket_materials()?; @@ -386,7 +376,7 @@ impl Probe { // Connect to the mixnet via the entry gateway let disconnected_mixnet_client = MixnetClientBuilder::new_with_storage(storage.clone()) .request_gateway(mixnet_entry_gateway_id.to_string()) - .network_details(NymNetworkDetails::new_from_env()) + .network_details(network_details.clone()) .debug_config(mixnet_debug_config( min_mixnet_performance, ignore_egress_epoch_role, @@ -417,6 +407,7 @@ impl Probe { test_mode, only_wireguard, false, // Not using mock ecash in regular probe mode + network_details, ) .await } @@ -434,6 +425,7 @@ impl Probe { test_lp_wg: bool, min_mixnet_performance: Option, use_mock_ecash: bool, + network_details: NymNetworkDetails, ) -> anyhow::Result { // Localnet mode - identity + LP address from CLI, no HTTP query // This path is used when --entry-gateway-identity is specified @@ -474,6 +466,7 @@ impl Probe { test_mode, only_wireguard, use_mock_ecash, + network_details, ) .await; } @@ -521,6 +514,7 @@ impl Probe { test_mode, only_wireguard, use_mock_ecash, + network_details, ) .await; } @@ -553,7 +547,7 @@ impl Probe { // and keeps its bandwidth between probe runs let disconnected_mixnet_client = MixnetClientBuilder::new_with_storage(storage.clone()) .request_gateway(mixnet_entry_gateway_id.to_string()) - .network_details(NymNetworkDetails::new_from_env()) + .network_details(network_details.clone()) .debug_config(mixnet_debug_config( min_mixnet_performance, ignore_egress_epoch_role, @@ -612,6 +606,7 @@ impl Probe { test_mode, only_wireguard, use_mock_ecash, + network_details, ) .await } @@ -699,6 +694,35 @@ impl Probe { }) } + async fn test_socks5_if_possible( + &self, + network_details: NymNetworkDetails, + network_requester_details: &Option, + directory: &NymApiDirectory, + ) -> Option { + if let Some(nr_details) = network_requester_details { + match do_socks5_connectivity_test( + &nr_details.address, + network_details, + directory, + self.socks5_args.socks5_json_rpc_url_list.clone(), + self.socks5_args.mixnet_client_timeout_sec, + self.socks5_args.test_count, + ) + .await + { + Ok(results) => Some(results), + Err(e) => { + error!("SOCKS5 test failed: {}", e); + None + } + } + } else { + info!("No NR available, skipping SOCKS5 tests"); + None + } + } + pub async fn lookup_gateway( &self, directory: &Option, @@ -771,11 +795,15 @@ impl Probe { test_mode: TestMode, only_wireguard: bool, use_mock_ecash: bool, + network_details: NymNetworkDetails, ) -> anyhow::Result where T: MixnetClientStorage + Clone + 'static, ::StorageError: Send + Sync, { + let Some(directory) = directory else { + bail!("You need to provide NYM API through environment") + }; // test_mode replaces the old only_lp_registration and test_lp_wg flags. // only_wireguard is kept separate as it controls ping behavior within Mixnet mode. let mut rng = rand::thread_rng(); @@ -919,8 +947,6 @@ impl Probe { // The tested node is the exit let exit_gateway = node_info.clone(); - let directory = directory - .ok_or_else(|| anyhow::anyhow!("Directory is required for LP-WG test mode"))?; let entry_gateway_node = directory.entry_gateway(&mixnet_entry_gateway_id)?; let entry_gateway = entry_gateway_node.to_testable_node()?; @@ -962,9 +988,8 @@ impl Probe { Arc::new(KeyPair::new(&mut rng)), ip_address, ); - let config = nym_validator_client::nyxd::Config::try_from_nym_network_details( - &NymNetworkDetails::new_from_env(), - )?; + let config = + nym_validator_client::nyxd::Config::try_from_nym_network_details(&network_details)?; let client = nym_validator_client::nyxd::NyxdClient::connect(config, nyxd_url.as_str())?; let bw_controller = nym_bandwidth_controller::BandwidthController::new( @@ -1037,28 +1062,13 @@ impl Probe { }; // test failure doesn't stop further tests - let socks5_outcome = { - if let Some(ref nr_details) = node_info.network_requester_details { - match do_socks5_connectivity_test( - &nr_details.address, - NymNetworkDetails::new_from_env(), - directory.as_deref(), - self.socks5_args.mixnet_client_timeout_sec, - self.socks5_args.test_count, - ) - .await - { - Ok(results) => Some(results), - Err(e) => { - error!("SOCKS5 test failed: {}", e); - None - } - } - } else { - info!("No NR available, skipping SOCKS5 tests"); - None - } - }; + let socks5_outcome = self + .test_socks5_if_possible( + network_details, + &node_info.network_requester_details, + directory, + ) + .await; // Disconnect the mixnet client gracefully outcome.map(|mut outcome| { @@ -1657,7 +1667,8 @@ async fn do_ping_exit( async fn do_socks5_connectivity_test( network_requester_address: &str, network_details: NymNetworkDetails, - directory: Option<&NymApiDirectory>, + directory: &NymApiDirectory, + json_rpc_endpoints: Vec, mixnet_client_timeout: u64, test_run_count: u64, ) -> anyhow::Result { @@ -1703,9 +1714,6 @@ async fn do_socks5_connectivity_test( // construction in case GW would get filtered out on topology refresh debug_config.topology.minimum_gateway_performance = 0; - let Some(directory) = directory else { - bail!("You need to provide Nym API directory through environment") - }; // Verify the NR gateway exists in the directory with exit_nr role let nr_gateway_id = nr_recipient.gateway(); if let Err(e) = directory.exit_gateway_nr(&nr_gateway_id) { @@ -1729,7 +1737,7 @@ async fn do_socks5_connectivity_test( // connect to mixnet via SOCKS5 let socks5_client = match socks5_client_builder.connect_to_mixnet_via_socks5().await { Ok(client) => { - info!("Successfully established SOCKS5 proxy connection"); + info!("🌐 Successfully connected to mixnet via SOCKS5 proxy"); info!( "Connected via entry gateway: {}", client.nym_address().gateway().to_base58_string() @@ -1746,9 +1754,23 @@ async fn do_socks5_connectivity_test( }; info!("Waiting for network topology to be ready..."); - tokio::time::sleep(Duration::from_secs(10)).await; + let topology_timeout = Duration::from_secs(60); + if let Err(e) = socks5_client.wait_for_topology(topology_timeout).await { + error!( + "Topology not available after {}s: {}", + topology_timeout.as_secs(), + e + ); + results.https_connectivity = + HttpsConnectivityResult::with_error(format!("Topology timeout: {}", e)); + socks5_client.disconnect().await; + return Ok(results); + } else { + info!("Network topology is ready") + } - let test = HttpsConnectivityTest::new(test_run_count, mixnet_client_timeout); + let test = + HttpsConnectivityTest::new(test_run_count, mixnet_client_timeout, json_rpc_endpoints); results.https_connectivity = test.run_tests(socks5_client.socks5_url()).await; // cleanup diff --git a/nym-gateway-probe/src/nodes.rs b/nym-gateway-probe/src/nodes.rs index cd0367a441..9f8d72dff3 100644 --- a/nym-gateway-probe/src/nodes.rs +++ b/nym-gateway-probe/src/nodes.rs @@ -385,7 +385,7 @@ impl NymApiDirectory { .iter() .filter(|(_, n)| n.described.description.ip_packet_router.is_some()) .choose(&mut rand::thread_rng()) - .ok_or(anyhow!("no gateways running IPR available")) + .context("no gateways running IPR available") .map(|(id, _)| *id) } @@ -395,7 +395,7 @@ impl NymApiDirectory { .iter() .filter(|(_, n)| n.described.description.ip_packet_router.is_some()) .choose(&mut rand::thread_rng()) - .ok_or(anyhow!("no gateways running NR available")) + .context("no gateways running NR available") .map(|(id, _)| *id) } @@ -405,7 +405,7 @@ impl NymApiDirectory { .iter() .filter(|(_, n)| n.described.description.declared_role.entry) .choose(&mut rand::thread_rng()) - .ok_or(anyhow!("no entry gateways available")) + .context("no entry gateways available") .map(|(id, _)| *id) } diff --git a/nym-gateway-probe/src/run.rs b/nym-gateway-probe/src/run.rs index 25aebbaa91..1bd7469994 100644 --- a/nym-gateway-probe/src/run.rs +++ b/nym-gateway-probe/src/run.rs @@ -333,6 +333,7 @@ pub(crate) async fn run() -> anyhow::Result { test_lp_wg, args.min_gateway_mixnet_performance, *use_mock_ecash, + network, )) .await } @@ -345,6 +346,7 @@ pub(crate) async fn run() -> anyhow::Result { only_lp_registration, test_lp_wg, args.min_gateway_mixnet_performance, + network, )) .await } @@ -487,6 +489,7 @@ pub(crate) async fn run() -> anyhow::Result { test_lp_wg, args.min_gateway_mixnet_performance, use_mock_ecash, + network, )) .await } @@ -499,6 +502,7 @@ pub(crate) async fn run() -> anyhow::Result { only_lp_registration, test_lp_wg, args.min_gateway_mixnet_performance, + network, )) .await } diff --git a/nym-gateway-probe/src/socks5_test.rs b/nym-gateway-probe/src/socks5_test.rs new file mode 100644 index 0000000000..47c04edd3c --- /dev/null +++ b/nym-gateway-probe/src/socks5_test.rs @@ -0,0 +1,318 @@ +use rand::Rng; +use serde::{Deserialize, Serialize}; +use std::time::Duration; +use tracing::{debug, error, info, warn}; + +pub struct HttpsConnectivityTest { + test_count: u64, + mixnet_client_timeout: Duration, + json_rpc_test_endpoints: Vec, +} + +impl HttpsConnectivityTest { + pub fn new( + test_count: u64, + mixnet_client_timeout: u64, + json_rpc_test_endpoints: Vec, + ) -> Self { + Self { + test_count: std::cmp::max(test_count, 1), + mixnet_client_timeout: Duration::from_secs(mixnet_client_timeout), + json_rpc_test_endpoints, + } + } + + pub async fn run_tests(self, socks5_url: String) -> HttpsConnectivityResult { + let mut result = HttpsConnectivityResult::default(); + + let proxy = match reqwest::Proxy::all(socks5_url) { + Ok(p) => p, + Err(e) => { + return HttpsConnectivityResult::with_error( + format!("Failed to create proxy: {e}",), + ); + } + }; + + let client = match reqwest::Client::builder() + .proxy(proxy) + .timeout(self.mixnet_client_timeout) + .build() + { + Ok(c) => c, + Err(e) => { + return HttpsConnectivityResult::with_error(format!( + "Failed to build HTTP client: {e}", + )); + } + }; + + let mut successful_runs = 0u64; + for i in 1..self.test_count + 1 { + info!("Running test {}/{}", i, self.test_count); + let interim_res = self.perform_https_request(&client).await; + if interim_res.https_success + && let Some(latency_ms) = interim_res.https_latency_ms + { + successful_runs += 1; + result.https_latency_ms = Some( + result + .https_latency_ms + .map_or(latency_ms, |existing| existing + latency_ms), + ); + result.https_success = true; + result.https_status_code = interim_res.https_status_code; + info!("{}/{} latency: {}ms", i, self.test_count, latency_ms); + } else if let Some(new_error) = interim_res.error { + result.error = Some(result.error.map_or(new_error.clone(), |existing| { + format!("{},{}", existing, new_error) + })) + } + + // too many failed runs: return early + let unsuccessful_runs = i - successful_runs; + if successful_runs < 2 && unsuccessful_runs > 2 { + // if < 2 runs, we don't have to calculate average before returning + warn!("Too many failed runs: returning early..."); + return result; + } + } + result.https_latency_ms = result + .https_latency_ms + .map(|latency| latency / successful_runs); + info!( + "AVG latency over {} runs (in ms): {:?}", + successful_runs, result.https_latency_ms + ); + + result + } + + async fn perform_https_request(&self, client: &reqwest::Client) -> HttpsConnectivityResult { + use tokio::time::Instant; + + // TODO dz instead of initializing a mutable default, then mutating fields, use constructors for outcome + let start = Instant::now(); + let mut error_msg = String::new(); + + // TODO dz utilize others as fallback + // let endpoint = self.json_rpc_test_endpoints.first().unwrap(); + for endpoint in self.json_rpc_test_endpoints.iter() { + info!( + "Testing against {} with timeout {}s", + endpoint, + self.mixnet_client_timeout.as_secs() + ); + match client + .post(endpoint) + .timeout(self.mixnet_client_timeout) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .json(&JsonRpcRequestBody::eth_chainid()) + .send() + .await + .and_then(reqwest::Response::error_for_status) + { + Ok(response) => { + let elapsed = start.elapsed(); + let status = response.status(); + + if status.is_success() { + // Deserialize body into JsonRpcResponse + match response.json::().await { + Ok(JsonRpcResponse::Ok { .. }) => { + let res = HttpsConnectivityResult::success( + status.as_u16(), + elapsed.as_millis() as u64, + endpoint.to_string(), + ); + debug!( + "HTTPS test completed: status={}, latency={}ms", + status.as_u16(), + elapsed.as_millis() + ); + return res; + } + Ok(JsonRpcResponse::Err { error, .. }) => { + warn!("JSON-RPC error: {} (code: {})", error.message, error.code); + error_msg = format!("JSON-RPC error: {}", error.message); + } + Err(e) => { + error!("Failed to parse JSON-RPC response: {}", e); + error_msg = format!("Failed to parse JSON-RPC response: {e}"); + } + } + } else { + error_msg = format!("HTTP error status: {}", status.as_u16()); + } + } + Err(e) => { + error!("HTTPS request failed: {}", e); + + error_msg = format!("HTTPS request failed: {}", e); + } + } + } + + HttpsConnectivityResult::with_error(error_msg) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct HttpsConnectivityResult { + /// successfully completed HTTPS request + https_success: bool, + + /// HTTPS status code received + https_status_code: Option, + + /// average HTTPS request latency in milliseconds + https_latency_ms: Option, + + /// among multiple endpoints available, list the one actually used + endpoint_used: Option, + + /// error message(s) (if any) + error: Option, +} + +impl HttpsConnectivityResult { + pub fn with_error(error: impl Into) -> Self { + Self { + https_success: false, + https_status_code: None, + https_latency_ms: None, + endpoint_used: None, + error: Some(error.into()), + } + } + + pub fn success(status_code: u16, latency: u64, endpoint_used: String) -> Self { + Self { + https_success: true, + https_status_code: Some(status_code), + https_latency_ms: Some(latency), + endpoint_used: Some(endpoint_used), + error: None, + } + } +} + +/// https://www.jsonrpc.org/specification +#[derive(Serialize)] +struct JsonRpcRequestBody { + // A String specifying the version of the JSON-RPC protocol. MUST be exactly "2.0". + jsonrpc: String, + method: String, + // A Structured value that holds the parameter values to be used during the invocation of the method. This member MAY be omitted. + params: serde_json::Value, + // The Server MUST reply with the same value in the Response object if included. + // This member is used to correlate the context between the two objects. + id: i64, +} + +impl JsonRpcRequestBody { + /// Very simple endpoint that requires no dynamic input + /// + /// https://ethereum.org/developers/docs/apis/json-rpc/#eth_chainId + pub fn eth_chainid() -> Self { + Self { + jsonrpc: String::from("2.0"), + method: String::from("eth_chainId"), + params: serde_json::json!([]), + id: rand::thread_rng().r#gen(), + } + } + + /// Create an eth_getBlockByNumber request with invalid params for testing error responses + #[cfg(test)] + pub fn eth_get_block_by_number_invalid() -> Self { + Self { + jsonrpc: String::from("2.0"), + method: String::from("eth_getBlockByNumber"), + // Invalid params: should be [blockNumber, boolean] but we pass garbage + params: serde_json::json!(["invalid_block_number"]), + id: rand::thread_rng().r#gen(), + } + } +} + +// dead code: we need these fields for deserialization, even if we don't read them explicitly +#[allow(dead_code)] +#[derive(Deserialize)] +#[serde(untagged)] +enum JsonRpcResponse { + Ok { + jsonrpc: String, + // have to use opaque Value because spec say this might be string, number or null (we don't care either way) + id: serde_json::Value, + // we don't really care for the exact result, just whether the response is OK or error + result: serde_json::Value, + }, + Err { + jsonrpc: String, + // have to use opaque Value because spec say this might be string, number or null (we don't care either way) + id: serde_json::Value, + error: JsonRpcError, + }, +} + +// dead code: we need these fields for deserialization, even if we don't read them explicitly +#[allow(dead_code)] +#[derive(Debug, Deserialize)] +struct JsonRpcError { + pub code: i64, + pub message: String, + #[serde(default)] + pub data: Option, +} + +#[cfg(test)] +mod test { + use super::*; + + const JSON_RPC_ENDPOINT: &str = "https://cloudflare-eth.com"; + + #[tokio::test] + async fn test_eth_chainid_returns_ok_response() { + let client = reqwest::Client::new(); + let response = client + .post(JSON_RPC_ENDPOINT) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .json(&JsonRpcRequestBody::eth_chainid()) + .send() + .await + .expect("Failed to send request"); + + assert!(response.status().is_success()); + + let json_response: JsonRpcResponse = + response.json().await.expect("Failed to parse response"); + + assert!( + matches!(json_response, JsonRpcResponse::Ok { .. }), + "Expected Ok variant for eth_chainId" + ); + } + + #[tokio::test] + async fn test_eth_get_block_by_number_invalid_returns_error_response() { + let client = reqwest::Client::new(); + let response = client + .post(JSON_RPC_ENDPOINT) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .json(&JsonRpcRequestBody::eth_get_block_by_number_invalid()) + .send() + .await + .expect("Failed to send request"); + + assert!(response.status().is_success()); // HTTP 200 but JSON-RPC error + + let json_response: JsonRpcResponse = + response.json().await.expect("Failed to parse response"); + + assert!( + matches!(json_response, JsonRpcResponse::Err { .. }), + "Expected Err variant for invalid params" + ); + } +} diff --git a/nym-gateway-probe/src/types.rs b/nym-gateway-probe/src/types.rs index 9710d8a0d0..f3c3ab28c6 100644 --- a/nym-gateway-probe/src/types.rs +++ b/nym-gateway-probe/src/types.rs @@ -1,8 +1,7 @@ -use std::time::Duration; - use nym_connection_monitor::ConnectionStatusEvent; use serde::{Deserialize, Serialize}; -use tracing::{debug, info, warn}; + +use crate::socks5_test::HttpsConnectivityResult; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ProbeResult { @@ -145,160 +144,6 @@ pub struct Socks5ProbeResults { pub https_connectivity: HttpsConnectivityResult, } -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct HttpsConnectivityResult { - /// successfully completed HTTPS request - https_success: bool, - - /// HTTPS status code received - https_status_code: Option, - - /// average HTTPS request latency in milliseconds - https_latency_ms: Option, - - /// error message(s) (if any) - error: Option, -} - -impl HttpsConnectivityResult { - pub fn with_error(error: impl Into) -> Self { - Self { - https_success: false, - https_status_code: None, - https_latency_ms: None, - error: Some(error.into()), - } - } -} - -pub struct HttpsConnectivityTest { - test_count: u64, - mixnet_client_timeout: Duration, -} - -/// endpoint to test against -/// https://www.quicknode.com/docs/ethereum/web3_clientVersion -const TARGET_URL: &str = "https://docs-demo.quiknode.pro"; -const POST_BODY: &str = r#"{"jsonrpc":"2.0","method":"web3_clientVersion","params":[],"id":1}"#; - -impl HttpsConnectivityTest { - pub fn new(test_count: u64, mixnet_client_timeout: u64) -> Self { - Self { - test_count: std::cmp::max(test_count, 1), - mixnet_client_timeout: Duration::from_secs(mixnet_client_timeout), - } - } - - pub async fn run_tests(self, socks5_url: String) -> HttpsConnectivityResult { - let mut result = HttpsConnectivityResult::default(); - - let proxy = match reqwest::Proxy::all(socks5_url) { - Ok(p) => p, - Err(e) => { - result.error = Some(format!("Failed to create proxy: {}", e)); - return result; - } - }; - - let client = match reqwest::Client::builder() - .proxy(proxy) - .timeout(self.mixnet_client_timeout) - .build() - { - Ok(c) => c, - Err(e) => { - result.error = Some(format!("Failed to build HTTP client: {}", e)); - return result; - } - }; - - let mut successful_runs = 0u64; - for i in 1..self.test_count + 1 { - info!("Running test {}/{}", i, self.test_count); - let interim_res = self.perform_https_request(&client).await; - if interim_res.https_success - && let Some(latency_ms) = interim_res.https_latency_ms - { - successful_runs += 1; - result.https_latency_ms = Some( - result - .https_latency_ms - .map_or(latency_ms, |existing| existing + latency_ms), - ); - result.https_success = true; - result.https_status_code = interim_res.https_status_code; - info!("{}/{} latency: {}ms", i, self.test_count, latency_ms); - } else if let Some(new_error) = interim_res.error { - result.error = Some(result.error.map_or(new_error.clone(), |existing| { - format!("{},{}", existing, new_error) - })) - } - - // too many failed runs: return early - if successful_runs < 2 && i - successful_runs > 2 { - // if < 2 runs, we don't have to calculate average before returning - return result; - } - } - result.https_latency_ms = result - .https_latency_ms - .map(|latency| latency / successful_runs); - info!( - "AVG latency over {} runs (in ms): {:?}", - successful_runs, result.https_latency_ms - ); - - result - } - - async fn perform_https_request(&self, client: &reqwest::Client) -> HttpsConnectivityResult { - use tokio::time::Instant; - - let mut result = HttpsConnectivityResult::default(); - let start = Instant::now(); - match tokio::time::timeout( - self.mixnet_client_timeout, - client - .post(TARGET_URL) - .header(reqwest::header::CONTENT_TYPE, "application/json") - .body(POST_BODY) - .send(), - ) - .await - { - Ok(Ok(response)) => { - let elapsed = start.elapsed(); - let status = response.status(); - result.https_success = status.is_success(); - result.https_status_code = Some(status.as_u16()); - result.https_latency_ms = Some(elapsed.as_millis() as u64); - debug!( - "HTTPS test completed: status={}, latency={}ms", - status.as_u16(), - elapsed.as_millis() - ); - } - Ok(Err(e)) => { - warn!("HTTPS request failed: {}", e); - if result.error.is_none() { - result.error = Some(format!("HTTPS request failed: {}", e)); - } - } - Err(_) => { - warn!( - "HTTPS request timed out after {}s", - self.mixnet_client_timeout.as_secs() - ); - if result.error.is_none() { - result.error = Some("HTTPS request timed out".to_string()); - } - } - } - - result - } -} - #[derive(Debug, Clone, Default)] pub struct IpPingReplies { pub ipr_tun_ip_v4: bool, diff --git a/sdk/rust/nym-sdk/src/mixnet/client.rs b/sdk/rust/nym-sdk/src/mixnet/client.rs index 7fe176b05b..c9b821acab 100644 --- a/sdk/rust/nym-sdk/src/mixnet/client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/client.rs @@ -8,7 +8,7 @@ use crate::mixnet::{CredentialStorage, MixnetClient, Recipient}; use crate::GatewayTransceiver; use crate::NymNetworkDetails; use crate::{Error, Result}; -use log::{debug, info, warn}; +use log::{debug, warn}; use nym_client_core::client::base_client::storage::gateways_storage::GatewayRegistration; use nym_client_core::client::base_client::storage::helpers::{ get_active_gateway_identity, get_all_registered_identities, has_gateway_details, @@ -601,9 +601,13 @@ where ); let available_gateways = self.available_gateways().await?; - info!("Listing all available gateways in topology:"); + debug!("Listing all available gateways in topology:"); for node in available_gateways.iter() { - info!("{}", node.identity_key.to_base58_string()); + debug!( + "node_id={}, identity_key={}", + node.node_id, + node.identity_key.to_base58_string() + ); } Ok(GatewaySetup::New { diff --git a/sdk/rust/nym-sdk/src/mixnet/socks5_client.rs b/sdk/rust/nym-sdk/src/mixnet/socks5_client.rs index 0a9ee37c90..1875924616 100644 --- a/sdk/rust/nym-sdk/src/mixnet/socks5_client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/socks5_client.rs @@ -1,10 +1,13 @@ +use std::time::Duration; + use nym_client_core::client::base_client::ClientState; use nym_socks5_client_core::config::Socks5; use nym_sphinx::addressing::clients::Recipient; use nym_task::connections::LaneQueueLengths; use nym_task::ShutdownTracker; +use tokio::sync::RwLockReadGuard; -use nym_topology::NymTopology; +use nym_topology::{NymRouteProvider, NymTopology, NymTopologyError}; use crate::mixnet::client::MixnetClientBuilder; use crate::Result; @@ -85,4 +88,28 @@ impl Socks5MixnetClient { pub async fn disconnect(self) { self.task_handle.shutdown().await; } + + /// Gets the current route provider if topology is available. + /// Returns `None` if topology is empty/not yet fetched. + async fn read_current_route_provider(&self) -> Option> { + self.client_state + .topology_accessor + .current_route_provider() + .await + } + + /// Wait for topology to become available, with a timeout. + /// Returns `Ok(())` when topology is ready, or `Err` if timeout is reached. + pub async fn wait_for_topology(&self, timeout: Duration) -> Result<(), NymTopologyError> { + let deadline = tokio::time::Instant::now() + timeout; + loop { + if self.read_current_route_provider().await.is_some() { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(NymTopologyError::EmptyNetworkTopology); + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + } }