queries for detailed node statuses with broken pagination

2024-05-17 17:02:29 +01:00
parent e5f41731ae
commit 89e34b4fd3
10 changed files with 611 additions and 15 deletions
@@ -53,6 +53,8 @@ pub(crate) fn node_status_routes(
            routes::get_active_set_detailed,
            routes::get_gateways_detailed,
            routes::get_gateways_detailed_unfiltered,
+            routes::unstable::mixnode_test_results,
+            routes::unstable::gateway_test_results,
        ]
    } else {
        // in the minimal variant we would not have access to endpoints relying on existence
@@ -1,6 +1,19 @@
-// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only

+use nym_api_requests::models::{
+    AllInclusionProbabilitiesResponse, ComputeRewardEstParam, GatewayBondAnnotated,
+    GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
+    GatewayUptimeResponse, InclusionProbabilityResponse, MixNodeBondAnnotated,
+    MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
+    MixnodeUptimeHistoryResponse, RewardEstimationResponse, StakeSaturationResponse,
+    UptimeResponse,
+};
+use nym_mixnet_contract_common::MixId;
+use rocket::serde::json::Json;
+use rocket::State;
+use rocket_okapi::openapi;
+
 use super::helpers::_get_gateways_detailed;
 use super::NodeStatusCache;
 use crate::node_status_api::helpers::{
@@ -15,18 +28,6 @@ use crate::node_status_api::helpers::{
 use crate::node_status_api::models::ErrorResponse;
 use crate::storage::NymApiStorage;
 use crate::NymContractCache;
-use nym_api_requests::models::{
-    AllInclusionProbabilitiesResponse, ComputeRewardEstParam, GatewayBondAnnotated,
-    GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
-    GatewayUptimeResponse, InclusionProbabilityResponse, MixNodeBondAnnotated,
-    MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
-    MixnodeUptimeHistoryResponse, RewardEstimationResponse, StakeSaturationResponse,
-    UptimeResponse,
-};
-use nym_mixnet_contract_common::MixId;
-use rocket::serde::json::Json;
-use rocket::State;
-use rocket_okapi::openapi;

 #[openapi(tag = "status")]
 #[get("/gateway/<identity>/report")]
@@ -227,3 +228,287 @@ pub async fn get_gateways_detailed_unfiltered(
 ) -> Json<Vec<GatewayBondAnnotated>> {
    Json(_get_gateways_detailed_unfiltered(cache).await)
 }
+
+pub mod unstable {
+    use crate::node_status_api::models::ErrorResponse;
+    use crate::support::http::helpers::PaginationRequest;
+    use crate::support::storage::NymApiStorage;
+    use nym_api_requests::models::{
+        GatewayTestResultResponse, MixnodeTestResultResponse, TestNode, TestResult, TestRoute,
+    };
+    use nym_api_requests::pagination::Pagination;
+    use nym_mixnet_contract_common::MixId;
+    use rocket::http::Status;
+    use rocket::serde::json::Json;
+    use rocket::State;
+    use rocket_okapi::openapi;
+    use std::cmp::min;
+    use std::collections::HashMap;
+    use std::sync::Arc;
+    use tokio::sync::RwLock;
+
+    pub type DbId = i64;
+
+    // a simply in-memory cache of node details
+    #[derive(Debug, Default)]
+    pub struct NodeInfoCache {
+        inner: Arc<RwLock<NodeInfoCacheInner>>,
+    }
+
+    impl NodeInfoCache {
+        async fn get_mix_node_details(&self, db_id: DbId, storage: &NymApiStorage) -> TestNode {
+            {
+                let read_guard = self.inner.read().await;
+                if let Some(cached) = read_guard.mixnodes.get(&db_id) {
+                    trace!("cache hit for mixnode {db_id}");
+                    return cached.clone();
+                }
+            }
+            trace!("cache miss for mixnode {db_id}");
+
+            let mut write_guard = self.inner.write().await;
+            // double-check the cache in case somebody already updated it while we were waiting for the lock
+            if let Some(cached) = write_guard.mixnodes.get(&db_id) {
+                return cached.clone();
+            }
+
+            let details = match storage.get_mixnode_details_by_db_id(db_id).await {
+                Ok(Some(details)) => details.into(),
+                Ok(None) => {
+                    error!("somebody has been messing with the database! details for mixnode with database id {db_id} have been removed!");
+                    TestNode::default()
+                }
+                Err(err) => {
+                    // don't insert into the cache in case another request is successful
+                    error!("failed to retrieve details for mixnode {db_id}: {err}");
+                    return TestNode::default();
+                }
+            };
+
+            write_guard.mixnodes.insert(db_id, details.clone());
+            details
+        }
+
+        async fn get_gateway_details(&self, db_id: DbId, storage: &NymApiStorage) -> TestNode {
+            {
+                let read_guard = self.inner.read().await;
+                if let Some(cached) = read_guard.gateways.get(&db_id) {
+                    trace!("cache hit for gateway {db_id}");
+                    return cached.clone();
+                }
+            }
+            trace!("cache miss for gateway {db_id}");
+
+            let mut write_guard = self.inner.write().await;
+            // double-check the cache in case somebody already updated it while we were waiting for the lock
+            if let Some(cached) = write_guard.gateways.get(&db_id) {
+                return cached.clone();
+            }
+
+            let details = match storage.get_gateway_details_by_db_id(db_id).await {
+                Ok(Some(details)) => details.into(),
+                Ok(None) => {
+                    error!("somebody has been messing with the database! details for gateway with database id {db_id} have been removed!");
+                    TestNode::default()
+                }
+                Err(err) => {
+                    // don't insert into the cache in case another request is successful
+                    error!("failed to retrieve details for gateway {db_id}: {err}");
+                    return TestNode::default();
+                }
+            };
+
+            write_guard.gateways.insert(db_id, details.clone());
+            details
+        }
+    }
+
+    #[derive(Debug, Default)]
+    struct NodeInfoCacheInner {
+        mixnodes: HashMap<DbId, TestNode>,
+        gateways: HashMap<DbId, TestNode>,
+    }
+
+    const MAX_TEST_RESULTS_PAGE_SIZE: u32 = 100;
+    const DEFAULT_TEST_RESULTS_PAGE_SIZE: u32 = 50;
+
+    async fn _mixnode_test_results(
+        mix_id: MixId,
+        page: u32,
+        per_page: u32,
+        info_cache: &State<NodeInfoCache>,
+        storage: &State<NymApiStorage>,
+    ) -> anyhow::Result<MixnodeTestResultResponse> {
+        // convert to db offset
+        // we're paging from page 0 like civilised people,
+        // so we have to skip (page * per_page) results
+        let offset = page * per_page;
+        let limit = per_page;
+
+        let raw_results = storage
+            .get_mixnode_detailed_statuses(mix_id, limit, offset)
+            .await?;
+        let total = match raw_results.first() {
+            None => 0,
+            Some(r) => storage.get_mixnode_detailed_statuses_count(r.db_id).await?,
+        };
+
+        let mut test_results = HashMap::new();
+
+        for result in raw_results {
+            let entry = test_results
+                .entry(result.monitor_run_id)
+                .or_insert(TestResult {
+                    monitor_run_id: result.monitor_run_id,
+                    timestamp: result.timestamp,
+                    reliability: result.reliability,
+                    test_routes: vec![],
+                });
+
+            let gateway = info_cache
+                .get_gateway_details(result.gateway_id, storage)
+                .await;
+            let layer1 = info_cache
+                .get_mix_node_details(result.layer1_mix_id, storage)
+                .await;
+            let layer2 = info_cache
+                .get_mix_node_details(result.layer2_mix_id, storage)
+                .await;
+            let layer3 = info_cache
+                .get_mix_node_details(result.layer3_mix_id, storage)
+                .await;
+
+            entry.test_routes.push(TestRoute {
+                gateway,
+                layer1,
+                layer2,
+                layer3,
+            })
+        }
+
+        Ok(MixnodeTestResultResponse {
+            pagination: Pagination {
+                total,
+                page,
+                size: test_results.len(),
+            },
+            data: test_results.into_values().collect(),
+        })
+    }
+
+    #[openapi(tag = "UNSTABLE - DO **NOT** USE")]
+    #[get("/mixnodes/<mix_id>/test-results?<pagination..>")]
+    pub async fn mixnode_test_results(
+        mix_id: MixId,
+        pagination: PaginationRequest,
+        info_cache: &State<NodeInfoCache>,
+        storage: &State<NymApiStorage>,
+    ) -> Result<Json<MixnodeTestResultResponse>, ErrorResponse> {
+        let page = pagination.page.unwrap_or_default();
+        let per_page = min(
+            pagination
+                .per_page
+                .unwrap_or(DEFAULT_TEST_RESULTS_PAGE_SIZE),
+            MAX_TEST_RESULTS_PAGE_SIZE,
+        );
+
+        match _mixnode_test_results(mix_id, page, per_page, info_cache, storage).await {
+            Ok(res) => Ok(Json(res)),
+            Err(err) => Err(ErrorResponse::new(
+                format!("failed to retrieve mixnode test results for node {mix_id}: {err}"),
+                Status::InternalServerError,
+            )),
+        }
+    }
+
+    async fn _gateway_test_results(
+        gateway_identity: &str,
+        page: u32,
+        per_page: u32,
+        info_cache: &State<NodeInfoCache>,
+        storage: &State<NymApiStorage>,
+    ) -> anyhow::Result<GatewayTestResultResponse> {
+        // convert to db offset
+        // we're paging from page 0 like civilised people,
+        // so we have to skip (page * per_page) results
+        let offset = page * per_page;
+        let limit = per_page;
+
+        let raw_results = storage
+            .get_gateway_detailed_statuses(gateway_identity, limit, offset)
+            .await?;
+        let total = match raw_results.first() {
+            None => 0,
+            Some(r) => storage.get_gateway_detailed_statuses_count(r.db_id).await?,
+        };
+
+        let mut test_results = HashMap::new();
+
+        for result in raw_results {
+            let entry = test_results
+                .entry(result.monitor_run_id)
+                .or_insert(TestResult {
+                    monitor_run_id: result.monitor_run_id,
+                    timestamp: result.timestamp,
+                    reliability: result.reliability,
+                    test_routes: vec![],
+                });
+
+            let gateway = info_cache
+                .get_gateway_details(result.gateway_id, storage)
+                .await;
+            let layer1 = info_cache
+                .get_mix_node_details(result.layer1_mix_id, storage)
+                .await;
+            let layer2 = info_cache
+                .get_mix_node_details(result.layer2_mix_id, storage)
+                .await;
+            let layer3 = info_cache
+                .get_mix_node_details(result.layer3_mix_id, storage)
+                .await;
+
+            entry.test_routes.push(TestRoute {
+                gateway,
+                layer1,
+                layer2,
+                layer3,
+            })
+        }
+
+        Ok(GatewayTestResultResponse {
+            pagination: Pagination {
+                total,
+                page,
+                size: test_results.len(),
+            },
+            data: test_results.into_values().collect(),
+        })
+    }
+
+    #[openapi(tag = "UNSTABLE - DO **NOT** USE")]
+    #[get("/gateways/<gateway_identity>/test-results?<pagination..>")]
+    pub async fn gateway_test_results(
+        gateway_identity: &str,
+        pagination: PaginationRequest,
+        info_cache: &State<NodeInfoCache>,
+        storage: &State<NymApiStorage>,
+    ) -> Result<Json<GatewayTestResultResponse>, ErrorResponse> {
+        let page = pagination.page.unwrap_or_default();
+        let per_page = min(
+            pagination
+                .per_page
+                .unwrap_or(DEFAULT_TEST_RESULTS_PAGE_SIZE),
+            MAX_TEST_RESULTS_PAGE_SIZE,
+        );
+
+        match _gateway_test_results(gateway_identity, page, per_page, info_cache, storage).await {
+            Ok(res) => Ok(Json(res)),
+            Err(err) => Err(ErrorResponse::new(
+                format!(
+                    "failed to retrieve mixnode test results for gateway {gateway_identity}: {err}"
+                ),
+                Status::InternalServerError,
+            )),
+        }
+    }
+}