From 235411a261441df6c0ddb1c4d7f99cf06b8f3c55 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 04:20:41 +0100 Subject: libp2p: explode! --- src/transport/libp2p/driver.rs | 527 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 527 insertions(+) create mode 100644 src/transport/libp2p/driver.rs (limited to 'src/transport/libp2p/driver.rs') 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, + done: oneshot::Sender>, + }, + /// Connect to `address` to find out which peer is there. + Identify { + address: Multiaddr, + done: oneshot::Sender>, + }, + /// 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, + mut commands: mpsc::UnboundedReceiver, + again: mpsc::UnboundedSender, + book: Arc>, + relaying: Relaying, + stopping: watch::Sender, +) { + 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>>> = HashMap::new(); + let mut identifying: HashMap< + ConnectionId, + (Multiaddr, oneshot::Sender>), + > = 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 = HashMap::new(); + // Connections closed as duplicates, until they are: their "Connection closed" says so. + let mut closing_duplicates: std::collections::HashSet = + 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 = 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); +} -- cgit v1.3.1