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

Check gateway supported versions (#5860)

Browse Source 
* Check gateway supported versions

* Fix boxed errors

* Fmt

* feat(client-core): integrate gateway protocol validation into SDK

Enable protocol version checking for all SDK-based clients by using
gateways_for_init_with_protocol_validation in MixnetClientBuilder.
This ensures clients only connect to gateways with compatible protocol
versions, preventing potential communication issues.

- Replace gateways_for_init with gateways_for_init_with_protocol_validation
- Add import for the new validation function
- Protocol validation now active for network monitor and all SDK users

* refactor(client-core): allow gateways with newer protocol versions

Change protocol validation to be more permissive - instead of rejecting
gateways with newer protocol versions, now logs a warning and continues.
This enables graceful degradation when gateways upgrade, relying on their
backward compatibility while signaling users to update their clients.

Changes:
- Accept gateways with protocol version > client version
- Log warning about version mismatch suggesting client update
- Update log messages from "validation" to "check" for clarity
- Add trace logging showing both gateway and client versions

This prevents connectivity issues when gateways upgrade before clients,
improving overall network resilience.

* feat(nym-network-monitor): add diagnostic logging for ConnectionRefused debugging

- Add detailed logging for client rotation lifecycle with [CLIENT_ROTATION_*] tags
- Add request tracking with unique IDs using the random message as identifier
- Add HTTP connection acceptance logging with [HTTP_REQUEST] tags
- Improve locust script with connection error handling and backoff strategy
- Add TCP backlog configuration support (default 1024)
- Add socket configuration with SO_REUSEADDR and configurable backlog
- Add HTTP timeout and tracing layers for better observability

These changes help diagnose when and why ConnectionRefused errors occur during load testing,
particularly around client rotation periods.

* fix(nym-api): calculate route average reliability as node mean instead of route-based

- Changed route average reliability calculation to use mean of all node reliabilities
- Added median calculation alongside mean for better statistical representation
- Fixed null average reliability for old method which doesn't analyze routes
- Rounded all float values to 2 decimal places in API responses for cleaner display
- Store median values in analysis_parameters JSON field
This commit is contained in:
Drazen Urch
2025-06-24 21:28:07 +02:00
committed by GitHub
parent d67a968e76
commit c1acef9bc8
12 changed files with 780 additions and 439 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+406 -376
View File
File diff suppressed because it is too large Load Diff
common/client-core/src/error.rs
+3
View File
@@ -166,6 +166,9 @@ pub enum ClientCoreError {
#[error("there are no gateways supporting the wss protocol available")]
NoWssGateways,
#[error("there are no gateways with compatible protocol versions available")]
NoGatewaysWithCompatibleProtocol,
#[error("the specified gateway '{gateway}' does not support the wss protocol")]
UnsupportedWssProtocol { gateway: String },
common/client-core/src/init/helpers.rs
+184
View File
@@ -7,6 +7,7 @@ use futures::{SinkExt, StreamExt};
use log::{debug, info, trace, warn};
use nym_crypto::asymmetric::ed25519;
use nym_gateway_client::GatewayClient;
use nym_gateway_requests::{ClientControlRequest, ServerResponse, CURRENT_PROTOCOL_VERSION};
use nym_topology::node::RoutingNode;
use nym_validator_client::client::IdentityKeyRef;
use nym_validator_client::UserAgent;
@@ -131,6 +132,63 @@ pub async fn gateways_for_init<R: Rng>(
Ok(valid_gateways)
}
pub async fn gateways_for_init_with_protocol_validation<R: Rng>(
rng: &mut R,
nym_apis: &[Url],
user_agent: Option<UserAgent>,
minimum_performance: u8,
ignore_epoch_roles: bool,
) -> Result<Vec<RoutingNode>, ClientCoreError> {
// First get the initial list of gateways
let gateways = gateways_for_init(
rng,
nym_apis,
user_agent,
minimum_performance,
ignore_epoch_roles,
)
.await?;
info!(
"Checking protocol compatibility for {} gateways...",
gateways.len()
);
// Filter out gateways with invalid protocols concurrently
let validated_gateways = Arc::new(tokio::sync::Mutex::new(Vec::new()));
futures::stream::iter(&gateways)
.for_each_concurrent(CONCURRENT_GATEWAYS_MEASURED, |gateway| async {
let id = gateway.identity();
trace!("validating protocol compatibility with {id}...");
match validate_gateway_protocol(gateway).await {
Ok(()) => {
debug!("{id}: protocol check successful");
validated_gateways.lock().await.push(gateway.clone());
}
Err(err) => {
warn!("failed to check protocol for {id}: {err}");
}
}
})
.await;
let validated_gateways = validated_gateways.lock().await;
info!(
"Protocol check complete: {}/{} gateways responded successfully",
validated_gateways.len(),
gateways.len()
);
if validated_gateways.is_empty() {
return Err(ClientCoreError::NoGatewaysWithCompatibleProtocol);
}
Ok(validated_gateways.clone())
}
#[cfg(not(target_arch = "wasm32"))]
async fn connect(endpoint: &str) -> Result<WsConn, ClientCoreError> {
match tokio::time::timeout(CONN_TIMEOUT, connect_async(endpoint)).await {
@@ -210,6 +268,132 @@ where
Ok(GatewayWithLatency::new(gateway, avg))
}
async fn validate_gateway_protocol<G>(gateway: &G) -> Result<(), ClientCoreError>
where
G: ConnectableGateway,
{
let Some(addr) = gateway.clients_address(false) else {
return Err(ClientCoreError::UnsupportedEntry {
id: gateway.node_id(),
identity: gateway.identity().to_string(),
});
};
trace!(
"validating protocol compatibility with {} ({addr})...",
gateway.identity(),
);
let mut stream = connect(&addr).await?;
// Send protocol version request
let protocol_request = ClientControlRequest::SupportedProtocol {};
// Send the request as JSON text message
stream.send(Message::from(protocol_request)).await?;
// Wait for response with timeout
let protocol_timeout = Duration::from_millis(2000);
let response_future = stream.next();
match tokio::time::timeout(protocol_timeout, response_future).await {
Err(_) => {
warn!("Gateway {} protocol check timed out", gateway.identity());
Err(ClientCoreError::GatewayConnectionTimeout)
}
Ok(Some(Ok(Message::Text(response_text)))) => {
// Try to deserialize the response
let response = ServerResponse::try_from(response_text).map_err(|_| {
ClientCoreError::GatewayClientError {
gateway_id: gateway.identity().to_base58_string(),
source: *Box::new(
nym_gateway_client::error::GatewayClientError::MalformedResponse,
),
}
})?;
match response {
ServerResponse::SupportedProtocol { version } => {
debug!(
"Gateway {} supports protocol version {}, ours: {}",
gateway.identity(),
version,
CURRENT_PROTOCOL_VERSION
);
// Check protocol compatibility
if version > CURRENT_PROTOCOL_VERSION {
warn!(
"Gateway {} uses newer protocol version {} (client supports {}). \
Gateway should gracefully degrade, but consider updating your client.",
gateway.identity(),
version,
CURRENT_PROTOCOL_VERSION
);
}
trace!(
"Gateway {} protocol validation successful (gateway: v{}, client: v{})",
gateway.identity(),
version,
CURRENT_PROTOCOL_VERSION
);
Ok(())
}
ServerResponse::Error { message } => {
warn!(
"Gateway {} returned error during protocol check: {}",
gateway.identity(),
message
);
Err(ClientCoreError::GatewayClientError {
gateway_id: gateway.identity().to_base58_string(),
source: *Box::new(
nym_gateway_client::error::GatewayClientError::GatewayError(message),
),
})
}
_ => {
warn!(
"Gateway {} returned unexpected response during protocol check",
gateway.identity()
);
Err(ClientCoreError::GatewayClientError {
gateway_id: gateway.identity().to_base58_string(),
source: *Box::new(
nym_gateway_client::error::GatewayClientError::UnexpectedResponse {
name: response.name().to_string(),
},
),
})
}
}
}
Ok(Some(Ok(_))) => {
warn!(
"Gateway {} sent non-text response during protocol check",
gateway.identity()
);
Err(ClientCoreError::GatewayConnectionAbruptlyClosed)
}
Ok(Some(Err(e))) => {
warn!(
"WebSocket error during protocol check with {}: {}",
gateway.identity(),
e
);
Err(e.into())
}
Ok(None) => {
warn!(
"Gateway {} closed connection during protocol check",
gateway.identity()
);
Err(ClientCoreError::GatewayConnectionAbruptlyClosed)
}
}
}
pub async fn choose_gateway_by_latency<R: Rng, G: ConnectableGateway + Clone>(
rng: &mut R,
gateways: &[G],
nym-api/src/epoch_operations/simulation.rs
+59 -20
View File
@@ -189,6 +189,20 @@ impl<'a> SimulationCoordinator<'a> {
)
.await;
// Calculate average reliability for old method (mean of all node reliabilities)
let node_reliabilities: Vec<f64> = performance_map.values()
.map(|p| p.naive_to_f64() * 100.0)
.filter(|&r| r > 0.0) // Only include nodes with non-zero reliability
.collect();
let (mean_reliability, median_reliability) = if !node_reliabilities.is_empty() {
let mean = node_reliabilities.iter().sum::<f64>() / node_reliabilities.len() as f64;
let median = calculate_median(&node_reliabilities);
(Some((mean * 100.0).round() / 100.0), Some((median * 100.0).round() / 100.0))
} else {
(None, None)
};
// Create route analysis for old method
let route_analysis = SimulatedRouteAnalysis {
id: 0, // Will be set by database
@@ -197,11 +211,13 @@ impl<'a> SimulationCoordinator<'a> {
total_routes_analyzed: 0, // Old method doesn't use route data
successful_routes: 0,
failed_routes: 0,
average_route_reliability: None,
average_route_reliability: mean_reliability,
time_window_hours: 24, // Old method uses 24h
analysis_parameters: Some(
"{\"method\":\"cache_based\",\"data_source\":\"status_cache\"}".to_string(),
),
analysis_parameters: Some(format!(
"{{\"method\":\"cache_based\",\"data_source\":\"status_cache\",\"median_reliability\":{},\"nodes_analyzed\":{}}}",
median_reliability.unwrap_or(0.0),
node_reliabilities.len()
)),
calculated_at: OffsetDateTime::now_utc().unix_timestamp(),
};
@@ -265,23 +281,16 @@ impl<'a> SimulationCoordinator<'a> {
let mut route_reliability_map = HashMap::new();
let mut total_routes = 0u32;
let mut successful_routes = 0u32;
let mut reliability_sum = 0.0;
let mut reliability_count = 0u32;
for node_reliability in &corrected_reliabilities {
let total_samples =
node_reliability.pos_samples_in_interval + node_reliability.neg_samples_in_interval;
total_routes += total_samples;
successful_routes += node_reliability.pos_samples_in_interval;
if total_samples > 0 {
reliability_sum += node_reliability.reliability;
reliability_count += 1;
}
performance_map.insert(
node_reliability.node_id,
Performance::naive_try_from_f64(node_reliability.reliability).unwrap_or_default(),
Performance::naive_try_from_f64(node_reliability.reliability / 100.0)
.unwrap_or_default(),
);
// Store sample counts for detailed performance records
@@ -298,6 +307,7 @@ impl<'a> SimulationCoordinator<'a> {
let rewarded_nodes =
self.calculate_rewards_for_nodes(rewarded_set, reward_params, &performance_map);
// Convert to simulation data structures
let node_performance = self
.convert_to_simulated_performance(
@@ -320,6 +330,20 @@ impl<'a> SimulationCoordinator<'a> {
)
.await;
// Calculate average reliability for new method (mean of all node reliabilities)
let node_reliabilities: Vec<f64> = corrected_reliabilities.iter()
.filter(|n| n.pos_samples_in_interval + n.neg_samples_in_interval > 0)
.map(|n| n.reliability)
.collect();
let (mean_reliability, median_reliability) = if !node_reliabilities.is_empty() {
let mean = node_reliabilities.iter().sum::<f64>() / node_reliabilities.len() as f64;
let median = calculate_median(&node_reliabilities);
(Some((mean * 100.0).round() / 100.0), Some((median * 100.0).round() / 100.0))
} else {
(None, None)
};
// Create route analysis for new method
let route_analysis = SimulatedRouteAnalysis {
id: 0, // Will be set by database
@@ -328,16 +352,14 @@ impl<'a> SimulationCoordinator<'a> {
total_routes_analyzed: total_routes,
successful_routes,
failed_routes: total_routes - successful_routes,
average_route_reliability: if reliability_count > 0 {
Some(reliability_sum * 100.0 / reliability_count as f64)
} else {
None
},
average_route_reliability: mean_reliability,
time_window_hours: self.config.new_method_time_window_hours,
analysis_parameters: Some(format!(
"{{\"method\":\"route_based\",\"time_window_hours\":{},\"corrected_routes\":{}}}",
"{{\"method\":\"route_based\",\"time_window_hours\":{},\"corrected_routes\":{},\"median_reliability\":{},\"nodes_analyzed\":{}}}",
self.config.new_method_time_window_hours,
corrected_reliabilities.len()
corrected_reliabilities.len(),
median_reliability.unwrap_or(0.0),
node_reliabilities.len()
)),
calculated_at: OffsetDateTime::now_utc().unix_timestamp(),
};
@@ -652,3 +674,20 @@ impl EpochAdvancer {
}
}
}
/// Calculate median of a vector of f64 values
fn calculate_median(values: &[f64]) -> f64 {
if values.is_empty() {
return 0.0;
}
let mut sorted = values.to_vec();
sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let len = sorted.len();
if len % 2 == 0 {
(sorted[len / 2 - 1] + sorted[len / 2]) / 2.0
} else {
sorted[len / 2]
}
}
nym-api/src/simulation_api/handlers/mod.rs
+17 -15
View File
@@ -594,13 +594,13 @@ fn build_node_comparisons_with_rankings(
// Calculate differences
let reliability_difference = match (&old_perf, &new_perf) {
(Some(old), Some(new)) => Some(new.reliability_score - old.reliability_score),
(Some(old), Some(new)) => Some(((new.reliability_score - old.reliability_score) * 100.0).round() / 100.0),
_ => None,
};
let performance_delta_percentage = match (&old_perf, &new_perf) {
(Some(old), Some(new)) if old.reliability_score != 0.0 => Some(
(new.reliability_score - old.reliability_score) / old.reliability_score * 100.0,
(((new.reliability_score - old.reliability_score) / old.reliability_score * 100.0) * 100.0).round() / 100.0,
),
_ => None,
};
@@ -707,14 +707,14 @@ fn calculate_summary_statistics(comparisons: &[NodeMethodComparison]) -> Compari
nodes_improved: improvements,
nodes_degraded: degradations,
nodes_unchanged: unchanged,
average_reliability_old,
average_reliability_new,
median_reliability_old,
median_reliability_new,
reliability_std_dev_old,
reliability_std_dev_new,
max_improvement,
max_degradation,
average_reliability_old: (average_reliability_old * 100.0).round() / 100.0,
average_reliability_new: (average_reliability_new * 100.0).round() / 100.0,
median_reliability_old: (median_reliability_old * 100.0).round() / 100.0,
median_reliability_new: (median_reliability_new * 100.0).round() / 100.0,
reliability_std_dev_old: (reliability_std_dev_old * 100.0).round() / 100.0,
reliability_std_dev_new: (reliability_std_dev_new * 100.0).round() / 100.0,
max_improvement: (max_improvement * 100.0).round() / 100.0,
max_degradation: (max_degradation * 100.0).round() / 100.0,
}
}
@@ -769,11 +769,13 @@ async fn get_route_analysis_comparison(
_ => 0,
};
let success_rate_difference = match (&old_analysis, &new_analysis) {
// For comparison, we now look at the average reliability difference between methods
let reliability_difference = match (&old_analysis, &new_analysis) {
(Some(old), Some(new)) => {
let old_rate = old.successful_routes as f64 / old.total_routes_analyzed as f64;
let new_rate = new.successful_routes as f64 / new.total_routes_analyzed as f64;
Some(new_rate - old_rate)
match (old.average_route_reliability, new.average_route_reliability) {
(Some(old_avg), Some(new_avg)) => Some(((new_avg - old_avg) * 100.0).round() / 100.0),
_ => None
}
}
_ => None,
};
@@ -783,7 +785,7 @@ async fn get_route_analysis_comparison(
new_method: new_analysis,
time_window_difference_hours,
route_coverage_difference,
success_rate_difference,
success_rate_difference: reliability_difference,
})
}
nym-api/src/simulation_api/models.rs
+15 -7
View File
@@ -80,10 +80,10 @@ impl From<SimulatedNodePerformance> for NodePerformanceData {
node_id: perf.node_id,
node_type: perf.node_type,
identity_key: perf.identity_key,
reliability_score: perf.reliability_score,
reliability_score: (perf.reliability_score * 100.0).round() / 100.0,
positive_samples: perf.positive_samples,
negative_samples: perf.negative_samples,
work_factor: perf.work_factor,
work_factor: perf.work_factor.map(|w| (w * 100.0).round() / 100.0),
calculation_method: perf.calculation_method,
calculated_at: perf.calculated_at,
}
@@ -109,12 +109,12 @@ impl From<SimulatedPerformanceComparison> for PerformanceComparisonData {
Self {
node_id: comparison.node_id,
node_type: comparison.node_type,
performance_score: comparison.performance_score,
work_factor: comparison.work_factor,
performance_score: (comparison.performance_score * 100.0).round() / 100.0,
work_factor: (comparison.work_factor * 100.0).round() / 100.0,
calculation_method: comparison.calculation_method,
positive_samples: comparison.positive_samples,
negative_samples: comparison.negative_samples,
route_success_rate: comparison.route_success_rate,
route_success_rate: comparison.route_success_rate.map(|r| (r * 100.0).round() / 100.0),
calculated_at: comparison.calculated_at,
}
}
@@ -128,6 +128,7 @@ pub struct RouteAnalysisData {
pub successful_routes: u32,
pub failed_routes: u32,
pub average_route_reliability: Option<f64>,
pub median_route_reliability: Option<f64>,
pub time_window_hours: u32,
pub analysis_parameters: Option<String>,
pub calculated_at: i64,
@@ -135,12 +136,19 @@ pub struct RouteAnalysisData {
impl From<SimulatedRouteAnalysis> for RouteAnalysisData {
fn from(analysis: SimulatedRouteAnalysis) -> Self {
// Extract median from analysis_parameters JSON if available
let median_route_reliability = analysis.analysis_parameters.as_ref()
.and_then(|params| serde_json::from_str::<serde_json::Value>(params).ok())
.and_then(|json| json.get("median_reliability").and_then(|v| v.as_f64()))
.map(|m| (m * 100.0).round() / 100.0);
Self {
calculation_method: analysis.calculation_method,
total_routes_analyzed: analysis.total_routes_analyzed,
successful_routes: analysis.successful_routes,
failed_routes: analysis.failed_routes,
average_route_reliability: analysis.average_route_reliability,
average_route_reliability: analysis.average_route_reliability.map(|a| (a * 100.0).round() / 100.0),
median_route_reliability,
time_window_hours: analysis.time_window_hours,
analysis_parameters: analysis.analysis_parameters,
calculated_at: analysis.calculated_at,
@@ -197,7 +205,7 @@ pub struct RouteAnalysisComparison {
pub new_method: Option<RouteAnalysisData>,
pub time_window_difference_hours: i32, // new - old
pub route_coverage_difference: i32, // new total routes - old total routes
pub success_rate_difference: Option<f64>, // new success rate - old success rate
pub success_rate_difference: Option<f64>, // new average reliability - old average reliability
}
/// Export format options
nym-network-monitor/Cargo.toml
+4
View File
@@ -23,11 +23,15 @@ rand_chacha = { workspace = true }
reqwest = { workspace = true, features = ["json"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
socket2 = "0.5"
tokio = { workspace = true, features = ["macros", "time"] }
tokio-util = { workspace = true }
tower = { workspace = true }
tower-http = { workspace = true, features = ["timeout", "trace"] }
utoipa = { workspace = true, features = ["axum_extras"] }
utoipa-swagger-ui = { workspace = true, features = ["axum"] }
tokio-postgres = { workspace = true }
tracing = { workspace = true }
# internal
nym-bin-common = { path = "../common/bin-common", features = ["basic_tracing"] }
nym-network-monitor/locustfile.py
+15 -3
View File
@@ -1,5 +1,11 @@
import time
import logging
from locust import HttpUser, task
from requests.exceptions import ConnectionError
# Configure logging to see what's happening
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class SendMsg(HttpUser):
@@ -8,8 +14,14 @@ class SendMsg(HttpUser):
try:
response = self.client.post("/v1/send")
if response.status_code == 503:
logger.warning(f"Got 503 Service Unavailable, sleeping for 1 second")
time.sleep(1)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
except Exception: # Catch other exceptions, including those raised by raise_for_status()
# You might want to log this error or handle it differently
pass
except ConnectionError as e:
# This catches ConnectionRefused errors
logger.error(f"Connection refused, backing off for 5 seconds: {e}")
time.sleep(5) # Longer pause for connection errors
except Exception as e:
# Log other errors but don't sleep as long
logger.warning(f"Request failed: {type(e).__name__}: {e}")
time.sleep(0.5) # Brief pause for other errors
nym-network-monitor/src/handlers.rs
+8 -2
View File
@@ -197,12 +197,14 @@ async fn send_receive_mixnet(state: AppState) -> Result<String, StatusCode> {
.collect();
let sent_msg = msg.clone();
debug!("[REQUEST_START] Processing send request {}", msg);
let client = {
let mut clients = state.clients().write().await;
if let Some(client) = clients.make_contiguous().choose(&mut rand::thread_rng()) {
Arc::clone(client)
} else {
error!("No clients currently available");
error!("No clients currently available - this may cause connection refused errors");
return Err(StatusCode::SERVICE_UNAVAILABLE);
}
};
@@ -249,12 +251,16 @@ async fn send_receive_mixnet(state: AppState) -> Result<String, StatusCode> {
match result {
Ok(_) => {}
Err(e) => {
error!("Failed to send/receive message: {e}");
error!(
"[REQUEST_ERROR] {} - Failed to send/receive message: {e}",
sent_msg
);
return Err(StatusCode::GATEWAY_TIMEOUT);
}
}
}
debug!("[REQUEST_SUCCESS] {} - Message sent successfully", sent_msg);
Ok(sent_msg)
}
nym-network-monitor/src/http.rs
+45 -6
View File
@@ -5,12 +5,15 @@ use crate::handlers::{
FragmentsSent,
};
use axum::routing::{get, post};
use axum::Router;
use log::info;
use axum::{Router};
use log::{debug, info, warn};
use nym_sphinx::chunking::fragment::FragmentHeader;
use nym_sphinx::chunking::{ReceivedFragment, SentFragment};
use std::net::SocketAddr;
use tokio_util::sync::CancellationToken;
use tower::ServiceBuilder;
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::TraceLayer;
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;
@@ -19,6 +22,7 @@ use crate::ClientsWrapper;
pub struct HttpServer {
listener: SocketAddr,
cancel: CancellationToken,
tcp_backlog: i32,
}
#[derive(OpenApi)]
@@ -59,8 +63,12 @@ impl AppState {
}
impl HttpServer {
pub fn new(listener: SocketAddr, cancel: CancellationToken) -> Self {
HttpServer { listener, cancel }
pub fn new(listener: SocketAddr, cancel: CancellationToken, tcp_backlog: i32) -> Self {
HttpServer {
listener,
cancel,
tcp_backlog,
}
}
pub async fn run(self, clients: ClientsWrapper) -> anyhow::Result<()> {
@@ -78,15 +86,46 @@ impl HttpServer {
.route("/v1/node_stats/:mix_id", get(node_stats_handler))
.route("/v1/node_stats", get(all_nodes_stats_handler))
.route("/v1/received", get(recv_handler))
.layer(
ServiceBuilder::new()
// Add request tracing
.layer(
TraceLayer::new_for_http()
.on_request(|_request: &axum::http::Request<_>, _span: &tracing::Span| {
debug!("[HTTP_REQUEST] New connection accepted");
})
.on_failure(|error: tower_http::classify::ServerErrorsFailureClass, latency: std::time::Duration, _span: &tracing::Span| {
warn!("[HTTP_ERROR] Request failed with error: {:?}, latency: {:?}", error, latency);
})
)
// Add a timeout layer to prevent hanging connections
.layer(TimeoutLayer::new(std::time::Duration::from_secs(30)))
)
.with_state(state);
let listener = tokio::net::TcpListener::bind(self.listener).await?;
// Configure socket with higher backlog to handle more concurrent connections
let socket = socket2::Socket::new(
socket2::Domain::for_address(self.listener),
socket2::Type::STREAM,
Some(socket2::Protocol::TCP),
)?;
// Enable SO_REUSEADDR to avoid "Address already in use" errors
socket.set_reuse_address(true)?;
// Set a higher backlog (default is often 128)
socket.bind(&self.listener.into())?;
socket.listen(self.tcp_backlog)?; // Use configurable backlog
let listener = tokio::net::TcpListener::from_std(socket.into())?;
let server_future =
axum::serve(listener, app).with_graceful_shutdown(self.cancel.cancelled_owned());
info!("##########################################################################################");
info!("######################### HTTP server running, with {} clients ############################################", n_clients);
info!("######################### HTTP server running on {} with {} clients ############################################", self.listener, n_clients);
info!("######################### TCP backlog set to {} connections ############################################", self.tcp_backlog);
info!("##########################################################################################");
info!("[HTTP_SERVER] Server started and ready to accept connections");
server_future.await?;
nym-network-monitor/src/main.rs
+22 -8
View File
@@ -50,8 +50,12 @@ async fn make_clients(
if spawned_clients >= n_clients {
info!("New client will be spawned in {} seconds", lifetime);
tokio::time::sleep(tokio::time::Duration::from_secs(lifetime)).await;
info!("Removing oldest client");
info!("[CLIENT_ROTATION_START] Beginning client rotation");
if let Some(dropped_client) = clients.write().await.pop_front() {
info!(
"[CLIENT_ROTATION] Popped client from queue, current ref count: {}",
Arc::strong_count(&dropped_client)
);
const CLIENT_DROP_TIMEOUT: Duration = Duration::from_secs(30);
let start = tokio::time::Instant::now();
@@ -60,18 +64,23 @@ async fn make_clients(
Ok(client) => {
let client_handle = client.into_inner();
client_handle.disconnect().await;
info!("Successfully disconnected client immediately");
info!("[CLIENT_ROTATION_END] Successfully disconnected client immediately");
}
Err(dropped_client) => {
// Fallback: wait with timeout for references to drop
info!("Client still has references, waiting for cleanup with timeout");
info!("[CLIENT_ROTATION_BLOCKING] Client still has {} references, waiting for cleanup with timeout", Arc::strong_count(&dropped_client));
while start.elapsed() < CLIENT_DROP_TIMEOUT {
if Arc::strong_count(&dropped_client) == 1 {
let elapsed = start.elapsed();
let ref_count = Arc::strong_count(&dropped_client);
if elapsed.as_secs() % 5 == 0 && elapsed.subsec_millis() < 100 {
info!("[CLIENT_ROTATION_WAITING] Still waiting for client cleanup, elapsed: {:?}, ref_count: {}", elapsed, ref_count);
}
if ref_count == 1 {
match Arc::try_unwrap(dropped_client) {
Ok(client) => {
let client_handle = client.into_inner();
client_handle.disconnect().await;
info!("Successfully disconnected client after waiting");
info!("[CLIENT_ROTATION_END] Successfully disconnected client after waiting {:?}", start.elapsed());
break;
}
Err(_) => {
@@ -85,7 +94,7 @@ async fn make_clients(
if start.elapsed() >= CLIENT_DROP_TIMEOUT {
warn!(
"Client drop timed out after {:?}, forcing drop",
"[CLIENT_ROTATION_TIMEOUT] Client drop timed out after {:?}, forcing drop.",
CLIENT_DROP_TIMEOUT
);
// Client will be dropped when Arc goes out of scope
@@ -94,7 +103,7 @@ async fn make_clients(
}
}
}
info!("Spawning new client");
info!("[CLIENT_SPAWN_START] Spawning new client");
let client = match make_client(topology.clone()).await {
Ok(client) => client,
Err(err) => {
@@ -106,6 +115,7 @@ async fn make_clients(
.write()
.await
.push_back(Arc::new(RwLock::new(client)));
info!("[CLIENT_SPAWN_END] New client added to pool");
}
}
@@ -165,6 +175,9 @@ struct Args {
#[arg(long, env = "NYM_APIS", value_delimiter = ',')]
nym_apis: Option<Vec<String>>,
#[arg(long, env = "TCP_BACKLOG", default_value_t = 1024)]
tcp_backlog: i32,
}
fn generate_key_pair() -> Result<()> {
@@ -273,9 +286,10 @@ async fn main() -> Result<()> {
let clients_server = clients.clone();
let tcp_backlog = args.tcp_backlog;
let server_handle = tokio::spawn(async move {
let socket = SocketAddr::new(IpAddr::V4(Ipv4Addr::from_str(&args.host)?), args.port);
let server = HttpServer::new(socket, server_cancel_token);
let server = HttpServer::new(socket, server_cancel_token, tcp_backlog);
server.run(clients_server).await
});
sdk/rust/nym-sdk/src/mixnet/client.rs
+2 -2
View File
@@ -25,7 +25,7 @@ use nym_client_core::client::{
};
use nym_client_core::config::{DebugConfig, ForgetMe, RememberMe, StatsReporting};
use nym_client_core::error::ClientCoreError;
use nym_client_core::init::helpers::gateways_for_init;
use nym_client_core::init::helpers::gateways_for_init_with_protocol_validation;
use nym_client_core::init::setup_gateway;
use nym_client_core::init::types::{GatewaySelectionSpecification, GatewaySetup};
use nym_credentials_interface::TicketType;
@@ -546,7 +546,7 @@ where
let topology_cfg = &self.config.debug_config.topology;
let mut rng = OsRng;
let available_gateways = gateways_for_init(
let available_gateways = gateways_for_init_with_protocol_validation(
&mut rng,
&nym_api_endpoints,
user_agent,