Skip to content

Commit

Permalink
Merge pull request #2377 from RolandSherwin/clean_external_addr_1
Browse files Browse the repository at this point in the history 
feat: remove external address on too many connection error
  • Loading branch information
@RolandSherwin
RolandSherwin authored Nov 28, 2024
2 parents 52db692 + 55f413f commit 87ee906
Show file tree
Hide file tree
Showing 5 changed files with 377 additions and 154 deletions.
35 changes: 27 additions & 8 deletions sn_networking/src/driver.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -690,11 +690,23 @@ impl NetworkBuilder {

let bootstrap = ContinuousNetworkDiscover::new();
let replication_fetcher = ReplicationFetcher::new(peer_id, network_event_sender.clone());
let mut relay_manager = RelayManager::new(peer_id);
if !is_client {
relay_manager.enable_hole_punching(self.is_behind_home_network);
}
let external_address_manager = ExternalAddressManager::new(peer_id);

// Enable relay manager for nodes behind home network
let relay_manager = if !is_client && self.is_behind_home_network {
let relay_manager = RelayManager::new(peer_id);
Some(relay_manager)
} else {
info!("Relay manager is disabled for this node.");
None
};
// Enable external address manager for public nodes and not behind nat
let external_address_manager = if !is_client && !self.local && !self.is_behind_home_network
{
Some(ExternalAddressManager::new(peer_id))
} else {
info!("External address manager is disabled for this node.");
None
};

let swarm_driver = SwarmDriver {
swarm,
Expand All @@ -707,6 +719,7 @@ impl NetworkBuilder {
peers_in_rt: 0,
bootstrap,
relay_manager,
connected_relay_clients: Default::default(),
external_address_manager,
replication_fetcher,
#[cfg(feature = "open-metrics")]
Expand Down Expand Up @@ -799,8 +812,10 @@ pub struct SwarmDriver {
pub(crate) close_group: Vec<PeerId>,
pub(crate) peers_in_rt: usize,
pub(crate) bootstrap: ContinuousNetworkDiscover,
pub(crate) external_address_manager: ExternalAddressManager,
pub(crate) relay_manager: RelayManager,
pub(crate) external_address_manager: Option<ExternalAddressManager>,
pub(crate) relay_manager: Option<RelayManager>,
/// The peers that are using our relay service.
pub(crate) connected_relay_clients: HashSet<PeerId>,
/// The peers that are closer to our PeerId. Includes self.
pub(crate) replication_fetcher: ReplicationFetcher,
#[cfg(feature = "open-metrics")]
Expand Down Expand Up @@ -933,7 +948,11 @@ impl SwarmDriver {
}
}
}
_ = relay_manager_reservation_interval.tick() => self.relay_manager.try_connecting_to_relay(&mut self.swarm, &self.bad_nodes),
_ = relay_manager_reservation_interval.tick() => {
if let Some(relay_manager) = &mut self.relay_manager {
relay_manager.try_connecting_to_relay(&mut self.swarm, &self.bad_nodes)
}
},
}
}
}
Expand Down
92 changes: 61 additions & 31 deletions sn_networking/src/event/swarm.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ use libp2p::mdns;
#[cfg(feature = "open-metrics")]
use libp2p::metrics::Recorder;
use libp2p::{
core::ConnectedPoint,
kad::K_VALUE,
multiaddr::Protocol,
swarm::{
Expand Down Expand Up @@ -62,8 +63,10 @@ impl SwarmDriver {
relay_peer_id, ..
} = *event
{
self.relay_manager
.on_successful_reservation_by_client(&relay_peer_id, &mut self.swarm);
if let Some(relay_manager) = self.relay_manager.as_mut() {
relay_manager
.on_successful_reservation_by_client(&relay_peer_id, &mut self.swarm);
}
}
}
#[cfg(feature = "upnp")]
Expand Down Expand Up @@ -97,11 +100,10 @@ impl SwarmDriver {
src_peer_id,
renewed: _,
} => {
self.relay_manager
.on_successful_reservation_by_server(src_peer_id);
self.connected_relay_clients.insert(src_peer_id);
}
libp2p::relay::Event::ReservationTimedOut { src_peer_id } => {
self.relay_manager.on_reservation_timeout(src_peer_id);
self.connected_relay_clients.remove(&src_peer_id);
}
_ => {}
}
Expand Down Expand Up @@ -173,13 +175,15 @@ impl SwarmDriver {
.any(|(_ilog2, peers)| peers.contains(&peer_id));

// Do not use an `already relayed` peer as `potential relay candidate`.
if !has_relayed && !is_bootstrap_peer && !self.is_client {
debug!("Adding candidate relay server {peer_id:?}, it's not a bootstrap node");
self.relay_manager.add_potential_candidates(
&peer_id,
&addrs,
&info.protocols,
);
if !has_relayed && !is_bootstrap_peer {
if let Some(relay_manager) = self.relay_manager.as_mut() {
debug!("Adding candidate relay server {peer_id:?}, it's not a bootstrap node");
relay_manager.add_potential_candidates(
&peer_id,
&addrs,
&info.protocols,
);
}
}

// When received an identify from un-dialed peer, try to dial it
Expand Down Expand Up @@ -300,13 +304,14 @@ impl SwarmDriver {
}
}
}

SwarmEvent::NewListenAddr {
mut address,
listener_id,
} => {
event_string = "new listen addr";

info!("Local node is listening {listener_id:?} on {address:?}");

let local_peer_id = *self.swarm.local_peer_id();
// Make sure the address ends with `/p2p/<local peer ID>`. In case of relay, `/p2p` is already there.
if address.iter().last() != Some(Protocol::P2p(local_peer_id)) {
Expand All @@ -320,16 +325,17 @@ impl SwarmDriver {
// all addresses are effectively external here...
// this is needed for Kad Mode::Server
self.swarm.add_external_address(address.clone());
} else if let Some(external_add_manager) =
self.external_address_manager.as_mut()
{
external_add_manager.on_new_listen_addr(address.clone(), &mut self.swarm);
} else {
self.external_address_manager
.add_listen_addr_as_external_address(address.clone(), &mut self.swarm);
// just for future reference.
warn!("External address manager is not enabled for a public node. This should not happen.");
}
}

self.send_event(NetworkEvent::NewListenAddr(address.clone()));

info!("Local node is listening {listener_id:?} on {address:?}");
println!("Local node is listening on {address:?}"); // TODO: make it print only once
}
SwarmEvent::ListenerClosed {
listener_id,
Expand All @@ -338,8 +344,9 @@ impl SwarmDriver {
} => {
event_string = "listener closed";
info!("Listener {listener_id:?} with add {addresses:?} has been closed for {reason:?}");
self.relay_manager
.on_listener_closed(&listener_id, &mut self.swarm);
if let Some(relay_manager) = self.relay_manager.as_mut() {
relay_manager.on_listener_closed(&listener_id, &mut self.swarm);
}
}
SwarmEvent::IncomingConnection {
connection_id,
Expand All @@ -359,6 +366,12 @@ impl SwarmDriver {
} => {
event_string = "ConnectionEstablished";
debug!(%peer_id, num_established, ?concurrent_dial_errors, "ConnectionEstablished ({connection_id:?}) in {established_in:?}: {}", endpoint_str(&endpoint));
if let Some(external_addr_manager) = self.external_address_manager.as_mut() {
if let ConnectedPoint::Listener { local_addr, .. } = &endpoint {
external_addr_manager
.on_established_incoming_connection(local_addr.clone());
}
}

let _ = self.live_connected_peers.insert(
connection_id,
Expand Down Expand Up @@ -529,6 +542,10 @@ impl SwarmDriver {
} else {
debug!("IncomingConnectionError from local_addr:?{local_addr:?}, send_back_addr {send_back_addr:?} on {connection_id:?} with error {error:?}");
}
if let Some(external_addr_manager) = self.external_address_manager.as_mut() {
external_addr_manager
.on_incoming_connection_error(local_addr.clone(), &mut self.swarm);
}
let _ = self.live_connected_peers.remove(&connection_id);
self.record_connection_metrics();
}
Expand All @@ -542,16 +559,8 @@ impl SwarmDriver {
SwarmEvent::NewExternalAddrCandidate { address } => {
event_string = "NewExternalAddrCandidate";

if !self.is_client
// If we are behind a home network, then our IP is returned here. We should be only having
// relay server as our external address
// todo: can our relay address be reported here? If so, maybe we should add them.
&& !self.is_behind_home_network
// When running a local network, we just need the local listen address to work.
&& !self.local
{
self.external_address_manager
.add_external_address_candidate(address, &mut self.swarm);
if let Some(external_addr_manager) = self.external_address_manager.as_mut() {
external_addr_manager.add_external_address_candidate(address, &mut self.swarm);
}
}
SwarmEvent::ExternalAddrConfirmed { address } => {
Expand All @@ -562,6 +571,20 @@ impl SwarmDriver {
event_string = "ExternalAddrExpired";
info!(%address, "external address: expired");
}
SwarmEvent::ExpiredListenAddr {
listener_id,
address,
} => {
event_string = "ExpiredListenAddr";
info!("Listen address has expired. {listener_id:?} on {address:?}");
if let Some(external_addr_manager) = self.external_address_manager.as_mut() {
external_addr_manager.on_expired_listen_addr(address, &self.swarm);
}
}
SwarmEvent::ListenerError { listener_id, error } => {
event_string = "ListenerError";
warn!("ListenerError {listener_id:?} with non-fatal error {error:?}");
}
other => {
event_string = "Other";

Expand Down Expand Up @@ -636,7 +659,14 @@ impl SwarmDriver {
}

// skip if the peer is a relay server that we're connected to
if self.relay_manager.keep_alive_peer(peer_id) {
if let Some(relay_manager) = self.relay_manager.as_ref() {
if relay_manager.keep_alive_peer(peer_id) {
return true; // retain peer
}
}

// skip if the peer is a node that is being relayed through us
if self.connected_relay_clients.contains(peer_id) {
return true; // retain peer
}

Expand Down
Loading

0 comments on commit 87ee906

Please sign in to comment.