libp2p: dial newly announced peer addresses

6 files changed, 118 insertions, 6 deletions

diff --git a/src/transport/libp2p/behaviour.rs b/src/transport/libp2p/behaviour.rs
index e9b156c..a2e8e91 100644
--- a/src/transport/libp2p/behaviour.rs
+++ b/src/transport/libp2p/behaviour.rs
@@ -244,7 +244,9 @@ impl Behaviour {
             dht
         });
         let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
-            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
+            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")))
+            // Connected peers hear of a new interface at once (`explore`).
+            .with_push_listen_addr_updates(true);
         let relay = relay.then(|| {
             ::libp2p::relay::Behaviour::new(
                 peer_id,
diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs
index 44b2694..7de4be7 100644
--- a/src/transport/libp2p/book.rs
+++ b/src/transport/libp2p/book.rs
@@ -41,6 +41,9 @@ pub(super) struct Book {
     /// connection to the peer closes, as a peer restarted elsewhere is found by what
     /// the others say.
     pub(super) identified: HashMap<PeerId, Vec<Multiaddr>>,
+    /// Addresses of directly connected peers dialled for a better path (`explore`):
+    /// not again while connected, unless our own interfaces change.
+    pub(super) explored: HashMap<PeerId, std::collections::HashSet<Multiaddr>>,
     pub(super) listening: Vec<Multiaddr>,
     /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`).
     pub(super) mapped: Vec<Multiaddr>,
@@ -213,6 +216,7 @@ impl Book {
         for peer in &stale {
             self.addresses.remove(peer);
             self.identified.remove(peer);
+            self.explored.remove(peer);
             self.relays.remove(peer);
             self.public_relays.remove(peer);
             self.continuum.remove(peer);
@@ -475,6 +479,38 @@ impl Book {
         addresses.truncate(ADDRESSES_PER_PEER);
     }
 
+    /// The addresses `explore` dials `peer` at, noted as dialled: a node of the swarm
+    /// connected directly, at addresses of its not dialled so yet, and with `again`
+    /// not only those at an IP it isn't connected at (an inbound connection's remote
+    /// port isn't where the peer listens).
+    pub(super) fn to_explore(&mut self, peer: PeerId, again: bool) -> Vec<Multiaddr> {
+        if !self.continuum.contains(&peer)
+            || book_connection(&self.connections, &peer) != Some(Connection::Direct)
+        {
+            return Vec::new();
+        }
+        let used: Vec<Option<std::net::IpAddr>> = self
+            .connections
+            .values()
+            .filter(|(connected, relayed, _)| *connected == peer && !relayed)
+            .map(|(_, _, address)| ip_of(address))
+            .collect();
+        let tried = self.explored.entry(peer).or_default();
+        let candidates: Vec<Multiaddr> = self
+            .addresses
+            .get(&peer)
+            .into_iter()
+            .flatten()
+            .filter(|address| is_ip(address) && !left_out(&self.ignored, address))
+            .filter(|address| again || !used.contains(&ip_of(address)))
+            .filter(|address| !tried.contains(*address))
+            .take(EXPLORED_AT_ONCE)
+            .cloned()
+            .collect();
+        tried.extend(candidates.iter().cloned());
+        candidates
+    }
+
     /// `peer`, connected, identified itself at `listen` (`published`): its direct
     /// addresses are those from now on, while connected (`remember`); those through
     /// relays and by name stay. Kept in order of `preference`: of a peer with many,
@@ -520,7 +556,7 @@ impl Book {
 }
 
 /// The IP address `address` starts with, if any.
-fn ip_of(address: &Multiaddr) -> Option<std::net::IpAddr> {
+pub(super) fn ip_of(address: &Multiaddr) -> Option<std::net::IpAddr> {
     match address.iter().next() {
         Some(Protocol::Ip4(ip)) => Some(ip.into()),
         Some(Protocol::Ip6(ip)) => Some(ip.into()),
diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs
index 145592d..9b4259f 100644
--- a/src/transport/libp2p/driver.rs
+++ b/src/transport/libp2p/driver.rs
@@ -220,6 +220,7 @@ pub(super) async fn run(
                         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 {
@@ -267,9 +268,23 @@ pub(super) async fn run(
                     if serving && !left_out && !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);
+                    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, .. } => {
@@ -501,6 +516,7 @@ pub(super) async fn run(
                         }
                     }
                     upgrade(&mut swarm, &book, peer_id, relaying.local);
+                    explore(&mut swarm, &book, peer_id, false);
                 }
                 SwarmEvent::Behaviour(BehaviourEvent::Mdns(
                     ::libp2p::mdns::Event::Discovered(found),
diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs
index 293558b..a4fefc3 100644
--- a/src/transport/libp2p/mod.rs
+++ b/src/transport/libp2p/mod.rs
@@ -32,7 +32,11 @@
 //! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`),
 //! by the side with the smaller peer ID: it keeps the direct one with the shortest round
 //! trip (libp2p's ping, on each connection), so a path over the LAN wins over one
-//! through a VPN, whichever opened first.
+//! through a VPN, whichever opened first. A peer connected directly is dialled again
+//! where it might be reached better (`explore`): at addresses it tells of later (a new
+//! interface: identify pushes them at once), and at all of them when this node gains
+//! an interface. Otherwise a connection made over the WLAN would stay there once
+//! ethernet comes.
 //!
 //! A relay says so where it publishes its addresses: next to each, the same address
 //! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the
@@ -166,6 +170,8 @@ const RELAY_PAUSE: Duration = Duration::from_secs(600);
 /// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to
 /// `RELAY_PAUSE`.
 const UPGRADE_RETRY: Duration = Duration::from_secs(30);
+/// A directly connected peer's addresses dialled at once for a better path (`explore`).
+const EXPLORED_AT_ONCE: usize = 4;
 /// A peer nobody mentioned for this long (the member list, identify, mDNS, a
 /// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted
 /// without their libp2p key, whose new ones the member list has since. Members are
diff --git a/src/transport/libp2p/relays.rs b/src/transport/libp2p/relays.rs
index dca031c..41d7e1d 100644
--- a/src/transport/libp2p/relays.rs
+++ b/src/transport/libp2p/relays.rs
@@ -175,6 +175,29 @@ pub(super) fn upgrade(
     }
 }
 
+/// Dials `peer`, connected directly, at its other addresses, in case one is a better
+/// path: one that came since (a new interface of its, identify tells), or, `again`
+/// (one of ours came), every one, as our side may reach them better now, the one in
+/// use too. Each once while connected (`Book::explored`); a connection that opens is
+/// pinged at once, and of several the fastest is kept (`Book::duplicates`). Without
+/// it, a connection stays on the path it was made on (the WLAN) when a better one comes
+/// (ethernet): a connected peer isn't dialled again.
+pub(super) fn explore(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, peer: PeerId, again: bool) {
+    let candidates = lock(book).to_explore(peer, again);
+    if candidates.is_empty() {
+        return;
+    }
+    debug!(peer = %Addressed(SCHEME, &peer), addresses = ?candidates, "Connected directly; dialling other addresses, for a better path");
+    if let Err(error) = swarm.dial(
+        DialOpts::peer_id(peer)
+            .addresses(candidates)
+            .condition(PeerCondition::Always)
+            .build(),
+    ) {
+        debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Couldn't dial for a better path");
+    }
+}
+
 /// Tries hole punching again with the peers it failed with that are due (`punching`).
 /// libp2p punches as a relayed connection opens, started by the side that was dialled
 /// through the relay; that side is where a punch's failure is told, so it is this one.
diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs
index 3d11913..ed23b34 100644
--- a/src/transport/libp2p/tests.rs
+++ b/src/transport/libp2p/tests.rs
@@ -696,6 +696,35 @@ fn ignored_addresses_not_kept() {
     assert!(!book.knows(them, &[via_vpn]));
 }
 
+/// A node connected directly over the WLAN dials the ethernet address it later hears
+/// of, once; and when an interface of ours comes, every address again, the one in use
+/// too.
+#[test]
+fn better_paths_explored() {
+    let mut book = Book::default();
+    let them = peer();
+    let wlan: Multiaddr = "/ip4/192.168.1.7/tcp/9661".parse().unwrap();
+    let ethernet: Multiaddr = "/ip4/192.168.2.7/tcp/9661".parse().unwrap();
+    book.continuum.insert(them);
+    book.remember(them, wlan.clone());
+    // Connected (inbound: from another port) over the WLAN.
+    book.connections.insert(
+        ConnectionId::new_unchecked(1),
+        (them, false, "/ip4/192.168.1.7/tcp/40000".parse().unwrap()),
+    );
+    assert!(book.to_explore(them, false).is_empty());
+    book.identified(them, vec![wlan.clone(), ethernet.clone()]);
+    assert_eq!(book.to_explore(them, false), [ethernet.clone()]);
+    assert!(book.to_explore(them, false).is_empty());
+    // An interface of ours came (`driver`): all of them again.
+    book.explored.clear();
+    let mut again = book.to_explore(them, true);
+    again.sort();
+    let mut both = vec![wlan, ethernet];
+    both.sort();
+    assert_eq!(again, both);
+}
+
 /// Peers nobody mentions any more are forgotten, but connected ones.
 #[test]
 fn stale_peers_are_forgotten() {