path: root/src/transport/libp2p/driver.rs

//! The swarm's task (`run`): it owns the swarm, takes `Command`s from the transport, and
//! answers the swarm's events (connections, identify, relays, hole punching, AutoNAT,
//! mDNS, UPnP, the DHT), keeping the book.

use super::*;

pub(super) enum Command {
    /// Connect to `peer` (at `addresses`) unless connected.
    Connect {
        peer: PeerId,
        addresses: Vec<Multiaddr>,
        done: oneshot::Sender<Result<(), String>>,
    },
    /// Connect to `address` to find out which peer is there.
    Identify {
        address: Multiaddr,
        done: oneshot::Sender<Result<PeerId, String>>,
    },
    /// See whether relays are needed now (one was just learned of).
    CheckRelays,
    /// Find out which peer relays at `address`, named without the relay's peer ID
    /// (`Book::finding`).
    FindRelay { address: Multiaddr },
    /// Where `--libp2p-internet-hosts` are now (`Routes::internet`).
    Internet(Vec<std::net::IpAddr>),
}

/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the
/// transport is dropped. `again`: to schedule commands.
pub(super) async fn run(
    mut swarm: Swarm<Behaviour>,
    mut commands: mpsc::UnboundedReceiver<Command>,
    again: mpsc::UnboundedSender<Command>,
    book: Arc<Mutex<Book>>,
    relaying: Relaying,
    stopping: watch::Sender<bool>,
) {
    let serving = relaying.serving;
    let mut nat = ::libp2p::autonat::NatStatus::Unknown;
    // Relays picked, by listener.
    let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new();
    let mut check = tokio::time::interval(RELAY_CHECK);
    check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP);
    dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    let mut connecting: HashMap<PeerId, Vec<oneshot::Sender<Result<(), String>>>> = HashMap::new();
    let mut identifying: HashMap<
        ConnectionId,
        (Multiaddr, oneshot::Sender<Result<PeerId, String>>),
    > = HashMap::new();
    let mut prune = tokio::time::interval(PRUNE_EVERY);
    prune.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    // Peers that may have several connections, and when to cut them to one. Only the
    // side with the smaller peer ID does that, so the two never close different ones.
    let mut dedupe_due: HashMap<PeerId, tokio::time::Instant> = HashMap::new();
    // Connections closed as duplicates, until they are: their "Connection closed" says so.
    let mut closing_duplicates: std::collections::HashSet<ConnectionId> =
        std::collections::HashSet::new();
    let mut dedupe = tokio::time::interval(DEDUPE_EVERY);
    dedupe.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
    loop {
        tokio::select! {
            _ = dedupe.tick(), if !dedupe_due.is_empty() => {
                let now = tokio::time::Instant::now();
                let due: Vec<PeerId> = dedupe_due
                    .iter()
                    .filter(|(_, at)| **at <= now)
                    .map(|(peer, _)| *peer)
                    .collect();
                for peer in due {
                    dedupe_due.remove(&peer);
                    let reserving = picked.values().any(|relay| *relay == peer);
                    let duplicates = {
                        let mut book = lock(&book);
                        let duplicates = book.duplicates(&peer);
                        if !duplicates.is_empty() && (reserving || book.slots.contains(&peer)) {
                            book.rehoming.insert(peer, now);
                        }
                        duplicates
                    };
                    for connection in duplicates {
                        if swarm.close_connection(connection) {
                            closing_duplicates.insert(connection);
                        }
                    }
                }
            }
            _ = prune.tick() => {
                let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer));
                if forgotten > 0 {
                    debug!(forgotten, "Forgot peers nobody mentions any more");
                }
            }
            _ = dht_bootstrap.tick(), if relaying.public => {
                if let Some(dht) = swarm.behaviour_mut().kad.as_mut()
                    && let Err(error) = dht.bootstrap()
                {
                    warn!(error = %error, "Couldn't join the IPFS DHT");
                }
            }
            _ = check.tick() => {
                let hosts = {
                    let mut book = lock(&book);
                    book.recheck_routes();
                    book.routes.internet_hosts.clone()
                };
                // Names are resolved aside, not to hold up the swarm.
                if !hosts.is_empty() {
                    let again = again.clone();
                    tokio::spawn(async move {
                        let _ = again.send(Command::Internet(resolve_internet_hosts(&hosts).await));
                    });
                }
                pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
                retry_punches(&mut swarm, &book, relaying.local);
            }
            command = commands.recv() => match command {
                None => break,
                Some(Command::CheckRelays) => {
                    pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked);
                }
                Some(Command::Internet(hosts)) => lock(&book).routes.internet(&hosts),
                Some(Command::Connect { peer, addresses, done }) => {
                    if swarm.is_connected(&peer) {
                        let _ = done.send(Ok(()));
                        upgrade(&mut swarm, &book, peer, relaying.local);
                        continue;
                    }
                    let waiting = connecting.entry(peer).or_default();
                    waiting.push(done);
                    if waiting.len() > 1 {
                        continue; // Already dialing.
                    }
                    let mut opts = DialOpts::peer_id(peer)
                        .addresses(addresses)
                        .condition(PeerCondition::DisconnectedAndNotDialing);
                    if relaying.public {
                        // Addresses the DHT knows too.
                        opts = opts.extend_addresses_through_behaviour();
                    }
                    if let Err(error) = swarm.dial(opts.build()) {
                        count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                        for done in connecting.remove(&peer).into_iter().flatten() {
                            let _ = done.send(Err(error.to_string()));
                        }
                        find_on_dht(&mut swarm, peer);
                    }
                }
                Some(Command::FindRelay { address }) => {
                    let again = again.clone();
                    let local = relaying.local;
                    let book = Arc::clone(&book);
                    tokio::spawn(async move {
                        let (done, identified) = oneshot::channel();
                        if again
                            .send(Command::Identify { address: address.clone(), done })
                            .is_err()
                        {
                            return;
                        }
                        match tokio::time::timeout(CONNECT_TIMEOUT, identified).await {
                            Ok(Ok(Ok(peer))) => {
                                if lock(&book).relay_found(address.clone(), peer, local) {
                                    debug!(relay = %Addressed(SCHEME, &address), peer = %Addressed(SCHEME, &peer), "Found a named relay");
                                    let _ = again.send(Command::CheckRelays);
                                }
                            }
                            failed => {
                                if !lock(&book).finding.contains_key(&address) {
                                    return; // No longer named.
                                }
                                let error = match failed {
                                    Ok(Ok(Err(error))) => error,
                                    Ok(_) => "the libp2p swarm stopped".to_owned(),
                                    Err(_) => format!("no answer within {CONNECT_TIMEOUT:?}"),
                                };
                                warn!(relay = %Addressed(SCHEME, &address), error = %error, retry = ?RELAY_RETRY, "Couldn't reach a named relay; trying again later");
                                tokio::time::sleep(RELAY_RETRY).await;
                                let _ = again.send(Command::FindRelay { address });
                            }
                        }
                    });
                }
                Some(Command::Identify { address, done }) => {
                    let opts = DialOpts::unknown_peer_id().address(address.clone()).build();
                    let id = opts.connection_id();
                    match swarm.dial(opts) {
                        Ok(()) => {
                            identifying.insert(id, (address, done));
                        }
                        Err(error) => {
                            let _ = done.send(Err(error.to_string()));
                        }
                    }
                }
            },
            event = swarm.select_next_some() => match event {
                SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => {
                    {
                        let mut book = lock(&book);
                        book.touch(peer_id);
                        let relayed = endpoint.is_relayed();
                        book.connections.insert(
                            connection_id,
                            (peer_id, relayed, endpoint.get_remote_address().clone()),
                        );
                        if !relayed {
                            book.upgrades.remove(&peer_id);
                            book.punching.remove(&peer_id);
                        }
                        let to_peer: Vec<ConnectionId> = book
                            .connections
                            .iter()
                            .filter(|(_, (connected, _, _))| *connected == peer_id)
                            .map(|(id, _)| *id)
                            .collect();
                        if to_peer.len() > 1 {
                            // A new path to compare: each is judged by its round trips
                            // from now on, not an old one by its best over hours
                            // (`Book::ranked`).
                            for connection in &to_peer {
                                book.rtts.remove(connection);
                            }
                        }
                        if relaying.local < peer_id && to_peer.len() > 1 {
                            // Only through relays: maybe a punch under way.
                            let grace = match book.connection(&peer_id) {
                                Some(Connection::Relayed) => PUNCH_GRACE,
                                _ => DUPLICATE_GRACE,
                            };
                            dedupe_due.insert(peer_id, tokio::time::Instant::now() + grace);
                        }
                        debug!(
                            peer = %Addressed(SCHEME, &peer_id),
                            connection = %connection_id,
                            dialled = endpoint.is_dialer(),
                            address = %endpoint.get_remote_address(),
                            relayed,
                            "Connection opened"
                        );
                    }
                    for done in connecting.remove(&peer_id).into_iter().flatten() {
                        let _ = done.send(Ok(()));
                    }
                    if let Some((address, done)) = identifying.remove(&connection_id) {
                        lock(&book).remember(peer_id, address);
                        let _ = done.send(Ok(peer_id));
                    }
                }
                SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => {
                    {
                        let mut book = lock(&book);
                        book.connections.remove(&connection_id);
                        book.rtts.remove(&connection_id);
                        if book.connection(&peer_id).is_none() {
                            book.identified.remove(&peer_id);
                            book.explored.remove(&peer_id);
                        }
                    }
                    let cause = match cause {
                        _ if closing_duplicates.remove(&connection_id) => "a duplicate".to_owned(),
                        // By us, or idle: libp2p doesn't tell which.
                        None => "closed here".to_owned(),
                        Some(error) => error.to_string(),
                    };
                    debug!(
                        peer = %Addressed(SCHEME, &peer_id),
                        connection = %connection_id,
                        cause = %cause,
                        "Connection closed"
                    );
                }
                SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => {
                    count_dial_failure(&mut lock(&book).counts.dial_failures, &error);
                    // Another peer there: not this one's address, though it said so (a
                    // NATed peer's LAN address, the same range as ours).
                    if let (Some(peer), ::libp2p::swarm::DialError::WrongPeerId { obtained, address }) =
                        (peer_id, &error)
                    {
                        debug!(
                            peer = %Addressed(SCHEME, &peer),
                            address = %address,
                            obtained = %obtained,
                            "Another peer at an address of this one; not dialling it there again"
                        );
                        lock(&book).not_theirs(peer, address.clone());
                    }
                    if let Some((_, done)) = identifying.remove(&connection_id) {
                        let _ = done.send(Err(describe_dial_error(&error)));
                    }
                    if let Some(peer) = peer_id
                        && !swarm.is_connected(&peer)
                        && let Some(waiting) = connecting.remove(&peer)
                    {
                        for done in waiting {
                            let _ = done.send(Err(describe_dial_error(&error)));
                        }
                        find_on_dht(&mut swarm, peer);
                    }
                    // A dial that kept ours from going (`upgrade`) is over.
                    if let Some(peer) = peer_id {
                        upgrade(&mut swarm, &book, peer, relaying.local);
                    }
                }
                SwarmEvent::NewListenAddr { address, .. } => {
                    // libp2p listens on every interface it's told to; the ranges left
                    // out are only kept out of what's published, told and connected.
                    let left_out = ignored(&lock(&book).ignored, &address);
                    if left_out {
                        debug!(address = %address, "Listening, but left out (libp2p-ignore): neither published nor connected over");
                    } else {
                        info!(address = %address, "Listening");
                    }
                    // A relay tells those reserving where it is: its external addresses.
                    if serving && !left_out && !is_local(&address) && !is_relayed(&address) {
                        swarm.add_external_address(address.clone());
                    }
                    let new_path = !left_out && !is_local(&address) && !is_relayed(&address);
                    {
                        let mut book = lock(&book);
                        if !book.listening.contains(&address) {
                            book.listening.push(address);
                        }
                        // A new interface (ethernet besides the WLAN, say): every
                        // address may be reached better now.
                        if new_path {
                            book.explored.clear();
                        }
                    }
                    if new_path {
                        let peers: Vec<PeerId> = swarm.connected_peers().copied().collect();
                        for peer in peers {
                            explore(&mut swarm, &book, peer, true);
                        }
                    }
                }
                SwarmEvent::ExpiredListenAddr { address, .. } => {
                    info!(address = %address, "No longer listening");
                    if serving {
                        swarm.remove_external_address(&address);
                    }
                    lock(&book).listening.retain(|known| *known != address);
                }
                SwarmEvent::ListenerClosed { listener_id, addresses, reason } => {
                    lock(&book)
                        .listening
                        .retain(|known| !addresses.contains(known));
                    // A relay picked: the next check picks again, another one if this one
                    // never granted the slot.
                    if let Some(relay) = picked.remove(&listener_id) {
                        let refused = match &reason {
                            Ok(()) => false,
                            Err(error) => relay_refused(&format!("{error:?}"), &innermost_causes(error)),
                        };
                        let (lost, pause, rehomed) = {
                            let mut book = lock(&book);
                            let lost = book.slots.remove(&relay);
                            // Its connection closed as a duplicate (`duplicates`): again
                            // at once, over the one kept.
                            let rehomed = lost
                                && book.rehoming.remove(&relay).is_some_and(|at| {
                                    tokio::time::Instant::now().saturating_duration_since(at)
                                        < REHOMING_WITHIN
                                });
                            let pause = match (refused, lost) {
                                (true, _) => Some(RELAY_PAUSE),
                                (false, true) => None,
                                (false, false) => Some(RELAY_RETRY),
                            };
                            if lost {
                                if !rehomed {
                                    book.counts.slots_lost += 1;
                                }
                            } else {
                                book.counts.slots_not_granted += 1;
                                // Named without its peer ID: maybe another one now.
                                if let Some(address) = book.find_again(relay) {
                                    let _ = again.send(Command::FindRelay { address });
                                }
                            }
                            if let Some(pause) = pause {
                                book.failed_relays.insert(relay, tokio::time::Instant::now() + pause);
                            }
                            (lost, pause, rehomed)
                        };
                        // Again at once, or others meanwhile.
                        let _ = again.send(Command::CheckRelays);
                        let reason = relay_failure(&reason);
                        if rehomed {
                            debug!(relay = %Addressed(SCHEME, &relay), "A duplicate connection to a relay closed with our slot on it; reserving again over the one kept");
                            continue;
                        }
                        match (lost, pause) {
                            (true, None) => {
                                info!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay; reserving again");
                            }
                            (true, Some(pause)) => {
                                info!(relay = %Addressed(SCHEME, &relay), pause = ?pause, reason = %reason, "Lost the slot on a relay; picking others for a while");
                            }
                            (false, pause) => {
                                info!(
                                    relay = %Addressed(SCHEME, &relay),
                                    pause = ?pause.unwrap_or_default(),
                                    reason = %reason,
                                    "A relay didn't grant a slot; picking others for a while"
                                );
                            }
                        }
                    }
                }
                SwarmEvent::ListenerError { error, .. } => {
                    warn!(error = %error, "Listener failed");
                }
                SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                    ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. },
                )) => {
                    {
                        let mut book = lock(&book);
                        book.slots.insert(relay_peer_id);
                        book.failed_relays.remove(&relay_peer_id);
                        if !renewal {
                            book.counts.slots_granted += 1;
                            // Peers that couldn't reach us may now, through it.
                            if let Some(reachable) = &book.reachable {
                                reachable.notify_one();
                            }
                        }
                    }
                    if !renewal {
                        info!(relay = %Addressed(SCHEME, &relay_peer_id), "Reachable through a relay");
                    }
                }
                SwarmEvent::Behaviour(BehaviourEvent::Autonat(
                    ::libp2p::autonat::Event::StatusChanged { old, new },
                )) => {
                    info!(old = ?old, new = ?new, "Reachability (AutoNAT)");
                    {
                        use ::libp2p::autonat::NatStatus;
                        let mut book = lock(&book);
                        (book.reachability, book.public_address) = match &new {
                            NatStatus::Unknown => (Reachability::Unknown, None),
                            NatStatus::Private => (Reachability::Private, None),
                            NatStatus::Public(address) => {
                                (Reachability::Public, Some(address.clone()))
                            }
                        };
                        if matches!(new, NatStatus::Public(_))
                            && !matches!(old, NatStatus::Public(_))
                            && let Some(reachable) = &book.reachable
                        {
                            reachable.notify_one();
                        }
                    }
                    nat = new;
                }
                SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => {
                    {
                        use ::libp2p::relay::Event;
                        let mut book = lock(&book);
                        match &event {
                            Event::ReservationReqAccepted { src_peer_id, .. } => {
                                book.slots_here.insert(*src_peer_id);
                            }
                            Event::ReservationClosed { src_peer_id }
                            | Event::ReservationTimedOut { src_peer_id } => {
                                book.slots_here.remove(src_peer_id);
                            }
                            _ => {}
                        }
                    }
                    debug!(event = ?event, "Relay event");
                }
                SwarmEvent::Behaviour(BehaviourEvent::Ping(::libp2p::ping::Event {
                    peer,
                    connection,
                    result,
                })) => match result {
                    Ok(rtt) => {
                        let mut book = lock(&book);
                        if book.connections.contains_key(&connection) {
                            let shortest = book.rtts.entry(connection).or_insert(rtt);
                            *shortest = (*shortest).min(rtt);
                        }
                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, rtt = ?rtt, "Connection's round trip");
                    }
                    Err(error) => {
                        trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, error = %error, "Connection's ping failed");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Upnp(event)) => match event {
                    ::libp2p::upnp::Event::NewExternalAddr {
                        local_addr,
                        external_addr: address,
                    } => {
                        info!(address = %address, to = %local_addr, "The router forwards to us (UPnP)");
                        let mut book = lock(&book);
                        if !book.mapped.contains(&address) {
                            book.mapped.push(address);
                        }
                    }
                    ::libp2p::upnp::Event::ExpiredExternalAddr {
                        local_addr,
                        external_addr: address,
                    } => {
                        info!(address = %address, to = %local_addr, "The router no longer forwards to us (UPnP)");
                        lock(&book).mapped.retain(|mapped| *mapped != address);
                    }
                    ::libp2p::upnp::Event::GatewayNotFound => {
                        info!("No router that maps ports (UPnP) answered");
                    }
                    ::libp2p::upnp::Event::NonRoutableGateway => {
                        warn!("The router that maps ports (UPnP) is behind another NAT itself; its mappings don't reach us from outside");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result {
                    Ok(_) => {
                        let mut book = lock(&book);
                        book.counts.hole_punches.succeeded += 1;
                        book.punching.remove(&event.remote_peer_id);
                        drop(book);
                        info!(peer = %Addressed(SCHEME, &event.remote_peer_id), "Connected directly to a peer behind a relay");
                    }
                    Err(error) => {
                        let retry = {
                            let mut book = lock(&book);
                            book.counts.hole_punches.failed += 1;
                            let failed = book
                                .punching
                                .get(&event.remote_peer_id)
                                .map_or(0, |(failed, _)| *failed);
                            let retry = PUNCH_RETRIES[failed.min(PUNCH_RETRIES.len() - 1)];
                            book.punching.insert(
                                event.remote_peer_id,
                                (failed + 1, tokio::time::Instant::now() + retry),
                            );
                            retry
                        };
                        debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, retry = ?retry, "Couldn't connect directly to a peer behind a relay; the relay carries on");
                    }
                },
                SwarmEvent::Behaviour(BehaviourEvent::Identify(
                    ::libp2p::identify::Event::Received { connection_id, peer_id, info },
                )) => {
                    {
                        let mut book = lock(&book);
                        let listen = published(&info.listen_addrs);
                        // Every new connection identifies again: logged when it says something new.
                        if book.knows(peer_id, &listen) {
                            trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again");
                        } else {
                            debug!(
                                peer = %Addressed(SCHEME, &peer_id),
                                connection = %connection_id,
                                listen = ?info.listen_addrs,
                                observed = %info.observed_addr,
                                agent = %info.agent_version,
                                "Peer identified"
                            );
                        }
                        book.identified(peer_id, listen);
                        if relaying.public
                            && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME)
                            && (book.public_relays.len() < MAX_PEERS
                                || book.public_relays.contains_key(&peer_id))
                            && let Some(address) = book
                                .addresses
                                .get(&peer_id)
                                .and_then(|known| known.first())
                                .and_then(|address| address.clone().with_p2p(peer_id).ok())
                        {
                            book.public_relays.insert(peer_id, address);
                        }
                        if !info.protocols.contains(&PROTOCOL)
                            && (book.not_nodes.len() < MAX_PEERS
                                || book.not_nodes.contains_key(&peer_id))
                        {
                            let relays = info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME);
                            book.not_nodes.insert(peer_id, relays);
                        }
                        if info.protocols.contains(&PROTOCOL) {
                            book.not_nodes.remove(&peer_id);
                            book.continuum.insert(peer_id);
                            if book.lan.remove(&peer_id) {
                                book.discover(peer_id);
                            }
                        }
                    }
                    upgrade(&mut swarm, &book, peer_id, relaying.local);
                    explore(&mut swarm, &book, peer_id, false);
                }
                SwarmEvent::Behaviour(BehaviourEvent::Mdns(
                    ::libp2p::mdns::Event::Discovered(found),
                )) => {
                    for (peer, address) in found {
                        if peer == relaying.local {
                            continue;
                        }
                        debug!(peer = %Addressed(SCHEME, &peer), address = %address, "Peer on the LAN");
                        let continuum = {
                            let mut book = lock(&book);
                            book.remember(peer, address.clone());
                            book.continuum.contains(&peer)
                        };
                        upgrade(&mut swarm, &book, peer, relaying.local);
                        if continuum {
                            lock(&book).discover(peer);
                        } else if lock(&book).lan.len() < MAX_PEERS {
                            // Connected, identify says whether it speaks continuum.
                            lock(&book).lan.insert(peer);
                            let _ = swarm.dial(
                                DialOpts::peer_id(peer)
                                    .addresses(vec![address])
                                    .condition(PeerCondition::DisconnectedAndNotDialing)
                                    .build(),
                            );
                        }
                    }
                }
                _ => {}
            },
        }
    }
    let _ = stopping.send(true);
}