bugfix: make gateways insert themselves into [local] topology (#5038)
* added explicit SP suffix to started tasks * added 'GatewayTopologyProvider' that always injects itself into the network * use the new topology provider to bypass described bootstrapping problem
This commit is contained in:
committed by
GitHub
parent
9234474565
commit
d626e7689f
Generated
+2
@@ -5293,9 +5293,11 @@ dependencies = [
|
||||
"nym-network-requester",
|
||||
"nym-node-http-api",
|
||||
"nym-pemstore",
|
||||
"nym-sdk",
|
||||
"nym-sphinx",
|
||||
"nym-statistics-common",
|
||||
"nym-task",
|
||||
"nym-topology",
|
||||
"nym-types",
|
||||
"nym-validator-client",
|
||||
"nym-wireguard",
|
||||
|
||||
@@ -6,7 +6,6 @@ pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit};
|
||||
use futures::StreamExt;
|
||||
use log::*;
|
||||
use nym_sphinx::addressing::nodes::NodeIdentity;
|
||||
use nym_topology::provider_trait::TopologyProvider;
|
||||
use nym_topology::NymTopologyError;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -18,7 +17,11 @@ use wasmtimer::tokio::sleep;
|
||||
|
||||
mod accessor;
|
||||
pub mod geo_aware_provider;
|
||||
pub(crate) mod nym_api_provider;
|
||||
pub mod nym_api_provider;
|
||||
|
||||
pub use geo_aware_provider::GeoAwareTopologyProvider;
|
||||
pub use nym_api_provider::{Config as NymApiTopologyProviderConfig, NymApiTopologyProvider};
|
||||
pub use nym_topology::provider_trait::TopologyProvider;
|
||||
|
||||
// TODO: move it to config later
|
||||
const MAX_FAILURE_COUNT: usize = 10;
|
||||
|
||||
@@ -14,9 +14,10 @@ use url::Url;
|
||||
pub const DEFAULT_MIN_MIXNODE_PERFORMANCE: u8 = 50;
|
||||
pub const DEFAULT_MIN_GATEWAY_PERFORMANCE: u8 = 50;
|
||||
|
||||
pub(crate) struct Config {
|
||||
pub(crate) min_mixnode_performance: u8,
|
||||
pub(crate) min_gateway_performance: u8,
|
||||
#[derive(Debug)]
|
||||
pub struct Config {
|
||||
pub min_mixnode_performance: u8,
|
||||
pub min_gateway_performance: u8,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
@@ -29,7 +30,7 @@ impl Default for Config {
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct NymApiTopologyProvider {
|
||||
pub struct NymApiTopologyProvider {
|
||||
config: Config,
|
||||
|
||||
validator_client: nym_validator_client::client::NymApiClient,
|
||||
@@ -40,7 +41,7 @@ pub(crate) struct NymApiTopologyProvider {
|
||||
}
|
||||
|
||||
impl NymApiTopologyProvider {
|
||||
pub(crate) fn new(
|
||||
pub fn new(
|
||||
config: Config,
|
||||
mut nym_api_urls: Vec<Url>,
|
||||
client_version: String,
|
||||
|
||||
@@ -286,6 +286,10 @@ impl NymTopology {
|
||||
self.get_gateway(gateway_identity).is_some()
|
||||
}
|
||||
|
||||
pub fn insert_gateway(&mut self, gateway: gateway::LegacyNode) {
|
||||
self.gateways.push(gateway)
|
||||
}
|
||||
|
||||
pub fn set_gateways(&mut self, gateways: Vec<gateway::LegacyNode>) {
|
||||
self.gateways = gateways
|
||||
}
|
||||
|
||||
@@ -76,9 +76,11 @@ nym-network-defaults = { path = "../common/network-defaults" }
|
||||
nym-network-requester = { path = "../service-providers/network-requester" }
|
||||
nym-node-http-api = { path = "../nym-node/nym-node-http-api" }
|
||||
nym-pemstore = { path = "../common/pemstore" }
|
||||
nym-sdk = { path = "../sdk/rust/nym-sdk" }
|
||||
nym-sphinx = { path = "../common/nymsphinx" }
|
||||
nym-statistics-common = { path = "../common/statistics" }
|
||||
nym-task = { path = "../common/task" }
|
||||
nym-topology = { path = "../common/topology" }
|
||||
nym-types = { path = "../common/types" }
|
||||
nym-validator-client = { path = "../common/client-libs/validator-client" }
|
||||
nym-ip-packet-router = { path = "../service-providers/ip-packet-router" }
|
||||
|
||||
@@ -3,13 +3,18 @@
|
||||
|
||||
use crate::config::Config;
|
||||
use crate::error::GatewayError;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use nym_crypto::asymmetric::encryption;
|
||||
use nym_gateway_storage::PersistentStorage;
|
||||
use nym_pemstore::traits::PemStorableKeyPair;
|
||||
use nym_pemstore::KeyPairPath;
|
||||
|
||||
use nym_sdk::{NymApiTopologyProvider, NymApiTopologyProviderConfig, UserAgent};
|
||||
use nym_topology::{gateway, NymTopology, TopologyProvider};
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::Mutex;
|
||||
use tracing::debug;
|
||||
use url::Url;
|
||||
|
||||
pub async fn load_network_requester_config<P: AsRef<Path>>(
|
||||
id: &str,
|
||||
@@ -93,3 +98,56 @@ pub(crate) fn load_sphinx_keys(config: &Config) -> Result<encryption::KeyPair, G
|
||||
);
|
||||
load_keypair(sphinx_paths, "gateway sphinx")
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct GatewayTopologyProvider {
|
||||
inner: Arc<Mutex<GatewayTopologyProviderInner>>,
|
||||
}
|
||||
|
||||
impl GatewayTopologyProvider {
|
||||
pub fn new(
|
||||
gateway_node: gateway::LegacyNode,
|
||||
user_agent: UserAgent,
|
||||
nym_api_url: Vec<Url>,
|
||||
) -> GatewayTopologyProvider {
|
||||
GatewayTopologyProvider {
|
||||
inner: Arc::new(Mutex::new(GatewayTopologyProviderInner {
|
||||
inner: NymApiTopologyProvider::new(
|
||||
NymApiTopologyProviderConfig {
|
||||
min_mixnode_performance: 50,
|
||||
min_gateway_performance: 0,
|
||||
},
|
||||
nym_api_url,
|
||||
env!("CARGO_PKG_VERSION").to_string(),
|
||||
Some(user_agent),
|
||||
),
|
||||
gateway_node,
|
||||
})),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct GatewayTopologyProviderInner {
|
||||
inner: NymApiTopologyProvider,
|
||||
gateway_node: gateway::LegacyNode,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl TopologyProvider for GatewayTopologyProvider {
|
||||
async fn get_new_topology(&mut self) -> Option<NymTopology> {
|
||||
let mut guard = self.inner.lock().await;
|
||||
match guard.inner.get_new_topology().await {
|
||||
None => None,
|
||||
Some(mut base) => {
|
||||
if !base.gateway_exists(&guard.gateway_node.identity_key) {
|
||||
debug!(
|
||||
"{} didn't exist in topology. inserting it.",
|
||||
guard.gateway_node.identity_key
|
||||
);
|
||||
base.insert_gateway(guard.gateway_node.clone());
|
||||
}
|
||||
Some(base)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+52
-2
@@ -12,9 +12,12 @@ use crate::http::HttpApiBuilder;
|
||||
use crate::node::client_handling::active_clients::ActiveClientsStore;
|
||||
use crate::node::client_handling::embedded_clients::{LocalEmbeddedClientHandle, MessageRouter};
|
||||
use crate::node::client_handling::websocket;
|
||||
use crate::node::helpers::{initialise_main_storage, load_network_requester_config};
|
||||
use crate::node::helpers::{
|
||||
initialise_main_storage, load_network_requester_config, GatewayTopologyProvider,
|
||||
};
|
||||
use crate::node::mixnet_handling::receiver::connection_handler::ConnectionHandler;
|
||||
use futures::channel::{mpsc, oneshot};
|
||||
use nym_bin_common::bin_info;
|
||||
use nym_credential_verification::ecash::{
|
||||
credential_sender::CredentialHandlerConfig, EcashManager,
|
||||
};
|
||||
@@ -25,13 +28,15 @@ use nym_network_requester::{LocalGateway, NRServiceProviderBuilder, RequestFilte
|
||||
use nym_node_http_api::state::metrics::SharedSessionStats;
|
||||
use nym_statistics_common::events::{self, StatsEventSender};
|
||||
use nym_task::{TaskClient, TaskHandle, TaskManager};
|
||||
use nym_topology::NetworkAddress;
|
||||
use nym_types::gateway::GatewayNodeDetailsResponse;
|
||||
use nym_validator_client::client::NodeId;
|
||||
use nym_validator_client::nyxd::{Coin, CosmWasmClient};
|
||||
use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient};
|
||||
use rand::seq::SliceRandom;
|
||||
use rand::thread_rng;
|
||||
use statistics::GatewayStatisticsCollector;
|
||||
use std::net::SocketAddr;
|
||||
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use tracing::*;
|
||||
@@ -225,6 +230,39 @@ impl<St> Gateway<St> {
|
||||
crate::helpers::node_details(&self.config).await
|
||||
}
|
||||
|
||||
fn gateway_topology_provider(&self) -> GatewayTopologyProvider {
|
||||
GatewayTopologyProvider::new(
|
||||
self.as_topology_node(),
|
||||
bin_info!().into(),
|
||||
self.config.gateway.nym_api_urls.clone(),
|
||||
)
|
||||
}
|
||||
|
||||
fn as_topology_node(&self) -> nym_topology::gateway::LegacyNode {
|
||||
let ip = self
|
||||
.config
|
||||
.host
|
||||
.public_ips
|
||||
.first()
|
||||
.copied()
|
||||
.unwrap_or(IpAddr::V4(Ipv4Addr::LOCALHOST));
|
||||
let mix_host = SocketAddr::new(ip, self.config.gateway.mix_port);
|
||||
|
||||
nym_topology::gateway::LegacyNode {
|
||||
// those fields are irrelevant for the purposes of routing so it's fine if they're inaccurate.
|
||||
// the only thing that matters is the identity key (and maybe version)
|
||||
node_id: NodeId::MAX,
|
||||
mix_host,
|
||||
host: NetworkAddress::IpAddr(ip),
|
||||
clients_ws_port: self.config.gateway.clients_port,
|
||||
clients_wss_port: self.config.gateway.clients_wss_port,
|
||||
sphinx_key: *self.sphinx_keypair.public_key(),
|
||||
|
||||
identity_key: *self.identity_keypair.public_key(),
|
||||
version: env!("CARGO_PKG_VERSION").into(),
|
||||
}
|
||||
}
|
||||
|
||||
fn start_mix_socket_listener(
|
||||
&self,
|
||||
ack_sender: MixForwardingSender,
|
||||
@@ -257,6 +295,7 @@ impl<St> Gateway<St> {
|
||||
async fn start_authenticator(
|
||||
&mut self,
|
||||
forwarding_channel: MixForwardingSender,
|
||||
topology_provider: GatewayTopologyProvider,
|
||||
shutdown: TaskClient,
|
||||
ecash_verifier: Arc<EcashManager<St>>,
|
||||
) -> Result<StartedAuthenticator, Box<dyn std::error::Error + Send + Sync>>
|
||||
@@ -304,6 +343,7 @@ impl<St> Gateway<St> {
|
||||
.with_shutdown(shutdown.fork("authenticator"))
|
||||
.with_wait_for_gateway(true)
|
||||
.with_minimum_gateway_performance(0)
|
||||
.with_custom_topology_provider(Box::new(topology_provider))
|
||||
.with_on_start(on_start_tx);
|
||||
|
||||
if let Some(custom_mixnet) = &opts.custom_mixnet_path {
|
||||
@@ -352,6 +392,7 @@ impl<St> Gateway<St> {
|
||||
async fn start_authenticator(
|
||||
&self,
|
||||
_forwarding_channel: MixForwardingSender,
|
||||
_topology_provider: GatewayTopologyProvider,
|
||||
_shutdown: TaskClient,
|
||||
_ecash_verifier: Arc<EcashManager<St>>,
|
||||
) -> Result<StartedAuthenticator, Box<dyn std::error::Error + Send + Sync>> {
|
||||
@@ -424,6 +465,7 @@ impl<St> Gateway<St> {
|
||||
async fn start_network_requester(
|
||||
&self,
|
||||
forwarding_channel: MixForwardingSender,
|
||||
topology_provider: GatewayTopologyProvider,
|
||||
shutdown: TaskClient,
|
||||
) -> Result<StartedNetworkRequester, GatewayError> {
|
||||
info!("Starting network requester...");
|
||||
@@ -451,6 +493,7 @@ impl<St> Gateway<St> {
|
||||
.with_custom_gateway_transceiver(Box::new(transceiver))
|
||||
.with_wait_for_gateway(true)
|
||||
.with_minimum_gateway_performance(0)
|
||||
.with_custom_topology_provider(Box::new(topology_provider))
|
||||
.with_on_start(on_start_tx);
|
||||
|
||||
if let Some(custom_mixnet) = &nr_opts.custom_mixnet_path {
|
||||
@@ -488,6 +531,7 @@ impl<St> Gateway<St> {
|
||||
async fn start_ip_packet_router(
|
||||
&self,
|
||||
forwarding_channel: MixForwardingSender,
|
||||
topology_provider: GatewayTopologyProvider,
|
||||
shutdown: TaskClient,
|
||||
) -> Result<LocalEmbeddedClientHandle, GatewayError> {
|
||||
info!("Starting IP packet provider...");
|
||||
@@ -516,6 +560,7 @@ impl<St> Gateway<St> {
|
||||
.with_custom_gateway_transceiver(Box::new(transceiver))
|
||||
.with_wait_for_gateway(true)
|
||||
.with_minimum_gateway_performance(0)
|
||||
.with_custom_topology_provider(Box::new(topology_provider))
|
||||
.with_on_start(on_start_tx);
|
||||
|
||||
if let Some(custom_mixnet) = &ip_opts.custom_mixnet_path {
|
||||
@@ -632,6 +677,8 @@ impl<St> Gateway<St> {
|
||||
shutdown.fork("statistics::GatewayStatisticsCollector"),
|
||||
);
|
||||
|
||||
let topology_provider = self.gateway_topology_provider();
|
||||
|
||||
let handler_config = CredentialHandlerConfig {
|
||||
revocation_bandwidth_penalty: self
|
||||
.config
|
||||
@@ -680,6 +727,7 @@ impl<St> Gateway<St> {
|
||||
let embedded_nr = self
|
||||
.start_network_requester(
|
||||
mix_forwarding_channel.clone(),
|
||||
topology_provider.clone(),
|
||||
shutdown.fork("NetworkRequester"),
|
||||
)
|
||||
.await?;
|
||||
@@ -695,6 +743,7 @@ impl<St> Gateway<St> {
|
||||
let embedded_ip_sp = self
|
||||
.start_ip_packet_router(
|
||||
mix_forwarding_channel.clone(),
|
||||
topology_provider.clone(),
|
||||
shutdown.fork("ip_service_provider"),
|
||||
)
|
||||
.await?;
|
||||
@@ -707,6 +756,7 @@ impl<St> Gateway<St> {
|
||||
let embedded_auth = self
|
||||
.start_authenticator(
|
||||
mix_forwarding_channel,
|
||||
topology_provider,
|
||||
shutdown.fork("authenticator"),
|
||||
ecash_verifier,
|
||||
)
|
||||
|
||||
@@ -10,7 +10,13 @@ pub mod mixnet;
|
||||
pub mod tcp_proxy;
|
||||
|
||||
pub use error::{Error, Result};
|
||||
pub use nym_client_core::client::mix_traffic::transceiver::*;
|
||||
pub use nym_client_core::client::{
|
||||
mix_traffic::transceiver::*,
|
||||
topology_control::{
|
||||
GeoAwareTopologyProvider, NymApiTopologyProvider, NymApiTopologyProviderConfig,
|
||||
TopologyProvider,
|
||||
},
|
||||
};
|
||||
pub use nym_network_defaults::{
|
||||
ChainDetails, DenomDetails, DenomDetailsOwned, NymContracts, NymNetworkDetails,
|
||||
ValidatorDetails,
|
||||
|
||||
@@ -128,7 +128,9 @@ impl<S: Storage + Clone + 'static> Authenticator<S> {
|
||||
// Connect to the mixnet
|
||||
let mixnet_client = crate::mixnet_client::create_mixnet_client(
|
||||
&self.config.base,
|
||||
task_handle.get_handle().named("nym_sdk::MixnetClient"),
|
||||
task_handle
|
||||
.get_handle()
|
||||
.named("nym_sdk::MixnetClient[AUTH]"),
|
||||
self.custom_gateway_transceiver,
|
||||
self.custom_topology_provider,
|
||||
self.wait_for_gateway,
|
||||
|
||||
@@ -133,7 +133,7 @@ impl IpPacketRouter {
|
||||
// Connect to the mixnet
|
||||
let mixnet_client = crate::mixnet_client::create_mixnet_client(
|
||||
&self.config.base,
|
||||
task_handle.get_handle().named("nym_sdk::MixnetClient"),
|
||||
task_handle.get_handle().named("nym_sdk::MixnetClient[IPR]"),
|
||||
self.custom_gateway_transceiver,
|
||||
self.custom_topology_provider,
|
||||
self.wait_for_gateway,
|
||||
|
||||
@@ -239,7 +239,7 @@ impl NRServiceProviderBuilder {
|
||||
// Connect to the mixnet
|
||||
let mixnet_client = create_mixnet_client(
|
||||
&self.config.base,
|
||||
shutdown.get_handle().named("nym_sdk::MixnetClient"),
|
||||
shutdown.get_handle().named("nym_sdk::MixnetClient[NR]"),
|
||||
self.custom_gateway_transceiver,
|
||||
self.custom_topology_provider,
|
||||
self.wait_for_gateway,
|
||||
|
||||
Reference in New Issue
Block a user