urma_node/

peer_gossip.rs

//! Peer gossip and explicit dialing. Gossiped addresses are untrusted routing
//! hints; a dial proves only the remote transport identity, and the Peers probe
//! that follows it only that the peer speaks this protocol on the same chain.
use crate::{
    Error,
    peer::{
        Answer, Behaviour, CALL_TIMEOUT, Command, PeerNetwork, Pending, bind_response, request,
    },
    peer_address::{AddressKind, Dialed, is_circuit, parse, relay},
    peer_limits::{Class, LimitDenied},
    peer_links::Links,
    peer_nat::{Nat, PeerLink, Reachability},
    peer_protocol::{
        GossipPeer, Kind, PEER_ADDRESSES_MAX, PEERS_LIMIT, PEERS_MAX, Reply, Response,
    },
    source::check_cancelled,
};
use bitcoin::BlockHash;
use libp2p::{
    Multiaddr, PeerId, Swarm,
    multiaddr::Protocol,
    swarm::{ConnectionId, DialError, dial_opts::DialOpts},
};
use rand::seq::SliceRandom;
use std::{
    collections::{BTreeMap, BTreeSet, HashMap, VecDeque},
    net::{Ipv4Addr, Ipv6Addr},
    sync::{
        Mutex, MutexGuard,
        atomic::{AtomicBool, Ordering},
        mpsc::{self, RecvTimeoutError},
    },
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

/// State shared between the network thread and the owner: the gossip snapshot
/// the scheduler hands over, public addresses peers observed for us, the live
/// connections and our bound listeners.
pub(crate) struct Shared {
    local: PeerId,
    gossip: Mutex<Vec<GossipPeer>>,
    observed: Mutex<ObservedAddrs>,
    connected: Mutex<BTreeMap<PeerId, Links>>,
    identified: Mutex<VecDeque<(PeerId, Vec<Multiaddr>)>>,
    listening: Mutex<Vec<Multiaddr>>,
    pub(crate) nat: Nat,
}

pub(crate) fn guard<T>(lock: &Mutex<T>) -> MutexGuard<'_, T> {
    match lock.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

/// Public addresses outbound peers reported observing for this node. A
/// candidate is confirmed only once WITNESSES distinct peers reported it.
#[derive(Default)]
pub struct ObservedAddrs {
    candidates: Vec<(Multiaddr, BTreeSet<PeerId>)>,
    confirmed: Vec<Multiaddr>,
}

impl ObservedAddrs {
    pub const WITNESSES: usize = 2;
    const CANDIDATES_MAX: usize = 32;
    const CONFIRMED_MAX: usize = 8;

    /// Record one report; non-public or non-literal addresses are ignored.
    /// Returns the address when this report is the one that confirms it.
    pub fn observe(&mut self, address: Multiaddr, reporter: PeerId) -> Option<Multiaddr> {
        if !is_public(&address)
            || validate_gossip_address(&address).is_err()
            || self.confirmed.contains(&address)
        {
            return None;
        }
        let index = match self
            .candidates
            .iter()
            .position(|(known, _)| *known == address)
        {
            Some(index) => index,
            None => {
                if self.candidates.len() == Self::CANDIDATES_MAX {
                    self.candidates.remove(0);
                }
                self.candidates.push((address, BTreeSet::new()));
                self.candidates.len() - 1
            }
        };
        self.candidates[index].1.insert(reporter);
        if self.candidates[index].1.len() < Self::WITNESSES {
            return None;
        }
        let (address, _) = self.candidates.remove(index);
        if self.confirmed.len() == Self::CONFIRMED_MAX {
            self.confirmed.remove(0);
        }
        self.confirmed.push(address.clone());
        Some(address)
    }

    pub fn confirmed(&self) -> Vec<Multiaddr> {
        self.confirmed.clone()
    }
}

impl Shared {
    const GOSSIP_MAX: usize = 256;
    const IDENTIFIED_MAX: usize = 64;
    const ENVELOPE: usize = 1024;

    pub(crate) fn new(local: PeerId) -> Self {
        Self {
            local,
            gossip: Mutex::new(Vec::new()),
            observed: Mutex::new(ObservedAddrs::default()),
            connected: Mutex::new(BTreeMap::new()),
            identified: Mutex::new(VecDeque::new()),
            listening: Mutex::new(Vec::new()),
            nat: Nat::default(),
        }
    }

    /// Queue a connected peer's own literal public addresses (newest 64 peers).
    pub(crate) fn identified(&self, peer: PeerId, advertised: Vec<Multiaddr>) {
        let addresses: Vec<Multiaddr> = advertised
            .into_iter()
            .filter(|address| validate_gossip_address(address).is_ok() && is_public(address))
            .take(PEER_ADDRESSES_MAX)
            .collect();
        if addresses.is_empty() || peer == self.local {
            return;
        }
        let mut queue = match self.identified.lock() {
            Ok(queue) => queue,
            Err(poisoned) => poisoned.into_inner(),
        };
        queue.retain(|(known, _)| *known != peer);
        if queue.len() == Self::IDENTIFIED_MAX {
            queue.pop_front();
        }
        queue.push_back((peer, addresses));
    }

    fn drain_identified(&self) -> Vec<(PeerId, Vec<Multiaddr>)> {
        match self.identified.lock() {
            Ok(mut queue) => queue.drain(..).collect(),
            Err(poisoned) => poisoned.into_inner().drain(..).collect(),
        }
    }

    fn set_gossip(&self, mut peers: Vec<GossipPeer>) {
        peers.truncate(Self::GOSSIP_MAX);
        if let Ok(mut gossip) = self.gossip.lock() {
            *gossip = peers;
        }
    }

    fn observed(&self) -> Vec<Multiaddr> {
        match self.observed.lock() {
            Ok(observed) => observed.confirmed(),
            Err(poisoned) => poisoned.into_inner().confirmed(),
        }
    }

    pub(crate) fn is_connected(&self, peer: PeerId) -> bool {
        guard(&self.connected).contains_key(&peer)
    }

    /// Whether the peer holds a data connection we dialed.
    pub(crate) fn is_outbound(&self, peer: PeerId) -> bool {
        guard(&self.connected)
            .get(&peer)
            .is_some_and(|links| links.outbound_data > 0)
    }

    fn connected(&self) -> Vec<PeerId> {
        guard(&self.connected).keys().copied().collect()
    }

    /// Directly connected peers that spoke this protocol version on our
    /// chain, those we dialed before those that only dialed us. A peer reached
    /// only through a relay is left out: data never rides a circuit.
    pub(crate) fn live(&self) -> Vec<PeerId> {
        let connected = guard(&self.connected);
        let (outbound, inbound): (Vec<_>, Vec<_>) = connected
            .iter()
            .filter(|(_, links)| links.on_chain && links.data > 0)
            .partition(|(_, links)| links.outbound_data > 0);
        outbound
            .into_iter()
            .chain(inbound)
            .map(|(peer, _)| *peer)
            .collect()
    }

    /// Whether the peer holds a direct data connection: the only kind that
    /// carries records, inventory, parents or objects (or Peers gossip).
    pub(crate) fn has_direct(&self, peer: PeerId) -> bool {
        guard(&self.connected)
            .get(&peer)
            .is_some_and(|links| links.data > 0)
    }

    /// Whether discovery's uniform pick may ask the peer for Peers: a data
    /// connection, or one we dialed that was demoted (it kept its record
    /// handlers). Inbound demoted links are cheap to hold, so they never steer
    /// what the book learns.
    pub(crate) fn can_gossip(&self, peer: PeerId) -> bool {
        guard(&self.connected)
            .get(&peer)
            .is_some_and(|links| links.data > 0 || links.outbound_demoted > 0)
    }

    /// Whether the peer holds any direct connection that carries records:
    /// data, or demoted from data.
    pub(crate) fn any_direct(&self, peer: PeerId) -> bool {
        guard(&self.connected)
            .get(&peer)
            .is_some_and(|links| links.data > 0 || links.demoted > 0)
    }

    /// Connected peers that have a link (`Links::link`), sorted by peer.
    pub(crate) fn links(&self) -> Vec<(PeerId, PeerLink)> {
        guard(&self.connected)
            .iter()
            .filter_map(|(peer, links)| Some((*peer, links.link()?)))
            .collect()
    }

    pub(crate) fn opened(&self, peer: PeerId, outbound: bool, class: Class) {
        let mut connected = guard(&self.connected);
        connected
            .entry(peer)
            .or_default()
            .count(outbound, class, true);
    }

    /// The peer leaves with its last connection, chain proof included.
    pub(crate) fn closed(&self, peer: PeerId, outbound: bool, class: Class) {
        let mut connected = guard(&self.connected);
        let Some(links) = connected.get_mut(&peer) else {
            return;
        };
        links.count(outbound, class, false);
        if links.inbound == 0 && links.outbound == 0 {
            connected.remove(&peer);
        }
    }

    /// One open connection of the peer changed class (a demotion or promotion).
    pub(crate) fn reclassed(&self, peer: PeerId, outbound: bool, (from, to): (Class, Class)) {
        if let Some(links) = guard(&self.connected).get_mut(&peer) {
            links.count(outbound, from, false);
            links.count(outbound, to, true);
        }
    }

    pub(crate) fn is_on_chain(&self, peer: PeerId) -> bool {
        guard(&self.connected)
            .get(&peer)
            .is_some_and(|links| links.on_chain)
    }

    /// A message carrying this protocol version and our genesis came from or
    /// was answered by the peer over a live connection.
    pub(crate) fn on_chain(&self, peer: PeerId) {
        if let Some(links) = guard(&self.connected).get_mut(&peer) {
            links.on_chain = true;
        }
    }

    pub(crate) fn listening(&self, address: Multiaddr, bound: bool) {
        let mut listening = guard(&self.listening);
        listening.retain(|known| *known != address);
        if bound {
            listening.push(address);
        }
    }

    pub(crate) fn listeners(&self) -> Vec<Multiaddr> {
        guard(&self.listening).clone()
    }

    /// One identify report, counted only from outbound connections. Returns
    /// the address once this report confirms it.
    pub(crate) fn observe(&self, address: Multiaddr, reporter: PeerId) -> Option<Multiaddr> {
        match self.observed.lock() {
            Ok(mut observed) => observed.observe(address, reporter),
            Err(poisoned) => poisoned.into_inner().observe(address, reporter),
        }
    }

    /// Answer a Peers request: peers holding a reservation on this relay
    /// alternate with the snapshot. Only IP-literal addresses are forwarded,
    /// private and loopback ones only to loopback requesters. Neither the
    /// requester nor this node is listed; future `last_seen` is clamped.
    pub(crate) fn serve(&self, requester: PeerId, loopback: bool, limit: u32) -> Reply {
        let Ok(limit) = usize::try_from(limit) else {
            return Reply::Refused;
        };
        if limit > usize::try_from(PEERS_MAX).unwrap_or(0) {
            return Reply::Refused;
        }
        let now = unix_now();
        let reserved = self.reserved(requester, now);
        let held: Vec<PeerId> = reserved.iter().map(|peer| peer.peer_id).collect();
        let Ok(snapshot) = self.gossip.lock() else {
            return Reply::Busy;
        };
        let mut listed = snapshot
            .iter()
            .filter(|entry| entry.peer_id != requester && entry.peer_id != self.local)
            .filter(|entry| !held.contains(&entry.peer_id))
            .filter_map(|entry| servable(entry, loopback, now));
        let mut reserved = reserved.into_iter();
        let mut peers = Vec::new();
        while peers.len() < limit {
            let (held, known) = (reserved.next(), listed.next());
            if held.is_none() && known.is_none() {
                break;
            }
            peers.extend(held.into_iter().chain(known));
        }
        peers.truncate(limit);
        loop {
            match serde_json::to_vec(&peers) {
                Ok(bytes) if bytes.len() + Self::ENVELOPE <= PEERS_LIMIT => break,
                Ok(_) => {
                    peers.pop();
                }
                Err(_) => return Reply::Busy,
            }
        }
        Reply::Peers(peers)
    }

    /// Peers holding an active reservation on this node's relay, in random
    /// order, each with its circuit addresses through this node's public IPv4
    /// external addresses. Empty while this node has none.
    fn reserved(&self, requester: PeerId, now: u64) -> Vec<GossipPeer> {
        let hops: Vec<Multiaddr> = self
            .nat
            .external()
            .into_iter()
            .map(|external| {
                external
                    .with(Protocol::P2p(self.local))
                    .with(Protocol::P2pCircuit)
            })
            .filter(|hop| parse(hop).is_ok())
            .take(PEER_ADDRESSES_MAX)
            .collect();
        if hops.is_empty() {
            return Vec::new();
        }
        let mut holders = self.nat.reserved_by_others();
        holders.retain(|holder| *holder != requester && *holder != self.local);
        holders.shuffle(&mut rand::thread_rng());
        holders
            .into_iter()
            .map(|peer_id| GossipPeer {
                peer_id,
                addrs: hops.clone(),
                last_seen: now,
            })
            .collect()
    }
}

/// A snapshot entry with only the addresses this requester may receive.
fn servable(entry: &GossipPeer, loopback: bool, now: u64) -> Option<GossipPeer> {
    let addrs: Vec<Multiaddr> = entry
        .addrs
        .iter()
        .filter(|address| validate_gossip_address(address).is_ok())
        .filter(|address| loopback || is_public(address))
        .take(PEER_ADDRESSES_MAX)
        .cloned()
        .collect();
    (!addrs.is_empty()).then(|| GossipPeer {
        peer_id: entry.peer_id,
        addrs,
        last_seen: entry.last_seen.min(now),
    })
}

/// Globally routable IP, or a DNS name other than localhost. IPv4-mapped IPv6
/// addresses are judged as the IPv4 address they carry.
pub fn is_public(address: &Multiaddr) -> bool {
    match address.iter().next() {
        Some(Protocol::Ip4(ip)) => public_v4(ip),
        Some(Protocol::Ip6(ip)) => match ip.to_ipv4_mapped() {
            Some(mapped) => public_v4(mapped),
            None => public_v6(ip),
        },
        Some(
            Protocol::Dns(name)
            | Protocol::Dns4(name)
            | Protocol::Dns6(name)
            | Protocol::Dnsaddr(name),
        ) => name != "localhost" && !name.ends_with(".localhost"),
        _ => false,
    }
}

fn public_v4(ip: Ipv4Addr) -> bool {
    let [first, second, third, _] = ip.octets();
    let reserved = first == 0
        || (first == 100 && (second & 0xc0) == 64)
        || (first == 192 && second == 0 && third == 0)
        || (first == 198 && (second & 0xfe) == 18)
        || first >= 240;
    !(reserved
        || ip.is_loopback()
        || ip.is_private()
        || ip.is_link_local()
        || ip.is_multicast()
        || ip.is_documentation())
}

fn public_v6(ip: Ipv6Addr) -> bool {
    let documentation = ip.segments()[..2] == [0x2001, 0x0db8];
    !(documentation
        || ip.is_loopback()
        || ip.is_unspecified()
        || ip.is_multicast()
        || ip.is_unique_local()
        || ip.is_unicast_link_local())
}

fn unix_now() -> u64 {
    match SystemTime::now().duration_since(UNIX_EPOCH) {
        Ok(elapsed) => elapsed.as_secs(),
        Err(_clock_before_epoch) => 0,
    }
}

pub(crate) fn is_loopback(address: &Multiaddr) -> bool {
    match address.iter().next() {
        Some(Protocol::Ip4(ip)) => ip.is_loopback(),
        Some(Protocol::Ip6(ip)) => ip.is_loopback(),
        _ => false,
    }
}

/// Gossip carries only IP-literal addresses, direct or circuit, each with an
/// optional `/p2p`: names come from operators (seeds, manual), never peers.
pub fn validate_gossip_address(address: &Multiaddr) -> Result<Option<PeerId>, Error> {
    let shape = parse(address)?;
    match shape.bare.iter().next() {
        Some(Protocol::Ip4(_) | Protocol::Ip6(_)) => Ok(shape.peer),
        _ => Err(Error::Invalid(
            "gossiped addresses must be IP literals".into(),
        )),
    }
}

/// Whether a failed chain probe closes the peer's connections: only when it
/// answered for another chain, version or protocol. Busy, transport and
/// admission failures say nothing about the peer, and its other connections
/// (a relay reservation among them) stay.
pub(crate) fn probe_closes(error: &Error) -> bool {
    matches!(error, Error::Invalid(_))
}

/// Dial targets: any direct address shape, or a circuit that names its
/// target. Returns the named identity.
pub fn validate_dial(address: &Multiaddr) -> Result<Option<PeerId>, Error> {
    let shape = parse(address)?;
    if shape.peer.is_none() && matches!(shape.kind, AddressKind::Circuit(_)) {
        return Err(Error::Invalid(
            "circuit dial must name its target peer".into(),
        ));
    }
    Ok(shape.peer)
}

/// Bind the reply to its request, then check every gossiped hint's shape.
pub(crate) fn validate_peers(
    genesis: BlockHash,
    peer: PeerId,
    pending: &Pending,
    response: Response,
) -> Result<Vec<GossipPeer>, Error> {
    bind_response(genesis, peer, pending, &response)?;
    let Kind::Peers { limit } = pending.kind else {
        return Err(Error::Invalid("peers reply to another request".into()));
    };
    match response.result {
        Reply::Peers(peers) => {
            let limit = usize::try_from(limit).map_err(|e| Error::Invalid(e.to_string()))?;
            if peers.len() > limit {
                return Err(Error::Invalid(
                    "peer gossip exceeds the requested limit".into(),
                ));
            }
            for entry in &peers {
                check_gossip(entry)?;
            }
            Ok(peers)
        }
        Reply::Busy => Err(Error::PeerBusy),
        _ => Err(Error::Invalid(
            "peer refused the peers request or answered another kind".into(),
        )),
    }
}

fn check_gossip(entry: &GossipPeer) -> Result<(), Error> {
    let peer = entry.peer_id;
    if entry.addrs.is_empty() || entry.addrs.len() > PEER_ADDRESSES_MAX {
        return Err(Error::Invalid(
            "peer gossip address count out of range".into(),
        ));
    }
    for address in &entry.addrs {
        if validate_gossip_address(address)?.is_some_and(|named| named != peer)
            || relay(address) == Some(peer)
        {
            return Err(Error::Invalid(
                "peer gossip address names another peer".into(),
            ));
        }
    }
    Ok(())
}

pub(crate) struct Dial {
    address: Multiaddr,
    deadline: Instant,
    answer: mpsc::SyncSender<Result<PeerId, Error>>,
}

impl Dial {
    fn answer(&self, result: Result<PeerId, Error>) {
        let _ = self.answer.try_send(result);
    }
}

/// Outstanding explicit dials on the network thread, keyed by connection.
#[derive(Default)]
pub(crate) struct Dials(HashMap<ConnectionId, (Option<PeerId>, Dial)>);

impl Dials {
    const MAX: usize = 8;

    pub(crate) fn start(&mut self, swarm: &mut Swarm<Behaviour>, dial: Dial) {
        let expected = match validate_dial(&dial.address) {
            Ok(expected) => expected,
            Err(error) => return dial.answer(Err(error)),
        };
        if Instant::now() >= dial.deadline || self.0.len() >= Self::MAX {
            return dial.answer(Err(Error::Busy("peer dial capacity or deadline".into())));
        }
        if let Some(peer) = expected.filter(|peer| swarm.is_connected(peer)) {
            return dial.answer(Ok(peer));
        }
        let options = match expected {
            Some(peer) => DialOpts::peer_id(peer)
                .addresses(vec![dial.address.clone()])
                .build(),
            None => DialOpts::unknown_peer_id()
                .address(dial.address.clone())
                .build(),
        };
        let connection = options.connection_id();
        match swarm.dial(options) {
            Ok(()) => {
                self.0.insert(connection, (expected, dial));
            }
            Err(error) => dial.answer(Err(dial_error(&error))),
        }
    }

    pub(crate) fn established(&mut self, connection: ConnectionId, peer: PeerId) {
        if let Some((expected, dial)) = self.0.remove(&connection) {
            if expected.is_some_and(|expected| expected != peer) {
                return dial.answer(Err(Error::Invalid(
                    "dialed peer authenticated as another identity".into(),
                )));
            }
            dial.answer(Ok(peer));
        }
    }

    pub(crate) fn failed(&mut self, connection: ConnectionId, error: &DialError) {
        if let Some((_, dial)) = self.0.remove(&connection) {
            dial.answer(Err(dial_error(error)));
        }
    }

    /// A dial past its deadline is answered now; a late connection idles out.
    pub(crate) fn expire(&mut self) {
        let now = Instant::now();
        self.0.retain(|_, (_, dial)| {
            if now < dial.deadline {
                return true;
            }
            dial.answer(Err(Error::Busy("peer dial deadline".into())));
            false
        });
    }

    pub(crate) fn close(&mut self) {
        for (_, (_, dial)) in self.0.drain() {
            dial.answer(Err(Error::Busy("peer network stopped".into())));
        }
    }
}

fn dial_error(error: &DialError) -> Error {
    match error {
        DialError::Denied { cause } => match cause.downcast_ref::<LimitDenied>() {
            Some(LimitDenied(limit)) => Error::DialDenied(*limit),
            None => Error::Invalid(format!("peer dial: {error}")),
        },
        DialError::LocalPeerId { .. } => Error::LocalPeer,
        DialError::DialPeerConditionFalse(condition) => {
            Error::Busy(format!("peer dial already pending ({condition:?})"))
        }
        other => Error::Invalid(format!("peer dial: {other}")),
    }
}

impl PeerNetwork {
    const DIAL_POLL: Duration = Duration::from_millis(50);

    /// Replace the snapshot served to Peers requests. Order is preserved; the
    /// first entries are the ones a bounded request receives.
    pub fn set_gossip(&self, peers: Vec<GossipPeer>) {
        self.shared.set_gossip(peers);
    }

    /// Public addresses connected peers reported observing for this node.
    pub fn observed_addrs(&self) -> Vec<Multiaddr> {
        self.shared.observed()
    }

    /// Peers with at least one established connection right now, sorted.
    pub fn connected_peer_ids(&self) -> Vec<PeerId> {
        self.shared.connected()
    }

    /// Connected peers with their link, sorted by peer.
    pub fn peer_links(&self) -> Vec<(PeerId, PeerLink)> {
        self.shared.links()
    }

    /// Whether the peer holds a direct connection; relayed-only peers never
    /// carry data.
    pub fn has_direct(&self, peer: PeerId) -> bool {
        self.shared.has_direct(peer)
    }

    /// Whether Peers gossip can ride an open connection to the peer.
    pub fn can_gossip(&self, peer: PeerId) -> bool {
        self.shared.can_gossip(peer)
    }

    /// Whether a direct connection that carries records is open to the peer.
    pub fn any_direct(&self, peer: PeerId) -> bool {
        self.shared.any_direct(peer)
    }

    pub fn reachability(&self) -> Reachability {
        self.shared.nat.reachability()
    }

    /// Addresses advertised to peers: AutoNAT-confirmed or announced.
    pub fn external_addresses(&self) -> Vec<Multiaddr> {
        self.shared.nat.external()
    }

    /// Relays holding an active reservation for this node.
    pub fn relays(&self) -> Vec<PeerId> {
        self.shared.nat.relays()
    }

    /// Peers holding an active reservation on this node's relay server.
    pub fn reserved_by_others(&self) -> Vec<PeerId> {
        self.shared.nat.reserved_by_others()
    }

    /// This node's `/p2p-circuit` listen addresses through its relays.
    pub fn circuit_addresses(&self) -> Vec<Multiaddr> {
        self.shared.nat.circuits()
    }

    /// Connected peers' own literal public addresses learned through identify
    /// since the last call, newest 64 peers; the queue is emptied.
    pub fn drain_identified(&self) -> Vec<(PeerId, Vec<Multiaddr>)> {
        self.shared.drain_identified()
    }

    /// `dial_any` for a peer that must be reached directly: a peer reached
    /// only through a relay is a Busy error until DCUtR links it directly.
    pub fn dial(
        &self,
        address: Multiaddr,
        cancelled: &AtomicBool,
        deadline: Instant,
    ) -> Result<PeerId, Error> {
        match self.dial_any(address, cancelled, deadline)? {
            Dialed::Direct(peer) => Ok(peer),
            Dialed::Relayed(peer) => Err(Error::Busy(format!("{peer} reachable only via relay"))),
        }
    }

    /// Dial an address (with or without `/p2p`); a direct peer gets the chain
    /// `probe`. A relayed-only peer is returned unprobed: records/5 never runs
    /// on relayed links. The connection stays for later requests.
    pub fn dial_any(
        &self,
        address: Multiaddr,
        cancelled: &AtomicBool,
        deadline: Instant,
    ) -> Result<Dialed, Error> {
        validate_dial(&address)?;
        check_cancelled(cancelled)?;
        if self.stop.load(Ordering::Acquire) {
            return Err(Error::Busy("peer network stopped".into()));
        }
        let deadline = deadline.min(Instant::now() + CALL_TIMEOUT);
        let circuit = is_circuit(&address);
        let (answer, receiver) = mpsc::sync_channel(1);
        self.commands
            .try_send(Command::Dial(Dial {
                address,
                deadline,
                answer,
            }))
            .map_err(|_| Error::Busy("peer command queue unavailable".into()))?;
        let peer = loop {
            check_cancelled(cancelled)?;
            let left = deadline.saturating_duration_since(Instant::now());
            if left.is_zero() {
                return Err(Error::Busy("peer dial deadline".into()));
            }
            match receiver.recv_timeout(left.min(Self::DIAL_POLL)) {
                Ok(result) => break result?,
                Err(RecvTimeoutError::Timeout) => (),
                Err(RecvTimeoutError::Disconnected) => {
                    return Err(Error::Busy("peer network stopped".into()));
                }
            }
        };
        if circuit && !self.any_direct(peer) {
            return Ok(Dialed::Relayed(peer));
        }
        self.probe(peer, cancelled, deadline).map(Dialed::Direct)
    }

    /// The chain probe: a zero-limit Peers request over a direct link. Only a
    /// peer proven to be elsewhere (`probe_closes`) loses its connections.
    pub fn probe(
        &self,
        peer: PeerId,
        cancelled: &AtomicBool,
        deadline: Instant,
    ) -> Result<PeerId, Error> {
        let error = match self.peers_of(peer, 0, cancelled, deadline) {
            Ok(_) => return Ok(peer),
            Err(error) => error,
        };
        if probe_closes(&error) {
            let _ = self.commands.try_send(Command::Disconnect(peer));
        }
        Err(error)
    }

    /// One bounded Peers request over an existing connection (or addresses the
    /// swarm already knows). `limit` is at most PEERS_MAX; 0 is a bare probe.
    pub fn peers_of(
        &self,
        peer: PeerId,
        limit: u32,
        cancelled: &AtomicBool,
        deadline: Instant,
    ) -> Result<Vec<GossipPeer>, Error> {
        if limit > PEERS_MAX {
            return Err(Error::Invalid("peers request limit exceeds 32".into()));
        }
        check_cancelled(cancelled)?;
        let kind = Kind::Peers { limit };
        request(
            &self.commands,
            &self.stop,
            (peer, Vec::new()),
            kind,
            deadline,
            Answer::Peers,
        )
    }
}