reth_discv5/

lib.rs

//! Wrapper around [`discv5::Discv5`].

#![doc(
    html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
    html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
    issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
)]
#![cfg_attr(not(test), warn(unused_crate_dependencies))]
#![cfg_attr(docsrs, feature(doc_cfg))]

use std::{
    collections::HashSet,
    fmt,
    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
    sync::Arc,
    time::Duration,
};

use alloy_primitives::bytes::Bytes;
use enr::{discv4_id_to_discv5_id, EnrCombinedKeyWrapper};
use futures::future::join_all;
use itertools::Itertools;
use rand::{Rng, RngCore};
use reth_ethereum_forks::{EnrForkIdEntry, ForkId};
use reth_network_peers::{NodeRecord, PeerId};
use secp256k1::SecretKey;
use tokio::{sync::mpsc, task};
use tracing::{debug, error, trace};

pub mod config;
pub mod enr;
pub mod error;
pub mod filter;
pub mod metrics;
pub mod network_stack_id;

pub use ::enr::{CombinedKey, Enr, EnrKey, NodeId};
pub use discv5::{self, IpMode};
pub use libp2p_identity;

pub use config::{
    BootNode, Config, ConfigBuilder, DEFAULT_COUNT_BOOTSTRAP_LOOKUPS, DEFAULT_DISCOVERY_V5_ADDR,
    DEFAULT_DISCOVERY_V5_ADDR_IPV6, DEFAULT_DISCOVERY_V5_LISTEN_CONFIG, DEFAULT_DISCOVERY_V5_PORT,
    DEFAULT_SECONDS_BOOTSTRAP_LOOKUP_INTERVAL, DEFAULT_SECONDS_LOOKUP_INTERVAL,
};
pub use enr::enr_to_discv4_id;
pub use error::Error;
pub use filter::{FilterOutcome, MustNotIncludeKeys};
pub use network_stack_id::NetworkStackId;

use metrics::{DiscoveredPeersMetrics, Discv5Metrics};

/// Max kbucket index is 255.
///
/// This is the max log2distance for 32 byte [`NodeId`] - 1. See <https://github.com/sigp/discv5/blob/e9e0d4f93ec35591832a9a8d937b4161127da87b/src/kbucket.rs#L586-L587>.
pub const MAX_KBUCKET_INDEX: usize = 255;

/// Default lowest kbucket index to attempt filling, in periodic look up query to populate kbuckets.
///
/// The peer at the 0th kbucket index is at log2distance 1 from the local node ID. See <https://github.com/sigp/discv5/blob/e9e0d4f93ec35591832a9a8d937b4161127da87b/src/kbucket.rs#L586-L587>.
///
/// Default is 0th index.
pub const DEFAULT_MIN_TARGET_KBUCKET_INDEX: usize = 0;

/// Transparent wrapper around [`discv5::Discv5`].
#[derive(Clone)]
pub struct Discv5 {
    /// sigp/discv5 node.
    discv5: Arc<discv5::Discv5>,
    /// [`IpMode`] of the `RLPx` network.
    rlpx_ip_mode: IpMode,
    /// Key used in kv-pair to ID chain, e.g. 'opstack' or 'eth'.
    fork_key: Option<&'static [u8]>,
    /// Filter applied to a discovered peers before passing it up to app.
    discovered_peer_filter: MustNotIncludeKeys,
    /// Metrics for underlying [`discv5::Discv5`] node and filtered discovered peers.
    metrics: Discv5Metrics,
    /// Returns the _local_ [`NodeRecord`] this service was started with.
    // Note: we must track this separately because the `discv5::Discv5` does not necessarily
    // provide this via its [`local_enr`](discv5::Discv5::local_enr()). This is intended for
    // obtaining the port this service was launched at
    local_node_record: NodeRecord,
}

impl Discv5 {
    ////////////////////////////////////////////////////////////////////////////////////////////////
    // Minimal interface with `reth_network::discovery`
    ////////////////////////////////////////////////////////////////////////////////////////////////

    /// Adds the node to the table, if it is not already present.
    pub fn add_node(&self, node_record: Enr<SecretKey>) -> Result<(), Error> {
        let EnrCombinedKeyWrapper(enr) = node_record.try_into().map_err(Error::InvalidEnr)?;
        self.discv5.add_enr(enr).map_err(Error::AddNodeFailed)
    }

    /// Sets the pair in the EIP-868 [`Enr`] of the node.
    ///
    /// If the key already exists, this will update it.
    ///
    /// CAUTION: The value **must** be rlp encoded
    pub fn set_eip868_in_local_enr(&self, key: Vec<u8>, rlp: Bytes) {
        let Ok(key_str) = std::str::from_utf8(&key) else {
            error!(target: "net::discv5",
                err="key not utf-8",
                "failed to update local enr"
            );
            return
        };
        if let Err(err) = self.discv5.enr_insert(key_str, &rlp) {
            error!(target: "net::discv5",
                %err,
                "failed to update local enr"
            );
        }
    }

    /// Sets the pair in the EIP-868 [`Enr`] of the node.
    ///
    /// If the key already exists, this will update it.
    pub fn encode_and_set_eip868_in_local_enr(
        &self,
        key: Vec<u8>,
        value: impl alloy_rlp::Encodable,
    ) {
        let mut buf = Vec::new();
        value.encode(&mut buf);
        self.set_eip868_in_local_enr(key, buf.into())
    }

    /// Adds the peer and id to the ban list.
    ///
    /// This will prevent any future inclusion in the table
    pub fn ban(&self, peer_id: PeerId, ip: IpAddr) {
        match discv4_id_to_discv5_id(peer_id) {
            Ok(node_id) => {
                self.discv5.ban_node(&node_id, None);
                self.ban_ip(ip);
            }
            Err(err) => error!(target: "net::discv5",
                %err,
                "failed to ban peer"
            ),
        }
    }

    /// Adds the ip to the ban list.
    ///
    /// This will prevent any future inclusion in the table
    pub fn ban_ip(&self, ip: IpAddr) {
        self.discv5.ban_ip(ip, None);
    }

    /// Returns the [`NodeRecord`] of the local node.
    ///
    /// This includes the currently tracked external IP address of the node.
    ///
    /// Returns `None` if the local ENR does not contain the required fields.
    pub fn node_record(&self) -> Option<NodeRecord> {
        let enr: Enr<_> = EnrCombinedKeyWrapper(self.discv5.local_enr()).into();
        enr.try_into().ok()
    }

    /// Returns the local [`Enr`] of the service.
    pub fn local_enr(&self) -> Enr<CombinedKey> {
        self.discv5.local_enr()
    }

    /// The port the discv5 service is listening on.
    pub const fn local_port(&self) -> u16 {
        self.local_node_record.udp_port
    }

    /// Spawns [`discv5::Discv5`]. Returns [`discv5::Discv5`] handle in reth compatible wrapper type
    /// [`Discv5`], a receiver of [`discv5::Event`]s from the underlying node, and the local
    /// [`Enr`](discv5::Enr) converted into the reth compatible [`NodeRecord`] type.
    pub async fn start(
        sk: &SecretKey,
        discv5_config: Config,
    ) -> Result<(Self, mpsc::Receiver<discv5::Event>), Error> {
        //
        // 1. make local enr from listen config
        //
        let (enr, local_node_record, fork_key, rlpx_ip_mode) = build_local_enr(sk, &discv5_config);

        trace!(target: "net::discv5", ?enr, "local ENR");

        //
        // 2. start discv5
        //
        let Config {
            discv5_config,
            bootstrap_nodes,
            lookup_interval,
            bootstrap_lookup_interval,
            bootstrap_lookup_countdown,
            discovered_peer_filter,
            ..
        } = discv5_config;

        let EnrCombinedKeyWrapper(enr) = enr.try_into().map_err(Error::InvalidEnr)?;
        let sk = CombinedKey::secp256k1_from_bytes(&mut sk.to_secret_bytes()).unwrap();
        let mut discv5 = match discv5::Discv5::new(enr, sk, discv5_config) {
            Ok(discv5) => discv5,
            Err(err) => return Err(Error::InitFailure(err)),
        };
        discv5.start().await.map_err(Error::Discv5Error)?;

        // start discv5 updates stream
        let discv5_updates = discv5.event_stream().await.map_err(Error::Discv5Error)?;

        let discv5 = Arc::new(discv5);

        //
        // 3. add boot nodes
        //
        bootstrap(bootstrap_nodes, &discv5).await?;

        let metrics = Discv5Metrics::default();

        //
        // 4. start bg kbuckets maintenance
        //
        spawn_populate_kbuckets_bg(
            lookup_interval,
            bootstrap_lookup_interval,
            bootstrap_lookup_countdown,
            metrics.clone(),
            Arc::downgrade(&discv5),
        );

        Ok((
            Self {
                discv5,
                rlpx_ip_mode,
                fork_key,
                discovered_peer_filter,
                metrics,
                local_node_record,
            },
            discv5_updates,
        ))
    }

    /// Process an event from the underlying [`discv5::Discv5`] node.
    pub fn on_discv5_update(&self, update: discv5::Event) -> Option<DiscoveredPeer> {
        #[expect(clippy::match_same_arms)]
        match update {
            discv5::Event::SocketUpdated(_) | discv5::Event::TalkRequest(_) |
            // `Discovered` not unique discovered peers
            discv5::Event::Discovered(_) |
            // Unrecognized frames are handled separately by the discovery layer
            discv5::Event::UnrecognizedFrame(_) => None,
            discv5::Event::NodeInserted { .. } => {

                // node has been inserted into kbuckets

                // `replaced` partly covers `reth_discv4::DiscoveryUpdate::Removed(_)`

                self.metrics.discovered_peers.increment_kbucket_insertions(1);

                None
            }
            discv5::Event::SessionEstablished(enr, remote_socket) => {
                // this branch is semantically similar to branches of
                // `reth_discv4::DiscoveryUpdate`: `DiscoveryUpdate::Added(_)` and
                // `DiscoveryUpdate::DiscoveredAtCapacity(_)

                // peer has been discovered as part of query, or, by incoming session (peer has
                // discovered us)

                self.metrics.discovered_peers.increment_established_sessions_raw(1);

                self.on_discovered_peer(&enr, remote_socket)
            }
            discv5::Event::UnverifiableEnr {
                enr,
                socket,
                node_id: _,
            } => {
                // this branch is semantically similar to branches of
                // `reth_discv4::DiscoveryUpdate`: `DiscoveryUpdate::Added(_)` and
                // `DiscoveryUpdate::DiscoveredAtCapacity(_)

                // peer has been discovered as part of query, or, by an outgoing session (but peer
                // is behind NAT and responds from a different socket)

                // NOTE: `discv5::Discv5` won't initiate a session with any peer with an
                // unverifiable node record, for example one that advertises a reserved LAN IP
                // address on a WAN network. This is in order to prevent DoS attacks, where some
                // malicious peers may advertise a victim's socket. We will still try and connect
                // to them over RLPx, to be compatible with EL discv5 implementations that don't
                // enforce this security measure.

                trace!(target: "net::discv5",
                    ?enr,
                    %socket,
                    "discovered unverifiable enr, source socket doesn't match socket advertised in ENR"
                );

                self.metrics.discovered_peers.increment_unverifiable_enrs_raw_total(1);

                self.on_discovered_peer(&enr, socket)
            }
            _ => None
        }
    }

    /// Processes a discovered peer. Returns `true` if peer is added to
    pub fn on_discovered_peer(
        &self,
        enr: &discv5::Enr,
        socket: SocketAddr,
    ) -> Option<DiscoveredPeer> {
        self.metrics.discovered_peers_advertised_networks.increment_once_by_network_type(enr);

        let node_record = match self.try_into_reachable(enr, socket) {
            Ok(enr_bc) => enr_bc,
            Err(err) => {
                trace!(target: "net::discv5",
                    %err,
                    ?enr,
                    "discovered peer is unreachable"
                );

                self.metrics.discovered_peers.increment_established_sessions_unreachable_enr(1);

                return None
            }
        };
        if let FilterOutcome::Ignore { reason } = self.filter_discovered_peer(enr) {
            trace!(target: "net::discv5",
                ?enr,
                reason,
                "filtered out discovered peer"
            );

            self.metrics.discovered_peers.increment_established_sessions_filtered(1);

            return None
        }

        let fork_id = self.get_fork_id(enr).ok();

        trace!(target: "net::discv5",
            ?fork_id,
            ?enr,
            "discovered peer"
        );

        Some(DiscoveredPeer { node_record, fork_id })
    }

    /// Tries to recover an unreachable [`Enr`](discv5::Enr) received via
    /// [`discv5::Event::UnverifiableEnr`], into a [`NodeRecord`] usable by `RLPx`.
    ///
    /// NOTE: Fallback solution to be compatible with Geth which includes peers into the discv5
    /// WAN topology which, for example, advertise in their ENR that localhost is their UDP IP
    /// address. These peers are only discovered if they initiate a connection attempt, and we by
    /// such means learn their reachable IP address. If we receive their ENR from any other peer
    /// as part of a lookup query, we won't find a reachable IP address on which to dial them by
    /// reading their ENR.
    pub fn try_into_reachable(
        &self,
        enr: &discv5::Enr,
        socket: SocketAddr,
    ) -> Result<NodeRecord, Error> {
        // Use the sender's address as received so discovery IP bans match source sockets.
        let address = socket.ip();
        let udp_port = socket.port();

        let id = enr_to_discv4_id(enr).ok_or(Error::IncompatibleKeyType)?;

        // The port must pair with the session socket's address, so select the
        // ENR field by the socket's address family, not by the local node's
        // RLPx listen family: an IPv4 session dials the `tcp` port, an IPv6
        // session the `tcp6` port (falling back to `tcp`).
        // Mapped IPv4 sessions use `tcp`, while the record retains the original socket address.
        let tcp_port = (match address.to_canonical() {
            IpAddr::V4(_) => enr.tcp4(),
            IpAddr::V6(_) => enr.tcp6().or_else(|| enr.tcp4()),
        })
        .unwrap_or(
            // tcp socket is missing from ENR for the session's address family.
            //
            // by default geth runs discv5 and discv4 behind the same udp port (the discv4 default
            // port 30303), so rlpx has a chance of successfully dialing the peer on its discv5
            // udp port if its running geth's p2p code.
            udp_port,
        );

        Ok(NodeRecord { address, tcp_port, udp_port, id })
    }

    /// Applies filtering rules on an ENR. Returns [`Ok`](FilterOutcome::Ok) if peer should be
    /// passed up to app, and [`Ignore`](FilterOutcome::Ignore) if peer should instead be dropped.
    pub fn filter_discovered_peer(&self, enr: &discv5::Enr) -> FilterOutcome {
        self.discovered_peer_filter.filter(enr)
    }

    /// Returns the [`ForkId`] of the given [`Enr`](discv5::Enr) w.r.t. the local node's network
    /// stack, if field is set.
    pub fn get_fork_id<K: EnrKey>(&self, enr: &Enr<K>) -> Result<ForkId, Error> {
        let Some(key) = self.fork_key else { return Err(Error::NetworkStackIdNotConfigured) };
        let fork_id = enr
            .get_decodable::<EnrForkIdEntry>(key)
            .or_else(|| {
                (key != NetworkStackId::ETH)
                    .then(|| {
                        // Fallback: trying to get fork id from Enr with 'eth' as network stack id
                        trace!(target: "net::discv5",
                            key = %String::from_utf8_lossy(key),
                            "Fork id not found for key, trying 'eth'..."
                        );
                        enr.get_decodable::<EnrForkIdEntry>(NetworkStackId::ETH)
                    })
                    .flatten()
            })
            .ok_or({
                trace!(target: "net::discv5", "Fork id not found for 'eth' network stack id");
                Error::ForkMissing(key)
            })?
            .map(Into::into)?;

        Ok(fork_id)
    }

    ////////////////////////////////////////////////////////////////////////////////////////////////
    // Interface with sigp/discv5
    ////////////////////////////////////////////////////////////////////////////////////////////////

    /// Exposes API of [`discv5::Discv5`].
    pub fn with_discv5<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&discv5::Discv5) -> R,
    {
        f(&self.discv5)
    }

    ////////////////////////////////////////////////////////////////////////////////////////////////
    // Complementary
    ////////////////////////////////////////////////////////////////////////////////////////////////

    /// Returns the `RLPx` [`IpMode`] of the local node.
    pub const fn ip_mode(&self) -> IpMode {
        self.rlpx_ip_mode
    }

    /// Returns the key to use to identify the [`ForkId`] kv-pair on the [`Enr`](discv5::Enr).
    pub const fn fork_key(&self) -> Option<&[u8]> {
        self.fork_key
    }
}

impl fmt::Debug for Discv5 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        "{ .. }".fmt(f)
    }
}

/// Result of successfully processing a peer discovered by [`discv5::Discv5`].
#[derive(Debug)]
pub struct DiscoveredPeer {
    /// A discovery v4 backwards compatible ENR.
    pub node_record: NodeRecord,
    /// [`ForkId`] extracted from ENR w.r.t. configured
    pub fork_id: Option<ForkId>,
}

/// Builds the local ENR with the supplied key.
pub fn build_local_enr(
    sk: &SecretKey,
    config: &Config,
) -> (Enr<SecretKey>, NodeRecord, Option<&'static [u8]>, IpMode) {
    let mut builder = Enr::builder();

    let Config {
        discv5_config,
        fork,
        tcp_socket,
        advertised_ipv4,
        advertised_ipv6,
        other_enr_kv_pairs,
        ..
    } = config;

    let socket = {
        let v4 = crate::config::ipv4(&discv5_config.listen_config);
        let v6 = crate::config::ipv6(&discv5_config.listen_config);

        // Prefer an explicit advertised IP for ENR IP fields. Listen sockets still supply UDP
        // ports and determine which address-family fields are emitted.
        if let Some(addr) = v4 {
            if let Some(ip) = advertised_ipv4 {
                builder.ip4(*ip);
            } else if *addr.ip() != Ipv4Addr::UNSPECIFIED {
                builder.ip4(*addr.ip());
            }
            builder.udp4(addr.port());
        }
        if let Some(addr) = v6 {
            if let Some(ip) = advertised_ipv6 {
                builder.ip6(*ip);
            } else if *addr.ip() != Ipv6Addr::UNSPECIFIED {
                builder.ip6(*addr.ip());
            }
            builder.udp6(addr.port());
        }
        // Advertise tcp4 when v4 is configured, else tcp6.
        if v4.is_some() {
            builder.tcp4(tcp_socket.port());
        } else if v6.is_some() {
            builder.tcp6(tcp_socket.port());
        }

        // Prefer v6 when both are configured
        v6.map(SocketAddr::V6)
            .or_else(|| v4.map(SocketAddr::V4))
            .unwrap_or_else(|| SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)))
    };

    let rlpx_ip_mode = if tcp_socket.is_ipv4() { IpMode::Ip4 } else { IpMode::Ip6 };

    // identifies which network node is on
    let network_stack_id = fork.as_ref().map(|(network_stack_id, fork_value)| {
        builder.add_value_rlp(network_stack_id, alloy_rlp::encode(fork_value).into());
        *network_stack_id
    });

    // add other data
    for (key, value) in other_enr_kv_pairs {
        builder.add_value_rlp(key, value.clone().into());
    }

    // enr v4 not to get confused with discv4, independent versioning enr and
    // discovery
    let enr = builder.build(sk).expect("should build enr v4");

    // backwards compatible enr
    let bc_enr = NodeRecord::from_secret_key(socket, sk);

    (enr, bc_enr, network_stack_id, rlpx_ip_mode)
}

/// Bootstraps underlying [`discv5::Discv5`] node with configured peers.
///
/// Boot nodes given as an ENR are added to the kbuckets right away. Boot nodes given as an enode
/// need their ENR requested over the wire first, which is driven in the background: an unreachable
/// boot node only answers after the discv5 request timeout, and waiting for that would stall node
/// startup.
///
/// Stays `async` because spawning those requests requires a tokio runtime context.
pub async fn bootstrap(
    bootstrap_nodes: HashSet<BootNode>,
    discv5: &Arc<discv5::Discv5>,
) -> Result<(), Error> {
    trace!(target: "net::discv5",
        ?bootstrap_nodes,
        "adding bootstrap nodes .."
    );

    let mut enr_requests = vec![];
    for node in bootstrap_nodes {
        match node {
            BootNode::Enr(node) => {
                if let Err(err) = discv5.add_enr(node) {
                    return Err(Error::AddNodeFailed(err))
                }
            }
            BootNode::Enode(enode) => {
                let discv5 = discv5.clone();
                enr_requests.push(async move {
                    if let Err(err) = discv5.request_enr(enode.to_string()).await {
                        debug!(target: "net::discv5",
                            ?enode,
                            %err,
                            "failed adding boot node"
                        );
                    }
                })
            }
        }
    }

    // If a session is established, the ENR is added straight away to discv5 kbuckets
    if !enr_requests.is_empty() {
        task::spawn(join_all(enr_requests));
    }

    Ok(())
}

/// Backgrounds regular look up queries, in order to keep kbuckets populated.
pub fn spawn_populate_kbuckets_bg(
    lookup_interval: u64,
    bootstrap_lookup_interval: u64,
    bootstrap_lookup_countdown: u64,
    metrics: Discv5Metrics,
    discv5: std::sync::Weak<discv5::Discv5>,
) {
    let lookup_interval = Duration::from_secs(lookup_interval);
    let metrics = metrics.discovered_peers;
    let mut kbucket_index = MAX_KBUCKET_INDEX;
    let pulse_lookup_interval = Duration::from_secs(bootstrap_lookup_interval);
    task::spawn(async move {
        let Some(discv5_handle) = discv5.upgrade() else {
            return;
        };
        let local_node_id = discv5_handle.local_enr().node_id();
        drop(discv5_handle);

        // make many fast lookup queries at bootstrap, trying to fill kbuckets at furthest
        // log2distance from local node
        for i in (0..bootstrap_lookup_countdown).rev() {
            let target = NodeId::random();

            trace!(target: "net::discv5",
                %target,
                bootstrap_boost_runs_countdown=i,
                lookup_interval=format!("{:#?}", pulse_lookup_interval),
                "starting bootstrap boost lookup query"
            );

            {
                let Some(discv5_handle) = discv5.upgrade() else {
                    return;
                };
                lookup(target, &discv5_handle, &metrics).await;
            }

            tokio::time::sleep(pulse_lookup_interval).await;
        }

        // initiate regular lookups to populate kbuckets
        loop {
            // make sure node is connected to each subtree in the network by target
            // selection (ref kademlia)
            let target = get_lookup_target(kbucket_index, local_node_id);

            trace!(target: "net::discv5",
                %target,
                lookup_interval=format!("{:#?}", lookup_interval),
                "starting periodic lookup query"
            );

            {
                let Some(discv5_handle) = discv5.upgrade() else {
                    return;
                };
                lookup(target, &discv5_handle, &metrics).await;
            }

            if kbucket_index > DEFAULT_MIN_TARGET_KBUCKET_INDEX {
                // try to populate bucket one step closer
                kbucket_index -= 1
            } else {
                // start over with bucket furthest away
                kbucket_index = MAX_KBUCKET_INDEX
            }

            tokio::time::sleep(lookup_interval).await;
        }
    });
}

/// Gets the next lookup target, based on which bucket is currently being targeted.
pub fn get_lookup_target(kbucket_index: usize, local_node_id: NodeId) -> NodeId {
    // init target
    let mut target = local_node_id.raw();

    // make sure target has a 'log2distance'-long suffix that differs from local node id
    let bit_offset = MAX_KBUCKET_INDEX.saturating_sub(kbucket_index);
    let (byte, bit) = (bit_offset / 8, bit_offset % 8);
    // Flip the target bit.
    target[byte] ^= 1 << (7 - bit);

    // Randomize the bits after the target.
    let mut rng = rand::rng();
    // Randomize remaining bits in the byte we modified.
    if bit < 7 {
        // Compute the mask of the bits that need to be randomized.
        let bits_to_randomize = 0xff >> (bit + 1);
        // Clear.
        target[byte] &= !bits_to_randomize;
        // Randomize.
        target[byte] |= rng.random::<u8>() & bits_to_randomize;
    }
    // Randomize remaining bytes.
    rng.fill_bytes(&mut target[byte + 1..]);

    target.into()
}

/// Runs a [`discv5::Discv5`] lookup query.
pub async fn lookup(target: NodeId, discv5: &discv5::Discv5, metrics: &DiscoveredPeersMetrics) {
    metrics.set_total_sessions(discv5::Discv5::metrics().active_sessions);
    metrics.set_total_kbucket_peers(
        discv5.with_kbuckets(|kbuckets| kbuckets.read().iter_ref().count()),
    );

    match discv5.find_node(target).await {
        Err(err) => trace!(target: "net::discv5",
            %err,
            "lookup query failed"
        ),
        Ok(peers) => trace!(target: "net::discv5",
            target=format!("{:#?}", target),
            peers_count=peers.len(),
            peers=format!("[{:#}]", peers.iter()
                .map(|enr| enr.node_id()
            ).format(", ")),
            "peers returned by lookup query"
        ),
    }

    // `Discv5::connected_peers` can be subset of sessions, not all peers make it
    // into kbuckets, e.g. incoming sessions from peers with
    // unreachable enrs
    debug!(target: "net::discv5",
        connected_peers=discv5.connected_peers(),
        "connected peers in routing table"
    );
}

#[cfg(test)]
mod test {
    #![allow(deprecated)]
    use super::*;
    use ::enr::{CombinedKey, EnrKey};
    use discv5::ListenConfig;
    use rand::rng;
    use reth_chainspec::MAINNET;
    use std::{
        net::UdpSocket,
        time::{Duration, Instant},
    };
    use tracing::trace;

    fn discv5_noop() -> Discv5 {
        let sk = CombinedKey::generate_secp256k1();
        Discv5 {
            discv5: Arc::new(
                discv5::Discv5::new(
                    Enr::empty(&sk).unwrap(),
                    sk,
                    discv5::ConfigBuilder::new(DEFAULT_DISCOVERY_V5_LISTEN_CONFIG).build(),
                )
                .unwrap(),
            ),
            rlpx_ip_mode: IpMode::Ip4,
            fork_key: None,
            discovered_peer_filter: MustNotIncludeKeys::default(),
            metrics: Discv5Metrics::default(),
            local_node_record: NodeRecord::new(
                (Ipv4Addr::LOCALHOST, 30303).into(),
                PeerId::random(),
            ),
        }
    }

    async fn start_discovery_node(udp_port_discv5: u16) -> (Discv5, mpsc::Receiver<discv5::Event>) {
        let secret_key = SecretKey::new(&mut rng());

        let discv5_addr: SocketAddr = format!("127.0.0.1:{udp_port_discv5}").parse().unwrap();
        let rlpx_addr: SocketAddr = "127.0.0.1:30303".parse().unwrap();

        let discv5_listen_config = ListenConfig::from(discv5_addr);
        let discv5_config = Config::builder(rlpx_addr)
            .discv5_config(discv5::ConfigBuilder::new(discv5_listen_config).build())
            .build();

        Discv5::start(&secret_key, discv5_config).await.expect("should build discv5")
    }

    async fn start_discovery_node_with_key(
        secret_key: &SecretKey,
        udp_port_discv5: u16,
    ) -> Result<(Discv5, mpsc::Receiver<discv5::Event>), Error> {
        let discv5_addr: SocketAddr = format!("127.0.0.1:{udp_port_discv5}").parse().unwrap();
        let rlpx_addr: SocketAddr = "127.0.0.1:30303".parse().unwrap();

        let discv5_listen_config = ListenConfig::from(discv5_addr);
        let discv5_config = Config::builder(rlpx_addr)
            .discv5_config(discv5::ConfigBuilder::new(discv5_listen_config).build())
            .build();

        Discv5::start(secret_key, discv5_config).await
    }

    fn unused_udp_port() -> u16 {
        UdpSocket::bind("127.0.0.1:0").unwrap().local_addr().unwrap().port()
    }

    async fn wait_for_udp_port_release(port: u16, timeout: Duration) {
        let deadline = Instant::now() + timeout;

        loop {
            match UdpSocket::bind(("127.0.0.1", port)) {
                Ok(socket) => {
                    drop(socket);
                    return;
                }
                Err(err) if Instant::now() < deadline => {
                    trace!(target: "net::discv5::test", %port, %err, "waiting for discv5 port release");
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                Err(err) => panic!("discv5 did not release port {port} before timeout: {err}"),
            }
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn discv5_releases_port_on_drop() {
        reth_tracing::init_test_tracing();

        let secret_key = SecretKey::new(&mut rng());
        let port = unused_udp_port();

        let (node, updates) =
            start_discovery_node_with_key(&secret_key, port).await.expect("should start discv5");
        drop(updates);
        drop(node);

        wait_for_udp_port_release(port, Duration::from_secs(1)).await;

        let restarted = start_discovery_node_with_key(&secret_key, port).await;
        assert!(restarted.is_ok(), "discv5 failed to rebind dropped port: {restarted:?}");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn discv5() {
        reth_tracing::init_test_tracing();

        // rig test

        // rig node_1
        let (node_1, mut stream_1) = start_discovery_node(30344).await;
        let node_1_enr = node_1.with_discv5(|discv5| discv5.local_enr());

        // rig node_2
        let (node_2, mut stream_2) = start_discovery_node(30355).await;
        let node_2_enr = node_2.with_discv5(|discv5| discv5.local_enr());

        trace!(target: "net::discv5::test",
            node_1_node_id=format!("{:#}", node_1_enr.node_id()),
            node_2_node_id=format!("{:#}", node_2_enr.node_id()),
            "started nodes"
        );

        // test

        // add node_2 to discovery handle of node_1 (should add node to discv5 kbuckets)
        let node_2_enr_reth_compatible_ty: Enr<SecretKey> =
            EnrCombinedKeyWrapper(node_2_enr.clone()).into();
        node_1.add_node(node_2_enr_reth_compatible_ty).unwrap();

        // verify node_2 is in KBuckets of node_1:discv5
        assert!(
            node_1.with_discv5(|discv5| discv5.table_entries_id().contains(&node_2_enr.node_id()))
        );

        // manually trigger connection from node_1 to node_2
        node_1.with_discv5(|discv5| discv5.send_ping(node_2_enr.clone())).await.unwrap();

        // verify node_1:discv5 is connected to node_2:discv5 and vv
        let event_1_v5 = stream_1.recv().await.unwrap();

        assert!(matches!(
            event_1_v5,
            discv5::Event::SessionEstablished(node, socket) if node == node_2_enr && socket == node_2_enr.udp4_socket().unwrap().into()
        ));

        // verify node_1 is in KBuckets of node_2:discv5
        let event_2_v5 = stream_2.recv().await.unwrap();
        assert!(matches!(
            event_2_v5,
            discv5::Event::NodeInserted { node_id, replaced } if node_id == node_1_enr.node_id() && replaced.is_none()
        ));
    }

    #[test]
    fn discovered_enr_disc_socket_missing() {
        reth_tracing::init_test_tracing();

        // rig test
        const REMOTE_RLPX_PORT: u16 = 30303;
        let remote_socket = "104.28.44.25:9000".parse().unwrap();
        let remote_key = CombinedKey::generate_secp256k1();
        let remote_enr = Enr::builder().tcp4(REMOTE_RLPX_PORT).build(&remote_key).unwrap();

        let discv5 = discv5_noop();

        // test
        let filtered_peer = discv5.on_discovered_peer(&remote_enr, remote_socket);

        assert_eq!(
            NodeRecord {
                address: remote_socket.ip(),
                udp_port: remote_socket.port(),
                tcp_port: REMOTE_RLPX_PORT,
                id: enr_to_discv4_id(&remote_enr).unwrap(),
            },
            filtered_peer.unwrap().node_record
        )
    }

    #[test]
    fn reachable_tcp_port_follows_session_address_family() {
        // Regression test for https://github.com/paradigmxyz/reth/issues/27734:
        // the TCP port paired with the session socket's IP must come from the
        // ENR field of the *session's* address family, not the local node's
        // RLPx listen family.
        const TCP4_PORT: u16 = 30304;
        const TCP6_PORT: u16 = 30306;
        let remote_key = CombinedKey::generate_secp256k1();
        let remote_enr = Enr::builder().tcp4(TCP4_PORT).tcp6(TCP6_PORT).build(&remote_key).unwrap();

        let v4_socket: SocketAddr = "127.0.0.1:30305".parse().unwrap();
        let v6_socket: SocketAddr = "[::1]:30307".parse().unwrap();

        for local_mode in [IpMode::Ip4, IpMode::Ip6] {
            let mut discv5 = discv5_noop();
            discv5.rlpx_ip_mode = local_mode;

            let record = discv5.try_into_reachable(&remote_enr, v4_socket).unwrap();
            assert_eq!(
                record.tcp_port, TCP4_PORT,
                "v4 session with local mode {local_mode:?} must dial the tcp4 port"
            );
            assert_eq!(record.address, v4_socket.ip());
            assert_eq!(record.udp_port, v4_socket.port());

            let record = discv5.try_into_reachable(&remote_enr, v6_socket).unwrap();
            assert_eq!(
                record.tcp_port, TCP6_PORT,
                "v6 session with local mode {local_mode:?} must dial the tcp6 port"
            );
            assert_eq!(record.address, v6_socket.ip());
            assert_eq!(record.udp_port, v6_socket.port());
        }
    }

    #[test]
    fn reachable_ipv4_mapped_session_uses_tcp4() {
        const TCP4_PORT: u16 = 30304;
        const TCP6_PORT: u16 = 30306;
        let remote_key = CombinedKey::generate_secp256k1();
        let remote_enr = Enr::builder().tcp4(TCP4_PORT).tcp6(TCP6_PORT).build(&remote_key).unwrap();
        let mapped_socket: SocketAddr = "[::ffff:104.28.44.25]:9000".parse().unwrap();

        for local_mode in [IpMode::Ip4, IpMode::Ip6] {
            let mut discv5 = discv5_noop();
            discv5.rlpx_ip_mode = local_mode;

            let record = discv5.try_into_reachable(&remote_enr, mapped_socket).unwrap();
            assert_eq!(record.tcp_port, TCP4_PORT);
            assert_eq!(record.address, mapped_socket.ip());
            assert_eq!(record.udp_port, mapped_socket.port());

            let peer = discv5.on_discovered_peer(&remote_enr, mapped_socket).unwrap();
            assert_eq!(peer.node_record, record);
        }
    }

    #[test]
    fn reachable_tcp_port_v6_session_falls_back_to_tcp4() {
        // A v6 session whose ENR carries only a `tcp` (v4) port uses that port
        // rather than the session's UDP port.
        const TCP4_PORT: u16 = 30304;
        let remote_key = CombinedKey::generate_secp256k1();
        let remote_enr = Enr::builder().tcp4(TCP4_PORT).build(&remote_key).unwrap();

        let v6_socket: SocketAddr = "[::1]:30307".parse().unwrap();

        let mut discv5 = discv5_noop();
        discv5.rlpx_ip_mode = IpMode::Ip6;
        let record = discv5.try_into_reachable(&remote_enr, v6_socket).unwrap();
        assert_eq!(record.tcp_port, TCP4_PORT);
    }

    #[test]
    fn reachable_tcp_port_missing_falls_back_to_udp_port() {
        // No TCP port in the ENR at all: keep the long-standing fallback to
        // the session's UDP port, for either session family.
        let remote_key = CombinedKey::generate_secp256k1();
        let remote_enr = Enr::builder().build(&remote_key).unwrap();

        let discv5 = discv5_noop();
        let v4_socket: SocketAddr = "127.0.0.1:30305".parse().unwrap();
        assert_eq!(discv5.try_into_reachable(&remote_enr, v4_socket).unwrap().tcp_port, 30305);
        let v6_socket: SocketAddr = "[::1]:30307".parse().unwrap();
        assert_eq!(discv5.try_into_reachable(&remote_enr, v6_socket).unwrap().tcp_port, 30307);
    }

    // Copied from sigp/discv5 with slight modification (U256 type)
    // <https://github.com/sigp/discv5/blob/master/src/kbucket/key.rs#L89-L101>
    #[expect(unreachable_pub)]
    #[expect(unused)]
    mod sigp {
        use alloy_primitives::U256;
        use enr::NodeId;

        /// A `Key` is a cryptographic hash, identifying both the nodes participating in
        /// the Kademlia DHT, as well as records stored in the DHT.
        ///
        /// The set of all `Key`s defines the Kademlia keyspace.
        ///
        /// `Key`s have an XOR metric as defined in the Kademlia paper, i.e. the bitwise XOR of
        /// the hash digests, interpreted as an integer. See [`Key::distance`].
        ///
        /// A `Key` preserves the preimage of type `T` of the hash function. See [`Key::preimage`].
        #[derive(Clone, Debug)]
        pub struct Key<T> {
            preimage: T,
            hash: [u8; 32],
        }

        impl<T> PartialEq for Key<T> {
            fn eq(&self, other: &Self) -> bool {
                self.hash == other.hash
            }
        }

        impl<T> Eq for Key<T> {}

        impl<TPeerId> AsRef<Self> for Key<TPeerId> {
            fn as_ref(&self) -> &Self {
                self
            }
        }

        impl<T> Key<T> {
            /// Construct a new `Key` by providing the raw 32 byte hash.
            pub const fn new_raw(preimage: T, hash: [u8; 32]) -> Self {
                Self { preimage, hash }
            }

            /// Borrows the preimage of the key.
            pub const fn preimage(&self) -> &T {
                &self.preimage
            }

            /// Converts the key into its preimage.
            pub fn into_preimage(self) -> T {
                self.preimage
            }

            /// Computes the distance of the keys according to the XOR metric.
            pub fn distance<U>(&self, other: &Key<U>) -> Distance {
                let a = U256::from_be_slice(self.hash.as_slice());
                let b = U256::from_be_slice(other.hash.as_slice());
                Distance(a ^ b)
            }

            // Used in the FINDNODE query outside of the k-bucket implementation.
            /// Computes the integer log-2 distance between two keys, assuming a 256-bit
            /// key. The output returns None if the key's are identical. The range is 1-256.
            pub fn log2_distance<U>(&self, other: &Key<U>) -> Option<u64> {
                let xor_dist = self.distance(other);
                let log_dist = (256 - xor_dist.0.leading_zeros() as u64);
                (log_dist != 0).then_some(log_dist)
            }
        }

        impl From<NodeId> for Key<NodeId> {
            fn from(node_id: NodeId) -> Self {
                Self { preimage: node_id, hash: node_id.raw() }
            }
        }

        /// A distance between two `Key`s.
        #[derive(Copy, Clone, PartialEq, Eq, Default, PartialOrd, Ord, Debug)]
        pub struct Distance(pub(super) U256);
    }

    #[test]
    fn select_lookup_target() {
        for bucket_index in 0..=MAX_KBUCKET_INDEX {
            let sk = CombinedKey::generate_secp256k1();
            let local_node_id = NodeId::from(sk.public());
            let target = get_lookup_target(bucket_index, local_node_id);

            let local_node_id = sigp::Key::from(local_node_id);
            let target = sigp::Key::from(target);

            assert_eq!(local_node_id.log2_distance(&target), Some(bucket_index as u64 + 1));
        }
    }

    #[test]
    fn build_enr_from_config() {
        const TCP_PORT: u16 = 30303;
        let fork_id = MAINNET.latest_fork_id();

        let config = Config::builder((Ipv4Addr::UNSPECIFIED, TCP_PORT).into())
            .fork(NetworkStackId::ETH, fork_id)
            .build();

        let sk = SecretKey::new(&mut rng());
        let (enr, _, _, _) = build_local_enr(&sk, &config);

        let decoded_fork_id = enr
            .get_decodable::<EnrForkIdEntry>(NetworkStackId::ETH)
            .unwrap()
            .map(Into::into)
            .unwrap();

        assert_eq!(fork_id, decoded_fork_id);
        assert_eq!(TCP_PORT, enr.tcp4().unwrap()); // listen config is defaulting to ip mode ipv4
    }

    #[test]
    fn get_fork_id_with_different_network_stack_ids() {
        let fork_id = MAINNET.latest_fork_id();
        let sk = SecretKey::new(&mut rng());

        // Test 1: ENR with OPEL fork ID, Discv5 configured for OPEL
        let enr_with_opel = Enr::builder()
            .add_value_rlp(
                NetworkStackId::OPEL,
                alloy_rlp::encode(EnrForkIdEntry::from(fork_id)).into(),
            )
            .build(&sk)
            .unwrap();

        let mut discv5 = discv5_noop();
        discv5.fork_key = Some(NetworkStackId::OPEL);
        assert_eq!(discv5.get_fork_id(&enr_with_opel).unwrap(), fork_id);

        // Test 2: ENR with ETH fork ID, Discv5 configured for OPEL (fallback to ETH)
        let enr_with_eth = Enr::builder()
            .add_value_rlp(
                NetworkStackId::ETH,
                alloy_rlp::encode(EnrForkIdEntry::from(fork_id)).into(),
            )
            .build(&sk)
            .unwrap();

        discv5.fork_key = Some(NetworkStackId::OPEL);
        assert_eq!(discv5.get_fork_id(&enr_with_eth).unwrap(), fork_id);

        // Test 3: ENR with neither OPEL nor ETH fork ID (should fail)
        let enr_without_network_stack_id = Enr::empty(&sk).unwrap();
        discv5.fork_key = Some(NetworkStackId::OPEL);
        assert!(matches!(
            discv5.get_fork_id(&enr_without_network_stack_id),
            Err(Error::ForkMissing(NetworkStackId::OPEL))
        ));

        // Test 4: discv5 without network stack id configured (should fail)
        let discv5 = discv5_noop();
        assert!(matches!(
            discv5.get_fork_id(&enr_without_network_stack_id),
            Err(Error::NetworkStackIdNotConfigured)
        ));
    }
}