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

additional wiring of nested connections control

Browse Source
This commit is contained in:
Jędrzej Stuczyński
2026-03-05 16:07:50 +00:00
parent 6a42a8dd49
commit 24891adddf
15 changed files with 262 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nym-node/src/node/lp/cleanup.rs
-4
View File
@@ -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};
nym-node/src/node/lp/control/egress/connection.rs
+3 -3
View File
@@ -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<S> {
impl<S> NestedNodeConnectionHandler<S>
where
// S: LpTransport + Unpin,
S: LpTransportChannel + Unpin,
{
/// Attempt to extract outer receiver index from the received message
/// (that is meant to be an `LpPacket`)
nym-node/src/node/lp/control/ingress/client_handler.rs
+7 -10
View File
@@ -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<S = TcpStream> {
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<S = TcpStream> {
/// 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<NestedClientConnection>,
}
impl<S> LpClientConnectionHandler<S>
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,
}
}
nym-node/src/node/lp/control/ingress/listener.rs
+9 -8
View File
@@ -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 {
nym-node/src/node/lp/control/ingress/node_handler.rs
+1 -1
View File
@@ -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(())
nym-node/src/node/lp/control/tests.rs
+2 -2
View File
@@ -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();
nym-node/src/node/lp/directory.rs
+7 -4
View File
@@ -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<LpNodeDetails> {
self.nodes.load().get(&node_ip).cloned()
}
pub(crate) fn get_node_id(&self, node_ip: IpAddr) -> Option<NodeId> {
self.nodes
.load()
.get(&node_ip)
.map(|details| details.node_id)
}
}
#[derive(Clone)]
nym-node/src/node/lp/error.rs
+10
View File
@@ -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<String>) -> Self {
LpHandlerError::Internal(msg.into())
}
pub fn other(msg: impl Into<String>) -> Self {
LpHandlerError::Other(msg.into())
}
nym-node/src/node/lp/forwarding/controller.rs
+73 -6
View File
@@ -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<Notify>),
}
/// 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<ed25519::PublicKey, NestedNodeControlSender>,
nodes_handles: HashMap<NodeId, NodeHandle>,
/// 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");
}
}
nym-node/src/node/lp/forwarding/manager.rs
+55 -9
View File
@@ -1,27 +1,73 @@
// Copyright 2026 - Nym Technologies SA <contact@nymtech.net>
// 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<ConnectionHandlerResponse, LpHandlerError> {
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<NestedClientConnection, LpHandlerError> {
// 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",
)),
}
}
}
nym-node/src/node/lp/forwarding/mod.rs
+13 -6
View File
@@ -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<GetConnectionHandler>;
pub(crate) type NodeConnectionControllerSender = Sender<GetConnectionHandler>;
pub(crate) type NodeConnectionControllerReceiver = Receiver<NestedConnectionControllerRequest>;
pub(crate) type NodeConnectionControllerSender = Sender<NestedConnectionControllerRequest>;
pub(crate) enum ConnectionControllerResponse<T> {
/// The response is immediately available
@@ -26,6 +25,16 @@ pub(crate) enum ConnectionControllerResponse<T> {
Pending(Arc<Notify>),
}
impl<T> ConnectionControllerResponse<T> {
pub fn new_pending() -> (Self, Arc<Notify>) {
let notify = Arc::new(Notify::new());
(
ConnectionControllerResponse::Pending(notify.clone()),
notify,
)
}
}
pub(crate) type ControllerResponse<T> = Result<ConnectionControllerResponse<T>, LpHandlerError>;
pub(crate) type ConnectionHandlerResponse = ControllerResponse<NestedClientConnection>;
@@ -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,
nym-node/src/node/lp/mod.rs
+61 -37
View File
@@ -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<PeerRegistrator>,
network_nodes: LpNodes,
mix_packet_sender: MixForwardingSender,
shutdown: ShutdownTracker,
) -> Result<Self, NymNodeError> {
// 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",
);
}
}
nym-node/src/node/lp/state.rs
+5 -1
View File
@@ -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<PeerRegistrator>,
/// 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<Semaphore>,
/// Common shared data
nym-node/src/node/mod.rs
+7 -1
View File
@@ -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<PeerRegistrator>,
mix_packet_sender: MixForwardingSender,
network_nodes: LpNodes,
) -> Result<LpSetup, NymNodeError> {
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,
nym-node/src/node/shared_network.rs
+9
View File
@@ -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::<HashMap<_, _>>();
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();