net: drastically simplify auto-addr logic by centralizing everything in Hosts with the ringbuffer

src/net/channel.rs

@@ -39,7 +39,7 @@ use url::Url;
 
 use super::{
     dnet::{self, dnetev, DnetEvent},
-    hosts::HostColor,
+    hosts::{HostColor, HostsPtr},
     message,
     message::{SerializedMessage, VersionMessage},
     message_publisher::{MessageSubscription, MessageSubsystem},
@@ -507,9 +507,14 @@ impl Channel {
         session.type_id()
     }
 
-    pub(in crate::net) fn p2p(&self) -> P2pPtr {
+    #[inline]
+    pub fn p2p(&self) -> P2pPtr {
         self.session().p2p()
     }
+    #[inline]
+    pub fn hosts(&self) -> HostsPtr {
+        self.p2p().hosts()
+    }
 
     fn is_eof_error(err: &Error) -> bool {
         match err {

src/net/hosts.rs

@@ -20,10 +20,10 @@ use log::{debug, error, info, trace, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
 use smol::lock::RwLock as AsyncRwLock;
 use std::{
-    collections::{HashMap, VecDeque},
+    collections::HashMap,
     fmt, fs,
     fs::File,
-    net::Ipv6Addr,
+    net::{IpAddr, Ipv6Addr},
     sync::{
         atomic::{AtomicBool, Ordering},
         Arc, Mutex as SyncMutex, RwLock,
@@ -41,7 +41,9 @@ use crate::{
     system::{Publisher, PublisherPtr, Subscription},
     util::{
         file::{load_file, save_file},
+        most_frequent_or_any,
         path::expand_path,
+        ringbuffer::RingBuffer,
     },
     Error, Result,
 };
@@ -873,7 +875,7 @@ pub struct Hosts {
     pub(in crate::net) ipv6_available: AtomicBool,
 
     /// Auto self discovered addresses. Used for filtering self connections.
-    auto_self_addrs: SyncMutex<VecDeque<Ipv6Addr>>,
+    auto_self_addrs: SyncMutex<RingBuffer<Ipv6Addr, 20>>,
 
     /// Pointer to configured P2P settings
     settings: Arc<AsyncRwLock<Settings>>,
@@ -890,7 +892,7 @@ impl Hosts {
             disconnect_publisher: Publisher::new(),
             last_connection: SyncMutex::new(Instant::now()),
             ipv6_available: AtomicBool::new(true),
-            auto_self_addrs: SyncMutex::new(VecDeque::new()),
+            auto_self_addrs: SyncMutex::new(RingBuffer::new()),
             settings,
         })
     }
@@ -987,7 +989,7 @@ impl Hosts {
         trace!(target: "net::hosts::check_addrs()", "[START]");
 
         let seeds = self.settings.read().await.seeds.clone();
-        let external_addrs = self.settings.read().await.external_addrs.clone();
+        let external_addrs = self.external_addrs().await;
 
         for (host, last_seen) in hosts {
             // Print a warning if we are trying to connect to a seed node in
@@ -1272,7 +1274,7 @@ impl Hosts {
 
             if !settings.localnet {
                 // Our own external addresses should never enter the hosts set.
-                for ext in &settings.external_addrs {
+                for ext in self.external_addrs().await {
                     if host == ext.host().unwrap() {
                         debug!(
                             target: "net::hosts::filter_addresses",
@@ -1281,22 +1283,6 @@ impl Hosts {
                         continue 'addr_loop
                     }
                 }
-
-                // Cloning this vec means a linear scan and copy here, so just doing a linear
-                // scan here is faster.
-                let auto_self_addrs = self.auto_self_addrs.lock().unwrap();
-                match host {
-                    Host::Ipv6(host_addr) => {
-                        if auto_self_addrs.contains(&host_addr) {
-                            debug!(
-                                target: "net::hosts::filter_addresses",
-                                "[{}] is our own auto discovered addr. Skipping", addr_,
-                            );
-                            continue 'addr_loop
-                        }
-                    }
-                    _ => {}
-                }
             } else {
                 // On localnet, make sure ours ports don't enter the host set.
                 for ext in &settings.external_addrs {
@@ -1497,13 +1483,63 @@ impl Hosts {
         Ok(())
     }
 
+    /// Upon version exchange, the node reports our external network address to us.
+    /// Accumulate them here in a ring buffer.
     pub(in crate::net) fn add_auto_addr(&self, addr: Ipv6Addr) {
         let mut auto_addrs = self.auto_self_addrs.lock().unwrap();
-        auto_addrs.push_back(addr);
-        // This is a ringbuf so remove front elements
-        while auto_addrs.len() > 20 {
-            auto_addrs.pop_front().unwrap();
+        auto_addrs.push(addr);
+    }
+
+    /// Pick the most frequent occuring reported external address from other nodes as
+    /// our auto ipv6 address.
+    pub fn guess_auto_addr(&self) -> Option<Ipv6Addr> {
+        let mut auto_addrs = self.auto_self_addrs.lock().unwrap();
+        let items = auto_addrs.make_contiguous();
+        most_frequent_or_any(items)
+    }
+
+    /// The external_addrs is set by the user but we need the actual addresses.
+    /// If the external_addr is set to `[::]` (unspecified), then replace it with the
+    /// the best guess from `guess_auto_addr()`.
+    /// Also if the port is 0, we lookup the port from the `InboundSession`.
+    pub async fn external_addrs(&self) -> Vec<Url> {
+        let mut external_addrs = self.settings.read().await.external_addrs.clone();
+        for ext_addr in &mut external_addrs {
+            let _ = self.patch_external_addr(ext_addr);
+        }
+        external_addrs
+    }
+
+    /// Make a best effort guess from the most frequently reported ipv6 auto address
+    /// to set any unspecified ipv6 addrs: `external_addrs = ["tcp://[::]:1365"]`.
+    fn patch_external_addr(&self, ext_addr: &mut Url) -> Option<()> {
+        if ext_addr.scheme() != "tcp" && ext_addr.scheme() != "tcp+tls" {
+            return None
+        }
+
+        let ext_host = ext_addr.host()?;
+        // Is it an Ipv6 listener?
+        let Host::Ipv6(ext_ip) = ext_host else { return None };
+        // We are only interested if it's [::]
+        if !ext_ip.is_unspecified() {
+            return None
+        }
+
+        // We should loop over the endpoints from the listeners
+        // But inbound session should be changed so the acceptors and listeners
+        // are accessible.
+        /*
+        let Some(mut port) = inbound.port() else { continue };
+        if port == 0 {
         }
+        */
+
+        // Get our auto-discovered IP
+        let auto_addr = self.guess_auto_addr()?;
+
+        // Do the actual replacement of the host part of the URL
+        ext_addr.set_ip_host(IpAddr::V6(auto_addr)).ok()?;
+        Some(())
     }
 }
 

src/net/protocol/protocol_address.rs

@@ -20,11 +20,10 @@ use async_trait::async_trait;
 use log::debug;
 use smol::{lock::RwLock as AsyncRwLock, Executor};
 use std::{sync::Arc, time::UNIX_EPOCH};
-use url::{Host, Url};
 
 use super::{
     super::{
-        channel::{Channel, ChannelPtr},
+        channel::ChannelPtr,
         hosts::{HostColor, HostsPtr},
         message::{AddrsMessage, GetAddrsMessage},
         message_publisher::MessageSubscription,
@@ -226,11 +225,7 @@ impl ProtocolAddress {
             return Ok(())
         }
 
-        let mut external_addrs = self.settings.read().await.external_addrs.clone();
-
-        for external_addr in &mut external_addrs {
-            let _ = Self::patch_external_addr(&self.channel, external_addr);
-        }
+        let external_addrs = self.channel.hosts().external_addrs().await;
 
         if external_addrs.is_empty() {
             debug!(
@@ -262,46 +257,6 @@ impl ProtocolAddress {
 
         Ok(())
     }
-
-    /// If the external_addr is set to `[::]` (unspecified), then replace it with the
-    /// ip address reported to us by the version exchange.
-    ///
-    /// Also used by ProtocolSeed.
-    pub(super) fn patch_external_addr(channel: &Channel, external_addr: &mut Url) -> Option<()> {
-        if external_addr.scheme() != "tcp" && external_addr.scheme() != "tcp+tls" {
-            return None
-        }
-
-        let external_addr_host = external_addr.host()?;
-        // Is it an Ipv6 listener?
-        match external_addr_host {
-            Host::Ipv6(addr) => {
-                // We are only interested if it's [::]
-                if !addr.is_unspecified() {
-                    return None
-                }
-            }
-            _ => return None,
-        }
-
-        // We should loop over the endpoints from the listeners
-        // But inbound session should be changed so the acceptors and listeners
-        // are accessible.
-        /*
-        let Some(mut port) = inbound.port() else { continue };
-        if port == 0 {
-        }
-        */
-
-        // Get our auto-discovered IP
-        let version = channel.get_version();
-        // Check if the returned value in the version exchange is ipv6.
-        // Maybe this outbound is using another transport or ipv4.
-        let _ = version.get_ipv6_addr()?;
-
-        external_addr.set_host(version.connect_recv_addr.host_str()).ok()?;
-        Some(())
-    }
 }
 
 #[async_trait]

src/net/protocol/protocol_seed.rs

@@ -30,7 +30,6 @@ use super::{
         p2p::P2pPtr,
         settings::Settings,
     },
-    protocol_address::ProtocolAddress,
     protocol_base::{ProtocolBase, ProtocolBasePtr},
 };
 use crate::Result;
@@ -63,11 +62,7 @@ impl ProtocolSeed {
             "[START] channel address={}", self.channel.address(),
         );
 
-        let mut external_addrs = self.settings.read().await.external_addrs.clone();
-
-        for external_addr in &mut external_addrs {
-            let _ = ProtocolAddress::patch_external_addr(&self.channel, external_addr);
-        }
+        let external_addrs = self.channel.hosts().external_addrs().await;
 
         if external_addrs.is_empty() {
             debug!(

src/net/protocol/protocol_version.rs

@@ -16,17 +16,16 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{
-    sync::Arc,
-    time::{Duration, UNIX_EPOCH},
-};
-
 use futures::{
     future::{join_all, select, Either},
     pin_mut,
 };
 use log::{debug, error};
 use smol::{lock::RwLock as AsyncRwLock, Executor, Timer};
+use std::{
+    sync::Arc,
+    time::{Duration, UNIX_EPOCH},
+};
 
 use super::super::{
     channel::ChannelPtr,
@@ -153,9 +152,10 @@ impl ProtocolVersion {
         let settings = self.settings.read().await;
         let node_id = settings.node_id.clone();
         let app_version = settings.app_version.clone();
-        let external_addrs = settings.external_addrs.clone();
         drop(settings);
 
+        let external_addrs = self.channel.hosts().external_addrs().await;
+
         let version = VersionMessage {
             node_id,
             version: app_version.clone(),

src/util/mod.rs

@@ -41,3 +41,30 @@ pub mod ringbuffer;
 /// This is an insecure PRNG used for simulations and tests.
 #[cfg(feature = "rand")]
 pub mod pcg;
+
+/// Return the most frequent element in vec or just any item.
+pub fn most_frequent_or_any<T: Eq + Clone>(items: &[T]) -> Option<T> {
+    if items.is_empty() {
+        return None;
+    }
+
+    let mut max_count = 0;
+    let mut most_freq = &items[0];
+
+    for i in 0..items.len() {
+        let mut count = 0;
+
+        for j in 0..items.len() {
+            if items[i] == items[j] {
+                count += 1;
+            }
+        }
+
+        if count > max_count {
+            max_count = count;
+            most_freq = &items[i];
+        }
+    }
+
+    Some(most_freq.clone())
+}