diff --git a/Cargo.lock b/Cargo.lock index 82ab2263c8..bcf25449f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4733,7 +4733,7 @@ checksum = "830b246a0e5f20af87141b25c173cd1b609bd7779a4617d6ec582abaf90870f3" [[package]] name = "nym-api" -version = "1.1.56" +version = "1.1.55" dependencies = [ "anyhow", "async-trait", @@ -6252,7 +6252,7 @@ dependencies = [ [[package]] name = "nym-node-status-api" -version = "2.2.0" +version = "2.3.0" dependencies = [ "ammonia", "anyhow", diff --git a/nym-node-status-api/nym-node-status-api/Cargo.toml b/nym-node-status-api/nym-node-status-api/Cargo.toml index 6ec2945e8f..587415579e 100644 --- a/nym-node-status-api/nym-node-status-api/Cargo.toml +++ b/nym-node-status-api/nym-node-status-api/Cargo.toml @@ -3,7 +3,7 @@ [package] name = "nym-node-status-api" -version = "2.2.0" +version = "2.3.0" authors.workspace = true repository.workspace = true homepage.workspace = true diff --git a/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs b/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs index 3cc221b15b..eebff95caf 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/api/nym_nodes.rs @@ -1,18 +1,26 @@ use axum::{ - extract::{Query, State}, + extract::{Path, Query, State}, Json, Router, }; +use nym_validator_client::client::NodeId; +use serde::Deserialize; use tracing::instrument; +use utoipa::IntoParams; use crate::http::{ error::{HttpError, HttpResult}, - models::ExtendedNymNode, + models::{ExtendedNymNode, NodeDelegation}, state::AppState, PagedResult, Pagination, }; pub(crate) fn routes() -> Router { - Router::new().route("/", axum::routing::get(nym_nodes)) + Router::new() + .route("/", axum::routing::get(nym_nodes)) + .route( + "/:node_id/delegations", + axum::routing::get(node_delegations), + ) } #[utoipa::path( @@ -27,7 +35,7 @@ pub(crate) fn routes() -> Router { (status = 200, body = PagedResult) ) )] -#[instrument(level = tracing::Level::DEBUG, skip_all, fields(page=pagination.page, size=pagination.size))] +#[instrument(level = tracing::Level::INFO, skip_all, fields(page=pagination.page, size=pagination.size))] async fn nym_nodes( Query(pagination): Query, State(state): State, @@ -46,3 +54,35 @@ async fn nym_nodes( Ok(Json(PagedResult::paginate(pagination, nodes))) } + +#[allow(dead_code)] // clippy doesn't detect usage in utoipa macros +#[derive(Deserialize, IntoParams)] +#[into_params(parameter_in = Path)] +struct NodeIdParam { + #[param(minimum = 0)] + node_id: NodeId, +} + +#[utoipa::path( + tag = "Nym Explorer", + get, + params( + NodeIdParam + ), + path = "/{node_id}/delegations", + context_path = "/explorer/v3/nym-nodes", + responses( + (status = 200, body = NodeDelegation) + ) +)] +#[instrument(level = tracing::Level::INFO, skip(state))] +async fn node_delegations( + Path(node_id): Path, + State(state): State, +) -> HttpResult>> { + state + .node_delegations(node_id) + .await + .ok_or_else(|| HttpError::no_delegations_for_node(node_id)) + .map(Json) +} diff --git a/nym-node-status-api/nym-node-status-api/src/http/error.rs b/nym-node-status-api/nym-node-status-api/src/http/error.rs index ed37966742..e36dc19125 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/error.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/error.rs @@ -1,3 +1,4 @@ +use nym_mixnet_contract_common::NodeId; use std::fmt::Display; pub(crate) type HttpResult = Result; @@ -40,6 +41,14 @@ impl HttpError { status: axum::http::StatusCode::SERVICE_UNAVAILABLE, } } + + pub(crate) fn no_delegations_for_node(node_id: NodeId) -> Self { + Self { + message: serde_json::json!({"message": format!("No delegation data for node_id={}", node_id)}) + .to_string(), + status: axum::http::StatusCode::NOT_FOUND, + } + } } impl axum::response::IntoResponse for HttpError { diff --git a/nym-node-status-api/nym-node-status-api/src/http/models.rs b/nym-node-status-api/nym-node-status-api/src/http/models.rs index be9aac0716..5633c6e1a6 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/models.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/models.rs @@ -1,4 +1,5 @@ -use cosmwasm_std::Decimal; +use cosmwasm_std::{Addr, Coin, Decimal}; +use nym_mixnet_contract_common::CoinSchema; use nym_node_requests::api::v1::node::models::NodeDescription; use nym_validator_client::client::NodeId; use serde::{Deserialize, Serialize}; @@ -120,3 +121,27 @@ pub struct SessionStats { pub mixnet_sessions: Option, pub unknown_sessions: Option, } + +#[derive(Debug, Clone, Deserialize, Serialize, ToSchema)] +pub struct NodeDelegation { + #[schema(value_type = CoinSchema)] + pub amount: Coin, + pub cumulative_reward_ratio: String, + pub block_height: u64, + #[schema(value_type = String)] + pub owner: Addr, + #[schema(value_type = Option)] + pub proxy: Option, +} + +impl From for NodeDelegation { + fn from(value: nym_mixnet_contract_common::Delegation) -> Self { + Self { + amount: value.amount, + cumulative_reward_ratio: value.cumulative_reward_ratio.to_string(), + block_height: value.height, + owner: value.owner, + proxy: value.proxy, + } + } +} diff --git a/nym-node-status-api/nym-node-status-api/src/http/server.rs b/nym-node-status-api/nym-node-status-api/src/http/server.rs index ce5abe2234..36c55a787a 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/server.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/server.rs @@ -1,13 +1,14 @@ use axum::Router; use core::net::SocketAddr; use nym_crypto::asymmetric::ed25519::PublicKey; -use tokio::{net::TcpListener, task::JoinHandle}; +use std::sync::Arc; +use tokio::{net::TcpListener, sync::RwLock, task::JoinHandle}; use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned}; use crate::{ db::DbPool, http::{api::RouterBuilder, state::AppState}, - monitor::NodeGeoCache, + monitor::{DelegationsCache, NodeGeoCache}, }; /// Return handles that allow for graceful shutdown of server + awaiting its @@ -19,6 +20,7 @@ pub(crate) async fn start_http_api( agent_key_list: Vec, agent_max_count: i64, node_geocache: NodeGeoCache, + node_delegations: Arc>, ) -> anyhow::Result { let router_builder = RouterBuilder::with_default_routes(); @@ -28,6 +30,7 @@ pub(crate) async fn start_http_api( agent_key_list, agent_max_count, node_geocache, + node_delegations, ) .await; let router = router_builder.with_state(state); diff --git a/nym-node-status-api/nym-node-status-api/src/http/state.rs b/nym-node-status-api/nym-node-status-api/src/http/state.rs index fdc58b385d..c2ebfafd15 100644 --- a/nym-node-status-api/nym-node-status-api/src/http/state.rs +++ b/nym-node-status-api/nym-node-status-api/src/http/state.rs @@ -1,17 +1,17 @@ -use std::{collections::HashMap, sync::Arc, time::Duration}; - use cosmwasm_std::Decimal; use moka::{future::Cache, Entry}; use nym_contracts_common::NaiveFloat; use nym_crypto::asymmetric::ed25519::PublicKey; +use nym_mixnet_contract_common::NodeId; use nym_validator_client::nym_api::SkimmedNode; +use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::sync::RwLock; use tracing::instrument; use crate::{ db::{queries, DbPool}, http::models::{DailyStats, ExtendedNymNode, Gateway, Mixnode, NodeGeoData, SummaryHistory}, - monitor::NodeGeoCache, + monitor::{DelegationsCache, NodeGeoCache}, }; use super::models::SessionStats; @@ -23,6 +23,7 @@ pub(crate) struct AppState { agent_key_list: Vec, agent_max_count: i64, node_geocache: NodeGeoCache, + node_delegations: Arc>, } impl AppState { @@ -32,6 +33,7 @@ impl AppState { agent_key_list: Vec, agent_max_count: i64, node_geocache: NodeGeoCache, + node_delegations: Arc>, ) -> Self { Self { db_pool, @@ -39,6 +41,7 @@ impl AppState { agent_key_list, agent_max_count, node_geocache, + node_delegations, } } @@ -61,6 +64,16 @@ impl AppState { pub(crate) fn node_geocache(&self) -> NodeGeoCache { self.node_geocache.clone() } + + pub(crate) async fn node_delegations( + &self, + node_id: NodeId, + ) -> Option> { + self.node_delegations + .read() + .await + .delegations_owned(node_id) + } } static GATEWAYS_LIST_KEY: &str = "gateways"; diff --git a/nym-node-status-api/nym-node-status-api/src/main.rs b/nym-node-status-api/nym-node-status-api/src/main.rs index 65128d8932..f0bd3b5b46 100644 --- a/nym-node-status-api/nym-node-status-api/src/main.rs +++ b/nym-node-status-api/nym-node-status-api/src/main.rs @@ -1,6 +1,9 @@ +use crate::monitor::DelegationsCache; use clap::Parser; use nym_crypto::asymmetric::ed25519::PublicKey; use nym_task::signal::wait_for_signal; +use nym_validator_client::nyxd::NyxdClient; +use std::sync::Arc; mod cli; mod db; @@ -41,19 +44,27 @@ async fn main() -> anyhow::Result<()> { let geocache = moka::future::Cache::builder() .time_to_live(args.geodata_ttl) .build(); + let delegations_cache = DelegationsCache::new(); // Start the monitor let args_clone = args.clone(); let geocache_clone = geocache.clone(); + let delegations_cache_clone = Arc::clone(&delegations_cache); + let config = nym_validator_client::nyxd::Config::try_from_nym_network_details( + &nym_network_defaults::NymNetworkDetails::new_from_env(), + )?; + let nyxd_client = NyxdClient::connect(config, args.nyxd_addr.as_str()) + .map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?; tokio::spawn(async move { monitor::spawn_in_background( db_pool, args_clone.nym_api_client_timeout, - args_clone.nyxd_addr, + nyxd_client, args_clone.monitor_refresh_interval, args_clone.ipinfo_api_token, geocache_clone, + delegations_cache_clone, ) .await; tracing::info!("Started monitor task"); @@ -74,6 +85,7 @@ async fn main() -> anyhow::Result<()> { agent_key_list.to_owned(), args.max_agent_count, geocache, + delegations_cache, ) .await .expect("Failed to start server"); diff --git a/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs b/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs index af70d01170..77d7199060 100644 --- a/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs +++ b/nym-node-status-api/nym-node-status-api/src/monitor/mod.rs @@ -12,25 +12,30 @@ use crate::utils::{decimal_to_i64, LogError, NumericalCheckedCast}; use anyhow::anyhow; use moka::future::Cache; use nym_network_defaults::NymNetworkDetails; -use nym_validator_client::client::{NodeId, NymApiClientExt, NymNodeDetails}; -use nym_validator_client::models::{ - LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription, +use nym_validator_client::{ + client::{NodeId, NymApiClientExt, NymNodeDetails}, + models::{LegacyDescribedMixNode, MixNodeBondAnnotated, NymNodeDescription}, }; -use nym_validator_client::nym_nodes::{NodeRole, SkimmedNode}; -use nym_validator_client::nyxd::contract_traits::PagedMixnetQueryClient; -use nym_validator_client::nyxd::{AccountId, NyxdClient}; -use nym_validator_client::NymApiClient; -use reqwest::Url; -use std::collections::{HashMap, HashSet}; -use std::str::FromStr; -use tokio::time::Duration; +use nym_validator_client::{ + nym_nodes::{NodeRole, SkimmedNode}, + nyxd::{contract_traits::PagedMixnetQueryClient, AccountId}, + NymApiClient, QueryHttpRpcNyxdClient, +}; +use std::{ + collections::{HashMap, HashSet}, + str::FromStr, + sync::Arc, +}; +use tokio::{sync::RwLock, time::Duration}; use tracing::instrument; pub(crate) use geodata::IpInfoClient; +pub(crate) use node_delegations::DelegationsCache; mod geodata; +mod node_delegations; -const FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); +const MONITOR_FAILURE_RETRY_DELAY: Duration = Duration::from_secs(60); static DELEGATION_PROGRAM_WALLET: &str = "n1rnxpdpx3kldygsklfft0gech7fhfcux4zst5lw"; pub(crate) type NodeGeoCache = Cache; @@ -38,9 +43,10 @@ struct Monitor { db_pool: DbPool, network_details: NymNetworkDetails, nym_api_client_timeout: Duration, - nyxd_addr: Url, + nyxd_client: QueryHttpRpcNyxdClient, ipinfo: IpInfoClient, geocache: NodeGeoCache, + node_delegations: Arc>, } // TODO dz: query many NYM APIs: @@ -49,19 +55,22 @@ struct Monitor { pub(crate) async fn spawn_in_background( db_pool: DbPool, nym_api_client_timeout: Duration, - nyxd_addr: Url, + nyxd_client: nym_validator_client::QueryHttpRpcNyxdClient, refresh_interval: Duration, ipinfo_api_token: String, geocache: NodeGeoCache, + node_delegations: Arc>, ) { let ipinfo = IpInfoClient::new(ipinfo_api_token.clone()); + let mut monitor = Monitor { db_pool, network_details: nym_network_defaults::NymNetworkDetails::new_from_env(), nym_api_client_timeout, - nyxd_addr, + nyxd_client, ipinfo, geocache, + node_delegations, }; loop { @@ -70,10 +79,9 @@ pub(crate) async fn spawn_in_background( if let Err(e) = monitor.run().await { tracing::error!( "Monitor run failed: {e}, retrying in {}s...", - FAILURE_RETRY_DELAY.as_secs() + MONITOR_FAILURE_RETRY_DELAY.as_secs() ); - // TODO dz implement some sort of backoff - tokio::time::sleep(FAILURE_RETRY_DELAY).await; + tokio::time::sleep(MONITOR_FAILURE_RETRY_DELAY).await; } else { tracing::info!( "Info successfully collected, sleeping for {}s...", @@ -193,8 +201,7 @@ impl Monitor { .nodes .data; - let delegation_program_members = - get_delegation_program_details(&self.network_details, &self.nyxd_addr).await?; + let delegation_program_members = self.get_delegation_program_details().await?; // keep stats for later let assigned_entry_count = nym_nodes @@ -233,7 +240,9 @@ impl Monitor { tracing::debug!("{} mixnode info written to DB!", mixnodes_count); })?; - let (all_historical_gateways, all_historical_mixnodes) = calculate_stats(&pool).await?; + self.refresh_node_delegations(&bonded_nym_nodes).await; + + let (all_historical_gateways, all_historical_mixnodes) = historical_count(&pool).await?; // // write summary keys and values to table @@ -457,9 +466,34 @@ impl Monitor { } } } + + #[instrument(level = "info", skip_all)] + async fn refresh_node_delegations(&mut self, bonded_nodes: &HashMap) { + let delegations_per_node = node_delegations::refresh(&self.nyxd_client, bonded_nodes).await; + + // update after refreshing all to avoid holding write lock for too long + *self.node_delegations.write().await = delegations_per_node; + } + + async fn get_delegation_program_details(&self) -> anyhow::Result> { + let account_id = AccountId::from_str(DELEGATION_PROGRAM_WALLET) + .map_err(|e| anyhow!("Invalid bech32 address: {}", e))?; + + let delegations = self + .nyxd_client + .get_all_delegator_delegations(&account_id) + .await?; + + let mix_ids: Vec = delegations + .iter() + .map(|delegation| delegation.node_id) + .collect(); + + Ok(mix_ids) + } } -async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> { +async fn historical_count(pool: &DbPool) -> anyhow::Result<(usize, usize)> { let mut conn = pool.acquire().await?; let all_historical_gateways = sqlx::query_scalar!(r#"SELECT count(id) FROM gateways"#) @@ -474,25 +508,3 @@ async fn calculate_stats(pool: &DbPool) -> anyhow::Result<(usize, usize)> { Ok((all_historical_gateways, all_historical_mixnodes)) } - -async fn get_delegation_program_details( - network_details: &NymNetworkDetails, - nyxd_addr: &Url, -) -> anyhow::Result> { - let config = nym_validator_client::nyxd::Config::try_from_nym_network_details(network_details)?; - - let client = NyxdClient::connect(config, nyxd_addr.as_str()) - .map_err(|err| anyhow::anyhow!("Couldn't connect: {}", err))?; - - let account_id = AccountId::from_str(DELEGATION_PROGRAM_WALLET) - .map_err(|e| anyhow!("Invalid bech32 address: {}", e))?; - - let delegations = client.get_all_delegator_delegations(&account_id).await?; - - let mix_ids: Vec = delegations - .iter() - .map(|delegation| delegation.node_id) - .collect(); - - Ok(mix_ids) -} diff --git a/nym-node-status-api/nym-node-status-api/src/monitor/node_delegations.rs b/nym-node-status-api/nym-node-status-api/src/monitor/node_delegations.rs new file mode 100644 index 0000000000..66f91a6082 --- /dev/null +++ b/nym-node-status-api/nym-node-status-api/src/monitor/node_delegations.rs @@ -0,0 +1,53 @@ +use nym_mixnet_contract_common::{NodeId, NymNodeDetails}; +use nym_validator_client::{nyxd::contract_traits::PagedMixnetQueryClient, QueryHttpRpcNyxdClient}; +use std::{collections::HashMap, sync::Arc}; +use tokio::{sync::RwLock, time::Instant}; +use tracing::{info, warn}; + +// abstracts away data structure that holds delegations +#[derive(Clone, Debug)] +pub(crate) struct DelegationsCache { + pub inner: HashMap>, +} + +impl DelegationsCache { + pub(crate) fn new() -> Arc> { + let a = Self { + inner: HashMap::new(), + }; + Arc::new(RwLock::new(a)) + } + + pub(crate) fn delegations_owned( + &self, + node_id: NodeId, + ) -> Option> { + self.inner.get(&node_id).cloned() + } +} + +pub(super) async fn refresh( + client: &QueryHttpRpcNyxdClient, + bonded_nodes: &HashMap, +) -> DelegationsCache { + info!("👥 Refreshing {} node delegations...", bonded_nodes.len()); + let now = Instant::now(); + + let mut delegations_per_node = HashMap::new(); + for node_id in bonded_nodes.keys() { + if let Ok(delegations) = client + .get_all_single_mixnode_delegations(*node_id) + .await + .inspect_err(|err| warn!("Failed to get delegations for {}: {}", node_id, err)) + { + delegations_per_node + .insert(*node_id, delegations.into_iter().map(From::from).collect()); + } + } + let time_taken = Instant::now() - now; + info!("👥 Node delegations refreshed in {}s", time_taken.as_secs(),); + + DelegationsCache { + inner: delegations_per_node, + } +}