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

Merge commit '6d44fe818ea4c74f476cc6d79434cc8619b45c0c' into simon/instrumented

Browse Source
This commit is contained in:
Simon Wicky
2023-03-28 13:17:09 +00:00
parent 5281895d5b 6d44fe818e
commit e95e33cd70
1686 changed files with 70374 additions and 44574 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/mixnode-common/src/packet_processor/error.rs
+17 -58
View File
@@ -1,66 +1,25 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use nymsphinx_acknowledgements::surb_ack::SurbAckRecoveryError;
use nymsphinx_addressing::nodes::NymNodeRoutingAddressError;
use nymsphinx_types::Error as SphinxError;
use std::fmt::{self, Display, Formatter};
use nym_sphinx_acknowledgements::surb_ack::SurbAckRecoveryError;
use nym_sphinx_addressing::nodes::NymNodeRoutingAddressError;
use nym_sphinx_types::Error as SphinxError;
use thiserror::Error;
#[derive(Debug)]
#[derive(Error, Debug)]
pub enum MixProcessingError {
SphinxProcessingError(SphinxError),
InvalidHopAddress(NymNodeRoutingAddressError),
NoSurbAckInFinalHop,
MalformedSurbAck(SurbAckRecoveryError),
#[error("failed to process received packet: {0}")]
SphinxProcessingError(#[from] SphinxError),
#[error("the forward hop address was malformed: {0}")]
InvalidForwardHopAddress(#[from] NymNodeRoutingAddressError),
#[error("the final hop did not contain a SURB-Ack")]
NoSurbAckInFinalHop,
#[error("failed to recover the expected SURB-Ack packet: {0}")]
MalformedSurbAck(#[from] SurbAckRecoveryError),
#[error("the received packet was set to use the very old and very much deprecated 'VPN' mode")]
ReceivedOldTypeVpnPacket,
}
impl From<SphinxError> for MixProcessingError {
// for the time being just have a single error instance for all possible results of SphinxError
fn from(err: SphinxError) -> Self {
use MixProcessingError::*;
SphinxProcessingError(err)
}
}
impl From<NymNodeRoutingAddressError> for MixProcessingError {
fn from(err: NymNodeRoutingAddressError) -> Self {
use MixProcessingError::*;
InvalidHopAddress(err)
}
}
impl From<SurbAckRecoveryError> for MixProcessingError {
fn from(err: SurbAckRecoveryError) -> Self {
use MixProcessingError::*;
MalformedSurbAck(err)
}
}
impl Display for MixProcessingError {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
MixProcessingError::SphinxProcessingError(sphinx_err) => {
write!(f, "Sphinx Processing Error - {}", sphinx_err)
}
MixProcessingError::InvalidHopAddress(address_err) => {
write!(f, "Invalid Hop Address - {:?}", address_err)
}
MixProcessingError::NoSurbAckInFinalHop => {
write!(f, "No SURBAck present in the final hop data")
}
MixProcessingError::MalformedSurbAck(surb_ack_err) => {
write!(f, "Malformed SURBAck - {:?}", surb_ack_err)
}
MixProcessingError::ReceivedOldTypeVpnPacket => {
write!(f, "Received an old-type unsafe 'VPN' mode packet")
}
}
}
}
impl std::error::Error for MixProcessingError {}
common/mixnode-common/src/packet_processor/processor.rs
+8 -8
View File
@@ -3,12 +3,12 @@
use crate::packet_processor::error::MixProcessingError;
use tracing::*;
use nymsphinx_acknowledgements::surb_ack::SurbAck;
use nymsphinx_addressing::nodes::NymNodeRoutingAddress;
use nymsphinx_forwarding::packet::MixPacket;
use nymsphinx_framing::packet::FramedSphinxPacket;
use nymsphinx_params::{PacketMode, PacketSize};
use nymsphinx_types::{
use nym_sphinx_acknowledgements::surb_ack::SurbAck;
use nym_sphinx_addressing::nodes::NymNodeRoutingAddress;
use nym_sphinx_forwarding::packet::MixPacket;
use nym_sphinx_framing::packet::FramedSphinxPacket;
use nym_sphinx_params::{PacketMode, PacketSize};
use nym_sphinx_types::{
Delay as SphinxDelay, DestinationAddressBytes, NodeAddressBytes, Payload, PrivateKey,
ProcessedPacket, SphinxPacket,
};
@@ -51,7 +51,7 @@ impl SphinxPacketProcessor {
packet: SphinxPacket,
) -> Result<ProcessedPacket, MixProcessingError> {
packet.process(&self.sphinx_key).map_err(|err| {
debug!("Failed to unwrap Sphinx packet: {:?}", err);
debug!("Failed to unwrap Sphinx packet: {err}");
MixProcessingError::SphinxProcessingError(err)
})
}
@@ -195,7 +195,7 @@ impl SphinxPacketProcessor {
#[cfg(test)]
mod tests {
use super::*;
use nymsphinx_types::crypto::keygen;
use nym_sphinx_types::crypto::keygen;
fn fixture() -> SphinxPacketProcessor {
let local_keys = keygen();
common/mixnode-common/src/verloc/error.rs
+5 -5
View File
@@ -53,19 +53,19 @@ impl Display for RttError {
write!(f, "The received reply packet had invalid signature")
}
RttError::UnreachableNode(id, err) => {
write!(f, "Could not establish connection to {} - {}", id, err)
write!(f, "Could not establish connection to {id} - {err}")
}
RttError::UnexpectedConnectionFailureWrite(id, err) => {
write!(f, "Failed to write echo packet to {} - {}", id, err)
write!(f, "Failed to write echo packet to {id} - {err}")
}
RttError::UnexpectedConnectionFailureRead(id, err) => {
write!(f, "Failed to read reply packet from {} - {}", id, err)
write!(f, "Failed to read reply packet from {id} - {err}")
}
RttError::ConnectionReadTimeout(id) => {
write!(f, "Timed out while trying to read reply packet from {}", id)
write!(f, "Timed out while trying to read reply packet from {id}")
}
RttError::ConnectionWriteTimeout(id) => {
write!(f, "Timed out while trying to write echo packet to {}", id)
write!(f, "Timed out while trying to write echo packet to {id}")
}
RttError::UnexpectedReplySequence => write!(
f,
common/mixnode-common/src/verloc/listener.rs
+10 -9
View File
@@ -4,14 +4,14 @@
use crate::verloc::error::RttError;
use crate::verloc::packet::{EchoPacket, ReplyPacket};
use bytes::{BufMut, BytesMut};
use crypto::asymmetric::identity;
use futures::StreamExt;
use log::*;
use nym_crypto::asymmetric::identity;
use nym_task::TaskClient;
use std::fmt::{Display, Formatter};
use std::net::SocketAddr;
use std::sync::Arc;
use std::{fmt, io, process};
use task::ShutdownListener;
use tokio::io::AsyncWriteExt;
use tokio::net::{TcpListener, TcpStream};
use tokio_util::codec::{Decoder, Encoder, Framed};
@@ -19,14 +19,14 @@ use tokio_util::codec::{Decoder, Encoder, Framed};
pub(crate) struct PacketListener {
address: SocketAddr,
connection_handler: Arc<ConnectionHandler>,
shutdown: ShutdownListener,
shutdown: TaskClient,
}
impl PacketListener {
pub(crate) fn new(
address: SocketAddr,
identity: Arc<identity::KeyPair>,
shutdown: ShutdownListener,
shutdown: TaskClient,
) -> Self {
PacketListener {
address,
@@ -56,7 +56,8 @@ impl PacketListener {
while !shutdown_listener.is_shutdown() {
// cloning the arc as each accepted socket is handled in separate task
let connection_handler = Arc::clone(&self.connection_handler);
let handler_shutdown_listener = self.shutdown.clone();
let mut handler_shutdown_listener = self.shutdown.clone();
handler_shutdown_listener.mark_as_success();
tokio::select! {
socket = listener.accept() => {
@@ -66,7 +67,7 @@ impl PacketListener {
tokio::spawn(connection_handler.handle_connection(socket, remote_addr, handler_shutdown_listener));
}
Err(err) => warn!("Failed to accept incoming connection - {:?}", err),
Err(err) => warn!("Failed to accept incoming connection - {err}"),
}
},
_ = shutdown_listener.recv() => {
@@ -91,7 +92,7 @@ impl ConnectionHandler {
self: Arc<Self>,
conn: TcpStream,
remote: SocketAddr,
mut shutdown_listener: ShutdownListener,
mut shutdown_listener: TaskClient,
) {
debug!("Starting connection handler for {:?}", remote);
@@ -141,9 +142,9 @@ enum EchoPacketCodecError {
impl Display for EchoPacketCodecError {
fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
match self {
EchoPacketCodecError::IoError(err) => write!(f, "encountered io error - {}", err),
EchoPacketCodecError::IoError(err) => write!(f, "encountered io error - {err}"),
EchoPacketCodecError::PacketRecoveryError(err) => {
write!(f, "failed to correctly decode an echo packet - {}", err)
write!(f, "failed to correctly decode an echo packet - {err}")
}
}
}
common/mixnode-common/src/verloc/measurement.rs
+1 -1
View File
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// SPDX-License-Identifier: Apache-2.0
use crypto::asymmetric::identity;
use nym_crypto::asymmetric::identity;
use serde::{Serialize, Serializer};
use std::cmp::Ordering;
use std::fmt::{self, Display, Formatter};
common/mixnode-common/src/verloc/mod.rs
+23 -24
View File
@@ -3,20 +3,22 @@
use crate::verloc::listener::PacketListener;
use crate::verloc::sender::{PacketSender, TestedNode};
use crypto::asymmetric::identity;
use futures::stream::FuturesUnordered;
use futures::StreamExt;
use log::*;
use nym_bin_common::version_checker::{self, parse_version};
use nym_crypto::asymmetric::identity;
use nym_network_defaults::mainnet::NYM_API;
use nym_task::TaskClient;
use rand::seq::SliceRandom;
use rand::thread_rng;
use std::net::{SocketAddr, ToSocketAddrs};
use std::net::SocketAddr;
use std::net::ToSocketAddrs;
use std::sync::Arc;
use std::time::Duration;
use task::ShutdownListener;
use tokio::task::JoinHandle;
use tokio::time::sleep;
use url::Url;
use version_checker::parse_version;
pub use crate::verloc::measurement::{AtomicVerlocResult, Verloc, VerlocResult};
@@ -69,8 +71,8 @@ pub struct Config {
/// due to being unable to get the list of nodes.
retry_timeout: Duration,
/// URLs to the validator apis for obtaining network topology.
validator_api_urls: Vec<Url>,
/// URLs to the nym apis for obtaining network topology.
nym_api_urls: Vec<Url>,
}
impl Config {
@@ -132,15 +134,15 @@ impl ConfigBuilder {
self
}
pub fn validator_api_urls(mut self, validator_api_urls: Vec<Url>) -> Self {
self.0.validator_api_urls = validator_api_urls;
pub fn nym_api_urls(mut self, nym_api_urls: Vec<Url>) -> Self {
self.0.nym_api_urls = nym_api_urls;
self
}
pub fn build(self) -> Config {
// panics here are fine as those are only ever constructed at the initial setup
assert!(
!self.0.validator_api_urls.is_empty(),
!self.0.nym_api_urls.is_empty(),
"at least one validator endpoint must be provided",
);
self.0
@@ -151,7 +153,7 @@ impl Default for ConfigBuilder {
fn default() -> Self {
ConfigBuilder(Config {
minimum_compatible_node_version: parse_version(MINIMUM_NODE_VERSION).unwrap(),
listening_address: format!("[::]:{}", DEFAULT_VERLOC_PORT).parse().unwrap(),
listening_address: format!("[::]:{DEFAULT_VERLOC_PORT}").parse().unwrap(),
packets_per_node: DEFAULT_PACKETS_PER_NODE,
packet_timeout: DEFAULT_PACKET_TIMEOUT,
connection_timeout: DEFAULT_CONNECTION_TIMEOUT,
@@ -159,7 +161,7 @@ impl Default for ConfigBuilder {
tested_nodes_batch_size: DEFAULT_BATCH_SIZE,
testing_interval: DEFAULT_TESTING_INTERVAL,
retry_timeout: DEFAULT_RETRY_TIMEOUT,
validator_api_urls: vec![],
nym_api_urls: vec![NYM_API.parse().expect("Invalid default API URL")],
})
}
}
@@ -168,7 +170,7 @@ pub struct VerlocMeasurer {
config: Config,
packet_sender: Arc<PacketSender>,
packet_listener: Arc<PacketListener>,
shutdown_listener: ShutdownListener,
shutdown_listener: TaskClient,
currently_used_api: usize,
@@ -176,7 +178,7 @@ pub struct VerlocMeasurer {
// It only does bunch of REST queries. If we update it at some point to a more sophisticated (maybe signing) client,
// then it definitely cannot be constructed here and probably will need to be passed from outside,
// as mixnodes/gateways would already be using an instance of said client.
validator_client: validator_client::ApiClient,
validator_client: validator_client::NymApiClient,
results: AtomicVerlocResult,
}
@@ -184,9 +186,9 @@ impl VerlocMeasurer {
pub fn new(
mut config: Config,
identity: Arc<identity::KeyPair>,
shutdown_listener: ShutdownListener,
shutdown_listener: TaskClient,
) -> Self {
config.validator_api_urls.shuffle(&mut thread_rng());
config.nym_api_urls.shuffle(&mut thread_rng());
VerlocMeasurer {
packet_sender: Arc::new(PacketSender::new(
@@ -204,24 +206,21 @@ impl VerlocMeasurer {
)),
shutdown_listener,
currently_used_api: 0,
validator_client: validator_client::ApiClient::new(
config.validator_api_urls[0].clone(),
),
validator_client: validator_client::NymApiClient::new(config.nym_api_urls[0].clone()),
config,
results: AtomicVerlocResult::new(),
}
}
fn use_next_validator_api(&mut self) {
if self.config.validator_api_urls.len() == 1 {
fn use_next_nym_api(&mut self) {
if self.config.nym_api_urls.len() == 1 {
warn!("There's only a single validator API available - it won't be possible to use a different one");
return;
}
self.currently_used_api =
(self.currently_used_api + 1) % self.config.validator_api_urls.len();
self.currently_used_api = (self.currently_used_api + 1) % self.config.nym_api_urls.len();
self.validator_client
.change_validator_api(self.config.validator_api_urls[self.currently_used_api].clone())
.change_nym_api(self.config.nym_api_urls[self.currently_used_api].clone())
}
pub fn get_verloc_results_pointer(&self) -> AtomicVerlocResult {
@@ -308,7 +307,7 @@ impl VerlocMeasurer {
"failed to obtain list of mixnodes from the validator - {}. Going to attempt to use another validator API in the next run",
err
);
self.use_next_validator_api();
self.use_next_nym_api();
sleep(self.config.retry_timeout).await;
continue;
}
common/mixnode-common/src/verloc/packet.rs
+1 -1
View File
@@ -2,7 +2,7 @@
// SPDX-License-Identifier: Apache-2.0
use crate::verloc::error::RttError;
use crypto::asymmetric::identity::{self, PUBLIC_KEY_LENGTH, SIGNATURE_LENGTH};
use nym_crypto::asymmetric::identity::{self, PUBLIC_KEY_LENGTH, SIGNATURE_LENGTH};
use std::convert::TryInto;
pub(crate) struct EchoPacket {
common/mixnode-common/src/verloc/sender.rs
+5 -4
View File
@@ -4,14 +4,14 @@
use crate::verloc::error::RttError;
use crate::verloc::measurement::Measurement;
use crate::verloc::packet::{EchoPacket, ReplyPacket};
use crypto::asymmetric::identity;
use log::*;
use nym_crypto::asymmetric::identity;
use nym_task::TaskClient;
use rand::{thread_rng, Rng};
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use std::{fmt, io};
use task::ShutdownListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::time::sleep;
@@ -45,7 +45,7 @@ pub(crate) struct PacketSender {
packet_timeout: Duration,
connection_timeout: Duration,
delay_between_packets: Duration,
shutdown_listener: ShutdownListener,
shutdown_listener: TaskClient,
}
impl PacketSender {
@@ -55,7 +55,7 @@ impl PacketSender {
packet_timeout: Duration,
connection_timeout: Duration,
delay_between_packets: Duration,
shutdown_listener: ShutdownListener,
shutdown_listener: TaskClient,
) -> Self {
PacketSender {
identity,
@@ -84,6 +84,7 @@ impl PacketSender {
tested_node: TestedNode,
) -> Result<Measurement, RttError> {
let mut shutdown_listener = self.shutdown_listener.clone();
shutdown_listener.mark_as_success();
let mut conn = match tokio::time::timeout(
self.connection_timeout,