Compare commits

..

22 Commits

Author SHA1 Message Date
benedetta davico 0d420fb0a5 remove explorer-api in workflow 2025-06-10 11:01:24 +02:00
benedettadavico fce195fdba update changelog 2025-06-10 10:28:47 +02:00
Bogdan-Ștefan Neacşu 3a528d3b89 No autoremoval of peers (#5831)
* No autoremoval

* Remove startup_timestamp
2025-06-06 12:48:34 +03:00
benedettadavico e9bb9792ab bump binaries 2025-06-04 14:42:04 +02:00
Bogdan-Ștefan Neacşu 88d4a9b111 Set cached storage counters to 0 (#5812)
* Set cached storage counters to 0

* u64 to i64 log possible error

* Check addition too
2025-06-04 12:11:46 +03:00
dynco-nym 4c67f01efb Make address cache configurable (#5784)
* Make address cache configurable

* TestFixture
2025-05-28 10:41:12 +02:00
Simon Wicky b69c2e1e94 Nym Statistics API (#5800)
* move stats types from vpn-client to here

* base stats api

* change storage schema

* add link to nymAPI for whitelisting

* remove outdated comment

* more comments update

* example of chrono vs time

* Add build.rs
- exports DATABASE_URL so cargo check works
- exports SQLX_OFFLINE for CI
- added pg_up.sh which spawns PG container
  - required for cargo sqlx prepare

* fixes time vs chrono issue and cleaner build with docker

* add correct swagger types, with feature locking where relevant

* apply dynco suggestions

---------

Co-authored-by: dynco-nym <173912580+dynco-nym@users.noreply.github.com>
2025-05-28 10:23:11 +02:00
benedetta davico d27e3b49db Merge pull request #5806 from nymtech/release/2025.10-brie 2025-05-28 09:38:36 +02:00
benedetta davico ac12455f97 add comment 2025-05-27 16:35:51 +02:00
Jędrzej Stuczyński 0b92a59f1a hack: temporarily use next.config.js instead of next.config.ts (#5805) 2025-05-27 11:41:51 +01:00
Jędrzej Stuczyński 474eff67fa chore: adjusted wallet storybook mocks to fix the build (#5804) 2025-05-27 11:38:13 +01:00
benedetta davico 1c6db86259 Merge pull request #5803 from nymtech/benny/change-rust-version
change rust version to fix ci
2025-05-27 12:11:13 +02:00
Jędrzej Stuczyński 4a1ce8154a chore: resolve 1.87 clippy warnings (#5802)
* Clippy in wallet & sdk

* Clippy in wallet

* Pin rust to 1.86 in builder

* apply changes from b7da75a18c

* missing nym-node features

* Box all the things

* additional boxes in the wallet

* post rebasing clippy

---------

Co-authored-by: dynco-nym <173912580+dynco-nym@users.noreply.github.com>
2025-05-27 11:08:36 +01:00
benedetta davico e126c1f7f1 Update publish-nym-binaries.yml 2025-05-27 11:45:53 +02:00
benedetta davico 31772019cd Update ci-contracts.yml 2025-05-27 11:44:01 +02:00
Bogdan-Ștefan Neacşu aca98ab04f Track wireguard credential retries (#5783)
* Add a cache for the credentials seen before on top-up

* Verify seen credentials on top ups

* Add warning log for timestamp subtraction

* Add unit test
2025-05-27 12:35:44 +03:00
Jędrzej Stuczyński f925c6caf0 QoL: RequestPath trait for http-api-client (#5788)
* qol: RequestPath trait for http-api-client

* additional test case

* applied the change to other trait methods
2025-05-27 10:30:13 +01:00
benedettadavico 5369e5eab9 update changelog 2025-05-27 10:03:22 +02:00
Andrej Mihajlov 2e634c59a7 Merge pull request #5801 from nymtech/am/backport-pr-5779 2025-05-26 21:03:29 +02:00
jmwample d7383d74f3 more relaxed usage of reqwest accept-encoding 2025-05-26 17:54:19 +02:00
dynco-nym e98d60d7ce Add node_bonded field to delegations (#5759)
* Add node_bonded field to delegations
- clarifies whether the delegation is to a bonded or unbonded node
- include delegations to unbonded nodes in the returned list

* PR feedback
2025-05-19 15:18:41 +02:00
benedettadavico f47650d6c8 bump binary versions 2025-05-16 13:03:37 +02:00
113 changed files with 2707 additions and 1591 deletions
+1
View File
@@ -13,6 +13,7 @@ on:
- 'nym-network-monitor/**'
- 'nym-node/**'
- 'nym-node-status-api/**'
- 'nym-statistics-api/**'
- 'nym-outfox/**'
- 'nym-validator-rewarder/**'
- 'nyx-chain-watcher/**'
+3 -1
View File
@@ -20,6 +20,7 @@ jobs:
runs-on: ubuntu-22.04
env:
CARGO_TERM_COLOR: always
RUSTUP_PERMIT_COPY_RENAME: 1
steps:
- uses: actions/checkout@v4
@@ -27,7 +28,8 @@ jobs:
uses: actions-rs/toolchain@v1
with:
profile: minimal
toolchain: stable
# pinned due to issues building contracts
toolchain: 1.86.0
target: wasm32-unknown-unknown
override: true
components: rustfmt, clippy
+1 -3
View File
@@ -52,7 +52,7 @@ jobs:
- name: Install Rust stable
uses: actions-rs/toolchain@v1
with:
toolchain: stable
toolchain: 1.86.0
override: true
- name: Build all binaries
@@ -66,7 +66,6 @@ jobs:
with:
name: my-artifact
path: |
target/release/explorer-api
target/release/nym-client
target/release/nym-socks5-client
target/release/nym-api
@@ -82,7 +81,6 @@ jobs:
if: github.event_name == 'release'
with:
files: |
target/release/explorer-api
target/release/nym-client
target/release/nym-socks5-client
target/release/nym-api
-1
View File
@@ -62,4 +62,3 @@ nym-api/redocly/formatted-openapi.json
**/settings.sql
**/enter_db.sh
CLAUDE.md
+76
View File
@@ -4,6 +4,82 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
## [Unreleased]
## [2025.11-cheddar] (2025-06-10)
- No autoremoval of peers ([#5831])
- Set cached storage counters to 0 ([#5812])
- hack: temporarily use next.config.js instead of next.config.ts ([#5805])
- chore: resolve 1.87 clippy warnings ([#5802])
- Nym Statistics API ([#5800])
- QoL: RequestPath trait for http-api-client ([#5788])
- Fix contains ticketbook function that always returned true ([#5787])
- swap a decode into a fromrow to please future postgres feature ([#5785])
- Make address cache configurable ([#5784])
- Track wireguard credential retries ([#5783])
[#5831]: https://github.com/nymtech/nym/pull/5831
[#5812]: https://github.com/nymtech/nym/pull/5812
[#5805]: https://github.com/nymtech/nym/pull/5805
[#5802]: https://github.com/nymtech/nym/pull/5802
[#5800]: https://github.com/nymtech/nym/pull/5800
[#5788]: https://github.com/nymtech/nym/pull/5788
[#5787]: https://github.com/nymtech/nym/pull/5787
[#5785]: https://github.com/nymtech/nym/pull/5785
[#5784]: https://github.com/nymtech/nym/pull/5784
[#5783]: https://github.com/nymtech/nym/pull/5783
## [2025.10-brie] (2025-05-27)
- Backport PR 5779 ([#5801])
- Expanded Accept Encoding for `reqwest` ([#5779])
- Teach HttpClientError how to report its status code and timeout ([#5770])
- Skip refreshing the topology on startup as we already have an initial set ([#5768])
- Fetch the topology from the nym-api concurrently ([#5767])
- feat: use bincode by default in NymApiClient + remove feature-lock ([#5761])
- Instrument create_request ([#5760])
- Add node_bonded field to delegations ([#5759])
- build(deps): bump mikefarah/yq from 4.45.1 to 4.45.4 ([#5758])
- Raw route submissions ([#5756])
- feat: expires header for `/active` nym-api responses ([#5755])
- Decrease default average packet delay to 15 ms ([#5754])
- build(deps): bump the patch-updates group across 1 directory with 12 updates ([#5753])
- Remove pretty_env_logger and switch remaining crates to use tracing ([#5749])
- Update pretty_env_logger to latest to not depend on unmaintained crate atty ([#5748])
- Upgrade prometheus crate to fix security warning ([#5747])
- Downgrade deranged crate to 0.4.0 ([#5746])
- feat: nym-api bincode + yaml support ([#5745])
- fix parallel feature in ecash crate with send + sync ([#5744])
- Remove old test directory - Update validator docker ([#5743])
- [Feature] `RememberMe` is the new don't `ForgetMe` ([#5742])
- build(deps): bump ammonia from 4.0.0 to 4.1.0 ([#5739])
- build(deps): bump base-x from 3.0.9 to 3.0.11 in /testnet-faucet ([#5737])
- build(deps): bump http-proxy-middleware from 2.0.8 to 2.0.9 ([#5730])
[#5801]: https://github.com/nymtech/nym/pull/5801
[#5779]: https://github.com/nymtech/nym/pull/5779
[#5770]: https://github.com/nymtech/nym/pull/5770
[#5768]: https://github.com/nymtech/nym/pull/5768
[#5767]: https://github.com/nymtech/nym/pull/5767
[#5761]: https://github.com/nymtech/nym/pull/5761
[#5760]: https://github.com/nymtech/nym/pull/5760
[#5759]: https://github.com/nymtech/nym/pull/5759
[#5758]: https://github.com/nymtech/nym/pull/5758
[#5756]: https://github.com/nymtech/nym/pull/5756
[#5755]: https://github.com/nymtech/nym/pull/5755
[#5754]: https://github.com/nymtech/nym/pull/5754
[#5753]: https://github.com/nymtech/nym/pull/5753
[#5749]: https://github.com/nymtech/nym/pull/5749
[#5748]: https://github.com/nymtech/nym/pull/5748
[#5747]: https://github.com/nymtech/nym/pull/5747
[#5746]: https://github.com/nymtech/nym/pull/5746
[#5745]: https://github.com/nymtech/nym/pull/5745
[#5744]: https://github.com/nymtech/nym/pull/5744
[#5743]: https://github.com/nymtech/nym/pull/5743
[#5742]: https://github.com/nymtech/nym/pull/5742
[#5739]: https://github.com/nymtech/nym/pull/5739
[#5737]: https://github.com/nymtech/nym/pull/5737
[#5730]: https://github.com/nymtech/nym/pull/5730
## [2025.9-appenzeller] (2025-05-13)
- build(deps): bump clap from 4.5.36 to 4.5.37 in the patch-updates group ([#5722])
Generated
+623 -553
View File
File diff suppressed because it is too large Load Diff
+2
View File
@@ -112,6 +112,7 @@ members = [
"nym-node/nym-node-metrics",
"nym-node/nym-node-requests",
"nym-outfox",
"nym-statistics-api",
"nym-validator-rewarder",
"nyx-chain-watcher",
"sdk/ffi/cpp",
@@ -152,6 +153,7 @@ default-members = [
"nym-node",
"nym-node-status-api/nym-node-status-agent",
"nym-node-status-api/nym-node-status-api",
"nym-statistics-api",
"nym-validator-rewarder",
"nyx-chain-watcher",
"service-providers/authenticator",
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-client"
version = "1.1.55"
version = "1.1.57"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
description = "Implementation of the Nym Client"
edition = "2021"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-socks5-client"
version = "1.1.55"
version = "1.1.57"
authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
description = "A SOCKS5 localhost proxy that converts incoming messages to Sphinx and sends them to a Nym address"
edition = "2021"
@@ -456,7 +456,7 @@ where
log::error!("Could not authenticate and start up the gateway connection - {err}");
ClientCoreError::GatewayClientError {
gateway_id: details.gateway_id.to_base58_string(),
source: err,
source: Box::new(err),
}
};
+10 -5
View File
@@ -18,7 +18,7 @@ pub enum ClientCoreError {
#[error("gateway client error ({gateway_id}): {source}")]
GatewayClientError {
gateway_id: String,
source: GatewayClientError,
source: Box<GatewayClientError>,
},
#[error("custom gateway client error: {source}")]
@@ -88,10 +88,7 @@ pub enum ClientCoreError {
},
#[error("failed to establish connection to gateway: {source}")]
GatewayConnectionFailure {
#[from]
source: tungstenite::Error,
},
GatewayConnectionFailure { source: Box<tungstenite::Error> },
#[cfg(target_arch = "wasm32")]
#[error("failed to establish gateway connection (wasm)")]
@@ -227,6 +224,14 @@ pub enum ClientCoreError {
HkdfDerivationError {},
}
impl From<tungstenite::Error> for ClientCoreError {
fn from(err: tungstenite::Error) -> ClientCoreError {
ClientCoreError::GatewayConnectionFailure {
source: Box::new(err),
}
}
}
/// Set of messages that the client can send to listeners via the task manager
#[derive(Debug)]
pub enum ClientCoreStatusMessage {
+2 -2
View File
@@ -329,7 +329,7 @@ pub(super) async fn register_with_gateway(
log::warn!("Failed to establish connection with gateway!");
ClientCoreError::GatewayClientError {
gateway_id: gateway_id.to_base58_string(),
source: err,
source: Box::new(err),
}
})?;
let auth_response = gateway_client
@@ -339,7 +339,7 @@ pub(super) async fn register_with_gateway(
log::warn!("Failed to register with the gateway {gateway_id}: {err}");
ClientCoreError::GatewayClientError {
gateway_id: gateway_id.to_base58_string(),
source: err,
source: Box::new(err),
}
})?;
+1 -1
View File
@@ -232,7 +232,7 @@ where
} => {
log::debug!("GatewaySetup::ReuseConnection");
Ok(reuse_gateway_connection(
authenticated_ephemeral_client,
*authenticated_ephemeral_client,
*gateway_details,
managed_keys,
))
+2 -2
View File
@@ -218,7 +218,7 @@ pub enum GatewaySetup {
ReuseConnection {
/// The authenticated ephemeral client that was created during `init`
authenticated_ephemeral_client: InitGatewayClient,
authenticated_ephemeral_client: Box<InitGatewayClient>,
// Details of this pre-initialised client (i.e. gateway and keys)
gateway_details: Box<GatewayRegistration>,
@@ -261,7 +261,7 @@ impl GatewaySetup {
pub fn try_reuse_connection(init_res: InitialisationResult) -> Result<Self, ClientCoreError> {
if let Some(authenticated_ephemeral_client) = init_res.authenticated_ephemeral_client {
Ok(GatewaySetup::ReuseConnection {
authenticated_ephemeral_client,
authenticated_ephemeral_client: Box::new(authenticated_ephemeral_client),
gateway_details: Box::new(init_res.gateway_registration),
client_keys: init_res.client_keys,
})
@@ -56,7 +56,7 @@ pub(crate) async fn connect_async(
}
.map_err(|err| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
source: Box::new(tungstenite::Error::from(err)),
})?;
#[cfg(unix)]
@@ -72,7 +72,7 @@ pub(crate) async fn connect_async(
Err(err) => {
stream = Err(GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: err.into(),
source: Box::new(tungstenite::Error::from(err)),
});
continue;
}
@@ -83,6 +83,6 @@ pub(crate) async fn connect_async(
.await
.map_err(|error| GatewayClientError::NetworkConnectionFailed {
address: endpoint.to_owned(),
source: error,
source: Box::new(error),
})
}
+12 -3
View File
@@ -25,7 +25,7 @@ pub enum GatewayClientError {
RequestError(#[from] GatewayRequestsError),
#[error("There was a network error: {0}")]
NetworkError(#[from] WsError),
NetworkError(Box<WsError>),
#[error("failed to upgrade our shared key - the gateway sent malformed response")]
FatalKeyUpgradeFailure,
@@ -41,7 +41,10 @@ pub enum GatewayClientError {
NetworkErrorWasm(#[from] JsError),
#[error("connection failed: {address}: {source}")]
NetworkConnectionFailed { address: String, source: WsError },
NetworkConnectionFailed {
address: String,
source: Box<WsError>,
},
#[error("no socket address for endpoint: {address}")]
NoEndpointForConnection { address: String },
@@ -127,10 +130,16 @@ pub enum GatewayClientError {
ShutdownInProgress,
}
impl From<WsError> for GatewayClientError {
fn from(error: WsError) -> Self {
GatewayClientError::NetworkError(Box::new(error))
}
}
impl GatewayClientError {
pub fn is_closed_connection(&self) -> bool {
match self {
GatewayClientError::NetworkError(ws_err) => match ws_err {
GatewayClientError::NetworkError(ws_err) => match ws_err.as_ref() {
WsError::AlreadyClosed | WsError::ConnectionClosed => true,
WsError::Io(io_err) => matches!(
io_err.kind(),
+2 -2
View File
@@ -28,7 +28,7 @@ pub(crate) fn cleanup_socket_message(
msg: Option<Result<Message, WsError>>,
) -> Result<Message, GatewayClientError> {
match msg {
Some(msg) => msg.map_err(GatewayClientError::NetworkError),
Some(msg) => msg.map_err(GatewayClientError::from),
None => Err(GatewayClientError::ConnectionAbruptlyClosed),
}
}
@@ -39,7 +39,7 @@ pub(crate) fn cleanup_socket_messages(
match msgs {
Some(msgs) => msgs
.into_iter()
.map(|msg| msg.map_err(GatewayClientError::NetworkError))
.map(|msg| msg.map_err(GatewayClientError::from))
.collect(),
None => Err(GatewayClientError::ConnectionAbruptlyClosed),
}
@@ -73,7 +73,6 @@ pub const STAKE_SATURATION: &str = "stake-saturation";
pub const INCLUSION_CHANCE: &str = "inclusion-probability";
pub const SUBMIT_GATEWAY: &str = "submit-gateway-monitoring-results";
pub const SUBMIT_NODE: &str = "submit-node-monitoring-results";
pub const SUBMIT_ROUTE: &str = "submit-route-monitoring-results";
pub const SERVICE_PROVIDERS: &str = "services";
+1 -2
View File
@@ -48,8 +48,7 @@ pub mod nym_config {
log::trace!("Loading from file: {:#?}", filepath.as_ref().to_owned());
let config_contents = fs::read_to_string(filepath)?;
toml::from_str(&config_contents)
.map_err(|toml_err| io::Error::new(io::ErrorKind::Other, toml_err))
toml::from_str(&config_contents).map_err(io::Error::other)
}
}
}
+1 -2
View File
@@ -151,8 +151,7 @@ where
let content = fs::read_to_string(path)?;
// TODO: should we be preserving original error type instead?
deserialize_config_from_toml_str(&content)
.map_err(|toml_err| io::Error::new(io::ErrorKind::Other, toml_err))
deserialize_config_from_toml_str(&content).map_err(io::Error::other)
}
//
+68 -48
View File
@@ -138,7 +138,12 @@
pub use reqwest::{IntoUrl, StatusCode};
use crate::path::RequestPath;
use async_trait::async_trait;
use bytes::Bytes;
use http::header::CONTENT_TYPE;
use http::HeaderMap;
use mime::Mime;
use reqwest::header::HeaderValue;
use reqwest::{RequestBuilder, Response};
use serde::de::DeserializeOwned;
@@ -149,10 +154,6 @@ use thiserror::Error;
use tracing::{debug, instrument, warn};
use url::Url;
use bytes::Bytes;
use http::header::CONTENT_TYPE;
use http::HeaderMap;
use mime::Mime;
#[cfg(not(target_arch = "wasm32"))]
use std::net::SocketAddr;
#[cfg(not(target_arch = "wasm32"))]
@@ -163,6 +164,8 @@ pub use user_agent::UserAgent;
#[cfg(not(target_arch = "wasm32"))]
mod dns;
mod path;
#[cfg(not(target_arch = "wasm32"))]
pub use dns::{HickoryDnsError, HickoryDnsResolver};
@@ -454,14 +457,15 @@ impl Client {
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait ApiClientCore {
/// Create an HTTP request using the host configured in this client.
fn create_request<B, K, V>(
fn create_request<P, B, K, V>(
&self,
method: reqwest::Method,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: Option<&B>,
) -> RequestBuilder
where
P: RequestPath,
B: Serialize + ?Sized,
K: AsRef<str>,
V: AsRef<str>;
@@ -512,7 +516,7 @@ pub trait ApiClientCore {
};
let params: Vec<(String, String)> = standin_url.query_pairs().into_owned().collect();
self.create_request(method, &path, &params, json_body)
self.create_request(method, path.as_slice(), &params, json_body)
}
/// Send a created HTTP request.
@@ -525,14 +529,15 @@ pub trait ApiClientCore {
E: Display;
/// Create and send a created HTTP request.
async fn send_request<B, K, V, E>(
async fn send_request<P, B, K, V, E>(
&self,
method: reqwest::Method,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: Option<&B>,
) -> Result<Response, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
B: Serialize + ?Sized + Sync,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -547,14 +552,15 @@ pub trait ApiClientCore {
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl ApiClientCore for Client {
#[instrument(level = "debug", skip_all, fields(path=?path))]
fn create_request<B, K, V>(
fn create_request<P, B, K, V>(
&self,
method: reqwest::Method,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: Option<&B>,
) -> RequestBuilder
where
P: RequestPath,
B: Serialize + ?Sized,
K: AsRef<str>,
V: AsRef<str>,
@@ -597,12 +603,9 @@ impl ApiClientCore for Client {
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
pub trait ApiClient: ApiClientCore {
/// Create an HTTP GET Request with the provided path and parameters
fn create_get_request<K, V>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> RequestBuilder
fn create_get_request<P, K, V>(&self, path: P, params: Params<'_, K, V>) -> RequestBuilder
where
P: RequestPath,
K: AsRef<str>,
V: AsRef<str>,
{
@@ -610,13 +613,14 @@ pub trait ApiClient: ApiClientCore {
}
/// Create an HTTP POST Request with the provided path, parameters, and json body
fn create_post_request<B, K, V>(
fn create_post_request<P, B, K, V>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> RequestBuilder
where
P: RequestPath,
B: Serialize + ?Sized,
K: AsRef<str>,
V: AsRef<str>,
@@ -625,12 +629,9 @@ pub trait ApiClient: ApiClientCore {
}
/// Create an HTTP DELETE Request with the provided path and parameters
fn create_delete_request<K, V>(
&self,
path: PathSegments<'_>,
params: Params<'_, K, V>,
) -> RequestBuilder
fn create_delete_request<P, K, V>(&self, path: P, params: Params<'_, K, V>) -> RequestBuilder
where
P: RequestPath,
K: AsRef<str>,
V: AsRef<str>,
{
@@ -638,13 +639,14 @@ pub trait ApiClient: ApiClientCore {
}
/// Create an HTTP PATCH Request with the provided path, parameters, and json body
fn create_patch_request<B, K, V>(
fn create_patch_request<P, B, K, V>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> RequestBuilder
where
P: RequestPath,
B: Serialize + ?Sized,
K: AsRef<str>,
V: AsRef<str>,
@@ -654,12 +656,13 @@ pub trait ApiClient: ApiClientCore {
/// Create and send an HTTP GET Request with the provided path and parameters
#[instrument(level = "debug", skip_all, fields(path=?path))]
async fn send_get_request<K, V, E>(
async fn send_get_request<P, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
) -> Result<Response, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
E: Display,
@@ -669,13 +672,14 @@ pub trait ApiClient: ApiClientCore {
}
/// Create and send an HTTP POST Request with the provided path, parameters, and json data
async fn send_post_request<B, K, V, E>(
async fn send_post_request<P, B, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> Result<Response, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
B: Serialize + ?Sized + Sync,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -686,12 +690,13 @@ pub trait ApiClient: ApiClientCore {
}
/// Create and send an HTTP DELETE Request with the provided path and parameters
async fn send_delete_request<K, V, E>(
async fn send_delete_request<P, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
) -> Result<Response, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
E: Display,
@@ -701,13 +706,14 @@ pub trait ApiClient: ApiClientCore {
}
/// Create and send an HTTP PATCH Request with the provided path, parameters, and json data
async fn send_patch_request<B, K, V, E>(
async fn send_patch_request<P, B, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> Result<Response, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
B: Serialize + ?Sized + Sync,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -722,12 +728,13 @@ pub trait ApiClient: ApiClientCore {
/// into the provided type `T`.
#[instrument(level = "debug", skip_all, fields(path=?path))]
// TODO: deprecate in favour of get_response that works based on mime type in the response
async fn get_json<T, K, V, E>(
async fn get_json<P, T, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -739,12 +746,13 @@ pub trait ApiClient: ApiClientCore {
/// 'get' data from the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple
/// defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the response
/// into the provided type `T` based on the content type header
async fn get_response<T, K, V, E>(
async fn get_response<P, T, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -759,13 +767,14 @@ pub trait ApiClient: ApiClientCore {
/// 'post' json data to the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple
/// defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the response
/// into the provided type `T`.
async fn post_json<B, T, K, V, E>(
async fn post_json<P, B, T, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> Result<T, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
B: Serialize + ?Sized + Sync,
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
@@ -781,12 +790,13 @@ pub trait ApiClient: ApiClientCore {
/// 'delete' json data from the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with
/// tuple defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the
/// response into the provided type `T`.
async fn delete_json<T, K, V, E>(
async fn delete_json<P, T, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
) -> Result<T, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
V: AsRef<str> + Sync,
@@ -801,13 +811,14 @@ pub trait ApiClient: ApiClientCore {
/// 'patch' json data at the segment-defined path, e.g. `["api", "v1", "mixnodes"]`, with tuple
/// defined key-value parameters, e.g. `[("since", "12345")]`. Attempt to parse the response
/// into the provided type `T`.
async fn patch_json<B, T, K, V, E>(
async fn patch_json<P, B, T, K, V, E>(
&self,
path: PathSegments<'_>,
path: P,
params: Params<'_, K, V>,
json_body: &B,
) -> Result<T, HttpClientError<E>>
where
P: RequestPath + Send + Sync,
B: Serialize + ?Sized + Sync,
for<'a> T: Deserialize<'a>,
K: AsRef<str> + Sync,
@@ -890,7 +901,7 @@ impl<C> ApiClient for C where C: ApiClientCore + Sync {}
/// utility function that should solve the double slash problem in API urls forever.
fn sanitize_url<K: AsRef<str>, V: AsRef<str>>(
base: &Url,
segments: PathSegments<'_>,
request_path: impl RequestPath,
params: Params<'_, K, V>,
) -> Url {
let mut url = base.clone();
@@ -900,10 +911,7 @@ fn sanitize_url<K: AsRef<str>, V: AsRef<str>>(
path_segments.pop_if_empty();
for segment in segments {
let segment = segment.strip_prefix('/').unwrap_or(segment);
let segment = segment.strip_suffix('/').unwrap_or(segment);
for segment in request_path.to_sanitized_segments() {
path_segments.push(segment);
}
@@ -1048,6 +1056,18 @@ mod tests {
fn sanitizing_urls() {
let base_url: Url = "http://foomp.com".parse().unwrap();
// works with a full string
assert_eq!(
"http://foomp.com/foo/bar",
sanitize_url(&base_url, "/foo//bar/", NO_PARAMS).as_str()
);
// (and leading slash doesn't matter)
assert_eq!(
"http://foomp.com/foo/bar",
sanitize_url(&base_url, "foo//bar/", NO_PARAMS).as_str()
);
// works with 1 segment
assert_eq!(
"http://foomp.com/foo",
+59
View File
@@ -0,0 +1,59 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use std::fmt::Debug;
/// Collection of URL Path Segments
pub type PathSegments<'a> = &'a [&'a str];
fn sanitize_fragment(segment: &str) -> &str {
segment.trim_matches(|c: char| c.is_whitespace() || c == '/')
}
pub trait RequestPath: Debug {
fn to_sanitized_segments(&self) -> Vec<&str>;
}
macro_rules! impl_stringified_sanitized_segments {
($frag_iter:expr) => {{
let mut path_segments = Vec::new();
for segment in $frag_iter {
if !segment.is_empty() {
path_segments.push(sanitize_fragment(segment));
}
}
path_segments
}};
}
impl RequestPath for PathSegments<'_> {
fn to_sanitized_segments(&self) -> Vec<&str> {
impl_stringified_sanitized_segments!(self.iter())
}
}
impl<const N: usize> RequestPath for &[&str; N] {
fn to_sanitized_segments(&self) -> Vec<&str> {
impl_stringified_sanitized_segments!(self.iter())
}
}
impl RequestPath for &str {
fn to_sanitized_segments(&self) -> Vec<&str> {
impl_stringified_sanitized_segments!(self.split('/'))
}
}
impl RequestPath for String {
fn to_sanitized_segments(&self) -> Vec<&str> {
impl_stringified_sanitized_segments!(self.split('/'))
}
}
impl RequestPath for &String {
fn to_sanitized_segments(&self) -> Vec<&str> {
impl_stringified_sanitized_segments!(self.split('/'))
}
}
@@ -124,7 +124,7 @@ impl ReconstructionBuffer {
// TODO: what to do in that case? give up on the message? overwrite it? panic?
// it *might* be due to lock ack-packet, but let's keep the `warn` level in case
// it could be somehow exploited
debug!(
warn!(
"duplicate fragment received! - frag - {} (set id: {})",
fragment.current_fragment(),
fragment.id()
+8 -8
View File
@@ -26,54 +26,54 @@ pub enum ScraperError {
WebSocketConnectionFailure {
url: String,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("failed to establish rpc connection to {url}: {source}")]
HttpConnectionFailure {
url: String,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("failed to create chain subscription: {source}")]
ChainSubscriptionFailure {
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain basic block information at height: {height}: {source}")]
BlockQueryFailure {
height: u32,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain block results information at height: {height}: {source}")]
BlockResultsQueryFailure {
height: u32,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain validators information at height: {height}: {source}")]
ValidatorsQueryFailure {
height: u32,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain tx results for tx: {hash}: {source}")]
TxResultsQueryFailure {
hash: Hash,
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not obtain current abci info: {source}")]
AbciInfoQueryFailure {
#[source]
source: tendermint_rpc::Error,
source: Box<tendermint_rpc::Error>,
},
#[error("could not parse tx {hash}: {source}")]
+30 -18
View File
@@ -29,7 +29,7 @@ impl RpcClient {
let http_client = HttpClient::new(url.as_str()).map_err(|source| {
ScraperError::HttpConnectionFailure {
url: url.to_string(),
source,
source: Box::new(source),
}
})?;
@@ -90,7 +90,10 @@ impl RpcClient {
self.inner
.block(height)
.await
.map_err(|source| ScraperError::BlockQueryFailure { height, source })
.map_err(|source| ScraperError::BlockQueryFailure {
height,
source: Box::new(source),
})
}
#[instrument(skip(self), err(Display))]
@@ -100,31 +103,37 @@ impl RpcClient {
) -> Result<block_results::Response, ScraperError> {
debug!("getting block results");
self.inner
.block_results(height)
.await
.map_err(|source| ScraperError::BlockResultsQueryFailure { height, source })
self.inner.block_results(height).await.map_err(|source| {
ScraperError::BlockResultsQueryFailure {
height,
source: Box::new(source),
}
})
}
pub(crate) async fn current_block_height(&self) -> Result<u64, ScraperError> {
debug!("getting current block height");
let info = self
.inner
.abci_info()
.await
.map_err(|source| ScraperError::AbciInfoQueryFailure { source })?;
let info =
self.inner
.abci_info()
.await
.map_err(|source| ScraperError::AbciInfoQueryFailure {
source: Box::new(source),
})?;
Ok(info.last_block_height.value())
}
pub(crate) async fn earliest_available_block_height(&self) -> Result<u64, ScraperError> {
debug!("getting earliest available block height");
let status = self
.inner
.status()
.await
.map_err(|source| ScraperError::AbciInfoQueryFailure { source })?;
let status =
self.inner
.status()
.await
.map_err(|source| ScraperError::AbciInfoQueryFailure {
source: Box::new(source),
})?;
Ok(status.sync_info.earliest_block_height.value())
}
@@ -167,7 +176,7 @@ impl RpcClient {
.await
.map_err(|source| ScraperError::TxResultsQueryFailure {
hash: tx_hash,
source,
source: Box::new(source),
})
}
@@ -181,6 +190,9 @@ impl RpcClient {
self.inner
.validators(height, Paging::All)
.await
.map_err(|source| ScraperError::ValidatorsQueryFailure { height, source })
.map_err(|source| ScraperError::ValidatorsQueryFailure {
height,
source: Box::new(source),
})
}
}
@@ -42,7 +42,7 @@ impl ChainSubscriber {
let websocket_url = websocket_endpoint.as_str().try_into().map_err(|source| {
ScraperError::WebSocketConnectionFailure {
url: websocket_endpoint.to_string(),
source,
source: Box::new(source),
}
})?;
@@ -52,7 +52,7 @@ impl ChainSubscriber {
.await
.map_err(|source| ScraperError::WebSocketConnectionFailure {
url: websocket_endpoint.to_string(),
source,
source: Box::new(source),
})?;
Ok(ChainSubscriber {
@@ -83,7 +83,7 @@ impl ChainSubscriber {
.await
.map_err(|source| ScraperError::WebSocketConnectionFailure {
url: self.websocket_endpoint.to_string(),
source,
source: Box::new(source),
})?;
self.websocket_client = client;
self.websocket_driver = Some(driver);
@@ -121,7 +121,9 @@ impl ChainSubscriber {
.websocket_client
.subscribe(EventType::NewBlock.into())
.await
.map_err(|source| ScraperError::ChainSubscriptionFailure { source })?;
.map_err(|source| ScraperError::ChainSubscriptionFailure {
source: Box::new(source),
})?;
let mut failures = 0;
+6 -9
View File
@@ -54,14 +54,11 @@ where
let key_pem = read_pem_file(path)?;
if T::pem_type() != key_pem.tag {
return Err(io::Error::new(
io::ErrorKind::Other,
format!(
"unexpected key pem tag. Got '{}', expected: '{}'",
key_pem.tag,
T::pem_type()
),
));
return Err(io::Error::other(format!(
"unexpected key pem tag. Got '{}', expected: '{}'",
key_pem.tag,
T::pem_type()
)));
}
let key = match T::from_bytes(&key_pem.contents) {
@@ -84,7 +81,7 @@ fn read_pem_file<P: AsRef<Path>>(filepath: P) -> io::Result<Pem> {
let mut pem_bytes = File::open(filepath)?;
let mut buf = Vec::new();
pem_bytes.read_to_end(&mut buf)?;
pem::parse(&buf).map_err(|e| io::Error::new(io::ErrorKind::Other, e))
pem::parse(&buf).map_err(io::Error::other)
}
fn write_pem_file<P: AsRef<Path>>(filepath: P, data: Vec<u8>, tag: &str) -> io::Result<()> {
+6
View File
@@ -28,5 +28,11 @@ nym-credentials-interface = { path = "../credentials-interface" }
nym-metrics = { path = "../nym-metrics" }
nym-task = { path = "../task" }
utoipa = { workspace = true, optional = true }
[target."cfg(target_arch = \"wasm32\")".dependencies.wasmtimer]
workspace = true
[features]
default = []
openapi = ["dep:utoipa"]
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crate::report::{ClientStatsReport, OsInformation};
use crate::report::client::{ClientStatsReport, OsInformation};
use nym_task::TaskClient;
use time::{OffsetDateTime, Time};
+5
View File
@@ -26,11 +26,16 @@ pub mod report;
pub mod types;
const CLIENT_ID_PREFIX: &str = "client_stats_id";
const VPN_CLIENT_ID_PREFIX: &str = "vpnclient_stats_id";
pub fn generate_client_stats_id(id_key: ed25519::PublicKey) -> String {
generate_stats_id(CLIENT_ID_PREFIX, id_key.to_base58_string())
}
pub fn generate_vpn_client_stats_id<M: AsRef<[u8]>>(seed: M) -> String {
generate_stats_id(VPN_CLIENT_ID_PREFIX, seed)
}
fn generate_stats_id<M: AsRef<[u8]>>(prefix: &str, id_seed: M) -> String {
let mut hasher = sha2::Sha256::new();
hasher.update(prefix);
@@ -6,7 +6,7 @@ use crate::clients::{
nym_api_statistics::NymApiStats, packet_statistics::PacketStatistics,
};
use super::error::StatsError;
use crate::error::StatsError;
use serde::{Deserialize, Serialize};
use sysinfo::System;
+5
View File
@@ -0,0 +1,5 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
pub mod client;
pub mod vpn_client;
@@ -0,0 +1,51 @@
// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use serde::{Deserialize, Serialize};
const KIND: &str = "vpn_client_stats_report";
const VERSION: &str = "v1";
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VpnClientStatsReport {
pub kind: String,
pub api_version: String,
pub stats_id: String,
pub static_information: StaticInformationReport,
//SW called it basic so we can swap it easily down the line for more data
pub basic_usage: Option<UsageReport>,
}
impl VpnClientStatsReport {
pub fn new(stats_id: String, static_information: StaticInformationReport) -> Self {
VpnClientStatsReport {
kind: KIND.into(),
api_version: VERSION.into(),
stats_id,
static_information,
basic_usage: None,
}
}
#[must_use]
pub fn with_usage_report(mut self, usage_report: UsageReport) -> Self {
self.basic_usage = Some(usage_report);
self
}
}
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StaticInformationReport {
pub os_type: String,
pub os_version: Option<String>,
pub os_arch: String,
pub app_version: String,
}
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UsageReport {
pub connection_time_ms: Option<i32>,
pub two_hop: bool,
}
-4
View File
@@ -224,10 +224,6 @@ impl NymTopology {
serde_json::from_reader(file).map_err(Into::into)
}
pub fn node_details(&self) -> &HashMap<NodeId, RoutingNode> {
&self.node_details
}
pub fn add_skimmed_nodes(&mut self, nodes: &[SkimmedNode]) {
self.add_additional_nodes(nodes.iter())
}
+7 -28
View File
@@ -11,27 +11,6 @@ static NETWORK_MONITORS: LazyLock<HashSet<String>> = LazyLock::new(|| {
nm
});
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, ToSchema)]
pub struct RouteResult {
pub layer1: u32,
pub layer2: u32,
pub layer3: u32,
pub gw: u32,
pub success: bool,
}
impl RouteResult {
pub fn new(layer1: u32, layer2: u32, layer3: u32, gw: u32, success: bool) -> Self {
RouteResult {
layer1,
layer2,
layer3,
gw,
success,
}
}
}
#[derive(Debug, Serialize, Deserialize, JsonSchema, Clone, ToSchema)]
pub struct NodeResult {
#[schema(value_type = u32)]
@@ -50,23 +29,23 @@ impl NodeResult {
}
}
#[derive(Serialize, Deserialize, JsonSchema, ToSchema)]
#[derive(Serialize, Deserialize, JsonSchema)]
#[serde(untagged)]
pub enum MonitorResults {
Node(Vec<NodeResult>),
Route(Vec<RouteResult>),
Mixnode(Vec<NodeResult>),
Gateway(Vec<NodeResult>),
}
#[derive(Serialize, Deserialize, JsonSchema, ToSchema)]
pub struct MonitorMessage {
results: MonitorResults,
results: Vec<NodeResult>,
signature: String,
signer: String,
timestamp: i64,
}
impl MonitorMessage {
fn message_to_sign(results: &MonitorResults, timestamp: i64) -> Vec<u8> {
fn message_to_sign(results: &[NodeResult], timestamp: i64) -> Vec<u8> {
let mut msg = serde_json::to_vec(results).unwrap_or_default();
msg.extend_from_slice(&timestamp.to_le_bytes());
msg
@@ -81,7 +60,7 @@ impl MonitorMessage {
now - self.timestamp < 5
}
pub fn new(results: MonitorResults, private_key: &PrivateKey) -> Self {
pub fn new(results: Vec<NodeResult>, private_key: &PrivateKey) -> Self {
let timestamp = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.expect("Time went backwards")
@@ -103,7 +82,7 @@ impl MonitorMessage {
NETWORK_MONITORS.contains(&self.signer)
}
pub fn results(&self) -> &MonitorResults {
pub fn results(&self) -> &[NodeResult] {
&self.results
}
+2 -4
View File
@@ -25,10 +25,8 @@ fn map_ws_error(err: WebSocketError) -> WsError {
// TODO: are we preserving correct semantics?
WebSocketError::ConnectionError => WsError::ConnectionClosed,
WebSocketError::ConnectionClose(_event) => WsError::ConnectionClosed,
WebSocketError::MessageSendError(err) => {
WsError::Io(io::Error::new(io::ErrorKind::Other, err.to_string()))
}
_ => WsError::Io(io::Error::new(io::ErrorKind::Other, "new websocket error")),
WebSocketError::MessageSendError(err) => WsError::Io(io::Error::other(err.to_string())),
_ => WsError::Io(io::Error::other("new websocket error")),
}
}
+46 -40
View File
@@ -13,12 +13,10 @@ use nym_gateway_storage::models::WireguardPeer;
use nym_task::TaskClient;
use nym_wireguard_types::DEFAULT_PEER_TIMEOUT_CHECK;
use std::sync::Arc;
use std::time::{Duration, SystemTime};
use tokio::sync::{mpsc, RwLock};
use tokio_stream::{wrappers::IntervalStream, StreamExt};
pub(crate) type SharedBandwidthStorageManager = Arc<RwLock<BandwidthStorageManager>>;
const AUTO_REMOVE_AFTER: Duration = Duration::from_secs(60 * 60); // 1 hour
pub struct PeerHandle {
public_key: Key,
@@ -28,7 +26,6 @@ pub struct PeerHandle {
request_tx: mpsc::Sender<PeerControlRequest>,
timeout_check_interval: IntervalStream,
task_client: TaskClient,
startup_timestamp: SystemTime,
}
impl PeerHandle {
@@ -53,7 +50,6 @@ impl PeerHandle {
request_tx,
timeout_check_interval,
task_client,
startup_timestamp: SystemTime::now(),
}
}
@@ -73,26 +69,49 @@ impl PeerHandle {
Ok(success)
}
async fn active_peer(
&mut self,
storage_peer: &WireguardPeer,
kernel_peer: &Peer,
) -> Result<bool, Error> {
if let Some(bandwidth_manager) = &self.bandwidth_storage_manager {
if kernel_peer.last_handshake.is_none()
&& SystemTime::now().duration_since(self.startup_timestamp)? >= AUTO_REMOVE_AFTER
{
let success = self.remove_peer().await?;
self.peer_storage_manager.remove_peer();
tracing::debug!(
"Peer {} has not been active for more then {} seconds, removing it",
kernel_peer.public_key.to_string(),
AUTO_REMOVE_AFTER.as_secs()
fn compute_spent_bandwidth(kernel_peer: &Peer, storage_peer: &WireguardPeer) -> Option<u64> {
let storage_peer_rx_bytes = u64::try_from(storage_peer.rx_bytes)
.inspect_err(|e| tracing::error!("Storage rx bytes could not be converted: {e}"))
.ok()?;
let storage_peer_tx_bytes = u64::try_from(storage_peer.tx_bytes)
.inspect_err(|e| tracing::error!("Storage tx bytes could not be converted: {e}"))
.ok()?;
let kernel_total = kernel_peer
.rx_bytes
.checked_add(kernel_peer.tx_bytes)
.or_else(|| {
tracing::error!(
"Overflow on kernel adding bytes: {} + {}",
kernel_peer.rx_bytes,
kernel_peer.tx_bytes
);
return Ok(!success);
}
let spent_bandwidth = (kernel_peer.rx_bytes + kernel_peer.tx_bytes)
.checked_sub(storage_peer.rx_bytes as u64 + storage_peer.tx_bytes as u64)
None
})?;
let storage_total = storage_peer_rx_bytes
.checked_add(storage_peer_tx_bytes)
.or_else(|| {
tracing::error!("Overflow on storage adding bytes: {storage_peer_rx_bytes} + {storage_peer_tx_bytes}");
None
})?;
kernel_total.checked_sub(storage_total).or_else(|| {
tracing::error!("Overflow on spent bandwidth subtraction: kernel - storage = {kernel_total} - {storage_total}");
None
})
}
async fn active_peer(&mut self, kernel_peer: &Peer) -> Result<bool, Error> {
let Some(storage_peer) = self.peer_storage_manager.get_peer() else {
log::debug!(
"Peer {:?} not in storage anymore, shutting down handle",
self.public_key
);
return Ok(false);
};
if let Some(bandwidth_manager) = &self.bandwidth_storage_manager {
let spent_bandwidth = Self::compute_spent_bandwidth(kernel_peer, &storage_peer)
.unwrap_or_else(|| {
// if gateway restarted, the kernel values restart from 0
// and we should restart from 0 in storage as well
@@ -100,8 +119,10 @@ impl PeerHandle {
self.peer_storage_manager.peer_information.as_mut()
{
peer_information.force_sync = true;
peer_information.peer.rx_bytes = kernel_peer.rx_bytes;
peer_information.peer.tx_bytes = kernel_peer.tx_bytes;
}
kernel_peer.rx_bytes + kernel_peer.tx_bytes
0
})
.try_into()
.map_err(|_| Error::InconsistentConsumedBytes)?;
@@ -124,14 +145,6 @@ impl PeerHandle {
}
}
} else {
if SystemTime::now().duration_since(self.startup_timestamp)? >= AUTO_REMOVE_AFTER {
log::debug!(
"Peer {} has been present for 30 days, removing it",
self.public_key
);
let success = self.remove_peer().await?;
return Ok(!success);
}
let spent_bandwidth = kernel_peer.rx_bytes + kernel_peer.tx_bytes;
if spent_bandwidth >= BANDWIDTH_CAP_PER_DAY {
log::debug!(
@@ -158,14 +171,7 @@ impl PeerHandle {
// the host information hasn't beed updated yet
return Ok(true);
};
let Some(storage_peer) = self.peer_storage_manager.get_peer() else {
log::debug!(
"Peer {:?} not in storage anymore, shutting down handle",
self.public_key
);
return Ok(false);
};
if !self.active_peer(&storage_peer, &kernel_peer).await? {
if !self.active_peer(&kernel_peer).await? {
log::debug!(
"Peer {:?} is not active anymore, shutting down handle",
self.public_key
+24
View File
@@ -0,0 +1,24 @@
// @ts-check
/** @type {import('next').NextConfig} */
const nextConfig = {
reactStrictMode: true,
basePath: "/explorer",
assetPrefix: "/explorer",
trailingSlash: false,
async redirects() {
return [
// Change the basePath to /explorer
{
source: "/",
destination: "/explorer",
basePath: false,
permanent: true,
},
];
},
};
module.exports = nextConfig
-23
View File
@@ -1,23 +0,0 @@
import type { NextConfig } from "next";
const nextConfig: NextConfig = {
reactStrictMode: true,
basePath: "/explorer",
assetPrefix: "/explorer",
trailingSlash: false,
async redirects() {
return [
// Change the basePath to /explorer
{
source: "/",
destination: "/explorer",
basePath: false,
permanent: true,
},
];
},
};
export default nextConfig;
+1 -1
View File
@@ -50,7 +50,7 @@ nym-api-requests = { path = "../nym-api/nym-api-requests" }
nym-credentials = { path = "../common/credentials" }
nym-credentials-interface = { path = "../common/credentials-interface" }
nym-credential-verification = { path = "../common/credential-verification" }
nym-crypto = { path = "../common/crypto" }
nym-crypto = { path = "../common/crypto", features = ["sphinx"] }
nym-gateway-storage = { path = "../common/gateway-storage" }
nym-gateway-stats-storage = { path = "../common/gateway-stats-storage" }
nym-gateway-requests = { path = "../common/gateway-requests" }
+9 -4
View File
@@ -69,10 +69,7 @@ pub enum GatewayError {
},
#[error("there was an issue with the local authenticator: {source}")]
AuthenticatorFailure {
#[from]
source: AuthenticatorError,
},
AuthenticatorFailure { source: Box<AuthenticatorError> },
#[error("failed to startup local {typ}")]
ServiceProviderStartupFailure { typ: &'static str },
@@ -138,3 +135,11 @@ impl From<ClientCoreError> for GatewayError {
}
}
}
impl From<AuthenticatorError> for GatewayError {
fn from(error: AuthenticatorError) -> Self {
GatewayError::AuthenticatorFailure {
source: Box::new(error),
}
}
}
@@ -22,7 +22,7 @@ enum ActiveClient {
Remote(RemoteClientData),
/// Handle to a locally (inside the same process) running client.
Embedded(LocalEmbeddedClientHandle),
Embedded(Box<LocalEmbeddedClientHandle>),
}
impl ActiveClient {
@@ -163,7 +163,7 @@ impl ActiveClientsStore {
/// Inserts a handle to the embedded client
pub fn insert_embedded(&self, local_client_handle: LocalEmbeddedClientHandle) {
let key = local_client_handle.client_destination();
let entry = ActiveClient::Embedded(local_client_handle);
let entry = ActiveClient::Embedded(Box::new(local_client_handle));
if self.inner.insert(key, entry).is_some() {
// this is literally impossible since we're starting the local embedded client before
// even spawning the websocket listener task
@@ -83,7 +83,7 @@ pub(crate) enum InitialAuthenticationError {
UnexpectedMessageType { typ: String },
#[error("Experienced connection error: {0}")]
ConnectionError(#[from] WsError),
ConnectionError(Box<WsError>),
#[error("Attempted to negotiate connection with client using incompatible protocol version. Ours is {current} and the client reports {client:?}")]
IncompatibleProtocol { client: Option<u8>, current: u8 },
@@ -91,7 +91,7 @@ pub(crate) enum InitialAuthenticationError {
#[error("failed to send authentication response: {source}")]
ResponseSendFailure {
#[source]
source: WsError,
source: Box<WsError>,
},
#[error("possibly received a sphinx packet without prior authentication. Request is going to be ignored")]
@@ -103,7 +103,7 @@ pub(crate) enum InitialAuthenticationError {
#[error("failed to obtain message from websocket stream: {source}")]
FailedToReadMessage {
#[source]
source: WsError,
source: Box<WsError>,
},
#[error("timed out while waiting for initial data")]
@@ -113,6 +113,12 @@ pub(crate) enum InitialAuthenticationError {
EmptyClientDetails,
}
impl From<WsError> for InitialAuthenticationError {
fn from(error: WsError) -> Self {
InitialAuthenticationError::ConnectionError(Box::new(error))
}
}
pub(crate) struct FreshHandler<R, S> {
rng: R,
pub(crate) shared_state: CommonHandlerState,
@@ -163,7 +169,7 @@ impl<R, S> FreshHandler<R, S> {
SocketStream::RawTcp(conn) => {
// TODO: perhaps in the future, rather than panic here (and uncleanly shut tcp stream)
// return a result with an error?
let ws_stream = tokio_tungstenite::accept_async(conn).await?;
let ws_stream = Box::new(tokio_tungstenite::accept_async(conn).await?);
SocketStream::UpgradedWebSocket(ws_stream)
}
other => other,
@@ -342,7 +348,7 @@ impl<R, S> FreshHandler<R, S> {
// push them to the client
if let Err(err) = self.push_packets_to_client(shared_keys, messages).await {
warn!("We failed to send stored messages to fresh client - {err}",);
return Err(InitialAuthenticationError::ConnectionError(err));
return Err(InitialAuthenticationError::ConnectionError(Box::new(err)));
} else {
// if it was successful - remove them from the store
self.shared_state.storage.remove_messages(ids).await?;
@@ -880,7 +886,9 @@ impl<R, S> FreshHandler<R, S> {
.await
{
debug!("failed to send authentication response: {source}");
return Err(InitialAuthenticationError::ResponseSendFailure { source });
return Err(InitialAuthenticationError::ResponseSendFailure {
source: Box::new(source),
});
}
let Some(client_details) = auth_result.client_details else {
@@ -963,7 +971,9 @@ impl<R, S> FreshHandler<R, S> {
Ok(Some(Ok(msg))) => msg,
Ok(Some(Err(source))) => {
debug!("failed to obtain message from websocket stream! stopping connection handler: {source}");
return Err(InitialAuthenticationError::FailedToReadMessage { source });
return Err(InitialAuthenticationError::FailedToReadMessage {
source: Box::new(source),
});
}
Ok(None) => return Err(InitialAuthenticationError::ClosedConnection),
Err(_timeout) => return Err(InitialAuthenticationError::Timeout),
@@ -31,7 +31,7 @@ const INITIAL_MESSAGE_TIMEOUT: Duration = Duration::from_millis(10_000);
pub(crate) enum SocketStream<S> {
RawTcp(S),
UpgradedWebSocket(WebSocketStream<S>),
UpgradedWebSocket(Box<WebSocketStream<S>>),
Invalid,
}
@@ -1,12 +0,0 @@
{
"db_name": "SQLite",
"query": "\n INSERT OR IGNORE INTO routes (layer1, layer2, layer3, gw, success) VALUES (?, ?, ?, ?, ?);\n ",
"describe": {
"columns": [],
"parameters": {
"Right": 5
},
"nullable": []
},
"hash": "66109c1d856e1ca2b5126e4bf4c58c7a27b8c303bfa079cf74909354202dcc49"
}
@@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "\n SELECT\n d.node_id as \"node_id: NodeId\",\n CASE WHEN count(*) > 3 THEN AVG(reliability) ELSE 100 END as \"value: f32\"\n FROM\n gateway_details d\n JOIN\n gateway_status s on d.id = s.gateway_details_id\n WHERE\n timestamp >= ? AND\n timestamp <= ?\n GROUP BY 1\n ",
"describe": {
"columns": [
{
"name": "node_id: NodeId",
"ordinal": 0,
"type_info": "Int64"
},
{
"name": "value: f32",
"ordinal": 1,
"type_info": "Int"
}
],
"parameters": {
"Right": 2
},
"nullable": [
false,
true
]
},
"hash": "676299beb2004ab89f7b38cf21ffb84ab5e7d7435297573523e2532560c2e302"
}
@@ -1,44 +0,0 @@
{
"db_name": "SQLite",
"query": "\n SELECT\n layer1 as \"layer1\",\n layer2 as \"layer2\",\n layer3 as \"layer3\",\n gw as \"gw\",\n success\n FROM routes\n WHERE timestamp >= ? AND timestamp <= ?\n ORDER BY timestamp ASC\n ",
"describe": {
"columns": [
{
"name": "layer1",
"ordinal": 0,
"type_info": "Int64"
},
{
"name": "layer2",
"ordinal": 1,
"type_info": "Int64"
},
{
"name": "layer3",
"ordinal": 2,
"type_info": "Int64"
},
{
"name": "gw",
"ordinal": 3,
"type_info": "Int64"
},
{
"name": "success",
"ordinal": 4,
"type_info": "Bool"
}
],
"parameters": {
"Right": 2
},
"nullable": [
false,
false,
false,
false,
false
]
},
"hash": "6b2479c02cf1ef5ae674ce0ab4d027595b91739f3579e1f289b0c722ea91bbcc"
}
@@ -0,0 +1,26 @@
{
"db_name": "SQLite",
"query": "\n SELECT\n d.mix_id as \"mix_id: NodeId\",\n AVG(s.reliability) as \"value: f32\"\n FROM\n mixnode_details d\n JOIN\n mixnode_status s on d.id = s.mixnode_details_id\n WHERE\n timestamp >= ? AND\n timestamp <= ?\n GROUP BY 1\n ",
"describe": {
"columns": [
{
"name": "mix_id: NodeId",
"ordinal": 0,
"type_info": "Int64"
},
{
"name": "value: f32",
"ordinal": 1,
"type_info": "Int64"
}
],
"parameters": {
"Right": 2
},
"nullable": [
false,
true
]
},
"hash": "c19e1b3768bf2929407599e6e8783ead09f4d7319b7997fa2a9bb628f9404166"
}
+1 -1
View File
@@ -4,7 +4,7 @@
[package]
name = "nym-api"
license = "GPL-3.0"
version = "1.1.58"
version = "1.1.60"
authors.workspace = true
edition = "2021"
rust-version.workspace = true
+1 -1
View File
@@ -2,4 +2,4 @@
set -e
/usr/src/nym/target/release/nym-api init --mnemonic "$MNEMONIC" && /usr/src/nym/target/release/nym-api run
/usr/src/nym/target/release/nym-api init && /usr/src/nym/target/release/nym-api run
@@ -1,17 +0,0 @@
-- Add routes table for storing route metrics data
CREATE TABLE IF NOT EXISTS routes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
layer1 INTEGER NOT NULL, -- NodeId of layer 1 mixnode
layer2 INTEGER NOT NULL, -- NodeId of layer 2 mixnode
layer3 INTEGER NOT NULL, -- NodeId of layer 3 mixnode
gw INTEGER NOT NULL, -- NodeId of gateway
success BOOLEAN NOT NULL, -- Whether the packet was delivered successfully
timestamp INTEGER NOT NULL DEFAULT (unixepoch()) -- When the measurement was taken
);
-- Add indexes for efficient querying
CREATE INDEX IF NOT EXISTS idx_routes_timestamp ON routes(timestamp);
CREATE INDEX IF NOT EXISTS idx_routes_layer1 ON routes(layer1);
CREATE INDEX IF NOT EXISTS idx_routes_layer2 ON routes(layer2);
CREATE INDEX IF NOT EXISTS idx_routes_layer3 ON routes(layer3);
CREATE INDEX IF NOT EXISTS idx_routes_gw ON routes(gw);
+1 -1
View File
@@ -16,7 +16,7 @@ pub enum DkgError {
StatePersistenceFailure {
path: PathBuf,
#[source]
source: EcashError,
source: Box<EcashError>,
},
#[error("failed to query for the current DKG epoch state: {source}")]
+1 -1
View File
@@ -73,7 +73,7 @@ impl<R: RngCore + CryptoRng + Clone> DkgController<R> {
persistent_state.save_to_file(save_path).map_err(|source| {
DkgError::StatePersistenceFailure {
path: save_path.to_path_buf(),
source,
source: Box::new(source),
}
})
}
+1 -1
View File
@@ -1275,7 +1275,7 @@ impl TestFixture {
AppState {
nyxd_client,
chain_status_cache: ChainStatusCache::new(Duration::from_secs(42)),
address_info_cache: AddressInfoCache::new(),
address_info_cache: AddressInfoCache::new(Duration::from_secs(42), 1000),
forced_refresh: ForcedRefresh::new(true),
nym_contract_cache: NymContractCache::new(),
node_status_cache: NodeStatusCache::new(),
@@ -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, MonitorResults};
use nym_types::monitoring::MonitorMessage;
use tracing::error;
pub(super) fn mandatory_routes() -> Router<AppState> {
@@ -33,10 +33,6 @@ 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()
@@ -62,53 +58,6 @@ 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,
@@ -138,21 +87,18 @@ pub(crate) async fn submit_gateway_monitoring_results(
return Err(AxumErrorResponse::bad_request("invalid signature"));
}
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",
))
}
}
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",
))
}
MonitorResults::Route(_results) => Err(AxumErrorResponse::bad_request(
"Gateway monitoring results not supported for this endpoint",
)),
}
}
@@ -185,21 +131,18 @@ pub(crate) async fn submit_node_monitoring_results(
return Err(AxumErrorResponse::bad_request("invalid signature"));
}
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",
))
}
}
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",
))
}
MonitorResults::Route(_results) => Err(AxumErrorResponse::bad_request(
"Node monitoring results not supported for this endpoint",
)),
}
}
+13 -1
View File
@@ -102,6 +102,15 @@ pub(crate) struct Args {
#[clap(long)]
pub(crate) bind_address: Option<SocketAddr>,
/// account/address cache TTL: should be lower than epoch length (1 hour)
/// because, at worst, data will be stale for <epoch_length> + <cache_ttl> seconds
#[clap(long, env = "ADDRESS_CACHE_REFRESH_INTERVAL_S")]
pub(crate) address_cache_ttl_seconds: Option<u64>,
/// number of addresses that are cached on account/address endpoint
#[clap(long, env = "ADDRESS_CACHE_CAPACITY")]
pub(crate) address_cache_capacity: Option<u64>,
#[clap(hide = true, long, default_value_t = false)]
pub(crate) allow_illegal_ips: bool,
}
@@ -194,7 +203,10 @@ async fn start_nym_api_tasks_axum(config: &Config) -> anyhow::Result<ShutdownHan
let router = router.with_state(AppState {
nyxd_client: nyxd_client.clone(),
chain_status_cache: ChainStatusCache::new(DEFAULT_CHAIN_STATUS_CACHE_TTL),
address_info_cache: AddressInfoCache::new(),
address_info_cache: AddressInfoCache::new(
config.address_cache.time_to_live,
config.address_cache.capacity,
),
forced_refresh: ForcedRefresh::new(
config.topology_cacher.debug.node_describe_allow_illegal_ips,
),
+28
View File
@@ -54,6 +54,9 @@ const DEFAULT_TOPOLOGY_CACHE_INTERVAL: Duration = Duration::from_secs(30);
const DEFAULT_NODE_STATUS_CACHE_INTERVAL: Duration = Duration::from_secs(120);
const DEFAULT_CIRCULATING_SUPPLY_CACHE_INTERVAL: Duration = Duration::from_secs(3600);
pub(crate) const DEFAULT_ADDRESS_CACHE_TTL: Duration = Duration::from_secs(60 * 15);
pub(crate) const DEFAULT_ADDRESS_CACHE_CAPACITY: u64 = 1000;
pub(crate) const DEFAULT_NODE_DESCRIBE_CACHE_INTERVAL: Duration = Duration::from_secs(4500);
pub(crate) const DEFAULT_NODE_DESCRIBE_BATCH_SIZE: usize = 50;
@@ -111,6 +114,9 @@ pub struct Config {
#[serde(alias = "coconut_signer")]
pub ecash_signer: EcashSigner,
#[serde(skip)]
pub address_cache: AddressCacheConfig,
}
impl NymConfigTemplate for Config {
@@ -130,6 +136,7 @@ impl Config {
circulating_supply_cacher: Default::default(),
rewarding: Default::default(),
ecash_signer: EcashSigner::new_default(id.as_ref()),
address_cache: Default::default(),
}
}
@@ -179,6 +186,12 @@ impl Config {
if args.allow_illegal_ips {
self.topology_cacher.debug.node_describe_allow_illegal_ips = true
}
if let Some(address_cache_ttl) = args.address_cache_ttl {
self.address_cache.time_to_live = address_cache_ttl;
}
if let Some(address_cache_capacity) = args.address_cache_capacity {
self.address_cache.capacity = address_cache_capacity;
}
self
}
@@ -292,6 +305,21 @@ impl Base {
}
}
#[derive(Debug, PartialEq, Eq)]
pub struct AddressCacheConfig {
pub time_to_live: Duration,
pub capacity: u64,
}
impl Default for AddressCacheConfig {
fn default() -> Self {
Self {
time_to_live: DEFAULT_ADDRESS_CACHE_TTL,
capacity: DEFAULT_ADDRESS_CACHE_CAPACITY,
}
}
}
// this got separated into 2 structs so that we could have a sane `default` implementation for the latter
#[derive(Debug, Deserialize, PartialEq, Eq, Serialize)]
pub struct NetworkMonitor {
+9 -1
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: GPL-3.0-only
use crate::support::cli::{init, run};
use std::net::SocketAddr;
use std::{net::SocketAddr, time::Duration};
// Configuration that can be overridden.
pub(crate) struct OverrideConfig {
@@ -32,6 +32,9 @@ pub(crate) struct OverrideConfig {
/// default: `127.0.0.1:8080` in `debug` builds and `0.0.0.0:8080` in `release`
pub(crate) bind_address: Option<SocketAddr>,
pub(crate) address_cache_ttl: Option<Duration>,
pub(crate) address_cache_capacity: Option<u64>,
pub(crate) allow_illegal_ips: bool,
}
@@ -46,6 +49,9 @@ impl From<init::Args> for OverrideConfig {
announce_address: args.announce_address,
monitor_credentials_mode: Some(args.monitor_credentials_mode),
bind_address: args.bind_address,
// irrelevant for --init command because we set the value in --run
address_cache_ttl: None,
address_cache_capacity: None,
allow_illegal_ips: args.allow_illegal_ips,
}
}
@@ -62,6 +68,8 @@ impl From<run::Args> for OverrideConfig {
announce_address: args.announce_address,
monitor_credentials_mode: args.monitor_credentials_mode,
bind_address: args.bind_address,
address_cache_ttl: args.address_cache_ttl_seconds.map(Duration::from_secs),
address_cache_capacity: args.address_cache_capacity,
allow_illegal_ips: args.allow_illegal_ips,
}
}
+43 -240
View File
@@ -10,9 +10,8 @@ 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 nym_types::monitoring::NodeResult;
use sqlx::FromRow;
use std::collections::HashMap;
use time::{Date, OffsetDateTime};
use tracing::info;
@@ -52,25 +51,6 @@ 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(
@@ -96,29 +76,27 @@ impl StorageManager {
start_ts_secs: i64,
end_ts_secs: i64,
) -> Result<Vec<AvgMixnodeReliability>, sqlx::Error> {
let corrected_reliabilities = self
.calculate_corrected_node_reliabilities_for_interval(start_ts_secs, end_ts_secs)
.await
.map_err(|_e| sqlx::Error::PoolClosed)?; // Example: map anyhow::Error to sqlx::Error; adjust as needed
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()
{
avg_mix_reliabilities.push(AvgMixnodeReliability {
mix_id: corrected_node_info.node_id,
value: Some(corrected_node_info.reliability as f32),
});
}
}
Ok(avg_mix_reliabilities)
let result = sqlx::query_as!(
AvgMixnodeReliability,
r#"
SELECT
d.mix_id as "mix_id: NodeId",
AVG(s.reliability) as "value: f32"
FROM
mixnode_details d
JOIN
mixnode_status s on d.id = s.mixnode_details_id
WHERE
timestamp >= ? AND
timestamp <= ?
GROUP BY 1
"#,
start_ts_secs,
end_ts_secs
)
.fetch_all(&self.connection_pool)
.await?;
Ok(result)
}
pub(super) async fn get_all_avg_gateway_reliability_in_interval(
@@ -126,35 +104,27 @@ impl StorageManager {
start_ts_secs: i64,
end_ts_secs: i64,
) -> Result<Vec<AvgGatewayReliability>, sqlx::Error> {
let corrected_reliabilities = self
.calculate_corrected_node_reliabilities_for_interval(start_ts_secs, end_ts_secs)
.await
.map_err(|_e| sqlx::Error::PoolClosed)?; // Example: map anyhow::Error to sqlx::Error; adjust as needed
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()
{
let total_samples = corrected_node_info.pos_samples_in_interval
+ corrected_node_info.neg_samples_in_interval;
let reliability_value = if total_samples <= 3 {
100.0 // Default to 100% if 3 or fewer samples
} else {
corrected_node_info.reliability as f32
};
avg_gateway_reliabilities.push(AvgGatewayReliability {
node_id: corrected_node_info.node_id, // AvgGatewayReliability uses node_id
value: Some(reliability_value),
});
}
}
Ok(avg_gateway_reliabilities)
let result = sqlx::query_as!(
AvgGatewayReliability,
r#"
SELECT
d.node_id as "node_id: NodeId",
CASE WHEN count(*) > 3 THEN AVG(reliability) ELSE 100 END as "value: f32"
FROM
gateway_details d
JOIN
gateway_status s on d.id = s.gateway_details_id
WHERE
timestamp >= ? AND
timestamp <= ?
GROUP BY 1
"#,
start_ts_secs,
end_ts_secs
)
.fetch_all(&self.connection_pool)
.await?;
Ok(result)
}
/// Tries to obtain row id of given mixnode given its identity.
@@ -694,33 +664,6 @@ 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],
@@ -1386,146 +1329,6 @@ 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: i64,
layer2: i64,
layer3: i64,
gw: i64,
success: Option<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 as NodeId,
r.layer2 as NodeId,
r.layer3 as NodeId,
r.gw as NodeId,
r.success.unwrap_or_default(),
)
})
.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 {
+1 -11
View File
@@ -17,7 +17,7 @@ use crate::support::storage::models::{
};
use dashmap::DashMap;
use nym_mixnet_contract_common::NodeId;
use nym_types::monitoring::{NodeResult, RouteResult};
use nym_types::monitoring::NodeResult;
use sqlx::sqlite::{SqliteAutoVacuum, SqliteSynchronous};
use sqlx::ConnectOptions;
use std::path::Path;
@@ -835,16 +835,6 @@ 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
+7 -13
View File
@@ -17,17 +17,11 @@ pub(crate) struct AddressInfoCache {
}
impl AddressInfoCache {
pub(crate) fn new() -> Self {
// epoch duration = 1 hour
// cache TTL is slightly lower than that to avoid too stale data in case
// cache was refreshed JUST BEFORE epoch transition
let cache_ttl = Duration::from_secs(60 * 30);
let max_capacity = 1000;
pub(crate) fn new(cache_ttl: Duration, capacity: u64) -> Self {
AddressInfoCache {
inner: Cache::builder()
.time_to_live(cache_ttl)
.max_capacity(max_capacity)
.max_capacity(capacity)
.build(),
}
}
@@ -108,13 +102,13 @@ impl AddressInfoCache {
address: account_id.to_string(),
balance: balance.into(),
delegations: delegation_data
.delegations()
.into_iter()
.map(|d| NyxAccountDelegationDetails {
delegated: d.amount,
height: d.height,
node_id: d.node_id,
proxy: d.proxy,
delegated: d.details().amount.clone(),
height: d.details().height,
node_id: d.details().node_id,
proxy: d.details().proxy.clone(),
node_bonded: d.is_node_bonded(),
})
.collect(),
accumulated_rewards,
@@ -8,10 +8,9 @@ use crate::{
};
use cosmwasm_std::{Coin, Decimal};
use nym_mixnet_contract_common::NodeRewarding;
use nym_topology::NodeId;
use nym_validator_client::nyxd::AccountId;
use std::collections::{HashMap, HashSet};
use tracing::warn;
use tracing::error;
pub(crate) struct AddressDataCollector {
nyxd_client: crate::nyxd::Client,
@@ -56,18 +55,18 @@ impl AddressDataCollector {
Ok(balance)
}
pub(crate) async fn get_delegations(&mut self) -> AxumResult<AddressDelegationInfo> {
pub(crate) async fn get_delegations(&mut self) -> AxumResult<Vec<AddressDelegationInfo>> {
let og_delegations = self
.nyxd_client
.get_all_delegator_delegations(&self.account_id)
.await?;
.await?
.into_iter()
.map(|delegation| (delegation.node_id, delegation))
.collect::<HashMap<_, _>>();
let delegated_to_nodes = og_delegations
.iter()
.map(|d| d.node_id)
.collect::<HashSet<_>>();
let mut node_delegation_info = Vec::new();
let nym_nodes = self
let delegated_to_nodes_bonded = self
.nym_contract_cache
.all_cached_nym_nodes()
.await
@@ -85,61 +84,75 @@ impl AddressDataCollector {
// add to totals
self.total_value += pending_operator_reward;
}
if delegated_to_nodes.contains(&node_details.node_id()) {
Some((
node_details.node_id(),
// avoid cloning node data which we don't need
(
node_details.rewarding_details.clone(),
node_details.is_unbonding(),
),
))
if let Some(delegation) = og_delegations.get(&node_details.node_id()) {
node_delegation_info.push(AddressDelegationInfo {
details: delegation.clone(),
node_reward_info: NodeBondStatus::Bonded {
rewarding_info: node_details.rewarding_details.to_owned(),
unbonding: node_details.is_unbonding(),
},
});
Some(node_details.node_id())
} else {
None
}
})
.collect::<HashMap<_, _>>();
.collect::<HashSet<_>>();
Ok(AddressDelegationInfo {
delegations: og_delegations,
delegated_to_nodes: nym_nodes,
})
for (node_id, delegation) in og_delegations {
if !delegated_to_nodes_bonded.contains(&node_id) {
node_delegation_info.push(AddressDelegationInfo {
details: delegation.clone(),
node_reward_info: NodeBondStatus::UnBonded,
});
}
}
Ok(node_delegation_info)
}
pub(crate) async fn calculate_rewards(
&mut self,
delegation_data: &AddressDelegationInfo,
delegation_data: &Vec<AddressDelegationInfo>,
) -> AxumResult<Vec<NyxAccountDelegationRewardDetails>> {
let mut accumulated_rewards = Vec::new();
for delegation in delegation_data.delegations.iter() {
let node_id = &delegation.node_id;
for delegation in delegation_data {
let node_id = delegation.details.node_id;
if let Some((rewarding_details, is_unbonding)) =
delegation_data.delegated_to_nodes.get(node_id)
{
match rewarding_details.determine_delegation_reward(delegation) {
Ok(delegation_reward) => {
let reward = NyxAccountDelegationRewardDetails {
node_id: delegation.node_id,
rewards: decimal_to_coin(delegation_reward, &self.base_denom),
amount_staked: delegation.amount.clone(),
node_still_fully_bonded: !is_unbonding,
};
// 4. sum the rewards and delegations
self.total_delegations += delegation.amount.amount.u128();
self.total_value += delegation.amount.amount.u128();
self.total_value += reward.rewards.amount.u128();
self.claimable_rewards += reward.rewards.amount.u128();
match &delegation.node_reward_info {
NodeBondStatus::Bonded {
rewarding_info,
unbonding,
} => {
match rewarding_info.determine_delegation_reward(&delegation.details) {
Ok(delegation_reward) => {
let reward = NyxAccountDelegationRewardDetails {
node_id,
rewards: decimal_to_coin(delegation_reward, &self.base_denom),
amount_staked: delegation.details.amount.clone(),
node_still_fully_bonded: !unbonding,
};
// 4. sum the rewards and delegations
self.total_delegations += delegation.details.amount.amount.u128();
self.total_value += delegation.details.amount.amount.u128();
self.total_value += reward.rewards.amount.u128();
self.claimable_rewards += reward.rewards.amount.u128();
accumulated_rewards.push(reward);
}
Err(err) => {
warn!(
"Couldn't determine delegations for {} on node {}: {}",
&self.account_id, node_id, err
)
accumulated_rewards.push(reward);
}
Err(err) => {
error!(
"Couldn't determine delegations for {} on node {}: {}",
&self.account_id, node_id, err
)
}
}
}
NodeBondStatus::UnBonded => {
// directory cache doesn't store node details required to
// calculate rewarding for unbonded nodes
}
}
}
@@ -171,17 +184,27 @@ impl AddressDataCollector {
}
pub(crate) struct AddressDelegationInfo {
delegations: Vec<nym_mixnet_contract_common::Delegation>,
delegated_to_nodes: HashMap<NodeId, RewardAndBondInfo>,
details: nym_mixnet_contract_common::Delegation,
node_reward_info: NodeBondStatus,
}
impl AddressDelegationInfo {
pub(crate) fn delegations(self) -> Vec<nym_mixnet_contract_common::Delegation> {
self.delegations
pub(crate) fn details(&self) -> &nym_mixnet_contract_common::Delegation {
&self.details
}
pub(crate) fn is_node_bonded(&self) -> bool {
matches!(self.node_reward_info, NodeBondStatus::Bonded { .. })
}
}
type RewardAndBondInfo = (NodeRewarding, bool);
pub(crate) enum NodeBondStatus {
Bonded {
rewarding_info: NodeRewarding,
unbonding: bool,
},
UnBonded,
}
fn decimal_to_coin(decimal: Decimal, denom: impl Into<String>) -> Coin {
Coin::new(decimal.to_uint_floor(), denom)
@@ -189,6 +212,7 @@ fn decimal_to_coin(decimal: Decimal, denom: impl Into<String>) -> Coin {
#[cfg(test)]
mod test {
use super::*;
#[tokio::test]
+4
View File
@@ -21,6 +21,7 @@ pub struct NyxAccountDelegationDetails {
pub height: u64,
#[schema(value_type = Option<String>)]
pub proxy: Option<Addr>,
pub node_bonded: bool,
}
#[derive(Clone, Debug, Serialize, Deserialize, utoipa::ToSchema, utoipa::ToResponse)]
@@ -41,6 +42,9 @@ pub struct NyxAccountDetails {
#[schema(value_type = CoinSchema)]
pub total_value: Coin,
pub delegations: Vec<NyxAccountDelegationDetails>,
/// Shows rewards from delegations to **currently** bonded nodes.
/// Rewards from nodes that user delegated to, but were later unbonded,
/// are claimable, but not shown here.
pub accumulated_rewards: Vec<NyxAccountDelegationRewardDetails>,
#[schema(value_type = String)]
pub total_delegations: Coin,
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "nym-network-monitor"
version = "1.1.0"
version = "1.0.2"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
-87
View File
@@ -4,14 +4,6 @@ Monitors the Nym network by sending itself packages across the mixnet.
Network monitor is running two tokio tasks, one manages mixnet clients and another manages monitoring itself. Monitor is designed to be driven externally, via an HTTP api. This means that it does not do any monitoring unless driven by something like [`locust`](https://locust.io/). This allows us to tailor the load externally, potentially distributing it across multiple monitors.
## Features
- **Continuous Monitoring**: Periodically sends test packets through the network
- **Node Performance**: Tracks individual node reliability metrics
- **Route Performance**: Records route-level success rates through specific node combinations
- **Multi-API Submission**: Capable of submitting metrics to multiple API endpoints (fanout)
- **Force Routing**: Can force packets through all mixnet nodes for comprehensive testing
### Client manager
On start network monitor will spawn `C` clients, with 10 being the default. Random client is dropped every `T`, defaults to 60 seconds, and a new one is created. Clients chose a random gateway to connect to the mixnet. Meaning that on average all gateways will be tested in `NUMBER_OF_GATEWAYS/N*T`, assuming at least one request per client per T.
@@ -48,87 +40,8 @@ Options:
-m, --mixnet-timeout <MIXNET_TIMEOUT> [default: 10]
--generate-key-pair
--private-key <PRIVATE_KEY>
--database-url <DATABASE_URL> SQLite database URL
--nym-apis <NYM_APIS> Comma-separated list of Nym API URLs
-h, --help Print help
-V, --version Print version
```
## Metrics Collection & Reporting
### Node Metrics
The Network Monitor tracks performance metrics for individual nodes:
- **Reliability**: Percentage of successful packet handling
- **Failure Sequences**: Tracking consecutive failures
- **Volume**: Number of packets handled
### Route Metrics
Since version 1.1.0, the Network Monitor also tracks route-level metrics:
- **Route Success Rates**: Tracking which specific combinations of nodes have successful packet delivery
- **Layer Analysis**: Identifying weak points in specific network layers
- **Path Correction**: Improved algorithms for attributing failures to specific nodes
### Metrics Fanout
The Network Monitor can submit metrics to multiple API endpoints simultaneously:
1. Metrics are collected during each monitoring cycle
2. The collected metrics are submitted to each configured API endpoint
3. This provides redundancy and allows for distributed metrics collection
To enable metrics fanout, use the `--nym-apis` parameter with a comma-separated list of API URLs:
```bash
cargo run -p nym-network-monitor -- --nym-apis https://api1.example.com,https://api2.example.com
```
## Route Data Structure
Route metrics use the following data structure:
```rust
// Route performance data
pub struct RouteResult {
pub layer1: u32, // NodeId of layer 1 mixnode
pub layer2: u32, // NodeId of layer 2 mixnode
pub layer3: u32, // NodeId of layer 3 mixnode
pub gw: u32, // NodeId of gateway
pub success: bool, // Whether the packet was successfully delivered
}
```
## Forced Routing
To ensure comprehensive testing of all nodes in the network, the Monitor supports forcing packets through all available nodes:
- Each node is assigned to a specific layer (1, 2, or 3) deterministically
- This ensures all nodes participate in route testing
- The routing algorithm cycles through all possible node combinations
Since version 1.1.0, Network Monitor automatically forces all available nodes to be active and distributes them evenly across the three layers (Layer 1, Layer 2, and Layer 3). This ensures every node in the network participates in testing, providing more comprehensive coverage and better metrics for all nodes, not just the popular ones.
## Node Performance Calculation
The Network Monitor uses a sophisticated algorithm for attributing failures to specific nodes:
1. For successful packet deliveries, all nodes in the path receive a positive sample
2. For failed deliveries:
- Nodes with more than 2 consecutive failures are considered "guilty"
- If no node is clearly guilty, all nodes in the path receive negative samples
3. Final node reliability is calculated as: positive_samples / (positive_samples + negative_samples)
## Changelog
### Version 1.1.0
- Added route-level metrics tracking and submission
- Implemented metrics fanout to multiple API endpoints
- Forced routing through all available nodes for comprehensive testing
- Improved reliability corrections with consecutive failure tracking
- Updated data structures for better metrics organization
### Version 1.0.2
- Initial public release with basic monitoring capabilities
+45 -99
View File
@@ -5,13 +5,11 @@ use std::{
use anyhow::Result;
use futures::{pin_mut, stream::FuturesUnordered, StreamExt};
use log::{debug, error, info, warn};
use log::{debug, error, info};
use nym_sphinx::chunking::{monitoring, SentFragment};
use nym_topology::{NymRouteProvider, RoutingNode};
use nym_types::monitoring::{MonitorMessage, MonitorResults, NodeResult, RouteResult};
use nym_validator_client::nym_api::routes::{
API_VERSION, STATUS, SUBMIT_GATEWAY, SUBMIT_NODE, SUBMIT_ROUTE,
};
use nym_types::monitoring::{MonitorMessage, NodeResult};
use nym_validator_client::nym_api::routes::{API_VERSION, STATUS, SUBMIT_GATEWAY, SUBMIT_NODE};
use rand::SeedableRng;
use rand_chacha::ChaCha8Rng;
use serde::{Deserialize, Serialize};
@@ -19,7 +17,7 @@ use tokio::task::JoinHandle;
use tokio_postgres::{binary_copy::BinaryCopyInWriter, types::Type, Client, NoTls};
use utoipa::ToSchema;
use crate::{NYM_API_URLS, PRIVATE_KEY, TOPOLOGY};
use crate::{NYM_API_URL, PRIVATE_KEY, TOPOLOGY};
struct HydratedRoute {
mix_nodes: Vec<RoutingNode>,
@@ -441,7 +439,6 @@ async fn db_connection(database_url: Option<&String>) -> Result<Option<(Client,
Ok(None)
}
}
pub async fn submit_metrics_to_db(database_url: Option<&String>) -> anyhow::Result<()> {
if let Some((client, handle)) = db_connection(database_url).await? {
let client = Arc::new(client);
@@ -494,100 +491,49 @@ pub async fn submit_metrics(database_url: Option<&String>) -> anyhow::Result<()>
}
if let Some(private_key) = PRIVATE_KEY.get() {
if let Some(nym_api_urls) = NYM_API_URLS.get() {
info!("Submitting metrics to {} nym apis", nym_api_urls.len());
for nym_api_url in nym_api_urls {
info!("Submitting metrics to {}", nym_api_url);
let node_stats = monitor_mixnode_results().await?;
let gateway_stats = monitor_gateway_results().await?;
let client = reqwest::Client::new();
let node_stats = monitor_mixnode_results().await?;
let gateway_stats = monitor_gateway_results().await?;
let node_submit_url =
format!("{}/api/{API_VERSION}/{STATUS}/{SUBMIT_NODE}", nym_api_url);
let gateway_submit_url = format!(
"{}/api/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}",
nym_api_url
info!("Submitting metrics to {}", *NYM_API_URL);
let client = reqwest::Client::new();
let node_submit_url = format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_NODE}", &*NYM_API_URL);
let gateway_submit_url =
format!("{}/{API_VERSION}/{STATUS}/{SUBMIT_GATEWAY}", &*NYM_API_URL);
info!("Submitting {} mixnode measurements", node_stats.len());
node_stats
.chunks(10)
.map(|chunk| {
let monitor_message = MonitorMessage::new(chunk.to_vec(), private_key);
client.post(&node_submit_url).json(&monitor_message).send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()?;
info!("Submitting {} gateway measurements", gateway_stats.len());
gateway_stats
.chunks(10)
.map(|chunk| {
let monitor_message = MonitorMessage::new(
chunk.to_vec(),
PRIVATE_KEY.get().expect("We've set this!"),
);
let route_submit_url =
format!("{}/api/{API_VERSION}/{STATUS}/{SUBMIT_ROUTE}", nym_api_url);
info!("Submitting {} mixnode measurements", node_stats.len());
node_stats
.chunks(10)
.map(|chunk| {
let monitor_results = MonitorResults::Node(chunk.to_vec());
let monitor_message = MonitorMessage::new(monitor_results, private_key);
client.post(&node_submit_url).json(&monitor_message).send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()?;
info!("Submitting {} gateway measurements", gateway_stats.len());
gateway_stats
.chunks(10)
.map(|chunk| {
let monitor_results = MonitorResults::Node(chunk.to_vec());
let monitor_message = MonitorMessage::new(
monitor_results,
PRIVATE_KEY.get().expect("We've set this!"),
);
client
.post(&gateway_submit_url)
.json(&monitor_message)
.send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()?;
let network_account = NetworkAccount::finalize()?;
let accounting_routes = network_account.accounting_routes;
info!("Submitting {} accounting routes", accounting_routes.len());
match accounting_routes
.chunks(10)
.map(|chunk| {
let route_results = chunk
.iter()
.map(|route| {
RouteResult::new(
route.mix_nodes.0,
route.mix_nodes.1,
route.mix_nodes.2,
route.gateway_node,
route.success,
)
})
.collect::<Vec<RouteResult>>();
let monitor_results = MonitorResults::Route(route_results);
let monitor_message = MonitorMessage::new(monitor_results, private_key);
client.post(&route_submit_url).json(&monitor_message).send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()
{
Ok(_) => info!(
"Successfully submitted accounting routes to {}",
nym_api_url
),
Err(e) => error!(
"Error submitting accounting routes to {}: {}",
nym_api_url, e
),
};
}
}
} else {
warn!("No private key or nym api urls found");
client
.post(&gateway_submit_url)
.json(&monitor_message)
.send()
})
.collect::<FuturesUnordered<_>>()
.collect::<Vec<Result<_, _>>>()
.await
.into_iter()
.collect::<Result<Vec<_>, _>>()?;
}
NetworkAccount::empty_buffers();
+7 -23
View File
@@ -10,9 +10,10 @@ use nym_network_defaults::setup_env;
use nym_network_defaults::var_names::NYM_API;
use nym_sdk::mixnet::{self, MixnetClient};
use nym_sphinx::chunking::monitoring;
use nym_topology::{HardcodedTopologyProvider, NymTopology, Role};
use nym_topology::{HardcodedTopologyProvider, NymTopology};
use std::fs::File;
use std::io::Write;
use std::sync::LazyLock;
use std::time::Duration;
use std::{
collections::VecDeque,
@@ -24,7 +25,9 @@ use tokio::sync::OnceCell;
use tokio::{signal::ctrl_c, sync::RwLock};
use tokio_util::sync::CancellationToken;
static NYM_API_URLS: OnceCell<Vec<String>> = OnceCell::const_new();
static NYM_API_URL: LazyLock<String> = LazyLock::new(|| {
std::env::var(NYM_API).unwrap_or_else(|_| panic!("{} env var not set", NYM_API))
});
static MIXNET_TIMEOUT: OnceCell<u64> = OnceCell::const_new();
static TOPOLOGY: OnceCell<NymTopology> = OnceCell::const_new();
@@ -135,9 +138,6 @@ struct Args {
#[arg(long, env = "DATABASE_URL")]
database_url: Option<String>,
#[arg(long, env = "NYM_APIS")]
nym_apis: Option<Vec<String>>,
}
fn generate_key_pair() -> Result<()> {
@@ -155,7 +155,7 @@ fn generate_key_pair() -> Result<()> {
Ok(())
}
async fn nym_topology_forced_all_from_env() -> anyhow::Result<NymTopology> {
async fn nym_topology_from_env() -> anyhow::Result<NymTopology> {
let api_url = std::env::var(NYM_API)?;
info!("Generating topology from {api_url}");
@@ -172,18 +172,6 @@ async fn nym_topology_forced_all_from_env() -> anyhow::Result<NymTopology> {
let mut topology = NymTopology::new_empty(rewarded_set);
topology.add_skimmed_nodes(&nodes);
let node_ids = topology.node_details().keys().cloned().collect::<Vec<_>>();
// Force all nodes to active to participate in route selection
for (idx, node_id) in node_ids.iter().enumerate() {
match idx % 3 {
0 => topology.force_set_active(*node_id, Role::Layer1),
1 => topology.force_set_active(*node_id, Role::Layer2),
2 => topology.force_set_active(*node_id, Role::Layer3),
_ => unreachable!(), // Unreachable since idx % 3 can only be 0, 1, or 2
}
}
Ok(topology)
}
@@ -212,15 +200,11 @@ async fn main() -> Result<()> {
PRIVATE_KEY.set(pk).ok();
}
if let Some(nym_apis) = args.nym_apis {
NYM_API_URLS.set(nym_apis).ok();
}
TOPOLOGY
.set(if let Some(topology_file) = args.topology {
NymTopology::new_from_file(topology_file)?
} else {
nym_topology_forced_all_from_env().await?
nym_topology_from_env().await?
})
.ok();
+2 -2
View File
@@ -3,7 +3,7 @@
[package]
name = "nym-node"
version = "1.11.0"
version = "1.13.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
@@ -84,7 +84,7 @@ tower-http = { workspace = true, features = ["fs"] }
utoipa = { workspace = true, features = ["axum_extras", "time"] }
utoipa-swagger-ui = { workspace = true, features = ["axum"] }
nym-http-api-common = { path = "../common/http-api-common", features = ["utoipa"] }
nym-http-api-common = { path = "../common/http-api-common", features = ["utoipa", "output", "middleware"] }
nym-node-requests = { path = "nym-node-requests", default-features = false, features = ["openapi"] }
nym-node-metrics = { path = "nym-node-metrics" }
+21 -3
View File
@@ -158,13 +158,13 @@ pub enum NymNodeError {
BloomfilterIoFailure { source: io::Error, path: PathBuf },
#[error(transparent)]
GatewayFailure(#[from] nym_gateway::GatewayError),
GatewayFailure(Box<nym_gateway::GatewayError>),
#[error(transparent)]
GatewayTasksStartupFailure(Box<dyn std::error::Error + Send + Sync>),
#[error(transparent)]
EntryGatewayFailure(#[from] EntryGatewayError),
EntryGatewayFailure(Box<EntryGatewayError>),
#[error(transparent)]
ServiceProvidersFailure(#[from] ServiceProvidersError),
@@ -177,6 +177,18 @@ pub enum NymNodeError {
FailedUpgrade,
}
impl From<EntryGatewayError> for NymNodeError {
fn from(error: EntryGatewayError) -> Self {
NymNodeError::EntryGatewayFailure(Box::new(error))
}
}
impl From<nym_gateway::GatewayError> for NymNodeError {
fn from(error: nym_gateway::GatewayError) -> Self {
NymNodeError::GatewayFailure(Box::new(error))
}
}
impl NymNodeError {
pub fn config_validation_failure<S: Into<String>>(error: S) -> Self {
NymNodeError::ConfigValidationFailure {
@@ -219,7 +231,13 @@ pub enum EntryGatewayError {
},
#[error("entry gateway failure: {0}")]
External(#[from] nym_gateway::GatewayError),
External(Box<nym_gateway::GatewayError>),
}
impl From<nym_gateway::GatewayError> for EntryGatewayError {
fn from(error: nym_gateway::GatewayError) -> Self {
EntryGatewayError::External(Box::new(error))
}
}
#[derive(Debug, Error)]
@@ -0,0 +1,21 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO active_device (\n day,\n device_id,\n os_type,\n os_version,\n architecture,\n app_version,\n user_agent,\n from_mixnet)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8)\n ON CONFLICT (device_id, day) DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Date",
"Text",
"Text",
"Text",
"Text",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "13bf07e42c49ea365e816eb94e4e4f607989ee95f68a0fcd95bc4a53f4e79cbb"
}
@@ -0,0 +1,19 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO connection_stats (\n received_at,\n connection_time_ms,\n two_hop,\n source_ip,\n country_code,\n from_mixnet) VALUES ($1::timestamptz, $2, $3, $4, $5, $6)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Timestamptz",
"Int4",
"Bool",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "dce9f3dae7ae0dc5953d1f69e843bea9553fb05a2f656cfff6598f12a21c99ba"
}
+55
View File
@@ -0,0 +1,55 @@
# Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
# SPDX-License-Identifier: Apache-2.0
[package]
name = "nym-statistics-api"
version = "0.1.0"
authors.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
edition.workspace = true
license.workspace = true
rust-version.workspace = true
[dependencies]
anyhow.workspace = true
axum = { workspace = true, features = ["tokio", "macros"] }
axum-extra = { workspace = true, features = ["typed-header"] }
celes.workspace = true
clap = { workspace = true, features = ["cargo", "derive", "env", "string"] }
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
sqlx = { workspace = true, features = [
"runtime-tokio-rustls",
"postgres",
"time",
] }
time.workspace = true
tokio = { workspace = true, features = ["rt-multi-thread"] }
tokio-util.workspace = true
tracing.workspace = true
tracing-subscriber = { workspace = true, features = ["env-filter"] }
tower-http = { workspace = true, features = ["cors", "trace"] }
url.workspace = true
utoipa = { workspace = true, features = ["axum_extras", "time"] }
utoipa-swagger-ui = { workspace = true, features = ["axum"] }
utoipauto.workspace = true
#internal
nym-bin-common = { path = "../common/bin-common" }
nym-http-api-client = { path = "../common/http-api-client" }
nym-http-api-common = { path = "../common/http-api-common", features = [
"middleware",
] }
nym-statistics-common = { path = "../common/statistics", features = [
"openapi",
] }
nym-task = { path = "../common/task" }
nym-validator-client = { path = "../common/client-libs/validator-client" }
[build-dependencies]
anyhow = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
sqlx = { workspace = true, features = ["runtime-tokio-rustls", "postgres"] }
+28
View File
@@ -0,0 +1,28 @@
# Nym-statistics-api
A simple API to collect and store statistics sent by nym-vpn-client.
## Build instructions
The statistics API is backed by a PostgreSQL database so you'll need a PostgreSQL server running if you want to add migrations or add/modify SQL queries. I recommend https://postgresapp.com on MacOS, very easy to use. If you're on another OS, it's up to you.
Assuming your database is running at `postgresql://user:password@host:port/database_name` you'll likely need to run the following :
```bash
DATABASE_URL="postgresql://user:password@host:port/database_name"
# if you don't have an existing datase
sqlx database create --database-url $DATABASE_URL
sqlx migrate run --database-url $DATABASE_URL
# reset it if you messed with migrations while developping
sqlx database reset --database-url $DATABASE_URL
# or just run new migrations
sqlx migrate run --database-url $DATABASE_URL
# then prepare queries for offline build mode
cargo sqlx prepare --database-url $DATABASE_URL
```
This should allow `cargo build` without having any postgreSQL server running.
Be sure to add the `.sqlx` directory to version control
@@ -0,0 +1,24 @@
CREATE TABLE active_device (
day DATE NOT NULL,
device_id TEXT NOT NULL,
os_type TEXT,
os_version TEXT,
architecture TEXT,
app_version TEXT,
user_agent TEXT,
from_mixnet BOOLEAN,
PRIMARY KEY (device_id, day)
);
CREATE TABLE connection_stats (
received_at TIMESTAMP WITH TIME ZONE NOT NULL,
connection_time_ms INTEGER,
two_hop BOOLEAN,
source_ip TEXT,
country_code TEXT,
from_mixnet BOOLEAN
);
CREATE INDEX idx_active_device_day ON active_device (day);
+30
View File
@@ -0,0 +1,30 @@
#!/bin/bash
set -e
export PGUSER="nym"
export PGPASSWORD="password1"
export PGPORT="5432"
export DB_NAME="nym_statistics_api"
export DATABASE_URL="postgres://${PGUSER}:${PGPASSWORD}@localhost:${PGPORT}/${DB_NAME}"
cat <<EOF > .env
SQLX_OFFLINE=true
POSTGRES_USER=$PGUSER
POSTGRES_PASSWORD=$PGPASSWORD
PGPORT=$PGPORT
DB_NAME=$DB_NAME
DATABASE_URL=$DATABASE_URL
EOF
docker run --rm -it \
--name ${DB_NAME} \
-e POSTGRES_USER=${PGUSER} \
-e POSTGRES_PASSWORD=${PGPASSWORD} \
-e POSTGRES_DB=${DB_NAME} \
-p ${PGPORT}:${PGPORT} \
postgres
# sqlx migrate run
# cargo sqlx prepare
+38
View File
@@ -0,0 +1,38 @@
use clap::Parser;
use nym_bin_common::bin_info;
use std::sync::OnceLock;
use url::Url;
// Helper for passing LONG_VERSION to clap
fn pretty_build_info_static() -> &'static str {
static PRETTY_BUILD_INFORMATION: OnceLock<String> = OnceLock::new();
PRETTY_BUILD_INFORMATION.get_or_init(|| bin_info!().pretty_print())
}
#[derive(Clone, Debug, Parser)]
#[clap(author = "Nymtech", version, long_version = pretty_build_info_static(), about)]
pub(crate) struct Cli {
/// URL for the NYM API to get a network view from
#[clap(long, env = "NYM_API_URL")]
pub(crate) nym_api_url: Option<Url>,
/// HTTP port on which to run statistics api.
#[clap(long, default_value_t = 8000, env = "NYM_STATISTICS_API_HTTP_PORT")]
pub(crate) http_port: u16,
/// Connection url for the database.
#[clap(long, env = "DATABASE_URL")]
pub(crate) database_url: String,
/// Username for the database.
#[clap(long, env = "POSTGRES_USER")]
pub(crate) username: String,
/// Password for the database.
#[clap(long, env = "POSTGRES_PASSWORD")]
pub(crate) password: String,
/// PgSQL port for the database.
#[clap(long, default_value_t = 5432, env = "PGPORT")]
pub(crate) pg_port: u16,
}
+74
View File
@@ -0,0 +1,74 @@
use anyhow::anyhow;
use axum::{response::Redirect, Router};
use nym_http_api_common::middleware::logging::log_request_info;
use tokio::net::ToSocketAddrs;
use tower_http::cors::CorsLayer;
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
use crate::http::{server::HttpServer, state::AppState};
pub(crate) mod stats;
pub(crate) struct RouterBuilder {
unfinished_router: Router<AppState>,
}
impl RouterBuilder {
pub(crate) fn with_default_routes() -> Self {
let router = Router::new()
.merge(
SwaggerUi::new("/swagger")
.url("/api-docs/openapi.json", super::api_docs::ApiDoc::openapi()),
)
.route(
"/",
axum::routing::get(|| async { Redirect::permanent("/swagger") }),
)
.nest("/v1", Router::new().nest("/stats", stats::routes()));
Self {
unfinished_router: router,
}
}
pub(crate) fn with_state(self, state: AppState) -> RouterWithState {
RouterWithState {
router: self.finalize_routes().with_state(state),
}
}
fn finalize_routes(self) -> Router<AppState> {
// layers added later wrap earlier layers
self.unfinished_router
// CORS layer needs to wrap other API layers
.layer(setup_cors())
// logger should be outermost layer
.layer(axum::middleware::from_fn(log_request_info))
}
}
pub(crate) struct RouterWithState {
router: Router,
}
impl RouterWithState {
pub(crate) async fn build_server<A: ToSocketAddrs>(
self,
bind_address: A,
) -> anyhow::Result<HttpServer> {
tokio::net::TcpListener::bind(bind_address)
.await
.map(|listener| HttpServer::new(self.router, listener))
.map_err(|err| anyhow!("Couldn't bind to address due to {}", err))
}
}
fn setup_cors() -> CorsLayer {
use axum::http::Method;
CorsLayer::new()
.allow_origin(tower_http::cors::Any)
.allow_methods([Method::POST, Method::GET, Method::PATCH, Method::OPTIONS])
.allow_headers(tower_http::cors::Any)
.allow_credentials(false)
}
+63
View File
@@ -0,0 +1,63 @@
use std::net::SocketAddr;
use axum::{
extract::{ConnectInfo, State},
Json, Router,
};
use axum_extra::{headers::UserAgent, TypedHeader};
use nym_statistics_common::report::vpn_client::VpnClientStatsReport;
use tracing::debug;
use crate::{
http::{
error::{HttpError, HttpResult},
state::AppState,
},
storage::models::{ConnectionInfoDto, DailyActiveDeviceDto},
};
pub(crate) fn routes() -> Router<AppState> {
Router::new().route("/report", axum::routing::post(submit_stats_report))
}
#[utoipa::path(
post,
request_body = VpnClientStatsReport,
tag = "Stats",
path = "/report",
context_path = "/v1/stats",
responses(
(status = 200)
)
)]
#[tracing::instrument(level = "info", skip_all)]
async fn submit_stats_report(
State(mut state): State<AppState>,
ConnectInfo(addr): ConnectInfo<SocketAddr>,
TypedHeader(user_agent): TypedHeader<UserAgent>,
Json(report): Json<VpnClientStatsReport>,
) -> HttpResult<Json<()>> {
let now = time::OffsetDateTime::now_utc();
let gateway_record = state.network_view().get_country_by_ip(&addr.ip()).await;
let from_mixnet = gateway_record.is_some();
let maybe_location = gateway_record.unwrap_or_default();
if from_mixnet {
debug!("Received a report from the network");
} else {
debug!("Received a report from outside of the network");
}
let active_device = DailyActiveDeviceDto::new(now, &report, user_agent, from_mixnet);
let maybe_connection_info =
ConnectionInfoDto::maybe_new(now, &report, addr, maybe_location, from_mixnet);
state
.storage()
.store_vpn_client_report(active_device, maybe_connection_info)
.await
.map_err(HttpError::internal_with_logging)?;
Ok(Json(()))
}
+10
View File
@@ -0,0 +1,10 @@
use utoipa::OpenApi;
use utoipauto::utoipauto;
// manually import external structs which are behind feature flags because they
// can't be automatically discovered
// https://github.com/ProbablyClem/utoipauto/issues/13#issuecomment-1974911829
#[utoipauto(paths = "./nym-statistics-api/src")]
#[derive(OpenApi)]
#[openapi(info(title = "Nym Statistics API"), tags(), components(schemas()))]
pub(super) struct ApiDoc;
+28
View File
@@ -0,0 +1,28 @@
use std::fmt::Display;
pub(crate) type HttpResult<T> = Result<T, HttpError>;
pub(crate) struct HttpError {
message: String,
status: axum::http::StatusCode,
}
impl HttpError {
pub(crate) fn internal_with_logging(msg: impl Display) -> Self {
tracing::error!("{}", msg.to_string());
Self::internal()
}
pub(crate) fn internal() -> Self {
Self {
message: serde_json::json!({"message": "Internal server error"}).to_string(),
status: axum::http::StatusCode::INTERNAL_SERVER_ERROR,
}
}
}
impl axum::response::IntoResponse for HttpError {
fn into_response(self) -> axum::response::Response {
(self.status, self.message).into_response()
}
}
+5
View File
@@ -0,0 +1,5 @@
pub(crate) mod api;
pub(crate) mod api_docs;
pub(crate) mod error;
pub(crate) mod server;
pub(crate) mod state;
+48
View File
@@ -0,0 +1,48 @@
use axum::Router;
use core::net::SocketAddr;
use nym_task::ShutdownToken;
use tokio::net::TcpListener;
use crate::{
http::{api::RouterBuilder, state::AppState},
network_view::NetworkView,
storage::StatisticsStorage,
};
pub(crate) async fn build_http_api(
storage: StatisticsStorage,
cached_network: NetworkView,
http_port: u16,
) -> anyhow::Result<HttpServer> {
let router_builder = RouterBuilder::with_default_routes();
let state = AppState::new(storage, cached_network).await;
let router = router_builder.with_state(state);
let bind_addr = format!("0.0.0.0:{}", http_port);
tracing::info!("Binding server to {bind_addr}");
router.build_server(bind_addr).await
}
pub(crate) struct HttpServer {
router: Router,
listener: TcpListener,
}
impl HttpServer {
pub(crate) fn new(router: Router, listener: TcpListener) -> Self {
Self { router, listener }
}
pub(crate) async fn run(self, shutdown_token: ShutdownToken) -> std::io::Result<()> {
// into_make_service_with_connect_info allows us to see client ip address
axum::serve(
self.listener,
self.router
.into_make_service_with_connect_info::<SocketAddr>(),
)
.with_graceful_shutdown(async move { shutdown_token.cancelled().await })
.await
}
}
+24
View File
@@ -0,0 +1,24 @@
use crate::{network_view::NetworkView, storage::StatisticsStorage};
#[derive(Debug, Clone)]
pub(crate) struct AppState {
storage_manager: StatisticsStorage,
network_view: NetworkView,
}
impl AppState {
pub(crate) async fn new(storage_manager: StatisticsStorage, network_view: NetworkView) -> Self {
Self {
storage_manager,
network_view,
}
}
pub(crate) fn storage(&mut self) -> &mut StatisticsStorage {
&mut self.storage_manager
}
pub(crate) fn network_view(&self) -> &NetworkView {
&self.network_view
}
}
+44
View File
@@ -0,0 +1,44 @@
use tracing::level_filters::LevelFilter;
use tracing_subscriber::{filter::Directive, EnvFilter};
pub(crate) fn setup_tracing_logger() -> anyhow::Result<()> {
fn directive_checked(directive: impl Into<String>) -> anyhow::Result<Directive> {
directive.into().parse().map_err(From::from)
}
let log_builder = tracing_subscriber::fmt()
// Use a more compact, abbreviated log format
.compact()
// Display source code file paths
.with_file(true)
// Display source code line numbers
.with_line_number(true)
.with_thread_ids(true)
// Don't display the event's target (module path)
.with_target(false);
let mut filter = EnvFilter::builder()
// if RUST_LOG isn't set, set default level
.with_default_directive(LevelFilter::DEBUG.into())
.from_env_lossy();
// these crates are more granularly filtered
let warn_crates = [
"rustls",
"sqlx",
"tower_http",
"axum",
"reqwest",
"hyper_util",
];
for crate_name in warn_crates {
filter = filter.add_directive(directive_checked(format!("{}=warn", crate_name))?);
}
let log_level_hint = filter.max_level_hint();
log_builder.with_env_filter(filter).init();
tracing::info!("Log level: {:?}", log_level_hint);
Ok(())
}
+53
View File
@@ -0,0 +1,53 @@
use clap::Parser;
use network_view::NetworkRefresher;
use nym_task::ShutdownManager;
mod cli;
mod http;
mod logging;
mod network_view;
mod storage;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
logging::setup_tracing_logger()?;
let args = cli::Cli::parse();
let connection_url = args.database_url.clone();
tracing::debug!("Using config:\n{:#?}", args);
let storage = storage::StatisticsStorage::init(
connection_url,
args.username,
args.password,
args.pg_port,
)
.await?;
tracing::info!("Connection to database successful");
let shutdown_manager = ShutdownManager::new("nym-statistics-api");
let network_refresher = NetworkRefresher::initialise_new(
args.nym_api_url,
shutdown_manager.child_token("network-refresher"),
)
.await?;
let http_server =
http::server::build_http_api(storage, network_refresher.network_view(), args.http_port)
.await
.expect("Failed to build http server");
let server_shutdown = shutdown_manager.clone_token("http-api-server");
// Starting tasks
shutdown_manager.spawn(async move { http_server.run(server_shutdown).await });
network_refresher.start();
tracing::info!("Started HTTP server on port {}", args.http_port);
shutdown_manager.close();
shutdown_manager.wait_for_shutdown_signal().await;
Ok(())
}
+152
View File
@@ -0,0 +1,152 @@
// Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: GPL-3.0-only
use anyhow::Result;
use nym_task::ShutdownToken;
use celes::Country;
use nym_validator_client::models::NymNodeDescription;
use nym_validator_client::NymApiClient;
use std::collections::HashMap;
use std::time::Duration;
use std::{net::IpAddr, sync::Arc};
use tokio::sync::RwLock;
use tokio::time::interval;
use url::Url;
use tracing::{error, info, trace, warn};
const NETWORK_CACHE_TTL: Duration = Duration::from_secs(600);
type IpToCountryMap = HashMap<IpAddr, Option<Country>>;
// SW this should use a proper NS API client once it exists
struct NodesQuerier {
client: NymApiClient,
}
impl NodesQuerier {
async fn current_nymnodes(&self) -> Result<Vec<NymNodeDescription>> {
Ok(self
.client
.get_all_described_nodes()
.await
.inspect_err(|err| error!("failed to get network nodes: {err}"))?)
}
}
#[derive(Clone, Debug)]
pub(crate) struct NetworkView {
inner: Arc<RwLock<NetworkViewInner>>,
}
impl NetworkView {
fn new_empty() -> Self {
NetworkView {
inner: Arc::new(RwLock::new(NetworkViewInner {
network_nodes: HashMap::new(),
})),
}
}
pub(crate) async fn get_country_by_ip(&self, ip_addr: &IpAddr) -> Option<Option<Country>> {
self.inner.read().await.network_nodes.get(ip_addr).copied()
}
}
#[derive(Debug)]
struct NetworkViewInner {
network_nodes: IpToCountryMap,
}
pub struct NetworkRefresher {
querier: Option<NodesQuerier>,
full_refresh_interval: Duration,
shutdown_token: ShutdownToken,
network: NetworkView,
}
impl NetworkRefresher {
pub(crate) async fn initialise_new(
maybe_nym_api_url: Option<Url>,
shutdown_token: ShutdownToken,
) -> Result<Self> {
let node_querier = match maybe_nym_api_url {
Some(url) => Some(NodesQuerier {
client: nym_http_api_client::Client::builder::<_, anyhow::Error>(url)?
.no_hickory_dns()
.with_user_agent("node-statistics-api")
.build::<anyhow::Error>()?
.into(),
}),
None => {
warn!("No Nym API specified, network view is unavailable");
None
}
};
let mut this = NetworkRefresher {
querier: node_querier,
full_refresh_interval: NETWORK_CACHE_TTL,
shutdown_token,
network: NetworkView::new_empty(),
};
this.refresh_network_nodes().await?;
Ok(this)
}
async fn refresh_network_nodes(&mut self) -> Result<()> {
if let Some(querier) = &self.querier {
let nodes = querier.current_nymnodes().await?;
// collect all known/allowed nodes information
let known_nodes = nodes
.iter()
.flat_map(|n| {
n.description
.host_information
.ip_address
.clone()
.into_iter()
.zip(std::iter::repeat(n.description.auxiliary_details.location))
})
.collect::<HashMap<_, _>>();
let mut network_guard = self.network.inner.write().await;
network_guard.network_nodes = known_nodes;
}
Ok(())
}
pub(crate) fn network_view(&self) -> NetworkView {
self.network.clone()
}
pub(crate) async fn run(&mut self) {
info!("NetworkRefresher started successfully");
let mut full_refresh_interval = interval(self.full_refresh_interval);
full_refresh_interval.reset();
while !self.shutdown_token.is_cancelled() {
tokio::select! {
biased;
_ = self.shutdown_token.cancelled() => {
trace!("NetworkRefresher: Received shutdown");
}
_ = full_refresh_interval.tick() => {
if self.refresh_network_nodes().await.is_err() {
warn!("Failed to refresh network nodes, we're gonna keep the same set");
}
}
}
}
trace!("NetworkRefresher: Exiting");
}
pub(crate) fn start(mut self) {
tokio::spawn(async move { self.run().await });
}
}
+102
View File
@@ -0,0 +1,102 @@
use anyhow::{anyhow, Result};
use models::{ConnectionInfoDto, DailyActiveDeviceDto};
use sqlx::{migrate::Migrator, postgres::PgConnectOptions};
use std::str::FromStr;
pub(crate) mod models;
pub(crate) type DbPool = sqlx::PgPool;
static MIGRATOR: Migrator = sqlx::migrate!("./migrations");
#[derive(Debug, Clone)]
pub(crate) struct StatisticsStorage {
connection_pool: DbPool,
}
impl StatisticsStorage {
pub async fn init(
connection_url: String,
user: String,
password: String,
port: u16,
) -> Result<Self> {
let connect_options = PgConnectOptions::from_str(&connection_url)?
.port(port)
.username(&user)
.password(&password)
.application_name(nym_bin_common::bin_info!().binary_name);
let pool = sqlx::PgPool::connect_with(connect_options)
.await
.map_err(|err| anyhow!("Failed to connect to {}: {}", &connection_url, err))?;
MIGRATOR.run(&pool).await?;
Ok(StatisticsStorage {
connection_pool: pool,
})
}
pub(crate) async fn store_vpn_client_report(
&mut self,
active_device: DailyActiveDeviceDto,
connection_info: Option<ConnectionInfoDto>,
) -> Result<()> {
self.store_device(active_device).await?;
if let Some(connection_info) = connection_info {
self.store_connection_stats(connection_info).await?;
}
Ok(())
}
// Interestingly enough, because gateway-storage is using the `chrono` feature of sqlx and in 0.7.4 it takes priority over the `time` one, we cannot use the query! macro here.
// Due to features unification, the binary will not compile when built from the workspace root because it will expect `chrono` types.
// As a consequence, there is no compile time verification of these queries.
async fn store_device(&self, active_device: DailyActiveDeviceDto) -> Result<()> {
sqlx::query!(
r#"INSERT INTO active_device (
day,
device_id,
os_type,
os_version,
architecture,
app_version,
user_agent,
from_mixnet)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (device_id, day) DO NOTHING"#,
active_device.day as time::Date,
active_device.stats_id,
active_device.os_type,
active_device.os_version,
active_device.os_arch,
active_device.app_version,
active_device.user_agent,
active_device.from_mixnet
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
async fn store_connection_stats(&self, connection_info: ConnectionInfoDto) -> Result<()> {
sqlx::query!(
r#"INSERT INTO connection_stats (
received_at,
connection_time_ms,
two_hop,
source_ip,
country_code,
from_mixnet) VALUES ($1::timestamptz, $2, $3, $4, $5, $6)"#,
connection_info.received_at as time::OffsetDateTime,
connection_info.connection_time_ms,
connection_info.two_hop,
connection_info.received_from,
connection_info.country_code,
connection_info.from_mixnet
)
.execute(&self.connection_pool)
.await?;
Ok(())
}
}
+69
View File
@@ -0,0 +1,69 @@
use std::net::SocketAddr;
use axum_extra::headers::UserAgent;
use celes::Country;
use nym_statistics_common::report::vpn_client::VpnClientStatsReport;
use time::{Date, OffsetDateTime};
pub type StatsId = String;
#[derive(Debug, Clone, sqlx::FromRow)]
pub(crate) struct DailyActiveDeviceDto {
pub(crate) day: Date,
pub(crate) stats_id: StatsId,
pub(crate) os_type: String,
pub(crate) os_version: Option<String>,
pub(crate) os_arch: String,
pub(crate) app_version: String,
pub(crate) user_agent: String,
pub(crate) from_mixnet: bool,
}
impl DailyActiveDeviceDto {
pub(crate) fn new(
received_at: OffsetDateTime,
stats_report: &VpnClientStatsReport,
user_agent: UserAgent,
from_mixnet: bool,
) -> Self {
Self {
day: received_at.date(),
stats_id: stats_report.stats_id.clone(),
os_type: stats_report.static_information.os_type.clone(),
os_version: stats_report.static_information.os_version.clone(),
os_arch: stats_report.static_information.os_arch.clone(),
app_version: stats_report.static_information.app_version.clone(),
user_agent: user_agent.to_string(),
from_mixnet,
}
}
}
#[derive(Debug, Clone, sqlx::FromRow)]
pub(crate) struct ConnectionInfoDto {
pub(crate) received_at: OffsetDateTime,
pub(crate) received_from: String,
pub(crate) connection_time_ms: Option<i32>,
pub(crate) two_hop: bool,
pub(crate) country_code: Option<String>,
pub(crate) from_mixnet: bool,
}
impl ConnectionInfoDto {
pub(crate) fn maybe_new(
received_at: OffsetDateTime,
stats_report: &VpnClientStatsReport,
received_from: SocketAddr,
maybe_country: Option<Country>,
from_mixnet: bool,
) -> Option<Self> {
stats_report.basic_usage.as_ref().map(|usage_report| Self {
received_at,
received_from: received_from.ip().to_string(),
connection_time_ms: usage_report.connection_time_ms,
two_hop: usage_report.two_hop,
country_code: maybe_country.map(|c| c.alpha2.into()),
from_mixnet,
})
}
}
@@ -0,0 +1,8 @@
/**
* This is a mock for Tauri's API package (@tauri-apps/api/app), to prevent stories from being excluded, because they either use
* or import dependencies that use Tauri.
*/
module.exports = {
invoke: () => undefined,
}
@@ -0,0 +1,4 @@
/**
* This is a mock for Tauri's API package (@tauri-apps/api/app), to prevent stories from being excluded, because they either use
* or import dependencies that use Tauri.
*/
+43
View File
@@ -415,11 +415,14 @@ version = "0.4.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59a194f9d963d8099596278594b3107448656ba73831c9d8c783e613ce86da64"
dependencies = [
"brotli",
"flate2",
"futures-core",
"memchr",
"pin-project-lite",
"tokio",
"zstd",
"zstd-safe",
]
[[package]]
@@ -940,6 +943,8 @@ version = "1.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fcb57c740ae1daf453ae85f16e37396f672b039e00d9d866e07ddb24e328e3a"
dependencies = [
"jobserver",
"libc",
"shlex",
]
@@ -3529,6 +3534,16 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8eaf4bc02d17cbdd7ff4c7438cafcdf7fb9a4613313ad11b4f8fefe7d3fa0130"
[[package]]
name = "jobserver"
version = "0.1.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "38f262f097c174adebe41eb73d66ae9c06b2844fb0da69969647bbddd9b0538a"
dependencies = [
"getrandom 0.3.2",
"libc",
]
[[package]]
name = "jpeg-decoder"
version = "0.3.1"
@@ -9270,6 +9285,34 @@ dependencies = [
"memchr",
]
[[package]]
name = "zstd"
version = "0.13.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a"
dependencies = [
"zstd-safe",
]
[[package]]
name = "zstd-safe"
version = "7.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d"
dependencies = [
"zstd-sys",
]
[[package]]
name = "zstd-sys"
version = "2.0.15+zstd.1.5.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb81183ddd97d0c74cedf1d50d85c8d08c1b8b68ee863bdee9e706eedba1a237"
dependencies = [
"cc",
"pkg-config",
]
[[package]]
name = "zvariant"
version = "5.4.0"
+4 -6
View File
@@ -142,7 +142,7 @@ impl Config {
let location = Self::config_file_path(None);
match toml::to_string_pretty(&global)
.map_err(|toml_err| io::Error::new(io::ErrorKind::Other, toml_err))
.map_err(io::Error::other)
.map(|toml| fs::write(location.clone(), toml))
{
Ok(_) => log::debug!("Writing to: {:#?}", location),
@@ -162,7 +162,7 @@ impl Config {
let location = Self::config_file_path(Some(network));
match toml::to_string_pretty(config)
.map_err(|toml_err| io::Error::new(io::ErrorKind::Other, toml_err))
.map_err(io::Error::other)
.map(|toml| fs::write(location.clone(), toml))
{
Ok(_) => log::debug!("Writing to: {:#?}", location),
@@ -367,10 +367,8 @@ fn load_from_file<T>(file: PathBuf) -> Result<T, io::Error>
where
T: DeserializeOwned,
{
fs::read_to_string(file).and_then(|contents| {
toml::from_str::<T>(&contents)
.map_err(|toml_err| io::Error::new(io::ErrorKind::Other, toml_err))
})
fs::read_to_string(file)
.and_then(|contents| toml::from_str::<T>(&contents).map_err(io::Error::other))
}
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq, Hash)]
+4 -4
View File
@@ -32,7 +32,7 @@ pub enum BackendError {
NyxdError {
pretty_error: String,
#[source]
source: NyxdError,
source: Box<NyxdError>,
},
#[error(transparent)]
CosmwasmStd {
@@ -192,18 +192,18 @@ impl From<NyxdError> for BackendError {
if let Some(pretty_log) = pretty_log {
Self::NyxdError {
pretty_error: pretty_log.to_string(),
source,
source: Box::new(source),
}
} else {
Self::NyxdError {
pretty_error: source.to_string(),
source,
source: Box::new(source),
}
}
}
nyxd_error => Self::NyxdError {
pretty_error: nyxd_error.to_string(),
source: nyxd_error,
source: Box::new(nyxd_error),
},
}
}
@@ -68,7 +68,7 @@ impl Substream {
}
fn check_closed(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Result<(), IoError> {
let closed_err = IoError::new(ErrorKind::Other, "stream closed");
let closed_err = IoError::other("stream closed");
// close_rx will return an error if the channel is closed (ie. sender was dropped),
// or if it's empty
@@ -172,12 +172,7 @@ impl AsyncWrite for Substream {
),
}),
})
.map_err(|e| {
IoError::new(
ErrorKind::Other,
format!("poll_write outbound_tx error: {}", e),
)
})?;
.map_err(|e| IoError::other(format!("poll_write outbound_tx error: {}", e)))?;
Poll::Ready(Ok(buf.len()))
}
@@ -195,7 +190,7 @@ impl AsyncWrite for Substream {
let mut closed = self.closed.lock();
if *closed {
return Poll::Ready(Err(IoError::new(ErrorKind::Other, "stream closed")));
return Poll::Ready(Err(IoError::other("stream closed")));
}
*closed = true;
@@ -210,12 +205,7 @@ impl AsyncWrite for Substream {
message: SubstreamMessage::new_close(self.substream_id.clone()),
}),
})
.map_err(|e| {
IoError::new(
ErrorKind::Other,
format!("poll_close outbound_rx error: {}", e),
)
})?;
.map_err(|e| IoError::other(format!("poll_close outbound_rx error: {}", e)))?;
Poll::Ready(Ok(()))
}
+8 -10
View File
@@ -212,19 +212,18 @@ where
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::result::Result<usize, std::io::Error>> {
ready!(self.tx.poll_ready_unpin(cx)).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::Other, "failed to send packet to mixnet")
})?;
ready!(self.tx.poll_ready_unpin(cx))
.map_err(|_| std::io::Error::other("failed to send packet to mixnet"))?;
let input_message = self
.message_translator
.to_input_message(buf)
.map_err(|err| std::io::Error::new(std::io::ErrorKind::Other, err))?;
.map_err(std::io::Error::other)?;
// Pass it to the mixnet sender
self.tx.start_send_unpin(input_message).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::Other, "failed to send packet to mixnet")
})?;
self.tx
.start_send_unpin(input_message)
.map_err(|_| std::io::Error::other("failed to send packet to mixnet"))?;
Poll::Ready(Ok(buf.len()))
}
@@ -233,9 +232,8 @@ where
mut self: Pin<&mut Self>,
cx: &mut Context,
) -> Poll<std::result::Result<(), std::io::Error>> {
ready!(self.tx.poll_flush_unpin(cx)).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::Other, "failed to send packet to mixnet")
})?;
ready!(self.tx.poll_flush_unpin(cx))
.map_err(|_| std::io::Error::other("failed to send packet to mixnet"))?;
Poll::Ready(Ok(()))
}
@@ -51,3 +51,6 @@ nym-task = { path = "../../common/task" }
nym-types = { path = "../../common/types" }
nym-wireguard = { path = "../../common/wireguard" }
nym-wireguard-types = { path = "../../common/wireguard-types" }
[dev-dependencies]
mock_instant = "0.5.3"
@@ -132,7 +132,9 @@ pub(crate) async fn execute(args: &Request) -> Result<(), AuthenticatorError> {
None,
))
.await
.map_err(|source| AuthenticatorError::FailedToSendPacketToMixnet { source })?;
.map_err(|source| AuthenticatorError::FailedToSendPacketToMixnet {
source: Box::new(source),
})?;
log::info!("Sent request, sleeping 60 seconds or until killed");
sleep(Duration::from_secs(60)).await;
+3 -3
View File
@@ -24,7 +24,7 @@ pub enum AuthenticatorError {
ShortPacket,
#[error("failed to connect to mixnet: {source}")]
FailedToConnectToMixnet { source: nym_sdk::Error },
FailedToConnectToMixnet { source: Box<nym_sdk::Error> },
#[error("failed to deserialize tagged packet: {source}")]
FailedToDeserializeTaggedPacket { source: bincode::Error },
@@ -33,13 +33,13 @@ pub enum AuthenticatorError {
FailedToLoadConfig(String),
#[error("failed to send packet to mixnet: {source}")]
FailedToSendPacketToMixnet { source: nym_sdk::Error },
FailedToSendPacketToMixnet { source: Box<nym_sdk::Error> },
#[error("failed to serialize response packet: {source}")]
FailedToSerializeResponsePacket { source: Box<bincode::ErrorKind> },
#[error("failed to setup mixnet client: {source}")]
FailedToSetupMixnetClient { source: nym_sdk::Error },
FailedToSetupMixnetClient { source: Box<nym_sdk::Error> },
#[error("{0}")]
GatewayStorageError(#[from] nym_gateway_storage::error::GatewayStorageError),
@@ -10,3 +10,4 @@ pub mod error;
pub mod mixnet_client;
pub mod mixnet_listener;
mod peer_manager;
mod seen_credential_cache;

Some files were not shown because too many files have changed in this diff Show More