Raw route handling and reliability corrections

2025-05-15 13:10:54 +02:00
parent 9c7d79683b
commit fcffebfe45
22 changed files with 1005 additions and 816 deletions
@@ -20,7 +20,7 @@ use nym_api_requests::models::{
 };
 use nym_http_api_common::{FormattedResponse, OutputParams};
 use nym_mixnet_contract_common::NodeId;
-use nym_types::monitoring::MonitorMessage;
+use nym_types::monitoring::{MonitorMessage, MonitorResults};
 use tracing::error;

 pub(super) fn mandatory_routes() -> Router<AppState> {
@@ -33,6 +33,10 @@ pub(super) fn mandatory_routes() -> Router<AppState> {
            "/submit-node-monitoring-results",
            post(submit_node_monitoring_results),
        )
+        .route(
+            "/submit-route-monitoring-results",
+            post(submit_route_monitoring_results),
+        )
        .nest(
            "/mixnode/:mix_id",
            Router::new()
@@ -58,6 +62,53 @@ pub(super) fn mandatory_routes() -> Router<AppState> {
        )
 }

+#[utoipa::path(
+    tag = "status",
+    post,
+    path = "/v1/status/submit-route-monitoring-results",
+    responses(
+        (status = 200),
+        (status = 400, body = String, description = "TBD"),
+        (status = 403, body = String, description = "TBD"),
+        (status = 500, body = String, description = "TBD"),
+    ),
+)]
+pub(crate) async fn submit_route_monitoring_results(
+    State(state): State<AppState>,
+    Json(message): Json<MonitorMessage>,
+) -> AxumResult<()> {
+    if !message.is_in_allowed() {
+        return Err(AxumErrorResponse::forbidden(
+            "Monitor not registered to submit results",
+        ));
+    }
+
+    if !message.timely() {
+        return Err(AxumErrorResponse::bad_request("Message is too old"));
+    }
+
+    if !message.verify() {
+        return Err(AxumErrorResponse::bad_request("invalid signature"));
+    }
+
+    match message.results() {
+        MonitorResults::Route(results) => {
+            match state.storage.submit_route_monitoring_results(results).await {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    error!("failed to submit node monitoring results: {err}");
+                    Err(AxumErrorResponse::internal_msg(
+                        "failed to submit node monitoring results",
+                    ))
+                }
+            }
+        }
+        MonitorResults::Node(_results) => Err(AxumErrorResponse::bad_request(
+            "Node monitoring results not supported for this endpoint",
+        )),
+    }
+}
+
 #[utoipa::path(
    tag = "status",
    post,
@@ -87,18 +138,21 @@ pub(crate) async fn submit_gateway_monitoring_results(
        return Err(AxumErrorResponse::bad_request("invalid signature"));
    }

-    match state
-        .storage
-        .submit_gateway_statuses_v2(message.results())
-        .await
-    {
-        Ok(_) => Ok(()),
-        Err(err) => {
-            error!("failed to submit gateway monitoring results: {err}");
-            Err(AxumErrorResponse::internal_msg(
-                "failed to submit gateway monitoring results",
-            ))
+    match message.results() {
+        MonitorResults::Node(results) => {
+            match state.storage.submit_gateway_statuses_v2(results).await {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    error!("failed to submit node monitoring results: {err}");
+                    Err(AxumErrorResponse::internal_msg(
+                        "failed to submit node monitoring results",
+                    ))
+                }
+            }
        }
+        MonitorResults::Route(_results) => Err(AxumErrorResponse::bad_request(
+            "Gateway monitoring results not supported for this endpoint",
+        )),
    }
 }

@@ -131,18 +185,21 @@ pub(crate) async fn submit_node_monitoring_results(
        return Err(AxumErrorResponse::bad_request("invalid signature"));
    }

-    match state
-        .storage
-        .submit_mixnode_statuses_v2(message.results())
-        .await
-    {
-        Ok(_) => Ok(()),
-        Err(err) => {
-            error!("failed to submit node monitoring results: {err}");
-            Err(AxumErrorResponse::internal_msg(
-                "failed to submit node monitoring results",
-            ))
+    match message.results() {
+        MonitorResults::Node(results) => {
+            match state.storage.submit_mixnode_statuses_v2(results).await {
+                Ok(_) => Ok(()),
+                Err(err) => {
+                    error!("failed to submit node monitoring results: {err}");
+                    Err(AxumErrorResponse::internal_msg(
+                        "failed to submit node monitoring results",
+                    ))
+                }
+            }
        }
+        MonitorResults::Route(_results) => Err(AxumErrorResponse::bad_request(
+            "Node monitoring results not supported for this endpoint",
+        )),
    }
 }

@@ -10,10 +10,11 @@ use crate::support::storage::models::{
 };
 use crate::support::storage::DbIdCache;
 use nym_mixnet_contract_common::{EpochId, IdentityKey, NodeId};
-use nym_types::monitoring::NodeResult;
+use nym_types::monitoring::{NodeResult, RouteResult};
 use sqlx::FromRow;
 use time::{Date, OffsetDateTime};
 use tracing::info;
+use std::collections::HashMap;

 #[derive(Clone)]
 pub(crate) struct StorageManager {
@@ -51,6 +52,25 @@ impl AvgGatewayReliability {
    }
 }

+// Helper struct for in-memory state during calculation for an interval
+#[derive(Debug, Default, Clone)]
+struct NodeCorrectionIntervalState {
+    pos_samples: u32,
+    neg_samples: u32,
+    fail_seq: u32,
+}
+
+// Output struct for the calculated corrected reliability for an interval
+#[derive(Debug, serde::Serialize)] // Add utoipa::ToSchema if this struct is directly exposed via API
+pub struct CorrectedNodeIntervalReliability {
+    pub node_id: NodeId, // nym_mixnet_contract_common::NodeId (typically u32)
+    pub identity: Option<String>, // Base58 public key
+    pub reliability: f64,
+    pub pos_samples_in_interval: u32,
+    pub neg_samples_in_interval: u32,
+    pub final_fail_seq_in_interval: u32,
+}
+
 // all SQL goes here
 impl StorageManager {
    pub(super) async fn get_all_avg_mix_reliability_in_last_24hr(
@@ -664,6 +684,33 @@ impl StorageManager {
        tx.commit().await
    }

+    pub(super) async fn submit_route_monitoring_results(
+        &self,
+        route_results: &[RouteResult],
+    ) -> Result<(), sqlx::Error> {
+        info!("Inserting {} route monitoring results", route_results.len());
+        // insert it all in a transaction to make sure all nodes are updated at the same time
+        // (plus it's a nice guard against new nodes)
+        let mut tx = self.connection_pool.begin().await?;
+
+        for route_result in route_results {
+            sqlx::query!(
+                r#"
+                    INSERT OR IGNORE INTO routes (layer1, layer2, layer3, gw, success) VALUES (?, ?, ?, ?, ?);
+                "#,
+                route_result.layer1,
+                route_result.layer2,
+                route_result.layer3,
+                route_result.gw,
+                route_result.success,
+            )
+            .execute(&mut *tx)
+            .await?;
+        }
+
+        tx.commit().await
+    }
+
    pub(super) async fn submit_gateway_statuses_v2(
        &self,
        gateway_results: &[NodeResult],
@@ -1329,6 +1376,137 @@ impl StorageManager {
        .fetch_all(&self.connection_pool)
        .await
    }
+
+    /// Fetches raw route results from the database for a given time interval.
+    /// Assumes the 'routes' table has layer1, layer2, layer3, gw, success, and a timestamp.
+    async fn get_raw_routes_in_interval(
+        &self,
+        start_ts_secs: i64,
+        end_ts_secs: i64,
+    ) -> Result<Vec<(NodeId, NodeId, NodeId, NodeId, bool)>, sqlx::Error> {
+        // Temporary struct to match the expected columns from the 'routes' table
+        // NodeId here is assumed to be compatible with how layer1, etc. are stored (e.g. u32/i32/i64)
+        struct RawRouteData {
+            layer1: NodeId,
+            layer2: NodeId,
+            layer3: NodeId,
+            gw: NodeId,
+            success: bool,
+            // timestamp: i64, // Not explicitly selected into struct, but used in WHERE and ORDER BY
+        }
+
+        // Ensure your 'routes' table has:
+        // layer1 (NodeId type), layer2 (NodeId type), layer3 (NodeId type),
+        // gw (NodeId type), success (bool/INTEGER), timestamp (BIGINT/INTEGER)
+        // The "NodeId:" type hint for sqlx::query_as! helps map to nym_mixnet_contract_common::NodeId
+        let db_routes = sqlx::query_as!(
+            RawRouteData,
+            r#"
+            SELECT
+                layer1 as "layer1",
+                layer2 as "layer2",
+                layer3 as "layer3",
+                gw as "gw",
+                success
+            FROM routes
+            WHERE timestamp >= ? AND timestamp <= ?
+            ORDER BY timestamp ASC
+            "#,
+            start_ts_secs,
+            end_ts_secs
+        )
+        .fetch_all(&self.connection_pool)
+        .await?;
+
+        Ok(db_routes
+            .into_iter()
+            .map(|r| (r.layer1, r.layer2, r.layer3, r.gw, r.success))
+            .collect())
+    }
+
+    pub async fn calculate_corrected_node_reliabilities_for_interval(
+        &self,
+        start_ts_secs: i64,
+        end_ts_secs: i64,
+    ) -> Result<Vec<CorrectedNodeIntervalReliability>, anyhow::Error> {
+        let raw_routes = self
+            .get_raw_routes_in_interval(start_ts_secs, end_ts_secs)
+            .await?;
+
+        let mut node_states: HashMap<NodeId, NodeCorrectionIntervalState> = HashMap::new();
+
+        for (l1, l2, l3, gw, success) in raw_routes {
+            let path_node_ids = [l1, l2, l3, gw];
+
+            if success {
+                for &node_id in &path_node_ids {
+                    let state = node_states.entry(node_id).or_default();
+                    state.pos_samples += 1;
+                    state.fail_seq = 0;
+                }
+            } else {
+                // Path test failed
+                let mut current_path_node_ids_for_blame: Vec<NodeId> = Vec::new();
+                for &node_id in &path_node_ids {
+                    let state = node_states.entry(node_id).or_default();
+                    state.fail_seq += 1;
+                    current_path_node_ids_for_blame.push(node_id);
+                }
+
+                let mut guilty_nodes_in_path: Vec<NodeId> = Vec::new();
+                for &node_id in &current_path_node_ids_for_blame {
+                    if let Some(state_after_update) = node_states.get(&node_id) {
+                        if state_after_update.fail_seq > 2 {
+                            guilty_nodes_in_path.push(node_id);
+                        }
+                    }
+                }
+
+                if !guilty_nodes_in_path.is_empty() {
+                    for &guilty_node_id in &guilty_nodes_in_path {
+                        if let Some(state) = node_states.get_mut(&guilty_node_id) {
+                            state.neg_samples += 1;
+                        }
+                    }
+                } else {
+                    // No single guilty party, distribute blame
+                    for &node_id_in_path in &current_path_node_ids_for_blame {
+                        if let Some(state) = node_states.get_mut(&node_id_in_path) {
+                            state.neg_samples += 1;
+                        }
+                    }
+                }
+            }
+        }
+
+        let mut final_reliabilities = Vec::new();
+        for (node_id, state) in node_states {
+            let total_samples = state.pos_samples + state.neg_samples;
+            let reliability = if total_samples == 0 {
+                0.0 // Default for no samples in this interval. Consider Option<f64> or filtering.
+            } else {
+                state.pos_samples as f64 / total_samples as f64
+            };
+
+            // Attempt to fetch identity, first as mixnode, then as gateway if not found.
+            // This assumes get_mixnode_identity_key and get_gateway_identity_key return Result<Option<String>, sqlx::Error>
+            let mut identity: Option<String> = self.get_mixnode_identity_key(node_id).await.unwrap_or(None);
+            if identity.is_none() {
+                identity = self.get_gateway_identity_key(node_id).await.unwrap_or(None);
+            }
+
+            final_reliabilities.push(CorrectedNodeIntervalReliability {
+                node_id,
+                identity,
+                reliability,
+                pos_samples_in_interval: state.pos_samples,
+                neg_samples_in_interval: state.neg_samples,
+                final_fail_seq_in_interval: state.fail_seq,
+            });
+        }
+
+        Ok(final_reliabilities)
+    }
 }

 pub(crate) mod v3_migration {
@@ -17,7 +17,7 @@ use crate::support::storage::models::{
 };
 use dashmap::DashMap;
 use nym_mixnet_contract_common::NodeId;
-use nym_types::monitoring::NodeResult;
+use nym_types::monitoring::{NodeResult, RouteResult};
 use sqlx::sqlite::{SqliteAutoVacuum, SqliteSynchronous};
 use sqlx::ConnectOptions;
 use std::path::Path;
@@ -835,6 +835,16 @@ impl NymApiStorage {
        Ok(())
    }

+    pub(crate) async fn submit_route_monitoring_results(
+        &self,
+        route_results: &[RouteResult],
+    ) -> Result<(), NymApiStorageError> {
+        self.manager
+            .submit_route_monitoring_results(route_results)
+            .await?;
+        Ok(())
+    }
+
    /// Obtains number of network monitor test runs that have occurred within the specified interval.
    ///
    /// # Arguments