diff --git a/Cargo.lock b/Cargo.lock index 949686c974..fb3c22f5e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4731,7 +4731,7 @@ checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3" [[package]] name = "nym-api" -version = "1.1.63" +version = "1.1.64" dependencies = [ "anyhow", "async-trait", diff --git a/common/client-core/src/client/mix_traffic/transceiver.rs b/common/client-core/src/client/mix_traffic/transceiver.rs index e4937fa007..729cbea23a 100644 --- a/common/client-core/src/client/mix_traffic/transceiver.rs +++ b/common/client-core/src/client/mix_traffic/transceiver.rs @@ -36,7 +36,7 @@ pub trait GatewayTransceiver: GatewaySender + GatewayReceiver { &mut self, message: ClientRequest, ) -> Result<(), GatewayClientError>; - + /// Check if the websocket connection to the gateway is alive fn is_connection_alive(&self) -> bool; } @@ -93,7 +93,7 @@ impl GatewayTransceiver for Box { log::debug!("Sent client request: {:?}", message); Ok(()) } - + #[inline] fn is_connection_alive(&self) -> bool { (**self).is_connection_alive() @@ -155,7 +155,7 @@ where ) -> Result<(), GatewayClientError> { self.gateway_client.send_client_request(message).await } - + fn is_connection_alive(&self) -> bool { self.gateway_client.is_connection_alive() } @@ -246,7 +246,7 @@ mod nonwasm_sealed { ) -> Result<(), GatewayClientError> { Ok(()) } - + fn is_connection_alive(&self) -> bool { // LocalGateway is always "connected" since it's in-process true @@ -333,7 +333,7 @@ impl GatewayTransceiver for MockGateway { ) -> Result<(), GatewayClientError> { Ok(()) } - + fn is_connection_alive(&self) -> bool { // MockGateway is always "connected" for testing purposes true diff --git a/common/client-libs/gateway-client/src/client/mod.rs b/common/client-libs/gateway-client/src/client/mod.rs index 12d16c63a8..b58ace7471 100644 --- a/common/client-libs/gateway-client/src/client/mod.rs +++ b/common/client-libs/gateway-client/src/client/mod.rs @@ -164,18 +164,18 @@ impl GatewayClient { pub fn remaining_bandwidth(&self) -> i64 { self.bandwidth.remaining() } - + pub fn is_connection_established(&self) -> bool { self.connection.is_established() } - + /// Check if the websocket connection is actually alive at the socket level pub fn is_connection_alive(&self) -> bool { // First check if we have an established connection if !self.connection.is_established() { return false; } - + // Get the file descriptor and check if the socket is alive match self.ws_fd() { Some(fd) => socket_is_alive(fd), @@ -1148,43 +1148,43 @@ impl GatewayClient { } /// Check if a socket file descriptor is alive and responsive -/// +/// /// This function performs socket-level checks to determine if the connection is actually alive. /// It's cross-platform compatible and works on both Unix and non-Unix systems. fn socket_is_alive(fd: RawFd) -> bool { #[cfg(unix)] { - use std::os::unix::io::FromRawFd; - use std::net::TcpStream; use std::io::ErrorKind; - + use std::net::TcpStream; + use std::os::unix::io::FromRawFd; + unsafe { // Create a TcpStream from the raw fd to perform socket operations let stream = TcpStream::from_raw_fd(fd); - + // Try to peek at the socket to see if it's still connected // We peek with a zero-length buffer to avoid consuming data let mut buf = [0u8; 0]; let result = match stream.peek(&mut buf) { - Ok(_) => true, // Socket is alive and readable + Ok(_) => true, // Socket is alive and readable Err(e) => match e.kind() { - ErrorKind::WouldBlock => true, // Socket is alive but no data available - ErrorKind::ConnectionReset - | ErrorKind::ConnectionAborted - | ErrorKind::BrokenPipe - | ErrorKind::NotConnected => false, // Socket is clearly dead - _ => true, // Other errors might be temporary, assume alive - } + ErrorKind::WouldBlock => true, // Socket is alive but no data available + ErrorKind::ConnectionReset + | ErrorKind::ConnectionAborted + | ErrorKind::BrokenPipe + | ErrorKind::NotConnected => false, // Socket is clearly dead + _ => true, // Other errors might be temporary, assume alive + }, }; - + // Prevent the TcpStream from closing the fd when it's dropped // since we don't own the fd std::mem::forget(stream); - + result } } - + #[cfg(not(unix))] { // On non-Unix systems, we can't easily check socket state diff --git a/common/client-libs/gateway-client/src/error.rs b/common/client-libs/gateway-client/src/error.rs index 8d3c8be79e..cf6be64ead 100644 --- a/common/client-libs/gateway-client/src/error.rs +++ b/common/client-libs/gateway-client/src/error.rs @@ -49,7 +49,6 @@ pub enum GatewayClientError { #[error("Invalid URL: {0}")] InvalidUrl(String), - #[error("No shared key was provided or obtained")] NoSharedKeyAvailable, diff --git a/nym-api/Cargo.toml b/nym-api/Cargo.toml index 1906ef30ba..673a3bde59 100644 --- a/nym-api/Cargo.toml +++ b/nym-api/Cargo.toml @@ -4,7 +4,7 @@ [package] name = "nym-api" license = "GPL-3.0" -version = "1.1.63" +version = "1.1.64" authors.workspace = true edition = "2021" rust-version.workspace = true diff --git a/nym-api/src/epoch_operations/error.rs b/nym-api/src/epoch_operations/error.rs index 5d9dccc920..1ce419d5a6 100644 --- a/nym-api/src/epoch_operations/error.rs +++ b/nym-api/src/epoch_operations/error.rs @@ -60,9 +60,7 @@ pub enum RewardingError { RewardingParamsRetrievalFailure, #[error("Database operation failed: {source}")] - DatabaseError { - source: anyhow::Error, - }, + DatabaseError { source: anyhow::Error }, #[error("{0}")] GenericError(#[from] anyhow::Error), diff --git a/nym-api/src/epoch_operations/mod.rs b/nym-api/src/epoch_operations/mod.rs index c62142436b..dad510a4fe 100644 --- a/nym-api/src/epoch_operations/mod.rs +++ b/nym-api/src/epoch_operations/mod.rs @@ -141,21 +141,34 @@ impl EpochAdvancer { if epoch_status.being_advanced_by.as_str() != address.as_ref() { // another nym-api is already handling - error!("another nym-api ({}) is already advancing the epoch... but we shouldn't have other nym-apis yet!", epoch_status.being_advanced_by); - return Ok(()); + // In simulation mode, we want to proceed anyway since we're not actually advancing the epoch + if self.simulation_config.is_some() { + info!("Another nym-api ({}) is advancing the epoch, but we're in simulation mode so proceeding anyway", epoch_status.being_advanced_by); + } else { + error!("another nym-api ({}) is already advancing the epoch... but we shouldn't have other nym-apis yet!", epoch_status.being_advanced_by); + return Ok(()); + } } else { warn!("we seem to have crashed mid-epoch advancement..."); } } else { - let should_continue = self.begin_epoch_transition().await?; - if !should_continue { - return Ok(()); + // In simulation mode, skip trying to begin epoch transition + if self.simulation_config.is_none() { + let should_continue = self.begin_epoch_transition().await?; + if !should_continue { + return Ok(()); + } + } else { + info!("Skipping epoch transition in simulation mode - proceeding to performance calculations"); } } // Run simulation if enabled (before actual rewarding) if let Some(simulation_config) = &self.simulation_config { - info!("Running reward simulation for epoch {}", interval.current_epoch_absolute_id()); + info!( + "Running reward simulation for epoch {}", + interval.current_epoch_absolute_id() + ); let rewarded_set = match self.nyxd_client.get_rewarded_set_nodes().await { Ok(rewarded_set) => rewarded_set, Err(err) => { @@ -174,26 +187,33 @@ impl EpochAdvancer { .await .into_inner() { - let _ = self.run_simulation_if_enabled( - &rewarded_set, - reward_params, - interval.current_epoch_absolute_id(), - simulation_config.clone(), - ).await; + let _ = self + .run_simulation_if_enabled( + &rewarded_set, + reward_params, + interval.current_epoch_absolute_id(), + simulation_config.clone(), + ) + .await; } else { warn!("Could not obtain reward parameters for simulation"); } } - // Reward all the nodes in the still current, soon to be previous rewarded set - info!("Rewarding the current rewarded set..."); - self.reward_current_rewarded_set(rewards, interval).await?; + // Skip actual rewarding operations in simulation mode + if self.simulation_config.is_none() { + // Reward all the nodes in the still current, soon to be previous rewarded set + info!("Rewarding the current rewarded set..."); + self.reward_current_rewarded_set(rewards, interval).await?; - // note: those operations don't really have to be atomic, so it's fine to send them - // as separate transactions - self.reconcile_epoch_events().await?; - self.update_rewarded_set_and_advance_epoch(&nym_nodes) - .await?; + // note: those operations don't really have to be atomic, so it's fine to send them + // as separate transactions + self.reconcile_epoch_events().await?; + self.update_rewarded_set_and_advance_epoch(&nym_nodes) + .await?; + } else { + info!("Skipping actual reward distribution in simulation mode"); + } info!("Purging old data (node statuses and routes) from the storage..."); let cutoff = (epoch_end - 2 * ONE_DAY).unix_timestamp(); diff --git a/nym-api/src/epoch_operations/simulation.rs b/nym-api/src/epoch_operations/simulation.rs index acd5091add..58704be0de 100644 --- a/nym-api/src/epoch_operations/simulation.rs +++ b/nym-api/src/epoch_operations/simulation.rs @@ -2,18 +2,21 @@ // SPDX-License-Identifier: GPL-3.0-only //! Performance methodology comparison system -//! +//! //! This module provides functionality to compare different performance calculation //! methodologies without affecting actual rewards, enabling analysis of: //! - Old method: 24-hour cache-based performance calculation //! - New method: 1-hour route-based performance calculation -//! +//! //! The system focuses on performance metrics and rankings rather than reward amounts, //! as actual rewards are calculated on-chain based on factors not available to the API. use crate::epoch_operations::error::RewardingError; use crate::epoch_operations::helpers::RewardedNodeWithParams; -use crate::storage::models::{SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedPerformanceRanking, SimulatedRouteAnalysis}; +use crate::storage::models::{ + SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedPerformanceRanking, + SimulatedRouteAnalysis, +}; use crate::support::storage::NymApiStorage; use crate::EpochAdvancer; use nym_contracts_common::types::NaiveFloat; @@ -44,7 +47,6 @@ impl Default for SimulationConfig { } } - /// Main simulation coordinator pub struct SimulationCoordinator<'a> { storage: &'a NymApiStorage, @@ -66,14 +68,16 @@ impl<'a> SimulationCoordinator<'a> { let now = OffsetDateTime::now_utc(); let end_timestamp = now.unix_timestamp(); let start_timestamp = end_timestamp - (24 * 3600); // 24 hours ago for baseline - + info!( "Starting simulation for epoch {} with time window {}h", current_epoch_id, self.config.new_method_time_window_hours ); // Create simulation epoch record - let epoch_db_id = self.storage.manager + let epoch_db_id = self + .storage + .manager .create_simulated_reward_epoch( current_epoch_id, "comparison", @@ -86,12 +90,10 @@ impl<'a> SimulationCoordinator<'a> { // Run old method simulation (24h cache-based) if self.config.run_both_methods { - match self.run_old_method_simulation( - rewarded_set, - reward_params, - epoch_db_id, - end_timestamp, - ).await { + match self + .run_old_method_simulation(rewarded_set, reward_params, epoch_db_id, end_timestamp) + .await + { Ok(_) => { info!("Old method simulation completed successfully"); } @@ -103,12 +105,10 @@ impl<'a> SimulationCoordinator<'a> { } // Run new method simulation (1h route-based) - match self.run_new_method_simulation( - rewarded_set, - reward_params, - epoch_db_id, - end_timestamp, - ).await { + match self + .run_new_method_simulation(rewarded_set, reward_params, epoch_db_id, end_timestamp) + .await + { Ok(_) => { info!("New method simulation completed successfully"); } @@ -132,57 +132,62 @@ impl<'a> SimulationCoordinator<'a> { debug!("Running old method simulation (24h cache-based)"); // Get 24h performance data using existing cache-based method - let mixnode_reliabilities = self.storage + let mixnode_reliabilities = self + .storage .get_all_avg_mix_reliability_in_last_24hr(end_timestamp) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - let gateway_reliabilities = self.storage + let gateway_reliabilities = self + .storage .get_all_avg_gateway_reliability_in_last_24hr(end_timestamp) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; // Convert to performance map let mut performance_map = HashMap::new(); - + for mix_reliability in mixnode_reliabilities { performance_map.insert( - mix_reliability.mix_id(), - Performance::from_percentage_value(mix_reliability.value() as u64).unwrap_or_default() + mix_reliability.mix_id(), + Performance::from_percentage_value(mix_reliability.value() as u64) + .unwrap_or_default(), ); } for gateway_reliability in gateway_reliabilities { performance_map.insert( gateway_reliability.node_id(), - Performance::from_percentage_value(gateway_reliability.value() as u64).unwrap_or_default() + Performance::from_percentage_value(gateway_reliability.value() as u64) + .unwrap_or_default(), ); } // Calculate rewards using old method logic - let rewarded_nodes = self.calculate_rewards_for_nodes( - rewarded_set, - reward_params, - &performance_map, - ); + let rewarded_nodes = + self.calculate_rewards_for_nodes(rewarded_set, reward_params, &performance_map); // Convert to simulation data structures - let node_performance = self.convert_to_simulated_performance( - &rewarded_nodes, - rewarded_set, - reward_params, - epoch_db_id, - "old", - None, // Old method doesn't have route sample data - ).await; + let node_performance = self + .convert_to_simulated_performance( + &rewarded_nodes, + rewarded_set, + reward_params, + epoch_db_id, + "old", + None, // Old method doesn't have route sample data + ) + .await; - let performance_comparisons = self.convert_to_performance_comparisons( - &rewarded_nodes, - rewarded_set, - reward_params, - epoch_db_id, - "old", - ).await; + let performance_comparisons = self + .convert_to_performance_comparisons( + &rewarded_nodes, + rewarded_set, + reward_params, + epoch_db_id, + "old", + ) + .await; // Create route analysis for old method let route_analysis = SimulatedRouteAnalysis { @@ -194,29 +199,36 @@ impl<'a> SimulationCoordinator<'a> { failed_routes: 0, average_route_reliability: None, time_window_hours: 24, // Old method uses 24h - analysis_parameters: Some("{\"method\":\"cache_based\",\"data_source\":\"status_cache\"}".to_string()), + analysis_parameters: Some( + "{\"method\":\"cache_based\",\"data_source\":\"status_cache\"}".to_string(), + ), calculated_at: OffsetDateTime::now_utc().unix_timestamp(), }; // Store results in database - self.storage.manager + self.storage + .manager .insert_simulated_node_performance(&node_performance) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - self.storage.manager + self.storage + .manager .insert_simulated_performance_comparisons(&performance_comparisons) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - + // Calculate and store performance rankings - let rankings = self.calculate_performance_rankings(&performance_comparisons, epoch_db_id, "old"); - self.storage.manager + let rankings = + self.calculate_performance_rankings(&performance_comparisons, epoch_db_id, "old"); + self.storage + .manager .insert_simulated_performance_rankings(&rankings) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - self.storage.manager + self.storage + .manager .insert_simulated_route_analysis(&route_analysis) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; @@ -232,13 +244,18 @@ impl<'a> SimulationCoordinator<'a> { epoch_db_id: i64, end_timestamp: i64, ) -> Result<(), RewardingError> { - debug!("Running new method simulation ({}h route-based)", self.config.new_method_time_window_hours); + debug!( + "Running new method simulation ({}h route-based)", + self.config.new_method_time_window_hours + ); let time_window_secs = (self.config.new_method_time_window_hours as i64) * 3600; let start_timestamp = end_timestamp - time_window_secs; // Get route-based performance data using new method - let corrected_reliabilities = self.storage.manager + let corrected_reliabilities = self + .storage + .manager .calculate_corrected_node_reliabilities_for_interval(start_timestamp, end_timestamp) .await .map_err(|e| RewardingError::DatabaseError { source: e })?; @@ -252,10 +269,11 @@ impl<'a> SimulationCoordinator<'a> { let mut reliability_count = 0u32; for node_reliability in &corrected_reliabilities { - let total_samples = node_reliability.pos_samples_in_interval + node_reliability.neg_samples_in_interval; + let total_samples = + node_reliability.pos_samples_in_interval + node_reliability.neg_samples_in_interval; total_routes += total_samples; successful_routes += node_reliability.pos_samples_in_interval; - + if total_samples > 0 { reliability_sum += node_reliability.reliability; reliability_count += 1; @@ -263,40 +281,45 @@ impl<'a> SimulationCoordinator<'a> { performance_map.insert( node_reliability.node_id, - Performance::naive_try_from_f64(node_reliability.reliability / 100.0).unwrap_or_default() + Performance::naive_try_from_f64(node_reliability.reliability / 100.0) + .unwrap_or_default(), ); - + // Store sample counts for detailed performance records route_reliability_map.insert( node_reliability.node_id, - (node_reliability.pos_samples_in_interval, node_reliability.neg_samples_in_interval) + ( + node_reliability.pos_samples_in_interval, + node_reliability.neg_samples_in_interval, + ), ); } // Calculate rewards using new method logic - let rewarded_nodes = self.calculate_rewards_for_nodes( - rewarded_set, - reward_params, - &performance_map, - ); + let rewarded_nodes = + self.calculate_rewards_for_nodes(rewarded_set, reward_params, &performance_map); - // Convert to simulation data structures - let node_performance = self.convert_to_simulated_performance( - &rewarded_nodes, - rewarded_set, - reward_params, - epoch_db_id, - "new", - Some(&route_reliability_map), // Pass route sample data for new method - ).await; + // Convert to simulation data structures + let node_performance = self + .convert_to_simulated_performance( + &rewarded_nodes, + rewarded_set, + reward_params, + epoch_db_id, + "new", + Some(&route_reliability_map), // Pass route sample data for new method + ) + .await; - let performance_comparisons = self.convert_to_performance_comparisons( - &rewarded_nodes, - rewarded_set, - reward_params, - epoch_db_id, - "new", - ).await; + let performance_comparisons = self + .convert_to_performance_comparisons( + &rewarded_nodes, + rewarded_set, + reward_params, + epoch_db_id, + "new", + ) + .await; // Create route analysis for new method let route_analysis = SimulatedRouteAnalysis { @@ -321,24 +344,29 @@ impl<'a> SimulationCoordinator<'a> { }; // Store results in database - self.storage.manager + self.storage + .manager .insert_simulated_node_performance(&node_performance) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - self.storage.manager + self.storage + .manager .insert_simulated_performance_comparisons(&performance_comparisons) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - + // Calculate and store performance rankings - let rankings = self.calculate_performance_rankings(&performance_comparisons, epoch_db_id, "new"); - self.storage.manager + let rankings = + self.calculate_performance_rankings(&performance_comparisons, epoch_db_id, "new"); + self.storage + .manager .insert_simulated_performance_rankings(&rankings) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; - self.storage.manager + self.storage + .manager .insert_simulated_route_analysis(&route_analysis) .await .map_err(|e| RewardingError::DatabaseError { source: e.into() })?; @@ -409,19 +437,33 @@ impl<'a> SimulationCoordinator<'a> { } /// Determine node type and work factor from rewarded set position - fn determine_node_info(&self, node_id: NodeId, rewarded_set: &EpochRewardedSet, reward_params: RewardingParams) -> (String, f64) { + fn determine_node_info( + &self, + node_id: NodeId, + rewarded_set: &EpochRewardedSet, + reward_params: RewardingParams, + ) -> (String, f64) { let nodes = &rewarded_set.assignment; - + // Check if node is in active mixnode layers - if nodes.layer1.contains(&node_id) || nodes.layer2.contains(&node_id) || nodes.layer3.contains(&node_id) { - return ("mixnode".to_string(), reward_params.active_node_work().naive_to_f64()); + if nodes.layer1.contains(&node_id) + || nodes.layer2.contains(&node_id) + || nodes.layer3.contains(&node_id) + { + return ( + "mixnode".to_string(), + reward_params.active_node_work().naive_to_f64(), + ); } - + // Check if node is in active gateways if nodes.entry_gateways.contains(&node_id) || nodes.exit_gateways.contains(&node_id) { - return ("gateway".to_string(), reward_params.active_node_work().naive_to_f64()); + return ( + "gateway".to_string(), + reward_params.active_node_work().naive_to_f64(), + ); } - + // Check if node is in standby (could be mixnode or gateway) if nodes.standby.contains(&node_id) { // Note: We cannot determine if standby nodes are mixnodes or gateways from the @@ -430,9 +472,12 @@ impl<'a> SimulationCoordinator<'a> { // 1. All standby nodes receive the same work factor regardless of type // 2. The gateway 3-sample rule can't be applied to standby nodes in either method // For consistency, we label all standby nodes as "mixnode" in the simulation data - return ("mixnode".to_string(), reward_params.standby_node_work().naive_to_f64()); + return ( + "mixnode".to_string(), + reward_params.standby_node_work().naive_to_f64(), + ); } - + // Default case (shouldn't happen) ("unknown".to_string(), 0.0) } @@ -448,40 +493,43 @@ impl<'a> SimulationCoordinator<'a> { route_reliability_map: Option<&HashMap>, // (positive_samples, negative_samples) ) -> Vec { let now = OffsetDateTime::now_utc().unix_timestamp(); - + let mut performance_records = Vec::with_capacity(rewarded_nodes.len()); - + // First collect all node IDs for batch identity key lookups let all_node_ids: Vec = rewarded_nodes.iter().map(|node| node.node_id).collect(); - + // Batch fetch identity keys for both mixnodes and gateways - let mixnode_identities = self.storage + let mixnode_identities = self + .storage .manager .get_mixnode_identity_keys_batch(&all_node_ids) .await .unwrap_or_default(); - let gateway_identities = self.storage + let gateway_identities = self + .storage .manager .get_gateway_identity_keys_batch(&all_node_ids) .await .unwrap_or_default(); for node in rewarded_nodes { - let (node_type, _) = self.determine_node_info(node.node_id, rewarded_set, reward_params); - + let (node_type, _) = + self.determine_node_info(node.node_id, rewarded_set, reward_params); + // Get identity key from our batch results let identity_key = match node_type.as_str() { "mixnode" => mixnode_identities.get(&node.node_id).cloned(), "gateway" => gateway_identities.get(&node.node_id).cloned(), _ => None, }; - + // Extract sample counts from route reliability if available let (positive_samples, negative_samples) = route_reliability_map .and_then(|map| map.get(&node.node_id)) .copied() .unwrap_or((0, 0)); - + performance_records.push(SimulatedNodePerformance { id: 0, // Will be set by database simulated_epoch_id: epoch_db_id, @@ -496,7 +544,7 @@ impl<'a> SimulationCoordinator<'a> { calculated_at: now, }); } - + performance_records } @@ -510,12 +558,13 @@ impl<'a> SimulationCoordinator<'a> { method: &str, ) -> Vec { let now = OffsetDateTime::now_utc().unix_timestamp(); - + let mut performance_records = Vec::with_capacity(rewarded_nodes.len()); - + for node in rewarded_nodes { - let (node_type, _) = self.determine_node_info(node.node_id, rewarded_set, reward_params); - + let (node_type, _) = + self.determine_node_info(node.node_id, rewarded_set, reward_params); + performance_records.push(SimulatedPerformanceComparison { id: 0, // Will be set by database simulated_epoch_id: epoch_db_id, @@ -530,10 +579,10 @@ impl<'a> SimulationCoordinator<'a> { calculated_at: now, }); } - + performance_records } - + /// Calculate performance rankings for a set of performance comparisons fn calculate_performance_rankings( &self, @@ -543,20 +592,22 @@ impl<'a> SimulationCoordinator<'a> { ) -> Vec { let now = OffsetDateTime::now_utc().unix_timestamp(); let mut rankings = Vec::with_capacity(performance_comparisons.len()); - + // Sort by performance score descending - let mut sorted_comparisons: Vec<_> = performance_comparisons.iter() + let mut sorted_comparisons: Vec<_> = performance_comparisons + .iter() .enumerate() .map(|(idx, comp)| (idx, comp.performance_score)) .collect(); - sorted_comparisons.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); - + sorted_comparisons + .sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); + let total_nodes = sorted_comparisons.len() as f64; - + for (rank, (original_idx, _score)) in sorted_comparisons.iter().enumerate() { let comparison = &performance_comparisons[*original_idx]; let percentile = ((total_nodes - rank as f64 - 1.0) / total_nodes) * 100.0; - + rankings.push(SimulatedPerformanceRanking { id: 0, // Will be set by database simulated_epoch_id: epoch_db_id, @@ -567,7 +618,7 @@ impl<'a> SimulationCoordinator<'a> { calculated_at: now, }); } - + rankings } } @@ -582,10 +633,16 @@ impl EpochAdvancer { simulation_config: SimulationConfig, ) -> Result<(), RewardingError> { let coordinator = SimulationCoordinator::new(&self.storage, simulation_config); - - match coordinator.run_simulation(rewarded_set, reward_params, current_epoch_id).await { + + match coordinator + .run_simulation(rewarded_set, reward_params, current_epoch_id) + .await + { Ok(()) => { - info!("Simulation completed successfully for epoch {}", current_epoch_id); + info!( + "Simulation completed successfully for epoch {}", + current_epoch_id + ); Ok(()) } Err(e) => { @@ -595,4 +652,4 @@ impl EpochAdvancer { } } } -} \ No newline at end of file +} diff --git a/nym-api/src/simulation_api/handlers/mod.rs b/nym-api/src/simulation_api/handlers/mod.rs index 2c322077f0..c1ad045230 100644 --- a/nym-api/src/simulation_api/handlers/mod.rs +++ b/nym-api/src/simulation_api/handlers/mod.rs @@ -4,10 +4,10 @@ //! Handlers for simulation API endpoints use crate::simulation_api::models::{ - SimulationApiError, SimulationEpochDetails, SimulationEpochSummary, SimulationEpochsResponse, - SimulationListQuery, NodeComparisonQuery, MethodComparisonResponse, NodeMethodComparison, - ComparisonSummaryStats, RouteAnalysisComparison, NodePerformanceData, PerformanceComparisonData, - RouteAnalysisData, ExportFormat, + ComparisonSummaryStats, ExportFormat, MethodComparisonResponse, NodeComparisonQuery, + NodeMethodComparison, NodePerformanceData, PerformanceComparisonData, RouteAnalysisComparison, + RouteAnalysisData, SimulationApiError, SimulationEpochDetails, SimulationEpochSummary, + SimulationEpochsResponse, SimulationListQuery, }; use crate::storage::models::SimulatedPerformanceRanking; use crate::support::http::state::AppState; @@ -33,7 +33,10 @@ pub(crate) fn simulation_routes() -> Router { .route("/epochs/:epoch_id", get(get_simulation_epoch_details)) .route("/epochs/:epoch_id/comparison", get(compare_methods)) .route("/epochs/:epoch_id/export", get(export_simulation_data)) - .route("/nodes/:node_id/performance", get(get_node_performance_history)) + .route( + "/nodes/:node_id/performance", + get(get_node_performance_history), + ) } #[derive(Deserialize, IntoParams)] @@ -73,48 +76,61 @@ async fn list_simulation_epochs( ) -> AxumResult> { let storage = state.storage(); let output = output.output.unwrap_or_default(); - + // Apply defaults and validation let limit = params.limit.unwrap_or(50).min(1000); let offset = params.offset.unwrap_or(0); - + let epochs = get_simulation_epochs_with_filters(storage, ¶ms, limit, offset) .await .map_err(to_axum_error)?; - + // Enhance epochs with additional metadata using batch operations let epoch_db_ids: Vec = epochs.iter().map(|e| e.id).collect(); let epoch_ids: Vec = epochs.iter().map(|e| e.epoch_id).collect(); - + // Batch fetch node counts and available methods for all epochs let node_counts = storage .manager .count_simulated_node_performance_for_epochs_batch(&epoch_db_ids) .await - .map_err(|e| to_axum_error(SimulationApiError::with_details("Database error", &e.to_string())))?; + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Database error", + &e.to_string(), + )) + })?; let available_methods = storage .manager .get_available_calculation_methods_for_epochs_batch(&epoch_ids) .await - .map_err(|e| to_axum_error(SimulationApiError::with_details("Database error", &e.to_string())))?; - + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Database error", + &e.to_string(), + )) + })?; + let mut enhanced_epochs = Vec::new(); for mut epoch in epochs { // Get metadata from our batch results epoch.nodes_analyzed = node_counts.get(&epoch.id).copied().unwrap_or(0); - epoch.available_methods = available_methods.get(&epoch.epoch_id).cloned().unwrap_or_default(); + epoch.available_methods = available_methods + .get(&epoch.epoch_id) + .cloned() + .unwrap_or_default(); enhanced_epochs.push(epoch); } - + let total_count = count_total_simulation_epochs(storage, ¶ms) .await .map_err(to_axum_error)?; - + let response = SimulationEpochsResponse { epochs: enhanced_epochs, total_count, }; - + Ok(output.to_response(response)) } @@ -141,39 +157,40 @@ async fn get_simulation_epoch_details( ) -> AxumResult> { let storage = state.storage(); let output = output.output.unwrap_or_default(); - + let epoch = get_simulation_epoch_by_id(storage, params.epoch_id) .await .map_err(to_axum_error)? .ok_or_else(|| to_axum_error(SimulationApiError::new("Simulation epoch not found")))?; - + let mut epoch_summary = SimulationEpochSummary::from(epoch); epoch_summary.nodes_analyzed = count_nodes_for_epoch(storage, params.epoch_id) .await .map_err(to_axum_error)?; - epoch_summary.available_methods = get_available_methods_for_epoch(storage, epoch_summary.epoch_id) - .await - .map_err(to_axum_error)?; - + epoch_summary.available_methods = + get_available_methods_for_epoch(storage, epoch_summary.epoch_id) + .await + .map_err(to_axum_error)?; + let node_performance = get_node_performance_for_epoch(storage, params.epoch_id) .await .map_err(to_axum_error)?; - + let performance_comparisons = get_performance_comparisons_for_epoch(storage, params.epoch_id) .await .map_err(to_axum_error)?; - + let route_analysis = get_route_analysis_for_epoch(storage, params.epoch_id) .await .map_err(to_axum_error)?; - + let details = SimulationEpochDetails { epoch: epoch_summary, node_performance, performance_comparisons, route_analysis, }; - + Ok(output.to_response(details)) } @@ -202,13 +219,13 @@ async fn compare_methods( ) -> AxumResult> { let storage = state.storage(); let output = output.output.unwrap_or_default(); - + // Get simulation epoch to extract actual epoch_id let sim_epoch = get_simulation_epoch_by_id(storage, params.epoch_id) .await .map_err(to_axum_error)? .ok_or_else(|| to_axum_error(SimulationApiError::new("Simulation epoch not found")))?; - + // Get performance data for both methods let old_performance = get_performance_by_method(storage, sim_epoch.epoch_id, "old") .await @@ -216,34 +233,46 @@ async fn compare_methods( let new_performance = get_performance_by_method(storage, sim_epoch.epoch_id, "new") .await .map_err(to_axum_error)?; - + // Get performance rankings for both methods - let old_rankings = storage.manager + let old_rankings = storage + .manager .get_simulated_performance_rankings(params.epoch_id, Some("old")) .await - .map_err(|e| to_axum_error(SimulationApiError::with_details("Database error", &e.to_string())))?; - let new_rankings = storage.manager + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Database error", + &e.to_string(), + )) + })?; + let new_rankings = storage + .manager .get_simulated_performance_rankings(params.epoch_id, Some("new")) .await - .map_err(|e| to_axum_error(SimulationApiError::with_details("Database error", &e.to_string())))?; - + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Database error", + &e.to_string(), + )) + })?; + // Build node comparisons let node_comparisons = build_node_comparisons_with_rankings( - old_performance, - new_performance, + old_performance, + new_performance, old_rankings, new_rankings, - &query + &query, ); - + // Calculate summary statistics let summary_statistics = calculate_summary_statistics(&node_comparisons); - + // Get route analysis comparison let route_analysis_comparison = get_route_analysis_comparison(storage, sim_epoch.epoch_id) .await .map_err(to_axum_error)?; - + let comparison = MethodComparisonResponse { epoch_id: sim_epoch.epoch_id, simulation_epoch_id: params.epoch_id, @@ -251,7 +280,7 @@ async fn compare_methods( summary_statistics, route_analysis_comparison, }; - + Ok(output.to_response(comparison)) } @@ -278,35 +307,65 @@ async fn export_simulation_data( ) -> Result)> { let storage = state.storage(); let format = query.format.unwrap_or(ExportFormat::Json); - + // Get detailed simulation data let epoch_details = get_simulation_epoch_details_internal(storage, params.epoch_id) .await .map_err(to_axum_error)? .ok_or_else(|| to_axum_error(SimulationApiError::new("Simulation epoch not found")))?; - + match format { ExportFormat::Json => { - let json_data = serde_json::to_string_pretty(&epoch_details) - .map_err(|e| to_axum_error(SimulationApiError::with_details("JSON serialization failed", &e.to_string())))?; - + let json_data = serde_json::to_string_pretty(&epoch_details).map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "JSON serialization failed", + &e.to_string(), + )) + })?; + Response::builder() .status(StatusCode::OK) .header("Content-Type", "application/json") - .header("Content-Disposition", format!("attachment; filename=\"simulation_epoch_{}.json\"", params.epoch_id)) + .header( + "Content-Disposition", + format!( + "attachment; filename=\"simulation_epoch_{}.json\"", + params.epoch_id + ), + ) .body(json_data.into()) - .map_err(|e| to_axum_error(SimulationApiError::with_details("Response building failed", &e.to_string()))) + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Response building failed", + &e.to_string(), + )) + }) } ExportFormat::Csv => { - let csv_data = convert_to_csv(&epoch_details) - .map_err(|e| to_axum_error(SimulationApiError::with_details("CSV conversion failed", &e.to_string())))?; - + let csv_data = convert_to_csv(&epoch_details).map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "CSV conversion failed", + &e.to_string(), + )) + })?; + Response::builder() .status(StatusCode::OK) .header("Content-Type", "text/csv") - .header("Content-Disposition", format!("attachment; filename=\"simulation_epoch_{}.csv\"", params.epoch_id)) + .header( + "Content-Disposition", + format!( + "attachment; filename=\"simulation_epoch_{}.csv\"", + params.epoch_id + ), + ) .body(csv_data.into()) - .map_err(|e| to_axum_error(SimulationApiError::with_details("Response building failed", &e.to_string()))) + .map_err(|e| { + to_axum_error(SimulationApiError::with_details( + "Response building failed", + &e.to_string(), + )) + }) } } } @@ -333,11 +392,11 @@ async fn get_node_performance_history( ) -> AxumResult>> { let storage = state.storage(); let output = output.output.unwrap_or_default(); - + let performance_history = get_node_performance_history_internal(storage, params.node_id) .await .map_err(to_axum_error)?; - + Ok(output.to_response(performance_history)) } @@ -351,7 +410,7 @@ async fn get_simulation_epochs_with_filters( ) -> SimulationResult> { let limit_i64 = limit as i64; let offset_i64 = offset as i64; - + let epochs = sqlx::query_as!( crate::support::storage::models::SimulatedRewardEpoch, "SELECT id as \"id!\", epoch_id as \"epoch_id!: u32\", calculation_method as \"calculation_method!\", @@ -366,8 +425,11 @@ async fn get_simulation_epochs_with_filters( .fetch_all(&storage.manager.connection_pool) .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - - Ok(epochs.into_iter().map(SimulationEpochSummary::from).collect()) + + Ok(epochs + .into_iter() + .map(SimulationEpochSummary::from) + .collect()) } async fn count_nodes_for_epoch(storage: &NymApiStorage, epoch_id: i64) -> SimulationResult { @@ -378,7 +440,10 @@ async fn count_nodes_for_epoch(storage: &NymApiStorage, epoch_id: i64) -> Simula .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string())) } -async fn get_available_methods_for_epoch(storage: &NymApiStorage, epoch_id: u32) -> SimulationResult> { +async fn get_available_methods_for_epoch( + storage: &NymApiStorage, + epoch_id: u32, +) -> SimulationResult> { storage .manager .get_available_calculation_methods_for_epoch(epoch_id) @@ -386,18 +451,22 @@ async fn get_available_methods_for_epoch(storage: &NymApiStorage, epoch_id: u32) .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string())) } -async fn count_total_simulation_epochs(storage: &NymApiStorage, _params: &SimulationListQuery) -> SimulationResult { - let result = sqlx::query!( - "SELECT COUNT(*) as count FROM simulated_reward_epochs" - ) - .fetch_one(&storage.manager.connection_pool) - .await - .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - +async fn count_total_simulation_epochs( + storage: &NymApiStorage, + _params: &SimulationListQuery, +) -> SimulationResult { + let result = sqlx::query!("SELECT COUNT(*) as count FROM simulated_reward_epochs") + .fetch_one(&storage.manager.connection_pool) + .await + .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; + Ok(result.count as usize) } -async fn get_simulation_epoch_by_id(storage: &NymApiStorage, id: i64) -> SimulationResult> { +async fn get_simulation_epoch_by_id( + storage: &NymApiStorage, + id: i64, +) -> SimulationResult> { storage .manager .get_simulated_reward_epoch(id) @@ -405,48 +474,66 @@ async fn get_simulation_epoch_by_id(storage: &NymApiStorage, id: i64) -> Simulat .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string())) } -async fn get_node_performance_for_epoch(storage: &NymApiStorage, epoch_id: i64) -> SimulationResult> { +async fn get_node_performance_for_epoch( + storage: &NymApiStorage, + epoch_id: i64, +) -> SimulationResult> { let performance = storage .manager .get_simulated_node_performance_for_epoch(epoch_id) .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - - Ok(performance.into_iter().map(NodePerformanceData::from).collect()) + + Ok(performance + .into_iter() + .map(NodePerformanceData::from) + .collect()) } -async fn get_performance_comparisons_for_epoch(storage: &NymApiStorage, epoch_id: i64) -> SimulationResult> { +async fn get_performance_comparisons_for_epoch( + storage: &NymApiStorage, + epoch_id: i64, +) -> SimulationResult> { let comparisons = storage .manager .get_simulated_performance_comparisons_for_epoch(epoch_id) .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - - Ok(comparisons.into_iter().map(PerformanceComparisonData::from).collect()) + + Ok(comparisons + .into_iter() + .map(PerformanceComparisonData::from) + .collect()) } -async fn get_route_analysis_for_epoch(storage: &NymApiStorage, epoch_id: i64) -> SimulationResult> { +async fn get_route_analysis_for_epoch( + storage: &NymApiStorage, + epoch_id: i64, +) -> SimulationResult> { let analysis = storage .manager .get_simulated_route_analysis_for_epoch(epoch_id) .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - + Ok(analysis.into_iter().map(RouteAnalysisData::from).collect()) } async fn get_performance_by_method( - storage: &NymApiStorage, - epoch_id: u32, - method: &str + storage: &NymApiStorage, + epoch_id: u32, + method: &str, ) -> SimulationResult> { let performance = storage .manager .get_simulated_node_performance_by_method(epoch_id, method) .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))?; - - Ok(performance.into_iter().map(NodePerformanceData::from).collect()) + + Ok(performance + .into_iter() + .map(NodePerformanceData::from) + .collect()) } fn build_node_comparisons_with_rankings( @@ -460,85 +547,88 @@ fn build_node_comparisons_with_rankings( .into_iter() .map(|p| (p.node_id, p)) .collect(); - + let mut new_map: HashMap = new_performance .into_iter() .map(|p| (p.node_id, p)) .collect(); - + // Create ranking maps let old_ranking_map: HashMap = old_rankings .into_iter() .map(|r| (r.node_id, r.performance_rank)) .collect(); - + let new_ranking_map: HashMap = new_rankings .into_iter() .map(|r| (r.node_id, r.performance_rank)) .collect(); - + let mut comparisons = Vec::new(); - + // Get all unique node IDs from both methods let mut all_node_ids: Vec<_> = old_map.keys().chain(new_map.keys()).cloned().collect(); all_node_ids.sort(); all_node_ids.dedup(); - + for node_id in all_node_ids { let old_perf = old_map.remove(&node_id); let new_perf = new_map.remove(&node_id); - + // Apply filters if let Some(filter_node_id) = query.node_id { if node_id != filter_node_id { continue; } } - + if let Some(ref filter_node_type) = query.node_type { - let node_type = old_perf.as_ref() + let node_type = old_perf + .as_ref() .or(new_perf.as_ref()) .map(|p| &p.node_type); if node_type != Some(filter_node_type) { continue; } } - + // Calculate differences let reliability_difference = match (&old_perf, &new_perf) { (Some(old), Some(new)) => Some(new.reliability_score - old.reliability_score), _ => None, }; - + let performance_delta_percentage = match (&old_perf, &new_perf) { - (Some(old), Some(new)) if old.reliability_score != 0.0 => { - Some((new.reliability_score - old.reliability_score) / old.reliability_score * 100.0) - } + (Some(old), Some(new)) if old.reliability_score != 0.0 => Some( + (new.reliability_score - old.reliability_score) / old.reliability_score * 100.0, + ), _ => None, }; - + // Apply delta filters if let Some(min_delta) = query.min_delta { if reliability_difference.map_or(true, |d| d < min_delta) { continue; } } - + if let Some(max_delta) = query.max_delta { if reliability_difference.map_or(true, |d| d > max_delta) { continue; } } - - let node_type = old_perf.as_ref() + + let node_type = old_perf + .as_ref() .or(new_perf.as_ref()) .map(|p| p.node_type.clone()) .unwrap_or_else(|| "unknown".to_string()); - - let identity_key = old_perf.as_ref() + + let identity_key = old_perf + .as_ref() .or(new_perf.as_ref()) .and_then(|p| p.identity_key.clone()); - + // Get rankings for this node let ranking_old = old_ranking_map.get(&node_id).copied(); let ranking_new = new_ranking_map.get(&node_id).copied(); @@ -546,7 +636,7 @@ fn build_node_comparisons_with_rankings( (Some(old), Some(new)) => Some(new - old), _ => None, }; - + comparisons.push(NodeMethodComparison { node_id, node_type, @@ -560,7 +650,7 @@ fn build_node_comparisons_with_rankings( ranking_delta, }); } - + comparisons } @@ -572,7 +662,7 @@ fn calculate_summary_statistics(comparisons: &[NodeMethodComparison]) -> Compari let mut unchanged = 0; let mut max_improvement: f64 = 0.0; let mut max_degradation: f64 = 0.0; - + for comparison in comparisons { if let Some(old) = &comparison.old_method { reliabilities_old.push(old.reliability_score); @@ -580,7 +670,7 @@ fn calculate_summary_statistics(comparisons: &[NodeMethodComparison]) -> Compari if let Some(new) = &comparison.new_method { reliabilities_new.push(new.reliability_score); } - + if let Some(diff) = comparison.reliability_difference { if diff > 0.001 { improvements += 1; @@ -593,23 +683,25 @@ fn calculate_summary_statistics(comparisons: &[NodeMethodComparison]) -> Compari } } } - + let average_reliability_old = if reliabilities_old.is_empty() { 0.0 } else { reliabilities_old.iter().sum::() / reliabilities_old.len() as f64 }; - + let average_reliability_new = if reliabilities_new.is_empty() { 0.0 } else { reliabilities_new.iter().sum::() / reliabilities_new.len() as f64 }; - + // Calculate medians and standard deviations - let (median_reliability_old, reliability_std_dev_old) = calculate_median_and_std(&reliabilities_old); - let (median_reliability_new, reliability_std_dev_new) = calculate_median_and_std(&reliabilities_new); - + let (median_reliability_old, reliability_std_dev_old) = + calculate_median_and_std(&reliabilities_old); + let (median_reliability_new, reliability_std_dev_new) = + calculate_median_and_std(&reliabilities_new); + ComparisonSummaryStats { total_nodes_compared: comparisons.len(), nodes_improved: improvements, @@ -630,22 +722,20 @@ fn calculate_median_and_std(values: &[f64]) -> (f64, f64) { if values.is_empty() { return (0.0, 0.0); } - + let mut sorted = values.to_vec(); sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); - + let median = if sorted.len() % 2 == 0 { (sorted[sorted.len() / 2 - 1] + sorted[sorted.len() / 2]) / 2.0 } else { sorted[sorted.len() / 2] }; - + let mean = values.iter().sum::() / values.len() as f64; - let variance = values.iter() - .map(|x| (x - mean).powi(2)) - .sum::() / values.len() as f64; + let variance = values.iter().map(|x| (x - mean).powi(2)).sum::() / values.len() as f64; let std_dev = variance.sqrt(); - + (median, std_dev) } @@ -659,24 +749,26 @@ async fn get_route_analysis_comparison( .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))? .map(RouteAnalysisData::from); - + let new_analysis = storage .manager .get_simulated_route_analysis_by_method(epoch_id, "new") .await .map_err(|e| SimulationApiError::with_details("Database error", &e.to_string()))? .map(RouteAnalysisData::from); - + let time_window_difference_hours = match (&old_analysis, &new_analysis) { (Some(old), Some(new)) => new.time_window_hours as i32 - old.time_window_hours as i32, _ => 0, }; - + let route_coverage_difference = match (&old_analysis, &new_analysis) { - (Some(old), Some(new)) => new.total_routes_analyzed as i32 - old.total_routes_analyzed as i32, + (Some(old), Some(new)) => { + new.total_routes_analyzed as i32 - old.total_routes_analyzed as i32 + } _ => 0, }; - + let success_rate_difference = match (&old_analysis, &new_analysis) { (Some(old), Some(new)) => { let old_rate = old.successful_routes as f64 / old.total_routes_analyzed as f64; @@ -685,7 +777,7 @@ async fn get_route_analysis_comparison( } _ => None, }; - + Ok(RouteAnalysisComparison { old_method: old_analysis, new_method: new_analysis, @@ -703,15 +795,16 @@ async fn get_simulation_epoch_details_internal( Some(epoch) => epoch, None => return Ok(None), }; - + let mut epoch_summary = SimulationEpochSummary::from(epoch); epoch_summary.nodes_analyzed = count_nodes_for_epoch(storage, epoch_id).await?; - epoch_summary.available_methods = get_available_methods_for_epoch(storage, epoch_summary.epoch_id).await?; - + epoch_summary.available_methods = + get_available_methods_for_epoch(storage, epoch_summary.epoch_id).await?; + let node_performance = get_node_performance_for_epoch(storage, epoch_id).await?; let performance_comparisons = get_performance_comparisons_for_epoch(storage, epoch_id).await?; let route_analysis = get_route_analysis_for_epoch(storage, epoch_id).await?; - + Ok(Some(SimulationEpochDetails { epoch: epoch_summary, node_performance, @@ -723,10 +816,12 @@ async fn get_simulation_epoch_details_internal( fn convert_to_csv(details: &SimulationEpochDetails) -> Result> { // Simple CSV conversion let mut csv = String::new(); - + // Header - csv.push_str("data_type,node_id,node_type,reliability_score,reward_amount,calculation_method\n"); - + csv.push_str( + "data_type,node_id,node_type,reliability_score,reward_amount,calculation_method\n", + ); + // Performance data for perf in &details.node_performance { csv.push_str(&format!( @@ -734,15 +829,19 @@ fn convert_to_csv(details: &SimulationEpochDetails) -> Result (StatusCode, Json) { @@ -766,9 +868,13 @@ fn to_axum_error(error: SimulationApiError) -> (StatusCode, Json NodePerformanceData { + use crate::simulation_api::models::{NodeMethodComparison, NodePerformanceData}; + + fn create_test_performance_data( + node_id: NodeId, + reliability: f64, + method: &str, + ) -> NodePerformanceData { NodePerformanceData { node_id, node_type: "mixnode".to_string(), @@ -862,7 +968,7 @@ mod tests { create_test_performance_data(1, 80.0, "old"), create_test_performance_data(2, 70.0, "old"), ]; - + let new_performance = vec![ create_test_performance_data(1, 90.0, "new"), create_test_performance_data(3, 85.0, "new"), // New node not in old @@ -896,7 +1002,7 @@ mod tests { calculated_at: 1234567890, }, ]; - + let new_rankings = vec![ SimulatedPerformanceRanking { id: 3, @@ -928,15 +1034,15 @@ mod tests { ]; let comparisons = build_node_comparisons_with_rankings( - old_performance, - new_performance, + old_performance, + new_performance, old_rankings, new_rankings, - &query + &query, ); assert_eq!(comparisons.len(), 3); // Nodes 1, 2, 3 - + // Find node 1 comparison let node1_comparison = comparisons.iter().find(|c| c.node_id == 1).unwrap(); assert!(node1_comparison.old_method.is_some()); @@ -963,7 +1069,7 @@ mod tests { create_test_performance_data(1, 80.0, "old"), create_test_performance_data(2, 70.0, "old"), ]; - + let new_performance = vec![ create_test_performance_data(1, 90.0, "new"), create_test_performance_data(2, 75.0, "new"), @@ -978,24 +1084,22 @@ mod tests { }; // Create test rankings for filter test - let rankings = vec![ - SimulatedPerformanceRanking { - id: 1, - simulated_epoch_id: 1, - node_id: 1, - calculation_method: "old".to_string(), - performance_rank: 1, - performance_percentile: 100.0, - calculated_at: 1234567890, - }, - ]; + let rankings = vec![SimulatedPerformanceRanking { + id: 1, + simulated_epoch_id: 1, + node_id: 1, + calculation_method: "old".to_string(), + performance_rank: 1, + performance_percentile: 100.0, + calculated_at: 1234567890, + }]; let comparisons = build_node_comparisons_with_rankings( - old_performance.clone(), - new_performance.clone(), + old_performance.clone(), + new_performance.clone(), rankings.clone(), rankings.clone(), - &query + &query, ); assert_eq!(comparisons.len(), 1); assert_eq!(comparisons[0].node_id, 1); @@ -1009,11 +1113,11 @@ mod tests { }; let comparisons = build_node_comparisons_with_rankings( - old_performance, + old_performance, new_performance, rankings.clone(), rankings, - &query + &query, ); assert_eq!(comparisons.len(), 1); assert_eq!(comparisons[0].node_id, 1); @@ -1037,28 +1141,28 @@ mod tests { create_test_performance_data(1, 80.0, "old"), create_test_performance_data(1, 90.0, "new"), ], - performance_comparisons: vec![ - PerformanceComparisonData { - node_id: 1, - node_type: "mixnode".to_string(), - performance_score: 80.0, - work_factor: 10.0, - calculation_method: "old".to_string(), - positive_samples: Some(100), - negative_samples: Some(20), - route_success_rate: Some(80.0), - calculated_at: 1234567890, - }, - ], + performance_comparisons: vec![PerformanceComparisonData { + node_id: 1, + node_type: "mixnode".to_string(), + performance_score: 80.0, + work_factor: 10.0, + calculation_method: "old".to_string(), + positive_samples: Some(100), + negative_samples: Some(20), + route_success_rate: Some(80.0), + calculated_at: 1234567890, + }], route_analysis: vec![], }; let csv = convert_to_csv(&details).unwrap(); println!("CSV: {}", csv); - - assert!(csv.contains("data_type,node_id,node_type,reliability_score,reward_amount,calculation_method")); + + assert!(csv.contains( + "data_type,node_id,node_type,reliability_score,reward_amount,calculation_method" + )); assert!(csv.contains("performance,1,mixnode,80,")); assert!(csv.contains("performance,1,mixnode,90,")); } -} \ No newline at end of file +} diff --git a/nym-api/src/simulation_api/mod.rs b/nym-api/src/simulation_api/mod.rs index be339c0736..caf626ba35 100644 --- a/nym-api/src/simulation_api/mod.rs +++ b/nym-api/src/simulation_api/mod.rs @@ -2,9 +2,9 @@ // SPDX-License-Identifier: GPL-3.0-only //! Simulation API for reward calculation analysis -//! +//! //! This module provides REST endpoints for accessing and analyzing //! simulated reward calculation data comparing old vs new methodologies. pub(crate) mod handlers; -pub(crate) mod models; \ No newline at end of file +pub(crate) mod models; diff --git a/nym-api/src/simulation_api/models.rs b/nym-api/src/simulation_api/models.rs index 7a974df544..8a018ac7dd 100644 --- a/nym-api/src/simulation_api/models.rs +++ b/nym-api/src/simulation_api/models.rs @@ -3,7 +3,10 @@ //! API models for simulation data responses -use crate::storage::models::{SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedRewardEpoch, SimulatedRouteAnalysis}; +use crate::storage::models::{ + SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedRewardEpoch, + SimulatedRouteAnalysis, +}; use nym_mixnet_contract_common::NodeId; use serde::{Deserialize, Serialize}; use time::OffsetDateTime; @@ -42,7 +45,7 @@ impl From for SimulationEpochSummary { end_timestamp: epoch.end_timestamp, description: epoch.description, created_at: epoch.created_at, - nodes_analyzed: 0, // Will be populated by handler + nodes_analyzed: 0, // Will be populated by handler available_methods: vec![], // Will be populated by handler } } @@ -174,17 +177,17 @@ pub struct NodeMethodComparison { #[derive(Serialize, Deserialize, ToSchema, Debug, Clone)] pub struct ComparisonSummaryStats { pub total_nodes_compared: usize, - pub nodes_improved: usize, // nodes with better performance in new method - pub nodes_degraded: usize, // nodes with worse performance in new method - pub nodes_unchanged: usize, // nodes with same performance + pub nodes_improved: usize, // nodes with better performance in new method + pub nodes_degraded: usize, // nodes with worse performance in new method + pub nodes_unchanged: usize, // nodes with same performance pub average_reliability_old: f64, pub average_reliability_new: f64, pub median_reliability_old: f64, pub median_reliability_new: f64, pub reliability_std_dev_old: f64, pub reliability_std_dev_new: f64, - pub max_improvement: f64, // highest positive delta - pub max_degradation: f64, // highest negative delta + pub max_improvement: f64, // highest positive delta + pub max_degradation: f64, // highest negative delta } /// Comparison of route analysis between methods @@ -192,8 +195,8 @@ pub struct ComparisonSummaryStats { pub struct RouteAnalysisComparison { pub old_method: Option, pub new_method: Option, - pub time_window_difference_hours: i32, // new - old - pub route_coverage_difference: i32, // new total routes - old total routes + pub time_window_difference_hours: i32, // new - old + pub route_coverage_difference: i32, // new total routes - old total routes pub success_rate_difference: Option, // new success rate - old success rate } @@ -252,4 +255,4 @@ impl SimulationApiError { timestamp: OffsetDateTime::now_utc().unix_timestamp(), } } -} \ No newline at end of file +} diff --git a/nym-api/src/support/cli/run.rs b/nym-api/src/support/cli/run.rs index 4ec8eb1bdd..706d82a574 100644 --- a/nym-api/src/support/cli/run.rs +++ b/nym-api/src/support/cli/run.rs @@ -9,7 +9,7 @@ use crate::ecash::dkg::controller::keys::{ }; use crate::ecash::dkg::controller::DkgController; use crate::ecash::state::EcashState; -use crate::epoch_operations::{EpochAdvancer, simulation::SimulationConfig}; +use crate::epoch_operations::{simulation::SimulationConfig, EpochAdvancer}; use crate::network::models::NetworkDetails; use crate::node_describe_cache::DescribedNodes; use crate::node_status_api::handlers::unstable; @@ -298,7 +298,10 @@ async fn start_nym_api_tasks_axum(config: &Config) -> anyhow::Result Result<(), sqlx::Error> { + pub(super) async fn purge_old_routes(&self, timestamp: i64) -> Result<(), sqlx::Error> { sqlx::query!("DELETE FROM routes WHERE timestamp < ?", timestamp) .execute(&self.connection_pool) .await?; @@ -2027,7 +2024,6 @@ impl StorageManager { &self, simulated_epoch_ids: &[i64], ) -> Result, sqlx::Error> { - if simulated_epoch_ids.is_empty() { return Ok(std::collections::HashMap::new()); } diff --git a/nym-api/src/support/storage/models.rs b/nym-api/src/support/storage/models.rs index 7165ac5ffd..eb4b2b9034 100644 --- a/nym-api/src/support/storage/models.rs +++ b/nym-api/src/support/storage/models.rs @@ -174,7 +174,7 @@ pub struct SimulatedNodePerformance { pub reliability_score: f64, // 0.0 to 100.0 pub positive_samples: u32, pub negative_samples: u32, - pub work_factor: Option, // 0.0 to 1.0 + pub work_factor: Option, // 0.0 to 1.0 pub calculation_method: String, // 'old' or 'new' pub calculated_at: i64, } @@ -186,9 +186,9 @@ pub struct SimulatedPerformanceComparison { pub id: i64, pub simulated_epoch_id: i64, pub node_id: NodeId, - pub node_type: String, // 'mixnode' or 'gateway' - pub performance_score: f64, // 0.0 to 100.0 - pub work_factor: f64, // Work factor applied (e.g., 10.0 for active, 1.0 for standby) + pub node_type: String, // 'mixnode' or 'gateway' + pub performance_score: f64, // 0.0 to 100.0 + pub work_factor: f64, // Work factor applied (e.g., 10.0 for active, 1.0 for standby) pub calculation_method: String, // 'old' or 'new' pub positive_samples: Option, pub negative_samples: Option, @@ -221,7 +221,7 @@ pub struct SimulatedRouteAnalysis { pub successful_routes: u32, pub failed_routes: u32, pub average_route_reliability: Option, // 0.0 to 100.0 - pub time_window_hours: u32, // 1 for new method, 24 for old method - pub analysis_parameters: Option, // JSON with analysis config + pub time_window_hours: u32, // 1 for new method, 24 for old method + pub analysis_parameters: Option, // JSON with analysis config pub calculated_at: i64, } diff --git a/nym-api/tests/public-api/simulation.rs b/nym-api/tests/public-api/simulation.rs index bad0310b18..009ec43637 100644 --- a/nym-api/tests/public-api/simulation.rs +++ b/nym-api/tests/public-api/simulation.rs @@ -13,7 +13,7 @@ async fn test_simulation_epochs_endpoint() { let url = format!("{}/v1/simulation/epochs", base); let response = make_request(&url).await; - + match response { Ok(res) => { let json = validate_json_response(res).await; @@ -23,10 +23,10 @@ async fn test_simulation_epochs_endpoint() { assert!(data.is_object()); assert!(data.get("epochs").is_some()); assert!(data.get("total_count").is_some()); - + let epochs = data.get("epochs").unwrap(); assert!(epochs.is_array()); - + // If there are epochs, verify their structure if let Some(epoch_array) = epochs.as_array() { if !epoch_array.is_empty() { @@ -34,7 +34,7 @@ async fn test_simulation_epochs_endpoint() { verify_epoch_summary_structure(first_epoch); } } - + println!("✓ Simulation epochs endpoint returned valid response"); } Err(e) => { @@ -58,7 +58,7 @@ async fn test_simulation_epochs_with_pagination() { let url = format!("{}/v1/simulation/epochs?limit=5&offset=0", base); let response = make_request(&url).await; - + match response { Ok(res) => { let json = validate_json_response(res).await; @@ -67,10 +67,10 @@ async fn test_simulation_epochs_with_pagination() { assert!(data.is_object()); assert!(data.get("epochs").is_some()); assert!(data.get("total_count").is_some()); - + let epochs = data.get("epochs").unwrap().as_array().unwrap(); assert!(epochs.len() <= 5); // Should respect limit - + println!("✓ Simulation epochs pagination works correctly"); } Err(_) => { @@ -94,7 +94,7 @@ async fn test_simulation_epoch_details_structure() { // First, get a list of epochs to find a valid ID let epochs_url = format!("{}/v1/simulation/epochs?limit=1", base); let epochs_response = make_request(&epochs_url).await; - + match epochs_response { Ok(res) => { let json = validate_json_response(res).await; @@ -103,11 +103,11 @@ async fn test_simulation_epoch_details_structure() { if !epochs.is_empty() { let first_epoch = &epochs[0]; let epoch_id = first_epoch.get("id").unwrap().as_i64().unwrap(); - + // Now test the details endpoint let details_url = format!("{}/v1/simulation/epochs/{}", base, epoch_id); let details_response = make_request(&details_url).await; - + match details_response { Ok(res) => { let json = validate_json_response(res).await; @@ -144,7 +144,7 @@ async fn test_simulation_comparison_endpoint() { // First, get a list of epochs to find a valid ID let epochs_url = format!("{}/v1/simulation/epochs?limit=1", base); let epochs_response = make_request(&epochs_url).await; - + match epochs_response { Ok(res) => { let json = validate_json_response(res).await; @@ -153,18 +153,21 @@ async fn test_simulation_comparison_endpoint() { if !epochs.is_empty() { let first_epoch = &epochs[0]; let epoch_id = first_epoch.get("id").unwrap().as_i64().unwrap(); - + // Test the comparison endpoint - let comparison_url = format!("{}/v1/simulation/epochs/{}/comparison", base, epoch_id); + let comparison_url = + format!("{}/v1/simulation/epochs/{}/comparison", base, epoch_id); let comparison_response = make_request(&comparison_url).await; - + match comparison_response { Ok(res) => { let json = validate_json_response(res).await; match json { Ok(data) => { verify_comparison_structure(&data); - println!("✓ Simulation comparison endpoint returned valid structure"); + println!( + "✓ Simulation comparison endpoint returned valid structure" + ); } Err(e) => { println!("✗ Failed to parse comparison data: {}", e); @@ -194,7 +197,7 @@ async fn test_simulation_export_endpoints() { // First, get a list of epochs to find a valid ID let epochs_url = format!("{}/v1/simulation/epochs?limit=1", base); let epochs_response = make_request(&epochs_url).await; - + match epochs_response { Ok(res) => { let json = validate_json_response(res).await; @@ -203,11 +206,14 @@ async fn test_simulation_export_endpoints() { if !epochs.is_empty() { let first_epoch = &epochs[0]; let epoch_id = first_epoch.get("id").unwrap().as_i64().unwrap(); - + // Test JSON export - let json_export_url = format!("{}/v1/simulation/epochs/{}/export?format=json", base, epoch_id); + let json_export_url = format!( + "{}/v1/simulation/epochs/{}/export?format=json", + base, epoch_id + ); let json_response = make_request(&json_export_url).await; - + match json_response { Ok(res) => { assert!(res.status().is_success()); @@ -221,11 +227,14 @@ async fn test_simulation_export_endpoints() { println!("✗ JSON export failed: {}", e); } } - + // Test CSV export - let csv_export_url = format!("{}/v1/simulation/epochs/{}/export?format=csv", base, epoch_id); + let csv_export_url = format!( + "{}/v1/simulation/epochs/{}/export?format=csv", + base, epoch_id + ); let csv_response = make_request(&csv_export_url).await; - + match csv_response { Ok(res) => { assert!(res.status().is_success()); @@ -258,7 +267,7 @@ async fn test_simulation_error_handling() { // Test 404 for non-existent epoch let invalid_url = format!("{}/v1/simulation/epochs/999999", base); let response = make_request(&invalid_url).await; - + match response { Ok(res) => { // Should get a successful response (empty or error structure) @@ -284,7 +293,7 @@ fn verify_epoch_summary_structure(epoch: &Value) { assert!(epoch.get("created_at").is_some()); assert!(epoch.get("nodes_analyzed").is_some()); assert!(epoch.get("available_methods").is_some()); - + // Verify types assert!(epoch.get("id").unwrap().is_i64()); assert!(epoch.get("epoch_id").unwrap().is_u64()); @@ -299,22 +308,22 @@ fn verify_epoch_details_structure(data: &Value) { assert!(data.get("node_performance").is_some()); assert!(data.get("rewards").is_some()); assert!(data.get("route_analysis").is_some()); - + // Verify epoch summary structure verify_epoch_summary_structure(data.get("epoch").unwrap()); - + // Verify arrays assert!(data.get("node_performance").unwrap().is_array()); assert!(data.get("rewards").unwrap().is_array()); assert!(data.get("route_analysis").unwrap().is_array()); - + // If there's performance data, verify its structure let performance_array = data.get("node_performance").unwrap().as_array().unwrap(); if !performance_array.is_empty() { let first_performance = &performance_array[0]; verify_performance_data_structure(first_performance); } - + // If there's reward data, verify its structure let rewards_array = data.get("rewards").unwrap().as_array().unwrap(); if !rewards_array.is_empty() { @@ -332,7 +341,7 @@ fn verify_performance_data_structure(performance: &Value) { assert!(performance.get("negative_samples").is_some()); assert!(performance.get("calculation_method").is_some()); assert!(performance.get("calculated_at").is_some()); - + // Verify types assert!(performance.get("node_id").unwrap().is_u64()); assert!(performance.get("node_type").unwrap().is_string()); @@ -350,7 +359,7 @@ fn verify_reward_data_structure(reward: &Value) { assert!(reward.get("calculated_reward_amount").is_some()); assert!(reward.get("reward_currency").is_some()); assert!(reward.get("calculation_method").is_some()); - + // Verify types assert!(reward.get("node_id").unwrap().is_u64()); assert!(reward.get("node_type").unwrap().is_string()); @@ -366,14 +375,14 @@ fn verify_comparison_structure(data: &Value) { assert!(data.get("node_comparisons").is_some()); assert!(data.get("summary_statistics").is_some()); assert!(data.get("route_analysis_comparison").is_some()); - + // Verify types assert!(data.get("epoch_id").unwrap().is_u64()); assert!(data.get("simulation_epoch_id").unwrap().is_i64()); assert!(data.get("node_comparisons").unwrap().is_array()); assert!(data.get("summary_statistics").unwrap().is_object()); assert!(data.get("route_analysis_comparison").unwrap().is_object()); - + // Verify summary statistics structure let stats = data.get("summary_statistics").unwrap(); assert!(stats.get("total_nodes_compared").is_some()); @@ -381,4 +390,4 @@ fn verify_comparison_structure(data: &Value) { assert!(stats.get("nodes_degraded").is_some()); assert!(stats.get("average_reliability_old").is_some()); assert!(stats.get("average_reliability_new").is_some()); -} \ No newline at end of file +} diff --git a/nym-network-monitor/src/accounting.rs b/nym-network-monitor/src/accounting.rs index 98fc385f09..05a980a641 100644 --- a/nym-network-monitor/src/accounting.rs +++ b/nym-network-monitor/src/accounting.rs @@ -504,10 +504,8 @@ pub async fn submit_metrics(database_url: Option<&String>) -> anyhow::Result<()> let node_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_NODE}", nym_api_url); - let gateway_submit_url = format!( - "{}/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}", - nym_api_url - ); + let gateway_submit_url = + format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}", nym_api_url); let route_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_ROUTE}", nym_api_url); diff --git a/nym-network-monitor/src/main.rs b/nym-network-monitor/src/main.rs index b709c1ca20..27fbcbd044 100644 --- a/nym-network-monitor/src/main.rs +++ b/nym-network-monitor/src/main.rs @@ -54,7 +54,7 @@ async fn make_clients( if let Some(dropped_client) = clients.write().await.pop_front() { const CLIENT_DROP_TIMEOUT: Duration = Duration::from_secs(30); let start = tokio::time::Instant::now(); - + // Try immediate unwrap first match Arc::try_unwrap(dropped_client) { Ok(client) => { @@ -82,9 +82,12 @@ async fn make_clients( } tokio::time::sleep(Duration::from_millis(100)).await; } - + if start.elapsed() >= CLIENT_DROP_TIMEOUT { - warn!("Client drop timed out after {:?}, forcing drop", CLIENT_DROP_TIMEOUT); + warn!( + "Client drop timed out after {:?}, forcing drop", + CLIENT_DROP_TIMEOUT + ); // Client will be dropped when Arc goes out of scope } } diff --git a/sdk/rust/nym-sdk/src/mixnet/native_client.rs b/sdk/rust/nym-sdk/src/mixnet/native_client.rs index 13a152851a..acdb96a00c 100644 --- a/sdk/rust/nym-sdk/src/mixnet/native_client.rs +++ b/sdk/rust/nym-sdk/src/mixnet/native_client.rs @@ -148,9 +148,9 @@ impl MixnetClient { pub fn gateway_connection(&self) -> GatewayConnection { self.client_state.gateway_connection } - + /// Check if the websocket connection to the gateway is alive at the socket level - /// + /// /// This performs a real socket-level health check to determine if the connection /// is actually alive and responsive, not just whether we have a file descriptor. pub fn is_gateway_connection_alive(&self) -> bool { @@ -353,43 +353,43 @@ impl MixnetMessageSender for MixnetClientSender { } /// Check if a socket file descriptor is alive and responsive -/// +/// /// This function performs socket-level checks to determine if the connection is actually alive. /// It's cross-platform compatible and works on both Unix and non-Unix systems. fn socket_is_alive(fd: std::os::raw::c_int) -> bool { #[cfg(unix)] { - use std::os::unix::io::FromRawFd; - use std::net::TcpStream; use std::io::ErrorKind; - + use std::net::TcpStream; + use std::os::unix::io::FromRawFd; + unsafe { // Create a TcpStream from the raw fd to perform socket operations let stream = TcpStream::from_raw_fd(fd); - + // Try to peek at the socket to see if it's still connected // We peek with a zero-length buffer to avoid consuming data let mut buf = [0u8; 0]; let result = match stream.peek(&mut buf) { - Ok(_) => true, // Socket is alive and readable + Ok(_) => true, // Socket is alive and readable Err(e) => match e.kind() { - ErrorKind::WouldBlock => true, // Socket is alive but no data available - ErrorKind::ConnectionReset - | ErrorKind::ConnectionAborted - | ErrorKind::BrokenPipe - | ErrorKind::NotConnected => false, // Socket is clearly dead - _ => true, // Other errors might be temporary, assume alive - } + ErrorKind::WouldBlock => true, // Socket is alive but no data available + ErrorKind::ConnectionReset + | ErrorKind::ConnectionAborted + | ErrorKind::BrokenPipe + | ErrorKind::NotConnected => false, // Socket is clearly dead + _ => true, // Other errors might be temporary, assume alive + }, }; - + // Prevent the TcpStream from closing the fd when it's dropped // since we don't own the fd std::mem::forget(stream); - + result } } - + #[cfg(not(unix))] { // On non-Unix systems, we can't easily check socket state