From 4f7fa557d5f4e80e4054af9410259f9db8e0cb39 Mon Sep 17 00:00:00 2001 From: durch Date: Thu, 5 Jun 2025 10:44:33 +0200 Subject: [PATCH] Optimize database queries by eliminating N+1 patterns in simulation system This commit addresses critical N+1 query performance issues identified in the reward simulation system. The optimizations significantly reduce database round trips and improve performance when processing large datasets. **Key Optimizations:** 1. **Batch Identity Key Lookups** - Added `get_mixnode_identity_keys_batch()` and `get_gateway_identity_keys_batch()` - Updated simulation performance conversion to use batch operations - Reduced from N individual queries to 2 batch queries 2. **Batch Node Classification** - Added `classify_nodes_batch()` method for mixnode/gateway determination - Updated reliability calculation methods to use batch classification - Reduced from N individual lookups to 2 batch queries 3. **Batch Epoch Metadata Enhancement** - Added `count_simulated_node_performance_for_epochs_batch()` - Added `get_available_calculation_methods_for_epochs_batch()` - Updated API handlers to use batch operations for metadata enhancement - Reduced from 2N queries to 2 batch queries for epoch data 4. **Bulk Insert Optimizations** - Converted individual INSERT operations to use `sqlx::QueryBuilder::push_values()` - Optimized simulation data insertion methods - Eliminated transaction overhead from individual inserts **Performance Impact:** - Before: N+2N database queries for N nodes/epochs - After: 2+2 batch queries regardless of dataset size - Significant performance improvement for large simulation datasets All changes maintain backward compatibility while providing substantial performance benefits for the reward simulation system. --- ...57d2300860e30920cfaea219da2c1b42ed8e8.json | 12 - ...f02a96ff0c723d45e695c17a7a422eac55916.json | 12 - ...942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json | 12 - nym-api/src/epoch_operations/simulation.rs | 29 +- nym-api/src/simulation_api/handlers/mod.rs | 30 +- nym-api/src/support/storage/manager.rs | 783 +++++++++++------- nym-api/src/support/storage/mod.rs | 2 + 7 files changed, 534 insertions(+), 346 deletions(-) delete mode 100644 nym-api/.sqlx/query-0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8.json delete mode 100644 nym-api/.sqlx/query-c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916.json delete mode 100644 nym-api/.sqlx/query-e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json diff --git a/nym-api/.sqlx/query-0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8.json b/nym-api/.sqlx/query-0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8.json deleted file mode 100644 index 043e208f90..0000000000 --- a/nym-api/.sqlx/query-0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n INSERT INTO simulated_performance_comparisons\n (simulated_epoch_id, node_id, node_type, performance_score, work_factor,\n calculation_method, positive_samples, negative_samples, route_success_rate)\n VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 9 - }, - "nullable": [] - }, - "hash": "0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8" -} diff --git a/nym-api/.sqlx/query-c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916.json b/nym-api/.sqlx/query-c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916.json deleted file mode 100644 index 7a6403aa40..0000000000 --- a/nym-api/.sqlx/query-c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n INSERT INTO simulated_performance_rankings\n (simulated_epoch_id, node_id, calculation_method, performance_rank,\n performance_percentile)\n VALUES (?, ?, ?, ?, ?)\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 5 - }, - "nullable": [] - }, - "hash": "c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916" -} diff --git a/nym-api/.sqlx/query-e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json b/nym-api/.sqlx/query-e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json deleted file mode 100644 index 8cb02ae2a9..0000000000 --- a/nym-api/.sqlx/query-e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "SQLite", - "query": "\n INSERT INTO simulated_node_performance\n (simulated_epoch_id, node_id, node_type, identity_key, reliability_score,\n positive_samples, negative_samples, work_factor, calculation_method)\n VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)\n ", - "describe": { - "columns": [], - "parameters": { - "Right": 9 - }, - "nullable": [] - }, - "hash": "e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0" -} diff --git a/nym-api/src/epoch_operations/simulation.rs b/nym-api/src/epoch_operations/simulation.rs index 466c366db9..acd5091add 100644 --- a/nym-api/src/epoch_operations/simulation.rs +++ b/nym-api/src/epoch_operations/simulation.rs @@ -451,21 +451,28 @@ impl<'a> SimulationCoordinator<'a> { 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 + .manager + .get_mixnode_identity_keys_batch(&all_node_ids) + .await + .unwrap_or_default(); + 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); - // Try to get identity key from storage + // Get identity key from our batch results let identity_key = match node_type.as_str() { - "mixnode" => self.storage - .get_mixnode_identity_key(node.node_id) - .await - .ok() - .flatten(), - "gateway" => self.storage - .get_gateway_identity_key(node.node_id) - .await - .ok() - .flatten(), + "mixnode" => mixnode_identities.get(&node.node_id).cloned(), + "gateway" => gateway_identities.get(&node.node_id).cloned(), _ => None, }; diff --git a/nym-api/src/simulation_api/handlers/mod.rs b/nym-api/src/simulation_api/handlers/mod.rs index 9baa209025..2c322077f0 100644 --- a/nym-api/src/simulation_api/handlers/mod.rs +++ b/nym-api/src/simulation_api/handlers/mod.rs @@ -82,19 +82,27 @@ async fn list_simulation_epochs( .await .map_err(to_axum_error)?; - // Enhance epochs with additional metadata + // 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())))?; + 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())))?; + let mut enhanced_epochs = Vec::new(); for mut epoch in epochs { - // Count nodes analyzed for this epoch - epoch.nodes_analyzed = count_nodes_for_epoch(storage, epoch.id) - .await - .map_err(to_axum_error)?; - - // Get available calculation methods for this epoch_id - epoch.available_methods = get_available_methods_for_epoch(storage, epoch.epoch_id) - .await - .map_err(to_axum_error)?; - + // 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(); enhanced_epochs.push(epoch); } diff --git a/nym-api/src/support/storage/manager.rs b/nym-api/src/support/storage/manager.rs index f714c18efb..1b0020ea89 100644 --- a/nym-api/src/support/storage/manager.rs +++ b/nym-api/src/support/storage/manager.rs @@ -11,7 +11,7 @@ use crate::support::storage::models::{ use crate::support::storage::DbIdCache; use nym_mixnet_contract_common::{EpochId, IdentityKey, NodeId}; use nym_types::monitoring::{NodeResult, RouteResult}; -use sqlx::FromRow; +use sqlx::{FromRow, Row}; use std::collections::HashMap; use time::{Date, OffsetDateTime}; use tracing::info; @@ -101,16 +101,21 @@ impl StorageManager { .await .map_err(|_e| sqlx::Error::PoolClosed)?; // Example: map anyhow::Error to sqlx::Error; adjust as needed + // Collect all node IDs for batch classification + let node_ids: Vec = corrected_reliabilities + .iter() + .map(|node| node.node_id) + .collect(); + + // Batch classify nodes to determine which are mixnodes + let (mixnode_ids, _) = self.classify_nodes_batch(&node_ids).await?; + let mixnode_set: std::collections::HashSet = mixnode_ids.into_iter().collect(); + let mut avg_mix_reliabilities = Vec::new(); for corrected_node_info in corrected_reliabilities { - // Check if this node_id is a mixnode by attempting to fetch its identity key as a mixnode. - // This relies on get_mixnode_identity_key returning Some for mixnodes and None (or error) for non-mixnodes. - if self - .get_mixnode_identity_key(corrected_node_info.node_id) - .await? - .is_some() - { + // Check if this node_id is a mixnode using our batch result + if mixnode_set.contains(&corrected_node_info.node_id) { avg_mix_reliabilities.push(AvgMixnodeReliability { mix_id: corrected_node_info.node_id, value: Some(corrected_node_info.reliability as f32), @@ -131,15 +136,21 @@ impl StorageManager { .await .map_err(|_e| sqlx::Error::PoolClosed)?; // Example: map anyhow::Error to sqlx::Error; adjust as needed + // Collect all node IDs for batch classification + let node_ids: Vec = corrected_reliabilities + .iter() + .map(|node| node.node_id) + .collect(); + + // Batch classify nodes to determine which are gateways + let (_, gateway_ids) = self.classify_nodes_batch(&node_ids).await?; + let gateway_set: std::collections::HashSet = gateway_ids.into_iter().collect(); + let mut avg_gateway_reliabilities = Vec::new(); for corrected_node_info in corrected_reliabilities { - // Check if this node_id is a gateway. - if self - .get_gateway_identity_key(corrected_node_info.node_id) - .await? - .is_some() - { + // Check if this node_id is a gateway using our batch result + if gateway_set.contains(&corrected_node_info.node_id) { let total_samples = corrected_node_info.pos_samples_in_interval + corrected_node_info.neg_samples_in_interval; let reliability_value = if total_samples <= 3 { @@ -232,6 +243,90 @@ impl StorageManager { Ok(identity_key) } + /// Batch version of get_gateway_identity_key for improved performance. + /// Returns a HashMap mapping node_id to identity for all found gateways. + pub(crate) async fn get_gateway_identity_keys_batch( + &self, + node_ids: &[NodeId], + ) -> Result, sqlx::Error> { + if node_ids.is_empty() { + return Ok(std::collections::HashMap::new()); + } + + // Create placeholders for the IN clause + let placeholders = node_ids.iter().map(|_| "?").collect::>().join(", "); + let query = format!( + "SELECT node_id, identity FROM gateway_details WHERE node_id IN ({})", + placeholders + ); + + let mut query_builder = sqlx::query(&query); + for node_id in node_ids { + query_builder = query_builder.bind(node_id); + } + + let rows = query_builder.fetch_all(&self.connection_pool).await?; + + let mut result = std::collections::HashMap::new(); + for row in rows { + let node_id: NodeId = row.try_get("node_id")?; + let identity: Option = row.try_get("identity")?; + if let Some(key) = identity { + result.insert(node_id, key); + } + } + + Ok(result) + } + + /// Batch method to classify nodes as mixnodes or gateways. + /// Returns (mixnode_ids, gateway_ids) as separate vectors for easier processing. + pub(crate) async fn classify_nodes_batch( + &self, + node_ids: &[NodeId], + ) -> Result<(Vec, Vec), sqlx::Error> { + if node_ids.is_empty() { + return Ok((Vec::new(), Vec::new())); + } + + // Query both tables to determine which nodes are mixnodes vs gateways + let placeholders = node_ids.iter().map(|_| "?").collect::>().join(", "); + + // First get all mixnode IDs + let mixnode_query = format!( + "SELECT mix_id FROM mixnode_details WHERE mix_id IN ({})", + placeholders + ); + let mut query_builder = sqlx::query(&mixnode_query); + for node_id in node_ids { + query_builder = query_builder.bind(node_id); + } + let mixnode_rows = query_builder.fetch_all(&self.connection_pool).await?; + + let mixnode_ids: Vec = mixnode_rows + .into_iter() + .map(|row| row.try_get("mix_id")) + .collect::, _>>()?; + + // Then get all gateway IDs + let gateway_query = format!( + "SELECT node_id FROM gateway_details WHERE node_id IN ({})", + placeholders + ); + let mut query_builder = sqlx::query(&gateway_query); + for node_id in node_ids { + query_builder = query_builder.bind(node_id); + } + let gateway_rows = query_builder.fetch_all(&self.connection_pool).await?; + + let gateway_ids: Vec = gateway_rows + .into_iter() + .map(|row| row.try_get("node_id")) + .collect::, _>>()?; + + Ok((mixnode_ids, gateway_ids)) + } + /// Tries to obtain identity value of given mixnode given its mix_id /// /// # Arguments @@ -252,6 +347,42 @@ impl StorageManager { Ok(identity_key) } + /// Batch version of get_mixnode_identity_key for improved performance. + /// Returns a HashMap mapping node_id to identity_key for all found mixnodes. + pub(crate) async fn get_mixnode_identity_keys_batch( + &self, + mix_ids: &[NodeId], + ) -> Result, sqlx::Error> { + if mix_ids.is_empty() { + return Ok(std::collections::HashMap::new()); + } + + // Create placeholders for the IN clause + let placeholders = mix_ids.iter().map(|_| "?").collect::>().join(", "); + let query = format!( + "SELECT mix_id, identity_key FROM mixnode_details WHERE mix_id IN ({})", + placeholders + ); + + let mut query_builder = sqlx::query(&query); + for mix_id in mix_ids { + query_builder = query_builder.bind(mix_id); + } + + let rows = query_builder.fetch_all(&self.connection_pool).await?; + + let mut result = std::collections::HashMap::new(); + for row in rows { + let mix_id: NodeId = row.try_get("mix_id")?; + let identity_key: Option = row.try_get("identity_key")?; + if let Some(key) = identity_key { + result.insert(mix_id, key); + } + } + + Ok(result) + } + /// Gets all reliability statuses for mixnode with particular identity that were inserted /// into the database after the specified unix timestamp. /// @@ -1528,70 +1659,70 @@ impl StorageManager { } } -pub(crate) mod v3_migration { - use crate::storage::models::{SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedPerformanceRanking, SimulatedRewardEpoch, SimulatedRouteAnalysis}; - use crate::support::storage::manager::StorageManager; - use crate::support::storage::models::GatewayDetailsBeforeMigration; - use nym_mixnet_contract_common::NodeId; +use crate::storage::models::{ + SimulatedNodePerformance, SimulatedPerformanceComparison, SimulatedPerformanceRanking, + SimulatedRewardEpoch, SimulatedRouteAnalysis, +}; +use crate::support::storage::models::GatewayDetailsBeforeMigration; - impl StorageManager { - pub(crate) async fn check_v3_migration(&self) -> Result { - sqlx::query!("SELECT EXISTS (SELECT 1 FROM v3_migration_info) AS 'exists'",) - .fetch_one(&self.connection_pool) - .await - .map(|result| result.exists == Some(1)) - } +impl StorageManager { + pub(crate) async fn check_v3_migration(&self) -> Result { + sqlx::query!("SELECT EXISTS (SELECT 1 FROM v3_migration_info) AS 'exists'",) + .fetch_one(&self.connection_pool) + .await + .map(|result| result.exists == Some(1)) + } - pub(crate) async fn set_v3_migration_completion(&self) -> Result<(), sqlx::Error> { - sqlx::query!("INSERT INTO v3_migration_info(id) VALUES (0)") - .execute(&self.connection_pool) - .await?; - Ok(()) - } - - pub(crate) async fn get_all_known_gateways( - &self, - ) -> Result, sqlx::Error> { - sqlx::query_as("SELECT * FROM gateway_details") - .fetch_all(&self.connection_pool) - .await - } - - pub(crate) async fn set_gateway_node_id( - &self, - identity: &str, - node_id: NodeId, - ) -> Result<(), sqlx::Error> { - sqlx::query!( - "UPDATE gateway_details SET node_id = ? WHERE identity = ?", - node_id, - identity - ) + pub(crate) async fn set_v3_migration_completion(&self) -> Result<(), sqlx::Error> { + sqlx::query!("INSERT INTO v3_migration_info(id) VALUES (0)") .execute(&self.connection_pool) .await?; - Ok(()) - } + Ok(()) + } - pub(crate) async fn purge_gateway(&self, db_id: i64) -> Result<(), sqlx::Error> { - sqlx::query!( - r#" + pub(crate) async fn get_all_known_gateways( + &self, + ) -> Result, sqlx::Error> { + sqlx::query_as("SELECT * FROM gateway_details") + .fetch_all(&self.connection_pool) + .await + } + + pub(crate) async fn set_gateway_node_id( + &self, + identity: &str, + node_id: NodeId, + ) -> Result<(), sqlx::Error> { + sqlx::query!( + "UPDATE gateway_details SET node_id = ? WHERE identity = ?", + node_id, + identity + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } + + pub(crate) async fn purge_gateway(&self, db_id: i64) -> Result<(), sqlx::Error> { + sqlx::query!( + r#" DELETE FROM gateway_historical_uptime WHERE gateway_details_id = ?; DELETE FROM gateway_status WHERE gateway_details_id = ?; DELETE FROM testing_route WHERE gateway_id = ?; DELETE FROM gateway_details WHERE id = ?; "#, - db_id, - db_id, - db_id, - db_id, - ) - .execute(&self.connection_pool) - .await?; - Ok(()) - } + db_id, + db_id, + db_id, + db_id, + ) + .execute(&self.connection_pool) + .await?; + Ok(()) + } - pub(crate) async fn make_node_id_not_null(&self) -> Result<(), sqlx::Error> { - sqlx::query( + pub(crate) async fn make_node_id_not_null(&self) -> Result<(), sqlx::Error> { + sqlx::query( r#" CREATE TABLE gateway_details_temp ( @@ -1607,136 +1738,130 @@ pub(crate) mod v3_migration { ) .execute(&self.connection_pool) .await?; - Ok(()) - } + Ok(()) + } - // Simulated Rewarding System CRUD Methods + // Simulated Rewarding System CRUD Methods - /// Creates a new simulated reward epoch and returns its ID - pub(crate) async fn create_simulated_reward_epoch( - &self, - epoch_id: u32, - calculation_method: &str, - start_timestamp: i64, - end_timestamp: i64, - description: Option<&str>, - ) -> Result { - let result = sqlx::query!( - r#" + /// Creates a new simulated reward epoch and returns its ID + pub(crate) async fn create_simulated_reward_epoch( + &self, + epoch_id: u32, + calculation_method: &str, + start_timestamp: i64, + end_timestamp: i64, + description: Option<&str>, + ) -> Result { + let result = sqlx::query!( + r#" INSERT INTO simulated_reward_epochs (epoch_id, calculation_method, start_timestamp, end_timestamp, description) VALUES (?, ?, ?, ?, ?) "#, - epoch_id, - calculation_method, - start_timestamp, - end_timestamp, - description, - ) - .execute(&self.connection_pool) - .await?; + epoch_id, + calculation_method, + start_timestamp, + end_timestamp, + description, + ) + .execute(&self.connection_pool) + .await?; - Ok(result.last_insert_rowid()) + Ok(result.last_insert_rowid()) + } + + /// Inserts simulated node performance data + pub(crate) async fn insert_simulated_node_performance( + &self, + performance_data: &[SimulatedNodePerformance], + ) -> Result<(), sqlx::Error> { + if performance_data.is_empty() { + return Ok(()); } - /// Inserts simulated node performance data - pub(crate) async fn insert_simulated_node_performance( - &self, - performance_data: &[SimulatedNodePerformance], - ) -> Result<(), sqlx::Error> { - let mut tx = self.connection_pool.begin().await?; - - for performance in performance_data { - sqlx::query!( - r#" - INSERT INTO simulated_node_performance - (simulated_epoch_id, node_id, node_type, identity_key, reliability_score, - positive_samples, negative_samples, work_factor, calculation_method) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - "#, - performance.simulated_epoch_id, - performance.node_id, - performance.node_type, - performance.identity_key, - performance.reliability_score, - performance.positive_samples, - performance.negative_samples, - performance.work_factor, - performance.calculation_method, - ) - .execute(&mut *tx) - .await?; - } - - tx.commit().await + // Use bulk insert with sqlx::QueryBuilder for better performance + let mut query_builder = sqlx::QueryBuilder::new( + "INSERT INTO simulated_node_performance (simulated_epoch_id, node_id, node_type, identity_key, reliability_score, positive_samples, negative_samples, work_factor, calculation_method) " + ); + + query_builder.push_values(performance_data, |mut b, performance| { + b.push_bind(performance.simulated_epoch_id) + .push_bind(performance.node_id) + .push_bind(&performance.node_type) + .push_bind(&performance.identity_key) + .push_bind(performance.reliability_score) + .push_bind(performance.positive_samples) + .push_bind(performance.negative_samples) + .push_bind(performance.work_factor) + .push_bind(&performance.calculation_method); + }); + + query_builder.build().execute(&self.connection_pool).await?; + Ok(()) + } + + /// Inserts performance comparison data for simulation analysis + pub(crate) async fn insert_simulated_performance_comparisons( + &self, + performance_data: &[SimulatedPerformanceComparison], + ) -> Result<(), sqlx::Error> { + if performance_data.is_empty() { + return Ok(()); } - /// Inserts performance comparison data for simulation analysis - pub(crate) async fn insert_simulated_performance_comparisons( - &self, - performance_data: &[SimulatedPerformanceComparison], - ) -> Result<(), sqlx::Error> { - let mut tx = self.connection_pool.begin().await?; - - for comparison in performance_data { - sqlx::query!( - r#" - INSERT INTO simulated_performance_comparisons - (simulated_epoch_id, node_id, node_type, performance_score, work_factor, - calculation_method, positive_samples, negative_samples, route_success_rate) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) - "#, - comparison.simulated_epoch_id, - comparison.node_id, - comparison.node_type, - comparison.performance_score, - comparison.work_factor, - comparison.calculation_method, - comparison.positive_samples, - comparison.negative_samples, - comparison.route_success_rate, - ) - .execute(&mut *tx) - .await?; - } - - tx.commit().await + // Use bulk insert with sqlx::QueryBuilder for better performance + let mut query_builder = sqlx::QueryBuilder::new( + "INSERT INTO simulated_performance_comparisons (simulated_epoch_id, node_id, node_type, performance_score, work_factor, calculation_method, positive_samples, negative_samples, route_success_rate) " + ); + + query_builder.push_values(performance_data, |mut b, comparison| { + b.push_bind(comparison.simulated_epoch_id) + .push_bind(comparison.node_id) + .push_bind(&comparison.node_type) + .push_bind(comparison.performance_score) + .push_bind(comparison.work_factor) + .push_bind(&comparison.calculation_method) + .push_bind(comparison.positive_samples) + .push_bind(comparison.negative_samples) + .push_bind(comparison.route_success_rate); + }); + + query_builder.build().execute(&self.connection_pool).await?; + Ok(()) + } + + /// Inserts performance rankings for simulation + pub(crate) async fn insert_simulated_performance_rankings( + &self, + rankings: &[SimulatedPerformanceRanking], + ) -> Result<(), sqlx::Error> { + if rankings.is_empty() { + return Ok(()); } - /// Inserts performance rankings for simulation - pub(crate) async fn insert_simulated_performance_rankings( - &self, - rankings: &[SimulatedPerformanceRanking], - ) -> Result<(), sqlx::Error> { - let mut tx = self.connection_pool.begin().await?; - - for ranking in rankings { - sqlx::query!( - r#" - INSERT INTO simulated_performance_rankings - (simulated_epoch_id, node_id, calculation_method, performance_rank, - performance_percentile) - VALUES (?, ?, ?, ?, ?) - "#, - ranking.simulated_epoch_id, - ranking.node_id, - ranking.calculation_method, - ranking.performance_rank, - ranking.performance_percentile, - ) - .execute(&mut *tx) - .await?; - } - - tx.commit().await - } - - /// Inserts route analysis metadata for a simulation - pub(crate) async fn insert_simulated_route_analysis( - &self, - analysis: &SimulatedRouteAnalysis, - ) -> Result<(), sqlx::Error> { - sqlx::query!( + // Use bulk insert with sqlx::QueryBuilder for better performance + let mut query_builder = sqlx::QueryBuilder::new( + "INSERT INTO simulated_performance_rankings (simulated_epoch_id, node_id, calculation_method, performance_rank, performance_percentile) " + ); + + query_builder.push_values(rankings, |mut b, ranking| { + b.push_bind(ranking.simulated_epoch_id) + .push_bind(ranking.node_id) + .push_bind(&ranking.calculation_method) + .push_bind(ranking.performance_rank) + .push_bind(ranking.performance_percentile); + }); + + query_builder.build().execute(&self.connection_pool).await?; + Ok(()) + } + + /// Inserts route analysis metadata for a simulation + pub(crate) async fn insert_simulated_route_analysis( + &self, + analysis: &SimulatedRouteAnalysis, + ) -> Result<(), sqlx::Error> { + sqlx::query!( r#" INSERT INTO simulated_route_analysis (simulated_epoch_id, calculation_method, total_routes_analyzed, successful_routes, @@ -1754,18 +1879,17 @@ pub(crate) mod v3_migration { ) .execute(&self.connection_pool) .await?; - - Ok(()) - } + Ok(()) + } - /// Retrieves node performance data for a specific simulation - pub(crate) async fn get_simulated_node_performance( - &self, - simulated_epoch_id: i64, - calculation_method: Option<&str>, - ) -> Result, sqlx::Error> { - match calculation_method { + /// Retrieves node performance data for a specific simulation + pub(crate) async fn get_simulated_node_performance( + &self, + simulated_epoch_id: i64, + calculation_method: Option<&str>, + ) -> Result, sqlx::Error> { + match calculation_method { Some(method) => { sqlx::query_as!( SimulatedNodePerformance, @@ -1802,15 +1926,15 @@ pub(crate) mod v3_migration { .await }, } - } + } - /// Retrieves performance comparison data for a specific simulation - pub(crate) async fn get_simulated_performance_comparisons( - &self, - simulated_epoch_id: i64, - calculation_method: Option<&str>, - ) -> Result, sqlx::Error> { - match calculation_method { + /// Retrieves performance comparison data for a specific simulation + pub(crate) async fn get_simulated_performance_comparisons( + &self, + simulated_epoch_id: i64, + calculation_method: Option<&str>, + ) -> Result, sqlx::Error> { + match calculation_method { Some(method) => { sqlx::query_as!( SimulatedPerformanceComparison, @@ -1849,44 +1973,126 @@ pub(crate) mod v3_migration { .await }, } + } + + /// Count simulated node performance records for a specific epoch + pub(crate) async fn count_simulated_node_performance_for_epoch( + &self, + simulated_epoch_id: i64, + ) -> Result { + let result = sqlx::query!( + "SELECT COUNT(*) as count FROM simulated_node_performance WHERE simulated_epoch_id = ?", + simulated_epoch_id + ) + .fetch_one(&self.connection_pool) + .await?; + + Ok(result.count as usize) + } + + /// Get available calculation methods for a specific epoch + pub(crate) async fn get_available_calculation_methods_for_epoch( + &self, + epoch_id: u32, + ) -> Result, sqlx::Error> { + let methods = sqlx::query!( + "SELECT DISTINCT calculation_method FROM simulated_reward_epochs WHERE epoch_id = ?", + epoch_id + ) + .fetch_all(&self.connection_pool) + .await?; + + Ok(methods.into_iter().map(|m| m.calculation_method).collect()) + } + + /// Batch version of count_simulated_node_performance_for_epoch for improved performance. + /// Returns a HashMap mapping epoch_id to node count for all requested epochs. + pub(crate) async fn count_simulated_node_performance_for_epochs_batch( + &self, + simulated_epoch_ids: &[i64], + ) -> Result, sqlx::Error> { + + if simulated_epoch_ids.is_empty() { + return Ok(std::collections::HashMap::new()); } - /// Count simulated node performance records for a specific epoch - pub(crate) async fn count_simulated_node_performance_for_epoch( - &self, - simulated_epoch_id: i64, - ) -> Result { - let result = sqlx::query!( - "SELECT COUNT(*) as count FROM simulated_node_performance WHERE simulated_epoch_id = ?", - simulated_epoch_id - ) - .fetch_one(&self.connection_pool) - .await?; - - Ok(result.count as usize) + let placeholders = simulated_epoch_ids + .iter() + .map(|_| "?") + .collect::>() + .join(", "); + let query = format!( + "SELECT simulated_epoch_id, COUNT(*) as count FROM simulated_node_performance WHERE simulated_epoch_id IN ({}) GROUP BY simulated_epoch_id", + placeholders + ); + + let mut query_builder = sqlx::query(&query); + for epoch_id in simulated_epoch_ids { + query_builder = query_builder.bind(epoch_id); } - /// Get available calculation methods for a specific epoch - pub(crate) async fn get_available_calculation_methods_for_epoch( - &self, - epoch_id: u32, - ) -> Result, sqlx::Error> { - let methods = sqlx::query!( - "SELECT DISTINCT calculation_method FROM simulated_reward_epochs WHERE epoch_id = ?", - epoch_id - ) - .fetch_all(&self.connection_pool) - .await?; - - Ok(methods.into_iter().map(|m| m.calculation_method).collect()) + let rows = query_builder.fetch_all(&self.connection_pool).await?; + + let mut result = std::collections::HashMap::new(); + for row in rows { + let epoch_id: i64 = row.try_get("simulated_epoch_id")?; + let count: i64 = row.try_get("count")?; + result.insert(epoch_id, count as usize); } - /// Get simulated reward epoch by ID - pub(crate) async fn get_simulated_reward_epoch( - &self, - id: i64, - ) -> Result, sqlx::Error> { - sqlx::query_as!( + // Initialize missing epochs with 0 count + for &epoch_id in simulated_epoch_ids { + result.entry(epoch_id).or_insert(0); + } + + Ok(result) + } + + /// Batch version of get_available_calculation_methods_for_epoch for improved performance. + /// Returns a HashMap mapping epoch_id to list of calculation methods for all requested epochs. + pub(crate) async fn get_available_calculation_methods_for_epochs_batch( + &self, + epoch_ids: &[u32], + ) -> Result>, sqlx::Error> { + if epoch_ids.is_empty() { + return Ok(std::collections::HashMap::new()); + } + + let placeholders = epoch_ids.iter().map(|_| "?").collect::>().join(", "); + let query = format!( + "SELECT epoch_id, calculation_method FROM simulated_reward_epochs WHERE epoch_id IN ({})", + placeholders + ); + + let mut query_builder = sqlx::query(&query); + for epoch_id in epoch_ids { + query_builder = query_builder.bind(epoch_id); + } + + let rows = query_builder.fetch_all(&self.connection_pool).await?; + + let mut result: std::collections::HashMap> = + std::collections::HashMap::new(); + for row in rows { + let epoch_id: u32 = row.try_get("epoch_id")?; + let method: String = row.try_get("calculation_method")?; + result.entry(epoch_id).or_default().push(method); + } + + // Initialize missing epochs with empty Vec + for &epoch_id in epoch_ids { + result.entry(epoch_id).or_default(); + } + + Ok(result) + } + + /// Get simulated reward epoch by ID + pub(crate) async fn get_simulated_reward_epoch( + &self, + id: i64, + ) -> Result, sqlx::Error> { + sqlx::query_as!( SimulatedRewardEpoch, r#" SELECT id as "id!", epoch_id as "epoch_id!: u32", calculation_method as "calculation_method!", @@ -1899,31 +2105,33 @@ pub(crate) mod v3_migration { ) .fetch_optional(&self.connection_pool) .await - } + } - /// Get simulated node performance for a specific epoch - pub(crate) async fn get_simulated_node_performance_for_epoch( - &self, - simulated_epoch_id: i64, - ) -> Result, sqlx::Error> { - self.get_simulated_node_performance(simulated_epoch_id, None).await - } + /// Get simulated node performance for a specific epoch + pub(crate) async fn get_simulated_node_performance_for_epoch( + &self, + simulated_epoch_id: i64, + ) -> Result, sqlx::Error> { + self.get_simulated_node_performance(simulated_epoch_id, None) + .await + } - /// Get performance comparisons for a specific epoch - pub(crate) async fn get_simulated_performance_comparisons_for_epoch( - &self, - simulated_epoch_id: i64, - ) -> Result, sqlx::Error> { - self.get_simulated_performance_comparisons(simulated_epoch_id, None).await - } + /// Get performance comparisons for a specific epoch + pub(crate) async fn get_simulated_performance_comparisons_for_epoch( + &self, + simulated_epoch_id: i64, + ) -> Result, sqlx::Error> { + self.get_simulated_performance_comparisons(simulated_epoch_id, None) + .await + } - /// Get performance rankings for a specific epoch and method - pub(crate) async fn get_simulated_performance_rankings( - &self, - simulated_epoch_id: i64, - calculation_method: Option<&str>, - ) -> Result, sqlx::Error> { - match calculation_method { + /// Get performance rankings for a specific epoch and method + pub(crate) async fn get_simulated_performance_rankings( + &self, + simulated_epoch_id: i64, + calculation_method: Option<&str>, + ) -> Result, sqlx::Error> { + match calculation_method { Some(method) => { sqlx::query_as!( SimulatedPerformanceRanking, @@ -1960,14 +2168,14 @@ pub(crate) mod v3_migration { .await }, } - } - - /// Get simulated route analysis for a specific epoch - pub(crate) async fn get_simulated_route_analysis_for_epoch( - &self, - simulated_epoch_id: i64, - ) -> Result, sqlx::Error> { - sqlx::query_as!( + } + + /// Get simulated route analysis for a specific epoch + pub(crate) async fn get_simulated_route_analysis_for_epoch( + &self, + simulated_epoch_id: i64, + ) -> Result, sqlx::Error> { + sqlx::query_as!( SimulatedRouteAnalysis, r#" SELECT id as "id!", simulated_epoch_id as "simulated_epoch_id!", @@ -1983,15 +2191,15 @@ pub(crate) mod v3_migration { ) .fetch_all(&self.connection_pool) .await - } + } - /// Get simulated node performance by method - pub(crate) async fn get_simulated_node_performance_by_method( - &self, - epoch_id: u32, - method: &str, - ) -> Result, sqlx::Error> { - sqlx::query_as!( + /// Get simulated node performance by method + pub(crate) async fn get_simulated_node_performance_by_method( + &self, + epoch_id: u32, + method: &str, + ) -> Result, sqlx::Error> { + sqlx::query_as!( SimulatedNodePerformance, r#" SELECT snp.id as "id!", snp.simulated_epoch_id as "simulated_epoch_id!", @@ -2009,15 +2217,15 @@ pub(crate) mod v3_migration { ) .fetch_all(&self.connection_pool) .await - } + } - /// Get simulated route analysis by method - pub(crate) async fn get_simulated_route_analysis_by_method( - &self, - epoch_id: u32, - method: &str, - ) -> Result, sqlx::Error> { - sqlx::query_as!( + /// Get simulated route analysis by method + pub(crate) async fn get_simulated_route_analysis_by_method( + &self, + epoch_id: u32, + method: &str, + ) -> Result, sqlx::Error> { + sqlx::query_as!( SimulatedRouteAnalysis, r#" SELECT sra.id as "id!", sra.simulated_epoch_id as "simulated_epoch_id!", @@ -2034,14 +2242,14 @@ pub(crate) mod v3_migration { ) .fetch_optional(&self.connection_pool) .await - } + } - /// Get node performance history across simulation epochs - pub(crate) async fn get_simulated_node_performance_history( - &self, - node_id: NodeId, - ) -> Result, sqlx::Error> { - sqlx::query_as!( + /// Get node performance history across simulation epochs + pub(crate) async fn get_simulated_node_performance_history( + &self, + node_id: NodeId, + ) -> Result, sqlx::Error> { + sqlx::query_as!( SimulatedNodePerformance, r#" SELECT snp.id as "id!", snp.simulated_epoch_id as "simulated_epoch_id!", @@ -2058,6 +2266,5 @@ pub(crate) mod v3_migration { ) .fetch_all(&self.connection_pool) .await - } } } diff --git a/nym-api/src/support/storage/mod.rs b/nym-api/src/support/storage/mod.rs index 0aa231d740..5f0b50cd83 100644 --- a/nym-api/src/support/storage/mod.rs +++ b/nym-api/src/support/storage/mod.rs @@ -1050,6 +1050,7 @@ pub(crate) mod v3_migration { } /// Get mixnode identity key by node ID + #[allow(dead_code)] pub(crate) async fn get_mixnode_identity_key( &self, mix_id: NodeId, @@ -1058,6 +1059,7 @@ pub(crate) mod v3_migration { } /// Get gateway identity key by node ID + #[allow(dead_code)] pub(crate) async fn get_gateway_identity_key( &self, node_id: NodeId,