libp2p: explode!

1 files changed, 527 insertions, 0 deletions

diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs
new file mode 100644
index 0000000..efcf109
--- /dev/null
+++ b/src/transport/libp2p/driver.rs
@@ -0,0 +1,527 @@
+//! 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 },
+}
+
+/// 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 = lock(&book).duplicates(&peer, reserving);
+                    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() => {
+                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::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);
+                        }
+                        if relaying.local < peer_id
+                            && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 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);
+                    }
+                    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);
+                    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);
+                    }
+                }
+                SwarmEvent::NewListenAddr { address, .. } => {
+                    info!(address = %address, "Listening");
+                    // A relay tells those reserving where it is: its external addresses.
+                    if serving && !is_local(&address) && !is_relayed(&address) {
+                        swarm.add_external_address(address.clone());
+                    }
+                    let mut book = lock(&book);
+                    if !book.listening.contains(&address) {
+                        book.listening.push(address);
+                    }
+                }
+                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) = {
+                            let mut book = lock(&book);
+                            let lost = book.slots.remove(&relay);
+                            let pause = match (refused, lost) {
+                                (true, _) => Some(RELAY_PAUSE),
+                                (false, true) => None,
+                                (false, false) => Some(RELAY_RETRY),
+                            };
+                            if lost {
+                                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)
+                        };
+                        // Again at once, or others meanwhile.
+                        let _ = again.send(Command::CheckRelays);
+                        let reason = relay_failure(&reason);
+                        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;
+                        }
+                    }
+                    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()))
+                            }
+                        };
+                    }
+                    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"
+                            );
+                        }
+                        // Last first: `remember` puts each in front.
+                        for address in listen.into_iter().rev() {
+                            book.remember(peer_id, address);
+                        }
+                        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);
+                }
+                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);
+}