fix(relay): relayed external address not expiring when stopping to li…

…sten through relay

Cargo.toml

@@ -97,7 +97,7 @@ libp2p-ping = { version = "0.45.0", path = "protocols/ping" }
 libp2p-plaintext = { version = "0.42.0", path = "transports/plaintext" }
 libp2p-pnet = { version = "0.25.0", path = "transports/pnet" }
 libp2p-quic = { version = "0.11.1", path = "transports/quic" }
-libp2p-relay = { version = "0.18.0", path = "protocols/relay" }
+libp2p-relay = { version = "0.18.1", path = "protocols/relay" }
 libp2p-rendezvous = { version = "0.15.0", path = "protocols/rendezvous" }
 libp2p-request-response = { version = "0.27.0", path = "protocols/request-response" }
 libp2p-server = { version = "0.12.7", path = "misc/server" }

protocols/dcutr/tests/lib.rs

@@ -57,8 +57,7 @@ async fn connect() {
 
     let dst_relayed_addr = relay_tcp_addr
         .with(Protocol::P2p(relay_peer_id))
-        .with(Protocol::P2pCircuit)
-        .with(Protocol::P2p(dst_peer_id));
+        .with(Protocol::P2pCircuit);
     dst.listen_on(dst_relayed_addr.clone()).unwrap();
 
     wait_for_reservation(
@@ -70,7 +69,8 @@ async fn connect() {
     .await;
     async_std::task::spawn(dst.loop_on_next());
 
-    src.dial_and_wait(dst_relayed_addr.clone()).await;
+    src.dial_and_wait(dst_relayed_addr.with(Protocol::P2p(dst_peer_id)))
+        .await;
 
     let dst_addr = dst_tcp_addr.with(Protocol::P2p(dst_peer_id));
 

protocols/relay/CHANGELOG.md

@@ -1,3 +1,9 @@
+## 0.18.1
+
+- Fix relayed `ExternalAddr` not expiring when stopping to listen through relay.
+  Removing `/p2p/<local_peer_id>` part from the listen and external addresses reported by the relay `Behaviour`.
+  See [PR 5577](https://github.com/libp2p/rust-libp2p/pull/5577).
+
 ## 0.18.0
 
 <!-- Update to libp2p-swarm v0.45.0 -->

protocols/relay/Cargo.toml

@@ -3,7 +3,7 @@ name = "libp2p-relay"
 edition = "2021"
 rust-version = { workspace = true }
 description = "Communications relaying for libp2p"
-version = "0.18.0"
+version = "0.18.1"
 authors = ["Parity Technologies <admin@parity.io>", "Max Inden <mail@max-inden.de>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

protocols/relay/src/priv_client.rs

@@ -33,15 +33,15 @@ use futures::future::{BoxFuture, FutureExt};
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::ready;
 use futures::stream::StreamExt;
-use libp2p_core::multiaddr::Protocol;
 use libp2p_core::transport::PortUse;
 use libp2p_core::{Endpoint, Multiaddr};
 use libp2p_identity::PeerId;
 use libp2p_swarm::behaviour::{ConnectionClosed, ConnectionEstablished, FromSwarm};
 use libp2p_swarm::dial_opts::DialOpts;
 use libp2p_swarm::{
-    dummy, ConnectionDenied, ConnectionHandler, ConnectionId, DialFailure, NetworkBehaviour,
-    NotifyHandler, Stream, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
+    dummy, ConnectionDenied, ConnectionHandler, ConnectionId, DialFailure, ExpiredListenAddr,
+    NetworkBehaviour, NewListenAddr, NotifyHandler, Stream, THandler, THandlerInEvent,
+    THandlerOutEvent, ToSwarm,
 };
 use std::collections::{hash_map, HashMap, VecDeque};
 use std::io::{Error, ErrorKind, IoSlice};
@@ -71,12 +71,6 @@ pub enum Event {
     },
 }
 
-#[derive(Debug, Copy, Clone, PartialEq, Eq)]
-enum ReservationStatus {
-    Pending,
-    Confirmed,
-}
-
 /// [`NetworkBehaviour`] implementation of the relay client
 /// functionality of the circuit relay v2 protocol.
 pub struct Behaviour {
@@ -87,11 +81,6 @@ pub struct Behaviour {
     /// connection.
     directly_connected_peers: HashMap<PeerId, Vec<ConnectionId>>,
 
-    /// Stores the address of a pending or confirmed reservation.
-    ///
-    /// This is indexed by the [`ConnectionId`] to a relay server and the address is the `/p2p-circuit` address we reserved on it.
-    reservation_addresses: HashMap<ConnectionId, (Multiaddr, ReservationStatus)>,
-
     /// Queue of actions to return when polled.
     queued_actions: VecDeque<ToSwarm<Event, Either<handler::In, Void>>>,
 
@@ -105,7 +94,6 @@ pub fn new(local_peer_id: PeerId) -> (Transport, Behaviour) {
         local_peer_id,
         from_transport,
         directly_connected_peers: Default::default(),
-        reservation_addresses: Default::default(),
         queued_actions: Default::default(),
         pending_handler_commands: Default::default(),
     };
@@ -140,12 +128,6 @@ impl Behaviour {
                     unreachable!("`on_connection_closed` for unconnected peer.")
                 }
             };
-            if let Some((addr, ReservationStatus::Confirmed)) =
-                self.reservation_addresses.remove(&connection_id)
-            {
-                self.queued_actions
-                    .push_back(ToSwarm::ExternalAddrExpired(addr));
-            }
         }
     }
 }
@@ -220,8 +202,19 @@ impl NetworkBehaviour for Behaviour {
             FromSwarm::ConnectionClosed(connection_closed) => {
                 self.on_connection_closed(connection_closed)
             }
+            FromSwarm::NewListenAddr(NewListenAddr { addr, .. }) => {
+                if addr.is_relayed() {
+                    self.queued_actions
+                        .push_back(ToSwarm::ExternalAddrConfirmed(addr.clone()));
+                }
+            }
+            FromSwarm::ExpiredListenAddr(ExpiredListenAddr { addr, .. }) => {
+                if addr.is_relayed() {
+                    self.queued_actions
+                        .push_back(ToSwarm::ExternalAddrExpired(addr.clone()));
+                }
+            }
             FromSwarm::DialFailure(DialFailure { connection_id, .. }) => {
-                self.reservation_addresses.remove(&connection_id);
                 self.pending_handler_commands.remove(&connection_id);
             }
             _ => {}
@@ -231,7 +224,7 @@ impl NetworkBehaviour for Behaviour {
     fn on_connection_handler_event(
         &mut self,
         event_source: PeerId,
-        connection: ConnectionId,
+        _: ConnectionId,
         handler_event: THandlerOutEvent<Self>,
     ) {
         let handler_event = match handler_event {
@@ -241,17 +234,6 @@ impl NetworkBehaviour for Behaviour {
 
         let event = match handler_event {
             handler::Event::ReservationReqAccepted { renewal, limit } => {
-                let (addr, status) = self
-                    .reservation_addresses
-                    .get_mut(&connection)
-                    .expect("Relay connection exist");
-
-                if !renewal && *status == ReservationStatus::Pending {
-                    *status = ReservationStatus::Confirmed;
-                    self.queued_actions
-                        .push_back(ToSwarm::ExternalAddrConfirmed(addr.clone()));
-                }
-
                 Event::ReservationReqAccepted {
                     relay_peer_id: event_source,
                     renewal,
@@ -292,42 +274,18 @@ impl NetworkBehaviour for Behaviour {
                     .get(&relay_peer_id)
                     .and_then(|cs| cs.first())
                 {
-                    Some(connection_id) => {
-                        self.reservation_addresses.insert(
-                            *connection_id,
-                            (
-                                relay_addr
-                                    .with(Protocol::P2p(relay_peer_id))
-                                    .with(Protocol::P2pCircuit)
-                                    .with(Protocol::P2p(self.local_peer_id)),
-                                ReservationStatus::Pending,
-                            ),
-                        );
-
-                        ToSwarm::NotifyHandler {
-                            peer_id: relay_peer_id,
-                            handler: NotifyHandler::One(*connection_id),
-                            event: Either::Left(handler::In::Reserve { to_listener }),
-                        }
-                    }
+                    Some(connection_id) => ToSwarm::NotifyHandler {
+                        peer_id: relay_peer_id,
+                        handler: NotifyHandler::One(*connection_id),
+                        event: Either::Left(handler::In::Reserve { to_listener }),
+                    },
                     None => {
                         let opts = DialOpts::peer_id(relay_peer_id)
                             .addresses(vec![relay_addr.clone()])
                             .extend_addresses_through_behaviour()
                             .build();
                         let relayed_connection_id = opts.connection_id();
 
-                        self.reservation_addresses.insert(
-                            relayed_connection_id,
-                            (
-                                relay_addr
-                                    .with(Protocol::P2p(relay_peer_id))
-                                    .with(Protocol::P2pCircuit)
-                                    .with(Protocol::P2p(self.local_peer_id)),
-                                ReservationStatus::Pending,
-                            ),
-                        );
-
                         self.pending_handler_commands
                             .insert(relayed_connection_id, handler::In::Reserve { to_listener });
                         ToSwarm::Dial { opts }

protocols/relay/src/priv_client/transport.rs

@@ -33,7 +33,7 @@ use futures::stream::{Stream, StreamExt};
 use libp2p_core::multiaddr::{Multiaddr, Protocol};
 use libp2p_core::transport::{DialOpts, ListenerId, TransportError, TransportEvent};
 use libp2p_identity::PeerId;
-use std::collections::VecDeque;
+use std::collections::{HashSet, VecDeque};
 use std::pin::Pin;
 use std::task::{Context, Poll, Waker};
 use thiserror::Error;
@@ -154,6 +154,7 @@ impl libp2p_core::Transport for Transport {
             from_behaviour,
             is_closed: false,
             waker: None,
+            listened_addrs: Default::default(),
         };
         self.listeners.push(listener);
         Ok(())
@@ -311,6 +312,8 @@ pub(crate) struct Listener {
     /// the sender side of the `from_behaviour` channel is dropped.
     is_closed: bool,
     waker: Option<Waker>,
+    /// Addresses listened through relay
+    listened_addrs: HashSet<Multiaddr>,
 }
 
 impl Listener {
@@ -368,14 +371,45 @@ impl Stream for Listener {
                         self.queued_events.is_empty(),
                         "Assert empty due to previous `pop_front` attempt."
                     );
-                    // Returned as [`ListenerEvent::NewAddress`] in next iteration of loop.
-                    self.queued_events = addrs
+
+                    let reserved_addresses = addrs
                         .into_iter()
-                        .map(|listen_addr| TransportEvent::NewAddress {
-                            listener_id: self.listener_id,
-                            listen_addr,
+                        .map(|mut addr| {
+                            // Every transport (tcp / quic / etc) gives addresses without the last
+                            // p2p part, so pop it if present.
+                            if matches!(addr.iter().last(), Some(Protocol::P2p(_))) {
+                                addr.pop();
+                            }
+                            addr
                         })
-                        .collect();
+                        .collect::<HashSet<_>>();
+
+                    let expired_addresses = self
+                        .listened_addrs
+                        .difference(&reserved_addresses)
+                        .map(Clone::clone)
+                        .collect::<Vec<_>>();
+                    let new_addresses = reserved_addresses
+                        .difference(&self.listened_addrs)
+                        .map(Clone::clone)
+                        .collect::<Vec<_>>();
+                    let listener_id = self.listener_id;
+
+                    for listen_addr in expired_addresses {
+                        self.listened_addrs.remove(&listen_addr);
+                        self.queued_events
+                            .push_back(TransportEvent::AddressExpired {
+                                listener_id,
+                                listen_addr,
+                            });
+                    }
+                    for listen_addr in new_addresses {
+                        self.listened_addrs.insert(listen_addr.clone());
+                        self.queued_events.push_back(TransportEvent::NewAddress {
+                            listener_id,
+                            listen_addr,
+                        });
+                    }
                 }
                 ToListenerMsg::IncomingRelayedConnection {
                     stream,