Skip to main content

hopr_transport_p2p/
swarm.rs

1use std::{num::NonZeroU8, sync::Arc};
2
3use dashmap::DashSet;
4use futures::{FutureExt, Stream, StreamExt, stream::BoxStream};
5use hopr_api::{Multiaddr, OffchainKeypair, network::NetworkBuilder};
6use hopr_network_types::prelude::is_public_address;
7use libp2p::{
8    autonat,
9    identity::PublicKey,
10    swarm::{NetworkInfo, SwarmEvent},
11};
12use tracing::{debug, error, info, trace, warn};
13
14#[cfg(feature = "runtime-tokio")]
15use crate::PeerDiscovery;
16use crate::{
17    HoprNetwork, HoprNetworkBehavior, HoprNetworkBehaviorEvent, constants,
18    errors::Result,
19    utils::{replace_transport_with_unspecified, resolve_dns_if_any},
20};
21
22#[cfg(all(feature = "prometheus", not(test)))]
23lazy_static::lazy_static! {
24    static ref METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT:  hopr_metrics::SimpleGauge =  hopr_metrics::SimpleGauge::new(
25        "hopr_transport_p2p_active_connection_count",
26        "Number of currently active p2p connections as observed from libp2p events"
27    ).unwrap();
28    static ref METRIC_TRANSPORT_NAT_STATUS: hopr_metrics::SimpleGauge = hopr_metrics::SimpleGauge::new(
29        "hopr_transport_p2p_nat_status",
30        "Current NAT status as reported by libp2p autonat. 0=Unknown, 1=Public, 2=Private"
31    ).unwrap();
32    static ref METRIC_NETWORK_HEALTH: hopr_metrics::SimpleGauge =
33         hopr_metrics::SimpleGauge::new("hopr_network_health", "Connectivity health indicator").unwrap();
34}
35
36pub struct InactiveNetwork {
37    swarm: libp2p::Swarm<HoprNetworkBehavior>,
38}
39
40#[cfg(any(target_os = "android", target_os = "ios"))]
41fn swarm_dns_config() -> (libp2p::dns::ResolverConfig, libp2p::dns::ResolverOpts) {
42    (
43        libp2p::dns::ResolverConfig::cloudflare(),
44        libp2p::dns::ResolverOpts::default(),
45    )
46}
47
48/// Build objects comprising an inactive p2p network.
49///
50/// Returns a built [libp2p::Swarm] object implementing the HoprNetworkBehavior functionality.
51impl InactiveNetwork {
52    #[cfg(feature = "runtime-tokio")]
53    pub async fn build(
54        me: libp2p::identity::Keypair,
55        external_discovery_events: BoxStream<'static, PeerDiscovery>,
56    ) -> Result<Self> {
57        let me_public: PublicKey = me.public();
58
59        let swarm = libp2p::SwarmBuilder::with_existing_identity(me)
60            .with_tokio()
61            .with_tcp(
62                libp2p::tcp::Config::default().nodelay(true),
63                libp2p::noise::Config::new,
64                // use default yamux configuration to enable auto-tuning
65                // see https://github.com/libp2p/rust-libp2p/pull/4970
66                libp2p::yamux::Config::default,
67            )
68            .map_err(|e| crate::errors::P2PError::Libp2p(e.to_string()))?;
69
70        #[cfg(feature = "transport-quic")]
71        let swarm = swarm.with_quic();
72
73        #[cfg(any(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(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(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(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(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/// Builder of the network view and an actual background process running the libp2p core
171/// event processing loop.
172///
173/// This object is primarily constructed to allow delayed starting of the background process,
174/// as well as setup all the interconnections with the underlying network views to allow complex
175/// functionality and signalling.
176pub struct HoprLibp2pNetworkBuilder {
177    subscribtions: (
178        async_broadcast::Sender<hopr_api::network::NetworkEvent>,
179        async_broadcast::InactiveReceiver<hopr_api::network::NetworkEvent>,
180    ),
181    bootstrap: std::pin::Pin<Box<dyn Stream<Item = PeerDiscovery> + Send + Sync>>,
182}
183
184impl HoprLibp2pNetworkBuilder {
185    pub fn new<T>(bootstrap: T) -> Self
186    where
187        T: Stream<Item = PeerDiscovery> + Send + Sync + 'static,
188    {
189        let (tx, rx) = async_broadcast::broadcast(1000);
190        Self {
191            subscribtions: (tx, rx.deactivate()),
192            bootstrap: Box::pin(bootstrap),
193        }
194    }
195
196    pub fn subscribe_network_events(
197        &self,
198    ) -> impl futures::Stream<Item = hopr_api::network::NetworkEvent> + Send + Sync + 'static {
199        let rx = self.subscribtions.1.clone();
200        rx.activate()
201    }
202}
203
204impl std::fmt::Debug for HoprLibp2pNetworkBuilder {
205    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
206        f.debug_struct("HoprSwarmBuilder").finish_non_exhaustive()
207    }
208}
209
210#[async_trait::async_trait]
211impl NetworkBuilder for HoprLibp2pNetworkBuilder {
212    type Network = HoprNetwork;
213
214    async fn build(
215        self,
216        identity: &OffchainKeypair,
217        my_multiaddresses: Vec<Multiaddr>,
218        protocol: &'static str,
219        allow_private_addresses: bool,
220    ) -> std::result::Result<(Self::Network, hopr_api::network::BoxedProcessFn), impl std::error::Error> {
221        #[cfg(all(feature = "prometheus", not(test)))]
222        {
223            METRIC_NETWORK_HEALTH.set(0.0);
224        }
225
226        let identity_p2p: libp2p::identity::Keypair = identity.into();
227        let me = identity_p2p.public().to_peer_id();
228        let swarm = InactiveNetwork::build(identity_p2p, self.bootstrap)
229            .await
230            .expect("swarm must be constructible");
231
232        let swarm = swarm
233            .with_listen_on(my_multiaddresses.clone())
234            .expect("swarm must be configurable");
235
236        let swarm = swarm.swarm;
237        let store = hopr_transport_network::store::NetworkPeerStore::new(me, my_multiaddresses.into_iter().collect());
238        let tracker: Arc<DashSet<libp2p::PeerId>> = Default::default();
239
240        let network = HoprNetwork {
241            tracker: tracker.clone(),
242            store: Arc::new(store.clone()),
243            control: swarm.behaviour().streams.new_control(),
244            protocol: libp2p::StreamProtocol::new(protocol),
245        };
246
247        #[cfg(all(feature = "prometheus", not(test)))]
248        let network_inner = network.clone();
249        let mut swarm = swarm;
250        let process = async move {
251            while let Some(event) = swarm.next().await {
252                match event {
253                    SwarmEvent::Behaviour(HoprNetworkBehaviorEvent::Discovery(_)) => {}
254                    SwarmEvent::Behaviour(
255                        HoprNetworkBehaviorEvent::Autonat(event)
256                    ) => {
257                        match event {
258                            autonat::Event::StatusChanged { old, new } => {
259                                info!(?old, ?new, "AutoNAT status changed");
260                                #[cfg(all(feature = "prometheus", not(test)))]
261                                {
262                                    let value = match new {
263                                        autonat::NatStatus::Unknown => 0.0,
264                                        autonat::NatStatus::Public(_) => 1.0,
265                                        autonat::NatStatus::Private => 2.0,
266                                    };
267                                    METRIC_TRANSPORT_NAT_STATUS.set(value);
268                                }
269                            }
270                            autonat::Event::InboundProbe { .. } => {}
271                            autonat::Event::OutboundProbe { .. } => {}
272                        }
273                    }
274                    SwarmEvent::Behaviour(HoprNetworkBehaviorEvent::Identify(_event)) => {}
275                    SwarmEvent::ConnectionEstablished {
276                        peer_id,
277                        connection_id,
278                        num_established,
279                        established_in,
280                        endpoint,
281                        ..
282                        // concurrent_dial_errors,
283                    } => {
284                        debug!(%peer_id, %connection_id, num_established, established_in_ms = established_in.as_millis(), transport="libp2p", "connection established");
285
286                        if num_established == std::num::NonZero::<u32>::new(1).expect("must be a non-zero value") {
287                            match endpoint {
288                                libp2p::core::ConnectedPoint::Dialer { address, .. } => {
289                                    if allow_private_addresses || is_public_address(&address) {
290                                        if let Err(error) = store.add(peer_id, std::collections::HashSet::from([address])) {
291                                            error!(peer = %peer_id, %error, direction = "outgoing", "failed to add connected peer to the peer store");
292                                        }
293                                    } else {
294                                        debug!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Private/local peer address encountered")
295                                    }
296                                    tracker.insert(peer_id);
297                                },
298                                libp2p::core::ConnectedPoint::Listener { send_back_addr, .. } => {
299                                    if allow_private_addresses || is_public_address(&send_back_addr) {
300                                        if let Err(error) = store.add(peer_id, std::collections::HashSet::from([send_back_addr])) {
301                                            error!(peer = %peer_id, %error, direction = "incoming", "failed to add connected peer to the peer store");
302                                        }
303                                    } else {
304                                        debug!(transport="libp2p", peer = %peer_id, multiaddress = %send_back_addr, "Private/local peer address ignored")
305                                    }
306                                    tracker.insert(peer_id);
307                                }
308                            }
309                        } else {
310                            trace!(transport="libp2p", peer = %peer_id, num_established, "Additional connection established")
311                        }
312
313                        print_network_info(swarm.network_info(), "connection established");
314
315                        #[cfg(all(feature = "prometheus", not(test)))]
316                        {
317                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
318                            METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT.increment(1.0);
319                        }
320                    }
321                    SwarmEvent::ConnectionClosed {
322                        peer_id,
323                        connection_id,
324                        cause,
325                        num_established,
326                        ..
327                        // endpoint,
328                    } => {
329                        debug!(%peer_id, %connection_id, num_established, transport="libp2p", "connection closed: {cause:?}");
330
331                        if num_established == 0 {
332                            tracker.remove(&peer_id);
333                        }
334
335                        print_network_info(swarm.network_info(), "connection closed");
336
337                        #[cfg(all(feature = "prometheus", not(test)))]
338                        {
339                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
340                            METRIC_TRANSPORT_P2P_OPEN_CONNECTION_COUNT.decrement(1.0);
341                        }
342                    }
343                    SwarmEvent::IncomingConnection {
344                        connection_id,
345                        local_addr,
346                        send_back_addr,
347                    } => {
348                        trace!(%local_addr, %send_back_addr, %connection_id, transport="libp2p", "incoming connection");
349                    }
350                    SwarmEvent::IncomingConnectionError {
351                        local_addr,
352                        connection_id,
353                        error,
354                        send_back_addr,
355                        peer_id
356                    } => {
357                        debug!(?peer_id, %local_addr, %send_back_addr, %connection_id, transport="libp2p", %error, "incoming connection error");
358                    }
359                    SwarmEvent::OutgoingConnectionError {
360                        connection_id,
361                        error,
362                        peer_id
363                    } => {
364                        debug!(peer = ?peer_id, %connection_id, transport="libp2p", %error, "outgoing connection error");
365
366                        if let Some(peer_id) = peer_id
367                            && !swarm.is_connected(&peer_id) {
368                                if let Err(error) = store.remove(&peer_id) {
369                                    error!(peer = %peer_id, %error, "failed to remove undialable peer from the peer store");
370                                }
371                                tracker.remove(&peer_id);
372                            }
373
374                        #[cfg(all(feature = "prometheus", not(test)))]
375                        {
376                            METRIC_NETWORK_HEALTH.set((hopr_api::network::NetworkView::health(&network_inner) as i32).into());
377                        }
378                    }
379                    SwarmEvent::NewListenAddr {
380                        listener_id,
381                        address,
382                    } => {
383                        debug!(%listener_id, %address, transport="libp2p", "new listen address")
384                    }
385                    SwarmEvent::ExpiredListenAddr {
386                        listener_id,
387                        address,
388                    } => {
389                        debug!(%listener_id, %address, transport="libp2p", "expired listen address")
390                    }
391                    SwarmEvent::ListenerClosed {
392                        listener_id,
393                        addresses,
394                        reason,
395                    } => {
396                        debug!(%listener_id, ?addresses, ?reason, transport="libp2p", "listener closed", )
397                    }
398                    SwarmEvent::ListenerError {
399                        listener_id,
400                        error,
401                    } => {
402                        debug!(%listener_id, transport="libp2p", %error, "listener error")
403                    }
404                    SwarmEvent::Dialing {
405                        peer_id,
406                        connection_id,
407                    } => {
408                        debug!(peer = ?peer_id, %connection_id, transport="libp2p", "dialing")
409                    }
410                    SwarmEvent::NewExternalAddrCandidate {address} => {
411                        debug!(%address, "Detected new external address candidate")
412                    }
413                    SwarmEvent::ExternalAddrConfirmed { address } => {
414                        info!(%address, "Detected external address")
415                    }
416                    SwarmEvent::ExternalAddrExpired {
417                        ..  // address: Multiaddr
418                    } => {}
419                    SwarmEvent::NewExternalAddrOfPeer {
420                        peer_id, address
421                    } => {
422                        // Only store public/routable addresses
423                        if allow_private_addresses || is_public_address(&address) {
424                            swarm.add_peer_address(peer_id, address.clone());
425                            trace!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Public peer address stored in swarm")
426                        } else {
427                            trace!(transport="libp2p", peer = %peer_id, multiaddress = %address, "Private/local peer address ignored")
428                        }
429                    },
430                    _ => trace!(transport="libp2p", "Unsupported enum option detected")
431                }
432            }
433        }.boxed();
434
435        Ok::<_, std::io::Error>((network, Box::new(move || process)))
436    }
437}
438
439fn print_network_info(network_info: NetworkInfo, event: &str) {
440    let num_peers = network_info.num_peers();
441    let connection_counters = network_info.connection_counters();
442    let num_incoming = connection_counters.num_established_incoming();
443    let num_outgoing = connection_counters.num_established_outgoing();
444    info!(
445        num_peers,
446        num_incoming, num_outgoing, "swarm network status after {event}"
447    );
448}
449
450#[cfg(test)]
451mod tests {
452    #[test]
453    #[cfg(any(target_os = "android", target_os = "ios"))]
454    fn uses_cloudflare_dns_resolver_config() {
455        let (resolver_config, resolver_opts) = super::swarm_dns_config();
456        assert_eq!(resolver_config, libp2p::dns::ResolverConfig::cloudflare());
457        assert_eq!(resolver_opts, libp2p::dns::ResolverOpts::default());
458    }
459}