diff --git a/nym-node/src/node/lp/cleanup.rs b/nym-node/src/node/lp/cleanup.rs index 0bc522a3d5..546d8426cd 100644 --- a/nym-node/src/node/lp/cleanup.rs +++ b/nym-node/src/node/lp/cleanup.rs @@ -3,11 +3,7 @@ use crate::config::LpDebug; use crate::node::lp::state::ActiveLpSessions; -use dashmap::DashMap; -use nym_lp::LpTransportSession; -use nym_lp::peer_config::LpReceiverIndex; use nym_metrics::inc_by; -use std::sync::Arc; use std::time::Duration; use tracing::{debug, info}; diff --git a/nym-node/src/node/lp/control/egress/connection.rs b/nym-node/src/node/lp/control/egress/connection.rs index 59f29215cf..5c2d72c03d 100644 --- a/nym-node/src/node/lp/control/egress/connection.rs +++ b/nym-node/src/node/lp/control/egress/connection.rs @@ -8,7 +8,7 @@ use crate::node::lp::forwarding::client_connection::NestedClientConnectionSender use crate::node::lp::state::SharedLpNodeControlState; use nym_lp::LpTransportSession; use nym_lp::peer_config::LpReceiverIndex; -use nym_lp::transport::LpHandshakeChannel; +use nym_lp::transport::{LpHandshakeChannel, LpTransportChannel}; use std::collections::HashMap; use std::net::SocketAddr; use tracing::{debug, warn}; @@ -103,7 +103,7 @@ where self.responder_details.node_id ); - // TODO: return session, etc. + // TODO: change return type into complete handler Some(Ok(session)) } } @@ -128,7 +128,7 @@ pub(crate) struct NestedNodeConnectionHandler { impl NestedNodeConnectionHandler where -// S: LpTransport + Unpin, + S: LpTransportChannel + Unpin, { /// Attempt to extract outer receiver index from the received message /// (that is meant to be an `LpPacket`) diff --git a/nym-node/src/node/lp/control/ingress/client_handler.rs b/nym-node/src/node/lp/control/ingress/client_handler.rs index 6a5628730c..2a25aec4df 100644 --- a/nym-node/src/node/lp/control/ingress/client_handler.rs +++ b/nym-node/src/node/lp/control/ingress/client_handler.rs @@ -4,6 +4,7 @@ use crate::node::lp::cleanup::TimestampedState; use crate::node::lp::control::{LP_DURATION_BUCKETS, LpConnectionStats}; use crate::node::lp::error::LpHandlerError; +use crate::node::lp::forwarding::client_connection::NestedClientConnection; use crate::node::lp::state::SharedLpClientControlState; use dashmap::mapref::one::RefMut; use nym_lp::packet::message::LpMessageType; @@ -13,7 +14,7 @@ use nym_lp::session::{LpAction, LpInput}; use nym_lp::transport::LpHandshakeChannel; use nym_lp::transport::traits::LpTransportChannel; use nym_lp::{LpTransportSession, packet::message::ExpectedResponseSize}; -use nym_metrics::{add_histogram_obs, inc, inc_by}; +use nym_metrics::{add_histogram_obs, inc}; use nym_node_metrics::NymNodeMetrics; use nym_registration_common::{LpRegistrationRequest, RegistrationStatus}; use std::net::SocketAddr; @@ -32,8 +33,6 @@ pub struct LpClientConnectionHandler { state: SharedLpClientControlState, stats: LpConnectionStats, - // /// Flag indicating whether this is a connection from an entry gateway serving as a proxy - // forwarded_connection: bool, /// Bound receiver_idx for this connection (set after first packet). /// All subsequent packets on this connection must use this receiver_idx. /// Set from ClientHello's proposed receiver_index, or from header for non-bootstrap packets. @@ -43,26 +42,24 @@ pub struct LpClientConnectionHandler { /// Opened on first forward, reused for subsequent forwards, closed when client disconnects. /// Tuple contains (stream, target_address) to verify subsequent forwards go to same exit. exit_stream: Option<(S, SocketAddr)>, + + #[allow(dead_code)] + forwarding_channel: Option, } impl LpClientConnectionHandler where S: LpTransportChannel + LpHandshakeChannel + Unpin, { - pub fn new( - stream: S, - // forwarded_connection: bool, - remote_addr: SocketAddr, - state: SharedLpClientControlState, - ) -> Self { + pub fn new(stream: S, remote_addr: SocketAddr, state: SharedLpClientControlState) -> Self { Self { stream, remote_addr, - // forwarded_connection, state, stats: LpConnectionStats::new(), bound_receiver_idx: None, exit_stream: None, + forwarding_channel: None, } } diff --git a/nym-node/src/node/lp/control/ingress/listener.rs b/nym-node/src/node/lp/control/ingress/listener.rs index ca316b57be..ea6e3ac37a 100644 --- a/nym-node/src/node/lp/control/ingress/listener.rs +++ b/nym-node/src/node/lp/control/ingress/listener.rs @@ -5,7 +5,7 @@ use crate::config::LpConfig; use crate::error::NymNodeError; use crate::node::lp::control::ingress::client_handler::LpClientConnectionHandler; use crate::node::lp::control::ingress::node_handler::InitialLpIngressNodeConnectionHandler; -use crate::node::lp::directory::{LpNodeDetails, LpNodes}; +use crate::node::lp::directory::LpNodeDetails; use crate::node::lp::state::{SharedLpClientControlState, SharedLpNodeControlState}; use nym_task::ShutdownTracker; use std::net::SocketAddr; @@ -30,15 +30,16 @@ pub struct LpControlListener { impl LpControlListener { pub fn new( bind_address: SocketAddr, - handler_state: SharedLpClientControlState, + clients_handler_state: SharedLpClientControlState, + nodes_handler_state: SharedLpNodeControlState, shutdown: ShutdownTracker, ) -> Self { - todo!() - // Self { - // bind_address, - // handler_state, - // shutdown, - // } + Self { + bind_address, + clients_handler_state, + nodes_handler_state, + shutdown, + } } fn lp_config(&self) -> LpConfig { diff --git a/nym-node/src/node/lp/control/ingress/node_handler.rs b/nym-node/src/node/lp/control/ingress/node_handler.rs index 8f08f4d69d..5215c699a9 100644 --- a/nym-node/src/node/lp/control/ingress/node_handler.rs +++ b/nym-node/src/node/lp/control/ingress/node_handler.rs @@ -134,7 +134,7 @@ where S: LpHandshakeChannel + LpTransportChannel + Unpin, { async fn handle(mut self) -> Result<(), LpHandlerError> { - todo!(); + // handle all the forwarding here self.stats.emit_lifecycle_node_metrics(true); Ok(()) diff --git a/nym-node/src/node/lp/control/tests.rs b/nym-node/src/node/lp/control/tests.rs index 3580dea28b..74d720ade6 100644 --- a/nym-node/src/node/lp/control/tests.rs +++ b/nym-node/src/node/lp/control/tests.rs @@ -45,8 +45,8 @@ mod tests { nym_test_utils::helpers::setup_test_logger(); let (init, resp) = mock_peers(); - let init_remote = resp.as_remote(); - let resp_remote = init.as_remote(); + let init_remote = init.as_remote(); + let resp_remote = resp.as_remote(); let conn_init = MockIOStream::default(); let conn_resp = conn_init.try_get_remote_handle(); diff --git a/nym-node/src/node/lp/directory.rs b/nym-node/src/node/lp/directory.rs index b1f7c0a79a..78e16357a2 100644 --- a/nym-node/src/node/lp/directory.rs +++ b/nym-node/src/node/lp/directory.rs @@ -18,13 +18,16 @@ pub struct LpNodes { } impl LpNodes { - pub(crate) fn is_from_known_node(&self, node_ip: IpAddr) -> bool { - self.nodes.load().contains_key(&node_ip) - } - pub(crate) fn get_node_details(&self, node_ip: IpAddr) -> Option { self.nodes.load().get(&node_ip).cloned() } + + pub(crate) fn get_node_id(&self, node_ip: IpAddr) -> Option { + self.nodes + .load() + .get(&node_ip) + .map(|details| details.node_id) + } } #[derive(Clone)] diff --git a/nym-node/src/node/lp/error.rs b/nym-node/src/node/lp/error.rs index f26b680a83..865d4ccaab 100644 --- a/nym-node/src/node/lp/error.rs +++ b/nym-node/src/node/lp/error.rs @@ -54,6 +54,12 @@ pub enum LpHandlerError { #[error("data channel is not yet implemented")] UnimplementedDataChannel, + #[error("{ip_addr} does not correspond to any known LP node")] + NotLpNode { ip_addr: IpAddr }, + + #[error("{0}")] + Internal(String), + #[error("{0}")] Other(String), } @@ -68,6 +74,10 @@ impl LpHandlerError { } } + pub fn internal(msg: impl Into) -> Self { + LpHandlerError::Internal(msg.into()) + } + pub fn other(msg: impl Into) -> Self { LpHandlerError::Other(msg.into()) } diff --git a/nym-node/src/node/lp/forwarding/controller.rs b/nym-node/src/node/lp/forwarding/controller.rs index 768edd7a22..59be7226f9 100644 --- a/nym-node/src/node/lp/forwarding/controller.rs +++ b/nym-node/src/node/lp/forwarding/controller.rs @@ -2,14 +2,25 @@ // SPDX-License-Identifier: GPL-3.0-only use crate::node::lp::control::egress::connection::NestedNodeControlSender; +use crate::node::lp::directory::LpNodes; +use crate::node::lp::error::LpHandlerError; use crate::node::lp::forwarding::{ - ConnectionHandlerResponse, GetConnectionHandler, NestedConnectionControllerRequest, + ConnectionControllerResponse, ConnectionHandlerResponse, ControllerResponse, + GetConnectionHandler, NestedConnectionControllerRequest, }; -use nym_crypto::asymmetric::ed25519; +use nym_topology::NodeId; use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::Notify; +use tracing::info; pub const CONTROL_CHANNEL_SIZE: usize = 64; +pub(crate) enum NodeHandle { + Established(NestedNodeControlSender), + Pending(Arc), +} + /// Keep track of connections to the exit gateway pub(crate) struct NestedConnectionsController { /// Handle channel for sending requests to this controller @@ -18,23 +29,63 @@ pub(crate) struct NestedConnectionsController { /// Channel for receiving requests in this controller receiver: super::NodeConnectionControllerReceiver, + /// Map of all LP node ip addresses to their details (and ids) + lp_nodes: LpNodes, + /// Handles to the active nested node connections - nodes_handles: HashMap, + nodes_handles: HashMap, + + /// Shutdown token + shutdown: nym_task::ShutdownToken, } impl NestedConnectionsController { + pub(crate) fn new(lp_nodes: LpNodes, shutdown: nym_task::ShutdownToken) -> Self { + let (sender, receiver) = tokio::sync::mpsc::channel(CONTROL_CHANNEL_SIZE); + + Self { + sender, + receiver, + lp_nodes, + nodes_handles: HashMap::new(), + shutdown, + } + } + pub(crate) fn request_sender(&self) -> super::NodeConnectionControllerSender { self.sender.clone() } async fn handle_get_connection_handler( - &self, + &mut self, request: GetConnectionHandler, ) -> ConnectionHandlerResponse { - todo!() + let ip = request.target_gateway_lp_address.ip(); + + let Some(node_id) = self.lp_nodes.get_node_id(ip) else { + return Err(LpHandlerError::NotLpNode { ip_addr: ip }); + }; + + match self.nodes_handles.get(&node_id) { + Some(NodeHandle::Established(handle)) => { + todo!() + } + Some(NodeHandle::Pending(notify)) => { + Ok(ConnectionControllerResponse::Pending(notify.clone())) + } + None => { + let (res, notify) = ConnectionControllerResponse::new_pending(); + self.nodes_handles + .insert(node_id, NodeHandle::Pending(notify.clone())); + + // create a new connection and return a pending response + todo!(); + return Ok(res); + } + } } - async fn handle_request(&self, request: NestedConnectionControllerRequest) { + async fn handle_request(&mut self, request: NestedConnectionControllerRequest) { match request { NestedConnectionControllerRequest::ConnectionHandler { request, @@ -46,4 +97,20 @@ impl NestedConnectionsController { } } } + + pub(crate) async fn run(&mut self) { + loop { + tokio::select! { + biased; + _ = self.shutdown.cancelled() => { + break; + } + Some(request) = self.receiver.recv() => { + self.handle_request(request).await; + } + } + } + + info!("Nested connection controller shutdown complete"); + } } diff --git a/nym-node/src/node/lp/forwarding/manager.rs b/nym-node/src/node/lp/forwarding/manager.rs index d455fd9587..4d02905827 100644 --- a/nym-node/src/node/lp/forwarding/manager.rs +++ b/nym-node/src/node/lp/forwarding/manager.rs @@ -1,27 +1,73 @@ // Copyright 2026 - Nym Technologies SA // SPDX-License-Identifier: GPL-3.0-only -use super::NodeConnectionControllerSender; +use super::{ + ConnectionControllerResponse, ConnectionHandlerResponse, GetConnectionHandler, + NestedConnectionControllerRequest, NodeConnectionControllerSender, +}; use crate::node::lp::error::LpHandlerError; use crate::node::lp::forwarding::client_connection::NestedClientConnection; -use nym_crypto::asymmetric::ed25519; +use nym_lp::peer_config::LpReceiverIndex; use std::net::SocketAddr; +use tokio::sync::oneshot; -pub(crate) struct NestedConnectionsManager { +#[derive(Clone)] +pub struct NestedConnectionsManager { sender: NodeConnectionControllerSender, } impl NestedConnectionsManager { + pub fn new(sender: NodeConnectionControllerSender) -> Self { + Self { sender } + } + + async fn send_connection_handler_request( + &self, + request: GetConnectionHandler, + ) -> Result { + let (response_tx, response_rx) = oneshot::channel(); + self.sender + .send(NestedConnectionControllerRequest::ConnectionHandler { + request, + response_tx, + }) + .await + .map_err(|_| LpHandlerError::internal("nested connection controller shut down"))?; + + response_rx.await.map_err(|_| { + LpHandlerError::internal("nested connection controller hasn't send a response") + }) + } + pub(crate) async fn get_connection_handle( &self, - target_gateway: ed25519::PublicKey, target_gateway_lp_address: SocketAddr, + inner_receiver_index: LpReceiverIndex, ) -> Result { - // let request = GetConnectionHandler { - // target_gateway, - // target_gateway_lp_address, - // }; + let request = GetConnectionHandler { + target_gateway_lp_address, + inner_receiver_index, + }; - todo!() + let notify = match self.send_connection_handler_request(request).await?? { + // if we have received a ready response, we can return the connection + ConnectionControllerResponse::Ready(conn) => return Ok(conn), + + // otherwise we need to wait for the notification when it becomes available + ConnectionControllerResponse::Pending(notify) => notify, + }; + + // TODO: timeout + notify.notified().await; + + match self.send_connection_handler_request(request).await?? { + // if we have received a ready response, we can return the connection + ConnectionControllerResponse::Ready(conn) => Ok(conn), + + // otherwise we need to wait for the notification when it becomes available + ConnectionControllerResponse::Pending(_) => Err(LpHandlerError::internal( + "unavailable connection handler after successful notification", + )), + } } } diff --git a/nym-node/src/node/lp/forwarding/mod.rs b/nym-node/src/node/lp/forwarding/mod.rs index efdd106364..ea5e73379d 100644 --- a/nym-node/src/node/lp/forwarding/mod.rs +++ b/nym-node/src/node/lp/forwarding/mod.rs @@ -3,7 +3,6 @@ use crate::node::lp::error::LpHandlerError; use crate::node::lp::forwarding::client_connection::NestedClientConnection; -use nym_crypto::asymmetric::ed25519; use nym_lp::peer_config::LpReceiverIndex; use std::net::SocketAddr; use std::sync::Arc; @@ -14,8 +13,8 @@ pub(crate) mod client_connection; pub(crate) mod controller; pub(crate) mod manager; -pub(crate) type NodeConnectionControllerReceiver = Receiver; -pub(crate) type NodeConnectionControllerSender = Sender; +pub(crate) type NodeConnectionControllerReceiver = Receiver; +pub(crate) type NodeConnectionControllerSender = Sender; pub(crate) enum ConnectionControllerResponse { /// The response is immediately available @@ -26,6 +25,16 @@ pub(crate) enum ConnectionControllerResponse { Pending(Arc), } +impl ConnectionControllerResponse { + pub fn new_pending() -> (Self, Arc) { + let notify = Arc::new(Notify::new()); + ( + ConnectionControllerResponse::Pending(notify.clone()), + notify, + ) + } +} + pub(crate) type ControllerResponse = Result, LpHandlerError>; pub(crate) type ConnectionHandlerResponse = ControllerResponse; @@ -39,10 +48,8 @@ pub(crate) enum NestedConnectionControllerRequest { }, } +#[derive(Copy, Clone)] pub(crate) struct GetConnectionHandler { - /// Target gateway's Ed25519 identity - pub target_gateway: ed25519::PublicKey, - /// Target gateway's LP address pub target_gateway_lp_address: SocketAddr, diff --git a/nym-node/src/node/lp/mod.rs b/nym-node/src/node/lp/mod.rs index 1e6d8d9c35..6352e49458 100644 --- a/nym-node/src/node/lp/mod.rs +++ b/nym-node/src/node/lp/mod.rs @@ -72,7 +72,6 @@ use crate::error::NymNodeError; use crate::node::lp::cleanup::CleanupTask; use crate::node::lp::control::ingress::listener::LpControlListener; use crate::node::lp::data::listener::LpDataListener; -use dashmap::DashMap; use nym_gateway::node::wireguard::PeerRegistrator; use nym_lp::peer::LpLocalPeer; use nym_mixnet_client::forwarder::MixForwardingSender; @@ -82,7 +81,10 @@ use std::sync::Arc; use tokio::sync::Semaphore; use tracing::error; -use crate::node::lp::state::ActiveLpSessions; +use crate::node::lp::directory::LpNodes; +use crate::node::lp::forwarding::controller::NestedConnectionsController; +use crate::node::lp::forwarding::manager::NestedConnectionsManager; +use crate::node::lp::state::{ActiveLpSessions, SharedLpNodeControlState}; pub use nym_mixnet_client::forwarder::{MixForwardingReceiver, mix_forwarding_channels}; pub use state::{SharedLpClientControlState, SharedLpDataState, SharedLpState}; @@ -99,6 +101,7 @@ pub struct LpSetup { control_listener: LpControlListener, data_listener: LpDataListener, cleanup_task: CleanupTask, + nested_connections_controller: NestedConnectionsController, /// Shutdown coordination shutdown: ShutdownTracker, @@ -110,54 +113,69 @@ impl LpSetup { lp_config: LpConfig, metrics: NymNodeMetrics, peer_registrator: Option, + network_nodes: LpNodes, mix_packet_sender: MixForwardingSender, shutdown: ShutdownTracker, ) -> Result { // TODO: this will require loading old states from disk in the future let session_states = ActiveLpSessions::new(); + let nested_connections_controller = NestedConnectionsController::new( + network_nodes.clone(), + shutdown.clone_shutdown_token(), + ); + let shared_lp_state = SharedLpState { metrics, lp_config, session_states: session_states.clone(), }; - todo!() + let client_control_state = SharedLpClientControlState { + local_lp_peer: local_lp_peer.clone(), + peer_registrator, + nested_connections_manager: NestedConnectionsManager::new( + nested_connections_controller.request_sender(), + ), + forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)), + shared: shared_lp_state.clone(), + }; - // let control_state = SharedLpControlState { - // local_lp_peer, - // peer_registrator, - // forward_semaphore: Arc::new(Semaphore::new(lp_config.debug.max_concurrent_forwards)), - // shared: shared_lp_state.clone(), - // }; - // - // let data_state = SharedLpDataState { - // outbound_mix_sender: mix_packet_sender, - // shared: shared_lp_state, - // }; - // - // let control_listener = LpControlListener::new( - // lp_config.control_bind_address, - // control_state, - // shutdown.clone(), - // ); - // let data_listener = LpDataListener::new( - // lp_config.data_bind_address, - // data_state, - // shutdown.clone_shutdown_token(), - // ); - // let cleanup_task = CleanupTask::new( - // session_states, - // lp_config.debug, - // shutdown.clone_shutdown_token(), - // ); - // - // Ok(LpSetup { - // control_listener, - // data_listener, - // cleanup_task, - // shutdown, - // }) + let nodes_control_state = SharedLpNodeControlState { + local_lp_peer, + nodes: network_nodes, + shared: shared_lp_state.clone(), + }; + + let data_state = SharedLpDataState { + outbound_mix_sender: mix_packet_sender, + shared: shared_lp_state, + }; + + let control_listener = LpControlListener::new( + lp_config.control_bind_address, + client_control_state, + nodes_control_state, + shutdown.clone(), + ); + let data_listener = LpDataListener::new( + lp_config.data_bind_address, + data_state, + shutdown.clone_shutdown_token(), + ); + let cleanup_task = CleanupTask::new( + session_states, + lp_config.debug, + shutdown.clone_shutdown_token(), + ); + + Ok(LpSetup { + control_listener, + data_listener, + cleanup_task, + nested_connections_controller, + shutdown, + }) } pub fn start_tasks(mut self) { @@ -192,5 +210,11 @@ impl LpSetup { async move { self.cleanup_task.run().await }, "LP::CleanupTask", ); + + // nested connections controller + self.shutdown.try_spawn_named( + async move { self.nested_connections_controller.run().await }, + "LP::NestedConnectionsController", + ); } } diff --git a/nym-node/src/node/lp/state.rs b/nym-node/src/node/lp/state.rs index 416e1f45a7..72f21dc8f5 100644 --- a/nym-node/src/node/lp/state.rs +++ b/nym-node/src/node/lp/state.rs @@ -5,6 +5,7 @@ use crate::config::LpConfig; use crate::node::lp::cleanup::TimestampedState; use crate::node::lp::directory::LpNodes; use crate::node::lp::error::LpHandlerError; +use crate::node::lp::forwarding::manager::NestedConnectionsManager; use dashmap::DashMap; use dashmap::mapref::one::RefMut; use nym_gateway::node::wireguard::PeerRegistrator; @@ -25,13 +26,16 @@ pub struct SharedLpClientControlState { /// Handle registering new wireguard peers pub peer_registrator: Option, + /// Controller for obtaining handles to forwarding channels + pub nested_connections_manager: NestedConnectionsManager, + /// Semaphore limiting concurrent forward connections /// /// Prevents file descriptor exhaustion when forwarding LP packets during /// telescope setup. When at capacity, forward requests return an error /// so clients can choose a different gateway. // this is temporary until there is persistent KKT/PSQ session between nodes - #[deprecated] + // #[deprecated] pub forward_semaphore: Arc, /// Common shared data diff --git a/nym-node/src/node/mod.rs b/nym-node/src/node/mod.rs index 6fdd5095ac..d755e7b12b 100644 --- a/nym-node/src/node/mod.rs +++ b/nym-node/src/node/mod.rs @@ -84,6 +84,7 @@ use tokio::sync::mpsc; use tracing::{debug, info, trace}; use zeroize::Zeroizing; +use crate::node::lp::directory::LpNodes; pub use nym_gateway::node::ActiveClientsStore; pub use nym_gateway::node::GatewayStorage; @@ -491,6 +492,7 @@ impl NymNode { &self, peer_registrator: Option, mix_packet_sender: MixForwardingSender, + network_nodes: LpNodes, ) -> Result { let lp_peer = LpLocalPeer::new(Ciphersuite::default(), self.x25519_lp_keys.clone()) .with_kem_keys(self.psq_kem_keys.clone()); @@ -500,6 +502,7 @@ impl NymNode { self.config.lp, self.metrics.clone(), peer_registrator, + network_nodes, mix_packet_sender, self.shutdown_manager.shutdown_tracker().clone(), ) @@ -683,6 +686,7 @@ impl NymNode { async fn start_gateway_tasks( &mut self, cached_network: CachedNetwork, + lp_nodes: LpNodes, metrics_sender: MetricEventsSender, active_clients_store: ActiveClientsStore, mix_packet_sender: MixForwardingSender, @@ -766,7 +770,7 @@ impl NymNode { self.config.lp.control_bind_address, self.config.lp.data_bind_address, ); let lp_tasks = self - .build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender) + .build_lp_tasks(wg_peer_registrator.clone(), mix_packet_sender, lp_nodes) .await?; lp_tasks.start_tasks(); } else { @@ -1355,6 +1359,7 @@ impl NymNode { let network_refresher = self.build_network_refresher().await?; let active_clients_store = ActiveClientsStore::new(); + let lp_nodes = network_refresher.lp_nodes(); let bloomfilters_manager = self.setup_replay_detection().await?; @@ -1381,6 +1386,7 @@ impl NymNode { self.start_gateway_tasks( network_refresher.cached_network(), + lp_nodes, metrics_sender, active_clients_store, mix_packet_sender, diff --git a/nym-node/src/node/shared_network.rs b/nym-node/src/node/shared_network.rs index 861d1a3e2b..b4911831e5 100644 --- a/nym-node/src/node/shared_network.rs +++ b/nym-node/src/node/shared_network.rs @@ -3,6 +3,7 @@ use crate::error::NymNodeError; use crate::node::key_rotation::active_keys::ActiveSphinxKeys; +use crate::node::lp::directory::LpNodes; use crate::node::routing_filter::network_filter::NetworkRoutingFilter; use async_trait::async_trait; use nym_crypto::asymmetric::ed25519; @@ -212,6 +213,7 @@ pub struct NetworkRefresher { network: CachedNetwork, routing_filter: NetworkRoutingFilter, noise_view: NoiseNetworkView, + lp_nodes: LpNodes, } impl NetworkRefresher { @@ -240,6 +242,7 @@ impl NetworkRefresher { network: CachedNetwork::new_empty(), routing_filter: NetworkRoutingFilter::new_empty(testnet), noise_view: NoiseNetworkView::new_empty(), + lp_nodes: Default::default(), }; this.obtain_initial_network().await?; @@ -335,6 +338,8 @@ impl NetworkRefresher { .collect::>(); self.noise_view.swap_view(noise_nodes); + warn!("unimplemented LP nodes update"); + let mut network_guard = self.network.inner.write().await; network_guard.topology_metadata = metadata.to_topology_metadata(); network_guard.network_nodes = nodes; @@ -373,6 +378,10 @@ impl NetworkRefresher { self.noise_view.clone() } + pub(crate) fn lp_nodes(&self) -> LpNodes { + self.lp_nodes.clone() + } + pub(crate) async fn run(&mut self) { let mut full_refresh_interval = interval(self.full_refresh_interval); full_refresh_interval.reset();