hotshot_libp2p_networking/network/

node.rs

// Copyright (c) 2021-2024 Espresso Systems (espressosys.com)
// This file is part of the HotShot repository.

// You should have received a copy of the MIT License
// along with the HotShot repository. If not, see <https://mit-license.org/>.

/// configuration for the libp2p network (e.g. how it should be built)
mod config;

/// libp2p network handle
/// allows for control over the libp2p network
mod handle;

use std::{
    collections::{HashMap, HashSet},
    iter,
    num::{NonZeroU32, NonZeroUsize},
    pin::pin,
    sync::Arc,
    time::{Duration, Instant},
};

use alloy::primitives::U256;
use bimap::BiMap;
use futures::{SinkExt, StreamExt, channel::mpsc};
use hotshot_types::{
    constants::KAD_DEFAULT_REPUB_INTERVAL_SEC, traits::node_implementation::NodeType,
};
use libp2p::{
    Multiaddr, StreamProtocol, Swarm, SwarmBuilder,
    core::{transport::ListenerId, upgrade::Version::V1Lazy},
    gossipsub::{
        Behaviour as Gossipsub, ConfigBuilder as GossipsubConfigBuilder, Event as GossipEvent,
        Message as GossipsubMessage, MessageAuthenticity, MessageId, Topic, ValidationMode,
    },
    identify::{
        Behaviour as IdentifyBehaviour, Config as IdentifyConfig, Event as IdentifyEvent,
        Info as IdentifyInfo,
    },
    identity::Keypair,
    kad::{Behaviour, Config, Mode, Record, store::MemoryStore},
    request_response::{
        Behaviour as RequestResponse, Config as Libp2pRequestResponseConfig, ProtocolSupport,
    },
    swarm::SwarmEvent,
};
use libp2p_identity::PeerId;
use parking_lot::Mutex;
use rand::{prelude::SliceRandom, thread_rng};
use tokio::{
    select, spawn,
    sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
    task::JoinHandle,
    time::sleep,
};
use tracing::{Instrument, debug, error, info, info_span, instrument, warn};

pub use self::{
    config::{
        DEFAULT_REPLICATION_FACTOR, GossipConfig, NetworkNodeConfig, NetworkNodeConfigBuilder,
        NetworkNodeConfigBuilderError, RequestResponseConfig,
    },
    handle::{NetworkNodeHandle, NetworkNodeReceiver, spawn_network_node},
};
use super::{
    BoxedTransport, ClientRequest, NetworkDef, NetworkError, NetworkEvent, NetworkEventInternal,
    behaviours::dht::{
        bootstrap::{DHTBootstrapTask, InputEvent},
        store::{
            persistent::{DhtPersistentStorage, PersistentStore},
            validated::ValidatedStore,
        },
    },
    cbor::Cbor,
    gen_transport,
};
use crate::network::{
    behaviours::{
        dht::{DHTBehaviour, DHTProgress, KadPutQuery},
        direct_message::{DMBehaviour, DMRequest},
        exponential_backoff::ExponentialBackoff,
    },
    log_summary::LogEvent,
};

/// Join handle for the spawned swarm event-loop task.
pub type SwarmTaskHandle = JoinHandle<Result<(), NetworkError>>;

/// Maximum size of a message
pub const MAX_GOSSIP_MSG_SIZE: usize = 2_000_000_000;

/// Wrapped num of connections
pub const ESTABLISHED_LIMIT: NonZeroU32 = NonZeroU32::new(ESTABLISHED_LIMIT_UNWR).unwrap();
/// Number of connections to a single peer before logging an error
pub const ESTABLISHED_LIMIT_UNWR: u32 = 10;

/// How long to wait for the first inbound connection.
const NO_INBOUND_CONNECTION_GRACE_PERIOD: Duration = Duration::from_secs(300);

/// Mainnet libp2p protocol identifiers. The snapshot tests below lock these down so a
/// change that would partition mainnet (e.g. a stray `protocol_id_prefix` call) is caught.
/// `None` for gossipsub means "do not call `protocol_id_prefix`" — libp2p's defaults
/// (`/meshsub/1.1.0` and `/meshsub/1.0.0`) are then used.
pub(crate) fn mainnet_gossipsub_prefix() -> Option<&'static str> {
    None
}
pub(crate) fn mainnet_kad_protocol() -> StreamProtocol {
    StreamProtocol::new("/ipfs/kad/1.0.0")
}
pub(crate) fn mainnet_direct_message_protocol() -> StreamProtocol {
    StreamProtocol::new("/HotShot/direct_message/1.0")
}
pub(crate) fn mainnet_identify_protocol() -> &'static str {
    "HotShot/identify/1.0"
}

/// Resolve the gossipsub `protocol_id_prefix` for the given network discriminator.
/// `None` returns the mainnet value (the libp2p default).
pub(crate) fn gossipsub_prefix(discriminator: Option<U256>) -> Option<String> {
    match discriminator {
        None => mainnet_gossipsub_prefix().map(String::from),
        Some(d) => Some(format!("/HotShot/gossipsub/1.0/{d:#x}")),
    }
}

/// Resolve the kademlia stream protocol for the given network discriminator.
pub(crate) fn kad_protocol(discriminator: Option<U256>) -> Result<StreamProtocol, NetworkError> {
    match discriminator {
        None => Ok(mainnet_kad_protocol()),
        Some(d) => StreamProtocol::try_from_owned(format!("/ipfs/kad/1.0.0/{d:#x}"))
            .map_err(|err| NetworkError::ConfigError(format!("invalid kademlia protocol: {err}"))),
    }
}

/// Resolve the identify protocol string for the given network discriminator.
pub(crate) fn identify_protocol(discriminator: Option<U256>) -> String {
    match discriminator {
        None => mainnet_identify_protocol().to_string(),
        Some(d) => format!("HotShot/identify/1.0/{d:#x}"),
    }
}

/// Resolve the direct-message stream protocol for the given network discriminator.
pub(crate) fn direct_message_protocol(
    discriminator: Option<U256>,
) -> Result<StreamProtocol, NetworkError> {
    match discriminator {
        None => Ok(mainnet_direct_message_protocol()),
        Some(d) => StreamProtocol::try_from_owned(format!("/HotShot/direct_message/1.0/{d:#x}"))
            .map_err(|err| {
                NetworkError::ConfigError(format!("invalid direct_message protocol: {err}"))
            }),
    }
}

/// A peer's network is identifiable only by its advertised kademlia `StreamProtocol`.
fn should_keep_peer(expected: &StreamProtocol, peer_protocols: &[StreamProtocol]) -> bool {
    peer_protocols.contains(expected)
}

fn resolve_put_quorum(
    quorum_override: Option<NonZeroUsize>,
    replication_factor: NonZeroUsize,
) -> NonZeroUsize {
    quorum_override.unwrap_or_else(|| {
        NonZeroUsize::new(replication_factor.get() / 2)
            .expect("replication factor should be bigger than 0")
    })
}

/// Network definition
#[derive(derive_more::Debug)]
pub struct NetworkNode<T: NodeType, D: DhtPersistentStorage> {
    /// peer id of network node
    peer_id: PeerId,
    /// the swarm of networkbehaviours
    #[debug(skip)]
    swarm: Swarm<NetworkDef<T::SignatureKey, D>>,
    /// The Kademlia record TTL
    kademlia_record_ttl: Duration,
    /// The map from consensus keys to peer IDs
    consensus_key_to_pid_map: Arc<Mutex<BiMap<T::SignatureKey, PeerId>>>,
    /// the listener id we are listening on, if it exists
    listener_id: Option<ListenerId>,
    /// Handler for direct messages
    direct_message_state: DMBehaviour,
    /// Handler for DHT Events
    dht_handler: DHTBehaviour<T::SignatureKey, D>,
    /// Channel to resend requests, set to Some when we call `spawn_listeners`
    resend_tx: Option<UnboundedSender<ClientRequest>>,
    /// Whether a peer ever connected to us.
    ///
    /// Used to warn the operator when the advertised address is likely not
    /// publicly reachable.
    saw_inbound_connection: bool,
    /// Kademlia `StreamProtocol` expected from same-network peers.
    expected_kad_protocol: StreamProtocol,
    /// Peers confirmed via identify to share our network discriminator. Gates
    /// `NewExternalAddrOfPeer`, whose event carries no protocols to check directly.
    same_network_peers: HashSet<PeerId>,
    dht_put_quorum: Option<NonZeroUsize>,
}

impl<T: NodeType, D: DhtPersistentStorage> NetworkNode<T, D> {
    /// Returns number of peers this node is connected to
    pub fn num_connected(&self) -> usize {
        self.swarm.connected_peers().count()
    }

    /// return hashset of PIDs this node is connected to
    pub fn connected_pids(&self) -> HashSet<PeerId> {
        self.swarm.connected_peers().copied().collect()
    }

    /// starts the swarm listening on `listen_addr`
    /// and optionally dials into peer `known_peer`
    /// returns the address the swarm is listening upon
    #[instrument(skip(self))]
    pub async fn start_listen(
        &mut self,
        listen_addr: Multiaddr,
    ) -> Result<Multiaddr, NetworkError> {
        self.listener_id = Some(self.swarm.listen_on(listen_addr).map_err(|err| {
            NetworkError::ListenError(format!("failed to listen for Libp2p: {err}"))
        })?);
        let addr = loop {
            if let Some(SwarmEvent::NewListenAddr { address, .. }) = self.swarm.next().await {
                break address;
            }
        };
        info!("Libp2p listening on {addr:?}");
        Ok(addr)
    }

    /// initialize the DHT with known peers
    /// add the peers to kademlia and then
    /// the `spawn_listeners` function
    /// will start connecting to peers
    #[instrument(skip(self))]
    pub fn add_known_peers(&mut self, known_peers: &[(PeerId, Multiaddr)]) {
        debug!("Adding {} known peers", known_peers.len());
        let behaviour = self.swarm.behaviour_mut();
        let mut bs_nodes = HashMap::<PeerId, HashSet<Multiaddr>>::new();
        let mut shuffled = known_peers.iter().collect::<Vec<_>>();
        shuffled.shuffle(&mut thread_rng());
        for (peer_id, addr) in shuffled {
            if *peer_id != self.peer_id {
                behaviour.dht.add_address(peer_id, addr.clone());
                bs_nodes.insert(*peer_id, iter::once(addr.clone()).collect());
            }
        }
    }

    /// Creates a new `Network` with the given settings.
    ///
    /// Currently:
    ///   * Generates a random key pair and associated [`PeerId`]
    ///   * Launches a hopefully production ready transport: QUIC v1 (RFC 9000) + DNS
    ///   * Generates a connection to the "broadcast" topic
    ///   * Creates a swarm to manage peers and events
    ///
    /// # Errors
    /// - If we fail to generate the transport or any of the behaviours
    ///
    /// # Panics
    /// If 5 < 0
    #[allow(clippy::too_many_lines)]
    pub async fn new(
        config: NetworkNodeConfig,
        dht_persistent_storage: D,
        consensus_key_to_pid_map: Arc<Mutex<BiMap<T::SignatureKey, PeerId>>>,
    ) -> Result<Self, NetworkError> {
        // Generate a random `KeyPair` if one is not specified
        let keypair = config
            .keypair
            .clone()
            .unwrap_or_else(Keypair::generate_ed25519);

        // Get the `PeerId` from the `KeyPair`
        let peer_id = PeerId::from(keypair.public());

        // Generate the transport from the keypair and auth message
        let transport: BoxedTransport = gen_transport::<T>(
            keypair.clone(),
            config.auth_message.clone(),
            Arc::clone(&consensus_key_to_pid_map),
        )
        .await?;

        // Calculate the record republication interval
        let kademlia_record_republication_interval = config
            .republication_interval
            .unwrap_or(Duration::from_secs(KAD_DEFAULT_REPUB_INTERVAL_SEC));

        // Calculate the Kademlia record TTL
        let kademlia_ttl = config
            .ttl
            .unwrap_or(16 * kademlia_record_republication_interval);

        let expected_kad_protocol = kad_protocol(config.network_discriminator)?;

        // Generate the swarm
        let mut swarm: Swarm<NetworkDef<T::SignatureKey, D>> = {
            // Use the `Blake3` hash of the message's contents as the ID
            let message_id_fn = |message: &GossipsubMessage| {
                let hash = blake3::hash(&message.data);
                MessageId::from(hash.as_bytes().to_vec())
            };

            // Derive a `Gossipsub` config from our gossip config
            let mut gossipsub_builder = GossipsubConfigBuilder::default();
            if let Some(prefix) = gossipsub_prefix(config.network_discriminator) {
                gossipsub_builder.protocol_id_prefix(prefix);
            }
            let gossipsub_config = gossipsub_builder
                .message_id_fn(message_id_fn) // Use the (blake3) hash of a message as its ID
                .validation_mode(ValidationMode::Strict) // Force all messages to have valid signatures
                .heartbeat_interval(config.gossip_config.heartbeat_interval) // Time between gossip heartbeats
                .history_gossip(config.gossip_config.history_gossip) // Number of heartbeats to gossip about
                .history_length(config.gossip_config.history_length) // Number of heartbeats to remember the full message for
                .mesh_n(config.gossip_config.mesh_n) // Target number of mesh peers
                .mesh_n_high(config.gossip_config.mesh_n_high) // Upper limit of mesh peers
                .mesh_n_low(config.gossip_config.mesh_n_low) // Lower limit of mesh peers
                .mesh_outbound_min(config.gossip_config.mesh_outbound_min) // Minimum number of outbound peers in mesh
                .max_transmit_size(config.gossip_config.max_transmit_size) // Maximum size of a message
                .max_ihave_length(config.gossip_config.max_ihave_length) // Maximum number of messages to include in an IHAVE message
                .max_ihave_messages(config.gossip_config.max_ihave_messages) // Maximum number of IHAVE messages to accept from a peer within a heartbeat
                .published_message_ids_cache_time(
                    config.gossip_config.published_message_ids_cache_time,
                ) // Cache duration for published message IDs
                .iwant_followup_time(config.gossip_config.iwant_followup_time) // Time to wait for a message requested through IWANT following an IHAVE advertisement
                .max_messages_per_rpc(config.gossip_config.max_messages_per_rpc) // The maximum number of messages we will process in a given RPC
                .gossip_retransimission(config.gossip_config.gossip_retransmission) // Controls how many times we will allow a peer to request the same message id through IWANT gossip before we start ignoring them.
                .flood_publish(config.gossip_config.flood_publish) // If enabled newly created messages will always be sent to all peers that are subscribed to the topic and have a good enough score.
                .duplicate_cache_time(config.gossip_config.duplicate_cache_time) // The time period that messages are stored in the cache
                .fanout_ttl(config.gossip_config.fanout_ttl) // Time to live for fanout peers
                .heartbeat_initial_delay(config.gossip_config.heartbeat_initial_delay) // Initial delay in each heartbeat
                .gossip_factor(config.gossip_config.gossip_factor) // Affects how many peers we will emit gossip to at each heartbeat
                .gossip_lazy(config.gossip_config.gossip_lazy) // Minimum number of peers to emit gossip to during a heartbeat
                .build()
                .map_err(|err| {
                    NetworkError::ConfigError(format!("error building gossipsub config: {err:?}"))
                })?;

            // - Build a gossipsub network behavior
            let gossipsub: Gossipsub = Gossipsub::new(
                MessageAuthenticity::Signed(keypair.clone()),
                gossipsub_config,
            )
            .map_err(|err| {
                NetworkError::ConfigError(format!("error building gossipsub behaviour: {err:?}"))
            })?;

            //   Build a identify network behavior needed for own
            //   node connection information
            //   E.g. this will answer the question: how are other nodes
            //   seeing the peer from behind a NAT
            let identify_cfg = IdentifyConfig::new(
                identify_protocol(config.network_discriminator),
                keypair.public(),
            );
            let identify = IdentifyBehaviour::new(identify_cfg);

            // - Build DHT needed for peer discovery
            let mut kconfig = Config::new(expected_kad_protocol.clone());
            kconfig
                .set_parallelism(NonZeroUsize::new(5).unwrap())
                .set_provider_publication_interval(Some(kademlia_record_republication_interval))
                .set_publication_interval(Some(kademlia_record_republication_interval))
                .set_record_ttl(Some(kademlia_ttl));

            // allowing panic here because something is very wrong if this fails
            #[allow(clippy::panic)]
            if let Some(factor) = config.replication_factor {
                kconfig.set_replication_factor(factor);
            } else {
                panic!("Replication factor not set");
            }

            // Create the DHT behaviour with the given persistent storage
            let mut kadem = Behaviour::with_config(
                peer_id,
                PersistentStore::new(
                    ValidatedStore::new(MemoryStore::new(peer_id)),
                    dht_persistent_storage,
                    5,
                )
                .await,
                kconfig,
            );
            kadem.set_mode(Some(Mode::Server));

            let rrconfig = Libp2pRequestResponseConfig::default();

            // Create a new `cbor` codec with the given request and response sizes
            let cbor = Cbor::new(
                config.request_response_config.request_size_maximum,
                config.request_response_config.response_size_maximum,
            );

            let direct_message: super::cbor::Behaviour<Vec<u8>, Vec<u8>> =
                RequestResponse::with_codec(
                    cbor,
                    [(
                        direct_message_protocol(config.network_discriminator)?,
                        ProtocolSupport::Full,
                    )],
                    rrconfig.clone(),
                );

            let network = NetworkDef::new(gossipsub, kadem, identify, direct_message);

            // build swarm
            let swarm = SwarmBuilder::with_existing_identity(keypair.clone());
            let swarm = swarm.with_tokio();

            swarm
                .with_other_transport(|_| transport)
                .unwrap()
                .with_behaviour(|_| network)
                .unwrap()
                .with_swarm_config(|cfg| {
                    cfg.with_idle_connection_timeout(Duration::from_secs(10))
                        .with_substream_upgrade_protocol_override(V1Lazy)
                })
                .build()
        };
        for (peer, addr) in &config.to_connect_addrs {
            if peer != swarm.local_peer_id() {
                swarm.behaviour_mut().add_address(peer, addr.clone());
                swarm.add_peer_address(*peer, addr.clone());
            }
        }

        for addr in &config.announce_addresses {
            info!("Adding announce address {addr}");
            swarm.add_external_address(addr.clone());
        }

        Ok(Self {
            peer_id,
            swarm,
            kademlia_record_ttl: kademlia_ttl,
            consensus_key_to_pid_map,
            listener_id: None,
            direct_message_state: DMBehaviour::default(),
            dht_handler: DHTBehaviour::new(
                peer_id,
                config
                    .replication_factor
                    .unwrap_or(NonZeroUsize::new(4).unwrap()),
            ),
            resend_tx: None,
            saw_inbound_connection: false,
            expected_kad_protocol,
            same_network_peers: HashSet::new(),
            dht_put_quorum: config.dht_put_quorum,
        })
    }

    /// Identify is the first point a peer's network is detectable; drop other networks.
    fn on_identify_received(&mut self, peer_id: PeerId, info: IdentifyInfo) {
        if should_keep_peer(&self.expected_kad_protocol, &info.protocols) {
            self.same_network_peers.insert(peer_id);
            let behaviour = self.swarm.behaviour_mut();
            // Deduplicate before inserting (duplicates are common in practice).
            for addr in info.listen_addrs.iter().collect::<HashSet<_>>() {
                behaviour.dht.add_address(&peer_id, addr.clone());
            }
        } else {
            debug!(
                "Dropping peer {peer_id}: its kad protocol does not match ours ({})",
                self.expected_kad_protocol
            );
            self.same_network_peers.remove(&peer_id);
            self.swarm.behaviour_mut().dht.remove_peer(&peer_id);
            let _ = self.swarm.disconnect_peer_id(peer_id);
        }
    }

    /// Publish a key/value to the record store.
    ///
    /// # Panics
    /// If the default replication factor is `None`
    pub fn put_record(&mut self, mut query: KadPutQuery) {
        // Create the new record
        let mut record = Record::new(query.key.clone(), query.value.clone());

        // Set the record's expiration time to the proper time
        record.expires = Some(Instant::now() + self.kademlia_record_ttl);

        let quorum = resolve_put_quorum(self.dht_put_quorum, self.dht_handler.replication_factor());
        match self
            .swarm
            .behaviour_mut()
            .dht
            .put_record(record, libp2p::kad::Quorum::N(quorum))
        {
            Err(e) => {
                // failed try again later
                query.progress = DHTProgress::NotStarted;
                query.backoff.start_next(false);
                error!("Error publishing to DHT: {e:?} for peer {:?}", self.peer_id);
            },
            Ok(qid) => {
                debug!("Published record to DHT with qid {qid:?}");
                let query = KadPutQuery {
                    progress: DHTProgress::InProgress(qid),
                    ..query
                };
                self.dht_handler.put_record(qid, query);
            },
        }
    }

    /// event handler for client events
    /// currently supported actions include
    /// - shutting down the swarm
    /// - gossipping a message to known peers on the `global` topic
    /// - returning the id of the current peer
    /// - subscribing to a topic
    /// - unsubscribing from a toipc
    /// - direct messaging a peer
    #[instrument(skip(self))]
    async fn handle_client_requests(
        &mut self,
        msg: Option<ClientRequest>,
    ) -> Result<bool, NetworkError> {
        let behaviour = self.swarm.behaviour_mut();
        match msg {
            Some(msg) => {
                match msg {
                    ClientRequest::BeginBootstrap => {
                        debug!("Beginning Libp2p bootstrap");
                        let _ = self.swarm.behaviour_mut().dht.bootstrap();
                    },
                    ClientRequest::LookupPeer(pid, chan) => {
                        let id = self.swarm.behaviour_mut().dht.get_closest_peers(pid);
                        self.dht_handler
                            .in_progress_get_closest_peers
                            .insert(id, chan);
                    },
                    ClientRequest::GetRoutingTable(chan) => {
                        self.dht_handler
                            .print_routing_table(&mut self.swarm.behaviour_mut().dht);
                        if chan.send(()).is_err() {
                            warn!("Tried to notify client but client not tracking anymore");
                        }
                    },
                    ClientRequest::PutDHT { key, value, notify } => {
                        let query = KadPutQuery {
                            progress: DHTProgress::NotStarted,
                            notify,
                            key,
                            value,
                            backoff: ExponentialBackoff::default(),
                        };
                        self.put_record(query);
                    },
                    ClientRequest::GetConnectedPeerNum(s) => {
                        if s.send(self.num_connected()).is_err() {
                            error!("error sending peer number to client");
                        }
                    },
                    ClientRequest::GetConnectedPeers(s) => {
                        if s.send(self.connected_pids()).is_err() {
                            error!("error sending peer set to client");
                        }
                    },
                    ClientRequest::GetKadRoutingPeers(s) => {
                        let peers: HashSet<PeerId> = self
                            .swarm
                            .behaviour_mut()
                            .dht
                            .kbuckets()
                            .flat_map(|b| {
                                b.iter().map(|e| *e.node.key.preimage()).collect::<Vec<_>>()
                            })
                            .collect();
                        if s.send(peers).is_err() {
                            error!("error sending kad routing peers to client");
                        }
                    },
                    ClientRequest::GetDHT {
                        key,
                        notify,
                        retry_count,
                    } => {
                        self.dht_handler.get_record(
                            key,
                            notify,
                            ExponentialBackoff::default(),
                            retry_count,
                            &mut self.swarm.behaviour_mut().dht,
                        );
                    },
                    ClientRequest::IgnorePeers(_peers) => {
                        // NOTE used by test with conductor only
                    },
                    ClientRequest::Shutdown => {
                        if let Some(listener_id) = self.listener_id {
                            self.swarm.remove_listener(listener_id);
                        }

                        return Ok(true);
                    },
                    ClientRequest::GossipMsg(topic, contents) => {
                        behaviour.publish_gossip(Topic::new(topic.clone()), contents.clone());
                    },
                    ClientRequest::Subscribe(t, chan) => {
                        behaviour.subscribe_gossip(&t);
                        if let Some(chan) = chan
                            && chan.send(()).is_err()
                        {
                            error!("finished subscribing but response channel dropped");
                        }
                    },
                    ClientRequest::Unsubscribe(t, chan) => {
                        behaviour.unsubscribe_gossip(&t);
                        if let Some(chan) = chan
                            && chan.send(()).is_err()
                        {
                            error!("finished unsubscribing but response channel dropped");
                        }
                    },
                    ClientRequest::DirectRequest {
                        pid,
                        contents,
                        retry_count,
                    } => {
                        debug!("Sending direct request to {pid:?}");
                        let id = behaviour.add_direct_request(pid, contents.clone());
                        let req = DMRequest {
                            peer_id: pid,
                            data: contents,
                            backoff: ExponentialBackoff::default(),
                            retry_count,
                        };
                        self.direct_message_state.add_direct_request(req, id);
                    },
                    ClientRequest::DirectResponse(chan, msg) => {
                        behaviour.add_direct_response(chan, msg);
                    },
                    ClientRequest::AddKnownPeers(peers) => {
                        self.add_known_peers(&peers);
                    },
                    ClientRequest::Prune(pid) => {
                        if self.swarm.disconnect_peer_id(pid).is_err() {
                            warn!("Could not disconnect from {pid:?}");
                        }
                    },
                }
            },
            None => {
                error!("Error receiving msg in main behaviour loop: channel closed");
            },
        }
        Ok(false)
    }

    /// event handler for events emitted from the swarm
    #[allow(clippy::type_complexity)]
    #[instrument(skip(self))]
    async fn handle_swarm_events(
        &mut self,
        event: SwarmEvent<NetworkEventInternal>,
        send_to_client: &UnboundedSender<NetworkEvent>,
    ) -> Result<(), NetworkError> {
        // Make the match cleaner
        debug!("Swarm event observed {:?}", event);

        #[allow(deprecated)]
        match event {
            SwarmEvent::ConnectionEstablished {
                connection_id: _,
                peer_id,
                endpoint,
                num_established,
                concurrent_dial_errors,
                established_in: _established_in,
            } => {
                if num_established > ESTABLISHED_LIMIT {
                    error!(
                        "Num concurrent connections to a single peer exceeding \
                         {ESTABLISHED_LIMIT:?} at {num_established:?}!"
                    );
                } else {
                    debug!(
                        "Connection established with {peer_id:?} at {endpoint:?} with \
                         {concurrent_dial_errors:?} concurrent dial errors"
                    );
                }

                if endpoint.is_listener() {
                    self.saw_inbound_connection = true;
                }

                // Send the number of connected peers to the client
                send_to_client
                    .send(NetworkEvent::ConnectedPeersUpdate(self.num_connected()))
                    .map_err(|err| NetworkError::ChannelSendError(err.to_string()))?;
            },
            SwarmEvent::ConnectionClosed {
                connection_id: _,
                peer_id,
                endpoint,
                num_established,
                cause,
            } => {
                if num_established > ESTABLISHED_LIMIT_UNWR {
                    error!(
                        "Num concurrent connections to a single peer exceeding \
                         {ESTABLISHED_LIMIT:?} at {num_established:?}!"
                    );
                } else {
                    debug!("Connection closed with {peer_id:?} at {endpoint:?} due to {cause:?}");
                }

                // If we are no longer connected to the peer, remove the consensus key from the map
                // and reset verified status so reconnecting peers are re-verified via identify.
                if num_established == 0 {
                    self.consensus_key_to_pid_map
                        .lock()
                        .remove_by_right(&peer_id);
                    self.same_network_peers.remove(&peer_id);
                }

                // Send the number of connected peers to the client
                send_to_client
                    .send(NetworkEvent::ConnectedPeersUpdate(self.num_connected()))
                    .map_err(|err| NetworkError::ChannelSendError(err.to_string()))?;
            },
            SwarmEvent::Dialing {
                peer_id,
                connection_id: _,
            } => {
                debug!("Attempting to dial {peer_id:?}");
            },
            SwarmEvent::ListenerClosed {
                listener_id: _,
                addresses: _,
                reason: _,
            }
            | SwarmEvent::NewListenAddr {
                listener_id: _,
                address: _,
            }
            | SwarmEvent::ExpiredListenAddr {
                listener_id: _,
                address: _,
            }
            | SwarmEvent::NewExternalAddrCandidate { .. }
            | SwarmEvent::ExternalAddrExpired { .. }
            | SwarmEvent::IncomingConnection {
                connection_id: _,
                local_addr: _,
                send_back_addr: _,
            } => {},
            SwarmEvent::Behaviour(b) => {
                let maybe_event = match b {
                    NetworkEventInternal::DHTEvent(e) => self
                        .dht_handler
                        .dht_handle_event(e, self.swarm.behaviour_mut().dht.store_mut()),
                    NetworkEventInternal::IdentifyEvent(e) => {
                        if let IdentifyEvent::Received {
                            peer_id,
                            info,
                            connection_id: _,
                        } = *e
                        {
                            self.on_identify_received(peer_id, info);
                        }
                        None
                    },
                    NetworkEventInternal::GossipEvent(e) => match *e {
                        GossipEvent::Message {
                            propagation_source: _peer_id,
                            message_id: _id,
                            message,
                        } => Some(NetworkEvent::GossipMsg(message.data)),
                        GossipEvent::Subscribed { peer_id, topic } => {
                            debug!("Peer {peer_id:?} subscribed to topic {topic:?}");
                            None
                        },
                        GossipEvent::Unsubscribed { peer_id, topic } => {
                            debug!("Peer {peer_id:?} unsubscribed from topic {topic:?}");
                            None
                        },
                        GossipEvent::GossipsubNotSupported { peer_id } => {
                            LogEvent::GossipsubNotSupported.record();
                            debug!("Peer {peer_id:?} does not support gossipsub");
                            None
                        },
                        GossipEvent::SlowPeer {
                            peer_id,
                            failed_messages: _,
                        } => {
                            LogEvent::GossipsubSlowPeer.record();
                            debug!("Peer {peer_id:?} is slow");
                            None
                        },
                    },
                    NetworkEventInternal::DMEvent(e) => self
                        .direct_message_state
                        .handle_dm_event(e, self.resend_tx.clone()),
                };

                if let Some(event) = maybe_event {
                    // forward messages directly to Client
                    send_to_client
                        .send(event)
                        .map_err(|err| NetworkError::ChannelSendError(err.to_string()))?;
                }
            },
            SwarmEvent::OutgoingConnectionError {
                connection_id: _,
                peer_id,
                error,
            } => {
                LogEvent::DialFailure.record();
                debug!("Outgoing connection error to {peer_id:?}: {error:?}");
            },
            SwarmEvent::IncomingConnectionError {
                connection_id: _,
                local_addr: _,
                send_back_addr: _,
                error,
                peer_id: _,
            } => {
                LogEvent::IncomingConnError.record();
                debug!("Incoming connection error: {error:?}");
            },
            SwarmEvent::ListenerError {
                listener_id: _,
                error,
            } => {
                LogEvent::ListenerError.record();
                debug!("Listener error: {error:?}");
            },
            SwarmEvent::ExternalAddrConfirmed { address } => {
                let my_id = *self.swarm.local_peer_id();
                self.swarm
                    .behaviour_mut()
                    .dht
                    .add_address(&my_id, address.clone());
            },
            SwarmEvent::NewExternalAddrOfPeer { peer_id, address } => {
                // Only same-network peers; foreign peers carry no protocols on this event.
                if self.same_network_peers.contains(&peer_id) {
                    self.swarm
                        .behaviour_mut()
                        .dht
                        .add_address(&peer_id, address.clone());
                }
            },
            _ => {
                debug!("Unhandled swarm event {event:?}");
            },
        }
        Ok(())
    }

    /// Spawn a task to listen for requests on the returned channel
    /// as well as any events produced by libp2p
    ///
    /// # Errors
    /// - If we fail to create the channels or the bootstrap channel
    pub fn spawn_listeners(
        mut self,
    ) -> Result<
        (
            UnboundedSender<ClientRequest>,
            UnboundedReceiver<NetworkEvent>,
            SwarmTaskHandle,
        ),
        NetworkError,
    > {
        let (s_input, mut s_output) = unbounded_channel::<ClientRequest>();
        let (r_input, r_output) = unbounded_channel::<NetworkEvent>();
        let (mut bootstrap_tx, bootstrap_rx) = mpsc::channel(100);
        self.resend_tx = Some(s_input.clone());
        self.dht_handler.set_bootstrap_sender(bootstrap_tx.clone());

        DHTBootstrapTask::run(bootstrap_rx, s_input.clone());
        // Keep the task's `JoinHandle` so callers can await the swarm loop
        // ending on shutdown, ensuring the listening socket is released.
        let task = spawn(
            async move {
                let mut inbound_check = pin!(sleep(NO_INBOUND_CONNECTION_GRACE_PERIOD));
                let mut inbound_check_pending = true;
                loop {
                    select! {
                        event = self.swarm.next() => {
                            debug!("peerid {:?}\t\thandling maybe event {:?}", self.peer_id, event);
                            if let Some(event) = event {
                                debug!("peerid {:?}\t\thandling event {:?}", self.peer_id, event);
                                self.handle_swarm_events(event, &r_input).await?;
                            }
                        },
                        msg = s_output.recv() => {
                            debug!("peerid {:?}\t\thandling msg {:?}", self.peer_id, msg);
                            let shutdown = self.handle_client_requests(msg).await?;
                            if shutdown {
                                let _ = bootstrap_tx.send(InputEvent::ShutdownBootstrap).await;
                                break
                            }
                        },
                        () = &mut inbound_check, if inbound_check_pending => {
                            inbound_check_pending = false;
                            self.warn_if_unreachable();
                        }
                    }
                }
                drop(self.swarm);
                Ok::<(), NetworkError>(())
            }
            .instrument(info_span!("Libp2p NetworkBehaviour Handler")),
        );
        Ok((s_input, r_output, task))
    }

    /// Get a reference to the network node's peer id.
    pub fn peer_id(&self) -> PeerId {
        self.peer_id
    }

    /// One-shot reachability check.
    ///
    /// Run [`NO_INBOUND_CONNECTION_GRACE_PERIOD`] after the swarm loop starts.
    /// A healthy, publicly reachable node receives inbound connections shortly
    /// after joining. If we are listening but no peer has connected to us by
    /// now, the advertised address is likely wrong or blocked, so point the
    /// operator at it. If we have no connections at all, not even outbound,
    /// the problem is more fundamental than the advertised address.
    fn warn_if_unreachable(&self) {
        if self.listener_id.is_none() || self.saw_inbound_connection {
            return;
        }
        if self.num_connected() == 0 {
            error!(
                "No connections at all within {NO_INBOUND_CONNECTION_GRACE_PERIOD:?} of startup. \
                 This node could not reach any peer. Check the bootstrap node configuration and \
                 that outbound UDP traffic is allowed (firewall/NAT/security group)."
            );
        } else {
            error!(
                "No peer has connected to us within {NO_INBOUND_CONNECTION_GRACE_PERIOD:?} of \
                 startup. This node is likely not publicly reachable: peers cannot direct-message \
                 us, so leader views may fail and we may accumulate missed slots. Verify \
                 --libp2p-advertise-address (env ESPRESSO_NODE_LIBP2P_ADVERTISE_ADDRESS) is set \
                 to a publicly reachable host:port, and ensure inbound UDP at that port is open \
                 from the public internet (firewall/NAT/security group)."
            );
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{collections::HashSet, num::NonZeroUsize, sync::Arc, time::Duration};

    use hotshot_example_types::node_types::TestTypes;
    use parking_lot::Mutex;

    use super::{
        U256, direct_message_protocol, gossipsub_prefix, identify_protocol, kad_protocol,
        resolve_put_quorum, should_keep_peer,
    };
    use crate::network::{
        NetworkNodeConfigBuilder, behaviours::dht::store::persistent::DhtNoPersistence,
        node::handle::spawn_network_node,
    };

    /// Spawn a node with the given discriminator and initial peer set. Returns the handle and
    /// a background task draining the event receiver so the swarm loop never stalls on backpressure.
    async fn spawn_node(
        discriminator: Option<U256>,
        connect_to: HashSet<(libp2p_identity::PeerId, libp2p::Multiaddr)>,
        id: usize,
    ) -> crate::network::node::handle::NetworkNodeHandle<TestTypes> {
        use bimap::BiMap;

        let config = NetworkNodeConfigBuilder::default()
            .network_discriminator(discriminator)
            .to_connect_addrs(connect_to)
            .build()
            .expect("config build");

        let key_map = Arc::new(Mutex::new(BiMap::default()));
        let (mut receiver, handle) = spawn_network_node::<TestTypes, DhtNoPersistence>(
            config,
            DhtNoPersistence,
            key_map,
            id,
        )
        .await
        .expect("spawn node");

        // Drain events so the swarm loop is never blocked on a full channel.
        tokio::spawn(async move {
            loop {
                if receiver.recv().await.is_err() {
                    break;
                }
            }
        });

        handle
    }

    fn snapshot_for(discriminator: Option<U256>) -> String {
        format!(
            "gossipsub_prefix: {:?}\nkad: {}\ndirect_message: {}\nidentify: {}",
            gossipsub_prefix(discriminator),
            kad_protocol(discriminator).unwrap(),
            direct_message_protocol(discriminator).unwrap(),
            identify_protocol(discriminator),
        )
    }

    #[test]
    fn mainnet_libp2p_protocol_identifiers() {
        insta::assert_snapshot!("mainnet_libp2p_protocol_identifiers", snapshot_for(None));
    }

    #[test]
    fn decaf_libp2p_protocol_identifiers() {
        insta::assert_snapshot!(
            "decaf_libp2p_protocol_identifiers",
            snapshot_for(Some(U256::from(0xdecafu64)))
        );
    }

    #[test]
    fn should_keep_peer_empty_protocols() {
        let expected = kad_protocol(None).unwrap();
        assert!(!should_keep_peer(&expected, &[]));
    }

    #[test]
    fn should_keep_peer_discriminator_matrix() {
        let discriminators = [None, Some(U256::from(1u64)), Some(U256::from(2u64))];
        for ours in discriminators {
            for theirs in discriminators {
                let expected = kad_protocol(ours).unwrap();
                let peer_protocols = [
                    direct_message_protocol(theirs).unwrap(),
                    kad_protocol(theirs).unwrap(),
                ];
                assert_eq!(
                    should_keep_peer(&expected, &peer_protocols),
                    ours == theirs,
                    "ours={ours:?} theirs={theirs:?}",
                );
            }
        }
    }

    /// Cross-network peer must not persist in a's Kademlia routing table after identify-triggered
    /// disconnect.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn evicts_cross_network_peer_from_routing_table() {
        tokio::time::timeout(Duration::from_secs(90), async {
            let a = spawn_node(Some(U256::from(1u64)), HashSet::new(), 0).await;

            let hub = HashSet::from([(a.peer_id(), a.listen_addr())]);
            let b = spawn_node(Some(U256::from(1u64)), hub.clone(), 1).await;
            let x = spawn_node(Some(U256::from(2u64)), hub, 2).await;

            b.begin_bootstrap().expect("begin_bootstrap b");
            x.begin_bootstrap().expect("begin_bootstrap x");

            let b_pid = b.peer_id();
            let x_pid = x.peer_id();

            let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
            let mut consecutive = 0usize;
            loop {
                assert!(
                    tokio::time::Instant::now() < deadline,
                    "steady state not reached",
                );
                tokio::time::sleep(Duration::from_secs(1)).await;
                let kad_peers = a.kad_routing_peers().await.expect("kad_routing_peers");
                if kad_peers.contains(&b_pid) && !kad_peers.contains(&x_pid) {
                    consecutive += 1;
                    if consecutive >= 3 {
                        break;
                    }
                } else {
                    consecutive = 0;
                }
            }

            x.shutdown().await.expect("shutdown x");
            b.shutdown().await.expect("shutdown b");
            a.shutdown().await.expect("shutdown a");
        })
        .await
        .expect("test timed out");
    }

    /// Hub `a` keeps same-network peer `b` and drops foreign peer `x`; proves selective,
    /// not wholesale, disconnection.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn drops_cross_network_peer() {
        tokio::time::timeout(Duration::from_secs(90), async {
            let a = spawn_node(Some(U256::from(1u64)), HashSet::new(), 0).await;

            let hub = HashSet::from([(a.peer_id(), a.listen_addr())]);
            let b = spawn_node(Some(U256::from(1u64)), hub.clone(), 1).await;
            let x = spawn_node(Some(U256::from(2u64)), hub, 2).await;

            b.begin_bootstrap().expect("begin_bootstrap b");
            x.begin_bootstrap().expect("begin_bootstrap x");

            let b_pid = b.peer_id();
            let expected = HashSet::from([b_pid]);

            let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
            // Require N consecutive 1s-apart checks to avoid transient matches.
            let mut consecutive = 0usize;
            loop {
                assert!(
                    tokio::time::Instant::now() < deadline,
                    "steady state not reached; a.connected_pids()={:?}",
                    a.connected_pids().await,
                );
                tokio::time::sleep(Duration::from_secs(1)).await;
                let a_pids = a.connected_pids().await.expect("connected_pids");
                let x_conn = x.num_connected().await.expect("num_connected");
                if a_pids == expected && x_conn == 0 {
                    consecutive += 1;
                    if consecutive >= 3 {
                        break;
                    }
                } else {
                    consecutive = 0;
                }
            }

            x.shutdown().await.expect("shutdown x");
            b.shutdown().await.expect("shutdown b");
            a.shutdown().await.expect("shutdown a");
        })
        .await
        .expect("test timed out");
    }

    #[test]
    fn put_quorum_override() {
        let nz = |n| NonZeroUsize::new(n).unwrap();

        // default path: quorum = replication_factor / 2
        assert_eq!(resolve_put_quorum(None, nz(20)), nz(10));
        assert_eq!(resolve_put_quorum(None, nz(2)), nz(1));

        // override below default
        assert_eq!(resolve_put_quorum(Some(nz(1)), nz(20)), nz(1));
        // override independent of replication factor
        assert_eq!(resolve_put_quorum(Some(nz(7)), nz(20)), nz(7));
        // resolver does not clamp; kad caps separately at min(n, rf)
        assert_eq!(resolve_put_quorum(Some(nz(50)), nz(20)), nz(50));
    }
}