GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimize database queries by eliminating N+1 patterns in simulation system

Browse Source 
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.
This commit is contained in:
durch
2025-06-05 10:44:33 +02:00
parent a96fb098c2
commit 4f7fa557d5
7 changed files with 534 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-api/.sqlx/query-0fe3846612980738c9c0f0b955657d2300860e30920cfaea219da2c1b42ed8e8.json
Generated
-12
View File
@@ -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"
}
nym-api/.sqlx/query-c8e2c9f82521b3cc149ff169605f02a96ff0c723d45e695c17a7a422eac55916.json
Generated
-12
View File
@@ -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"
}
nym-api/.sqlx/query-e9629265275e6484b4ce2c8a9ce942d2138c00ac4b05cfe5a01eabec3cd2a9a0.json
Generated
-12
View File
@@ -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"
}
nym-api/src/epoch_operations/simulation.rs
+18 -11
View File
@@ -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<NodeId> = 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,
};
nym-api/src/simulation_api/handlers/mod.rs
+19 -11
View File
@@ -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<i64> = epochs.iter().map(|e| e.id).collect();
let epoch_ids: Vec<u32> = 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);
}
nym-api/src/support/storage/manager.rs
+495 -288
View File
@@ -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<NodeId> = 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<NodeId> = 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<NodeId> = 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<NodeId> = 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<std::collections::HashMap<NodeId, IdentityKey>, 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::<Vec<_>>().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<String> = 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<NodeId>, Vec<NodeId>), 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::<Vec<_>>().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<NodeId> = mixnode_rows
.into_iter()
.map(|row| row.try_get("mix_id"))
.collect::<Result<Vec<_>, _>>()?;
// 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<NodeId> = gateway_rows
.into_iter()
.map(|row| row.try_get("node_id"))
.collect::<Result<Vec<_>, _>>()?;
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<std::collections::HashMap<NodeId, IdentityKey>, 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::<Vec<_>>().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<String> = 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<bool, sqlx::Error> {
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<bool, sqlx::Error> {
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<Vec<GatewayDetailsBeforeMigration>, 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<Vec<GatewayDetailsBeforeMigration>, 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<i64, sqlx::Error> {
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<i64, sqlx::Error> {
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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedPerformanceComparison>, 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<Vec<SimulatedPerformanceComparison>, 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<usize, sqlx::Error> {
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<Vec<String>, 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<std::collections::HashMap<i64, usize>, 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<usize, sqlx::Error> {
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::<Vec<_>>()
.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<Vec<String>, 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<Option<SimulatedRewardEpoch>, 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<std::collections::HashMap<u32, Vec<String>>, sqlx::Error> {
if epoch_ids.is_empty() {
return Ok(std::collections::HashMap::new());
}
let placeholders = epoch_ids.iter().map(|_| "?").collect::<Vec<_>>().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<u32, Vec<String>> =
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<Option<SimulatedRewardEpoch>, 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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedPerformanceComparison>, 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<Vec<SimulatedPerformanceComparison>, 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<Vec<SimulatedPerformanceRanking>, 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<Vec<SimulatedPerformanceRanking>, 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<Vec<SimulatedRouteAnalysis>, 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<Vec<SimulatedRouteAnalysis>, 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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedNodePerformance>, 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<Option<SimulatedRouteAnalysis>, 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<Option<SimulatedRouteAnalysis>, 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<Vec<SimulatedNodePerformance>, 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<Vec<SimulatedNodePerformance>, 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
}
}
}
nym-api/src/support/storage/mod.rs
+2
View File
@@ -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,