hopr_transport_p2p/
swarm.rs

1#[cfg(feature = "runtime-tokio")]
2use std::num::NonZeroU8;
3use std::sync::Arc;
4
5use dashmap::DashSet;
6use futures::{FutureExt, Stream, StreamExt};
7use hopr_api::{Multiaddr, OffchainKeypair, network::BoxedProcessFn};
8use hopr_utils::network_types::prelude::is_public_address;
9use libp2p::{
10    autonat,
11    swarm::{NetworkInfo, SwarmEvent},
12};
13use tracing::{debug, error, info, trace, warn};
14
15use crate::{
16    HoprNetwork, HoprNetworkBehavior, HoprNetworkBehaviorEvent, PeerDiscovery,
17    errors::Result,
18    utils::{replace_transport_with_unspecified, resolve_dns_if_any},
19};
20
21#[cfg(all(feature = "telemetry", not(test)))]
22lazy_static::lazy_static! {
23    static ref METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT:  hopr_api::types::telemetry::SimpleGauge =  hopr_api::types::telemetry::SimpleGauge::new(
24        "hopr_transport_p2p_active_connection_count",
25        "Number of currently active p2p connections as observed from libp2p events"
26    ).unwrap();
27    static ref METRIC_TRANSPORT_NAT_STATUS: hopr_api::types::telemetry::SimpleGauge = hopr_api::types::telemetry::SimpleGauge::new(
28        "hopr_transport_p2p_nat_status",
29        "Current NAT status as reported by libp2p autonat. 0=Unknown, 1=Public, 2=Private"
30    ).unwrap();
31    static ref METRIC_NETWORK_HEALTH: hopr_api::types::telemetry::SimpleGauge =
32         hopr_api::types::telemetry::SimpleGauge::new("hopr_network_health", "Connectivity health indicator").unwrap();
33}
34
35pub struct InactiveNetwork {
36    swarm: libp2p::Swarm<HoprNetworkBehavior>,
37}
38
39#[cfg(any(feature = "testing", target_os = "android", target_os = "ios"))]
40fn swarm_dns_config() -> (libp2p::dns::ResolverConfig, libp2p::dns::ResolverOpts) {
41    (
42        libp2p::dns::ResolverConfig::cloudflare(),
43        libp2p::dns::ResolverOpts::default(),
44    )
45}
46
47/// Build objects comprising an inactive p2p network.
48///
49/// Returns a built [libp2p::Swarm] object implementing the HoprNetworkBehavior functionality.
50impl InactiveNetwork {
51    #[cfg(feature = "runtime-tokio")]
52    pub async fn build(
53        me: libp2p::identity::Keypair,
54        external_discovery_events: futures::stream::BoxStream<'static, PeerDiscovery>,
55    ) -> Result<Self> {
56        let me_public: libp2p::identity::PublicKey = me.public();
57
58        let swarm = libp2p::SwarmBuilder::with_existing_identity(me)
59            .with_tokio()
60            .with_tcp(
61                libp2p::tcp::Config::default().nodelay(true),
62                libp2p::noise::Config::new,
63                // use default yamux configuration to enable auto-tuning
64                // see https://github.com/libp2p/rust-libp2p/pull/4970
65                libp2p::yamux::Config::default,
66            )
67            .map_err(|e| crate::errors::P2PError::Libp2p(e.to_string()))?;
68
69        #[cfg(feature = "transport-quic")]
70        let swarm = swarm.with_quic();
71
72        // Nix test sandboxes do not provide /etc/resolv.conf, so use an explicit resolver configuration.
73        #[cfg(any(feature = "testing", target_os = "android", target_os = "ios"))]
74        let swarm = {
75            let (dns_resolver_config, dns_resolver_opts) = swarm_dns_config();
76            swarm.with_dns_config(dns_resolver_config, dns_resolver_opts)
77        };
78
79        #[cfg(not(any(feature = "testing", target_os = "android", target_os = "ios")))]
80        let swarm = swarm
81            .with_dns()
82            .map_err(|e| crate::errors::P2PError::Libp2p(e.to_string()))?;
83
84        Ok(Self {
85            swarm: swarm
86                .with_behaviour(|_key| HoprNetworkBehavior::new(me_public, external_discovery_events))
87                .map_err(|e| crate::errors::P2PError::Libp2p(e.to_string()))?
88                .with_swarm_config(|cfg| {
89                    cfg.with_dial_concurrency_factor(
90                        NonZeroU8::new({
91                            let v = std::env::var("HOPR_INTERNAL_LIBP2P_MAX_CONCURRENTLY_DIALED_PEER_COUNT")
92                                .ok()
93                                .and_then(|v| v.trim().parse::<u8>().ok())
94                                .unwrap_or(crate::constants::HOPR_SWARM_CONCURRENTLY_DIALED_PEER_COUNT);
95                            v.max(1)
96                        })
97                        .expect("clamped to >= 1, will never fail"),
98                    )
99                    .with_max_negotiating_inbound_streams(
100                        std::env::var("HOPR_INTERNAL_LIBP2P_MAX_NEGOTIATING_INBOUND_STREAM_COUNT")
101                            .and_then(|v| v.parse::<usize>().map_err(|_e| std::env::VarError::NotPresent))
102                            .unwrap_or(crate::constants::HOPR_SWARM_CONCURRENTLY_NEGOTIATING_INBOUND_PEER_COUNT),
103                    )
104                    .with_idle_connection_timeout(
105                        std::env::var("HOPR_INTERNAL_LIBP2P_SWARM_IDLE_TIMEOUT")
106                            .and_then(|v| v.parse::<u64>().map_err(|_e| std::env::VarError::NotPresent))
107                            .map(std::time::Duration::from_secs)
108                            .unwrap_or(crate::constants::HOPR_SWARM_IDLE_CONNECTION_TIMEOUT),
109                    )
110                })
111                .build(),
112        })
113    }
114
115    #[cfg(not(feature = "runtime-tokio"))]
116    pub async fn build<T>(_me: libp2p::identity::Keypair, _external_discovery_events: T) -> Result<Self>
117    where
118        T: Stream<Item = PeerDiscovery> + Send + 'static,
119    {
120        Err(crate::errors::P2PError::Libp2p(
121            "InactiveNetwork::build requires the runtime-tokio feature".to_string(),
122        ))
123    }
124
125    pub fn with_listen_on(mut self, multiaddresses: Vec<Multiaddr>) -> Result<InactiveConfiguredNetwork> {
126        for multiaddress in multiaddresses.iter() {
127            match resolve_dns_if_any(multiaddress) {
128                Ok(ma) => {
129                    if let Err(e) = self.swarm.listen_on(ma.clone()) {
130                        warn!(%multiaddress, listen_on=%ma, error = %e, "Failed to listen_on, will try to use an unspecified address");
131
132                        match replace_transport_with_unspecified(&ma) {
133                            Ok(ma) => {
134                                if let Err(e) = self.swarm.listen_on(ma.clone()) {
135                                    warn!(multiaddress = %ma, error = %e, "Failed to listen_on using the unspecified multiaddress",);
136                                } else {
137                                    info!(
138                                        listen_on = ?ma,
139                                        multiaddress = ?multiaddress,
140                                        "Listening for p2p connections"
141                                    );
142                                    self.swarm.add_external_address(multiaddress.clone());
143                                }
144                            }
145                            Err(e) => {
146                                error!(multiaddress = %ma, error = %e, "Failed to transform the multiaddress")
147                            }
148                        }
149                    } else {
150                        info!(
151                            listen_on = ?ma,
152                            multiaddress = ?multiaddress,
153                            "Listening for p2p connections"
154                        );
155                        self.swarm.add_external_address(multiaddress.clone());
156                    }
157                }
158                Err(error) => error!(%multiaddress, %error, "Failed to transform the multiaddress"),
159            }
160        }
161
162        Ok(InactiveConfiguredNetwork { swarm: self.swarm })
163    }
164}
165
166pub struct InactiveConfiguredNetwork {
167    swarm: libp2p::Swarm<HoprNetworkBehavior>,
168}
169
170/// Factory for constructing the libp2p network and its background process.
171///
172/// Accepts a peer discovery stream and produces a [`HoprNetwork`] + background
173/// process function. The network supports event subscription via
174/// [`NetworkView::subscribe_network_events`](hopr_api::network::NetworkView::subscribe_network_events).
175pub struct HoprLibp2pNetworkBuilder {
176    bootstrap: std::pin::Pin<Box<dyn Stream<Item = PeerDiscovery> + Send>>,
177}
178
179impl HoprLibp2pNetworkBuilder {
180    pub fn new<T>(bootstrap: T) -> Self
181    where
182        T: Stream<Item = PeerDiscovery> + Send + 'static,
183    {
184        Self {
185            bootstrap: Box::pin(bootstrap),
186        }
187    }
188}
189
190impl std::fmt::Debug for HoprLibp2pNetworkBuilder {
191    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
192        f.debug_struct("HoprSwarmBuilder").finish_non_exhaustive()
193    }
194}
195
196impl HoprLibp2pNetworkBuilder {
197    /// Build the network and return it along with its background process.
198    ///
199    /// The returned [`HoprNetwork`] supports event subscription — subscribe
200    /// before starting the process to avoid missing events.
201    pub async fn build(
202        self,
203        identity: &OffchainKeypair,
204        my_multiaddresses: Vec<Multiaddr>,
205        protocol: &'static str,
206        allow_private_addresses: bool,
207    ) -> std::result::Result<(HoprNetwork, BoxedProcessFn), impl std::error::Error> {
208        #[cfg(all(feature = "telemetry", not(test)))]
209        {
210            METRIC_NETWORK_HEALTH.set(0.0);
211        }
212
213        let identity_p2p: libp2p::identity::Keypair = identity.into();
214        let me = identity_p2p.public().to_peer_id();
215        let swarm = InactiveNetwork::build(identity_p2p, self.bootstrap)
216            .await
217            .expect("swarm must be constructible");
218
219        let swarm = swarm
220            .with_listen_on(my_multiaddresses.clone())
221            .expect("swarm must be configurable");
222
223        let swarm = swarm.swarm;
224        let store = crate::peer_store::NetworkPeerStore::new(me, my_multiaddresses.into_iter().collect());
225        let tracker: Arc<DashSet<libp2p::PeerId>> = Default::default();
226        let liveness: crate::liveness::LivenessRegistry = Default::default();
227
228        let (notifier, event_rx) = async_broadcast::broadcast(1000);
229
230        let network = HoprNetwork {
231            tracker: tracker.clone(),
232            store: Arc::new(store.clone()),
233            control: swarm.behaviour().streams.new_control(),
234            protocol: libp2p::StreamProtocol::new(protocol),
235            event_rx: event_rx.deactivate(),
236            liveness: liveness.clone(),
237        };
238
239        #[cfg(all(feature = "telemetry", not(test)))]
240        let network_inner = network.clone();
241        let mut swarm = swarm;
242        let process = async move {
243            // Shared teardown for both ConnectionClosed and OutgoingConnectionError: remove
244            // from the tracker and liveness registry and broadcast the disconnection event.
245            let disconnect_peer = |peer_id: libp2p::PeerId| {
246                tracker.remove(&peer_id);
247                liveness.remove(&peer_id);
248                if let Err(error) = notifier.try_broadcast(hopr_api::network::NetworkEvent::PeerDisconnected(peer_id)) {
249                    error!(peer = %peer_id, %error, "failed to broadcast peer disconnected event");
250                }
251            };
252
253            while let Some(event) = swarm.next().await {
254                match event {
255                    SwarmEvent::Behaviour(HoprNetworkBehaviorEvent::Discovery(_)) => {}
256                    SwarmEvent::Behaviour(
257                        HoprNetworkBehaviorEvent::Autonat(event)
258                    ) => {
259                        match event {
260                            autonat::Event::StatusChanged { old, new } => {
261                                info!(?old, ?new, "AutoNAT status changed");
262                                #[cfg(all(feature = "telemetry", not(test)))]
263                                {
264                                    let value = match new {
265                                        autonat::NatStatus::Unknown => 0.0,
266                                        autonat::NatStatus::Public(_) => 1.0,
267                                        autonat::NatStatus::Private => 2.0,
268                                    };
269                                    METRIC_TRANSPORT_NAT_STATUS.set(value);
270                                }
271                            }
272                            autonat::Event::InboundProbe { .. } => {}
273                            autonat::Event::OutboundProbe { .. } => {}
274                        }
275                    }
276                    SwarmEvent::Behaviour(HoprNetworkBehaviorEvent::Identify(_event)) => {}
277                    SwarmEvent::ConnectionEstablished {
278                        peer_id,
279                        connection_id,
280                        num_established,
281                        established_in,
282                        endpoint,
283                        ..
284                        // concurrent_dial_errors,
285                    } => {
286                        debug!(%peer_id, %connection_id, num_established, established_in_ms = established_in.as_millis(), transport="libp2p", "connection established");
287
288                        if num_established == std::num::NonZero::<u32>::new(1).expect("must be a non-zero value") {
289                            match endpoint {
290                                libp2p::core::ConnectedPoint::Dialer { address, .. } => {
291                                    if allow_private_addresses || is_public_address(&address) {
292                                        if let Err(error) = store.add(peer_id, std::collections::HashSet::from([address])) {
293                                            error!(peer = %peer_id, %error, direction = "outgoing", "failed to add connected peer to the peer store");
294                                        }
295                                    } else {
296                                        debug!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Private/local peer address encountered")
297                                    }
298                                    tracker.insert(peer_id);
299                                    if let Err(error) = notifier.try_broadcast(hopr_api::network::NetworkEvent::PeerConnected(peer_id)) {
300                                        error!(peer = %peer_id, %error, "failed to broadcast peer connected event");
301                                    }
302                                },
303                                libp2p::core::ConnectedPoint::Listener { send_back_addr, .. } => {
304                                    if allow_private_addresses || is_public_address(&send_back_addr) {
305                                        if let Err(error) = store.add(peer_id, std::collections::HashSet::from([send_back_addr])) {
306                                            error!(peer = %peer_id, %error, direction = "incoming", "failed to add connected peer to the peer store");
307                                        }
308                                    } else {
309                                        debug!(transport="libp2p", peer = %peer_id, multiaddress = %send_back_addr, "Private/local peer address ignored")
310                                    }
311                                    tracker.insert(peer_id);
312                                    if let Err(error) = notifier.try_broadcast(hopr_api::network::NetworkEvent::PeerConnected(peer_id)) {
313                                        error!(peer = %peer_id, %error, "failed to broadcast peer connected event");
314                                    }
315                                }
316                            }
317                        } else {
318                            trace!(transport="libp2p", peer = %peer_id, num_established, "Additional connection established")
319                        }
320
321                        print_network_info(swarm.network_info(), "connection established");
322
323                        #[cfg(all(feature = "telemetry", not(test)))]
324                        {
325                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
326                            METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT.increment(1.0);
327                        }
328                    }
329                    SwarmEvent::ConnectionClosed {
330                        peer_id,
331                        connection_id,
332                        cause,
333                        num_established,
334                        ..
335                        // endpoint,
336                    } => {
337                        debug!(%peer_id, %connection_id, num_established, transport="libp2p", "connection closed: {cause:?}");
338
339                        if num_established == 0 {
340                            disconnect_peer(peer_id);
341                        }
342
343                        print_network_info(swarm.network_info(), "connection closed");
344
345                        #[cfg(all(feature = "telemetry", not(test)))]
346                        {
347                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
348                            METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT.decrement(1.0);
349                        }
350                    }
351                    SwarmEvent::IncomingConnection {
352                        connection_id,
353                        local_addr,
354                        send_back_addr,
355                    } => {
356                        trace!(%local_addr, %send_back_addr, %connection_id, transport="libp2p", "incoming connection");
357                    }
358                    SwarmEvent::IncomingConnectionError {
359                        local_addr,
360                        connection_id,
361                        error,
362                        send_back_addr,
363                        peer_id
364                    } => {
365                        debug!(?peer_id, %local_addr, %send_back_addr, %connection_id, transport="libp2p", %error, "incoming connection error");
366                    }
367                    SwarmEvent::OutgoingConnectionError {
368                        connection_id,
369                        error,
370                        peer_id
371                    } => {
372                        debug!(peer = ?peer_id, %connection_id, transport="libp2p", %error, "outgoing connection error");
373
374                        if let Some(peer_id) = peer_id
375                            && !swarm.is_connected(&peer_id) {
376                                if let Err(error) = store.remove(&peer_id) {
377                                    error!(peer = %peer_id, %error, "failed to remove undialable peer from the peer store");
378                                }
379                                disconnect_peer(peer_id);
380                            }
381
382                        #[cfg(all(feature = "telemetry", not(test)))]
383                        {
384                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
385                        }
386                    }
387                    SwarmEvent::NewListenAddr {
388                        listener_id,
389                        address,
390                    } => {
391                        debug!(%listener_id, %address, transport="libp2p", "new listen address")
392                    }
393                    SwarmEvent::ExpiredListenAddr {
394                        listener_id,
395                        address,
396                    } => {
397                        debug!(%listener_id, %address, transport="libp2p", "expired listen address")
398                    }
399                    SwarmEvent::ListenerClosed {
400                        listener_id,
401                        addresses,
402                        reason,
403                    } => {
404                        debug!(%listener_id, ?addresses, ?reason, transport="libp2p", "listener closed", )
405                    }
406                    SwarmEvent::ListenerError {
407                        listener_id,
408                        error,
409                    } => {
410                        debug!(%listener_id, transport="libp2p", %error, "listener error")
411                    }
412                    SwarmEvent::Dialing {
413                        peer_id,
414                        connection_id,
415                    } => {
416                        debug!(peer = ?peer_id, %connection_id, transport="libp2p", "dialing")
417                    }
418                    SwarmEvent::NewExternalAddrCandidate {address} => {
419                        debug!(%address, "Detected new external address candidate")
420                    }
421                    SwarmEvent::ExternalAddrConfirmed { address } => {
422                        info!(%address, "Detected external address")
423                    }
424                    SwarmEvent::ExternalAddrExpired {
425                        ..  // address: Multiaddr
426                    } => {}
427                    SwarmEvent::NewExternalAddrOfPeer {
428                        peer_id, address
429                    } => {
430                        // Only store public/routable addresses
431                        if allow_private_addresses || is_public_address(&address) {
432                            swarm.add_peer_address(peer_id, address.clone());
433                            trace!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Public peer address stored in swarm")
434                        } else {
435                            trace!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Private/local peer address ignored")
436                        }
437                    },
438                    _ => trace!(transport="libp2p", "Unsupported enum option detected")
439                }
440            }
441        }.boxed();
442
443        Ok::<_, std::io::Error>((network, Box::new(move || process)))
444    }
445}
446
447fn print_network_info(network_info: NetworkInfo, event: &str) {
448    let num_peers = network_info.num_peers();
449    let connection_counters = network_info.connection_counters();
450    let num_incoming = connection_counters.num_established_incoming();
451    let num_outgoing = connection_counters.num_established_outgoing();
452    info!(
453        num_peers,
454        num_incoming, num_outgoing, "swarm network status after {event}"
455    );
456}
457
458#[cfg(test)]
459mod tests {
460    use std::net::{IpAddr, Ipv4Addr};
461
462    #[test]
463    #[cfg(any(feature = "testing", target_os = "android", target_os = "ios"))]
464    fn uses_cloudflare_dns_resolver_config() {
465        let (resolver_config, _) = super::swarm_dns_config();
466        assert!(
467            resolver_config
468                .name_servers()
469                .iter()
470                .any(|server| server.socket_addr.ip() == IpAddr::V4(Ipv4Addr::new(1, 1, 1, 1)))
471        );
472    }
473}
hopr_transport_p2p/swarm.rs

hopr_transport_p2p/
swarm.rs
