Testing branch with unreleased fixes

Cargo.toml

@@ -22,6 +22,7 @@ ed25519-dalek = { version = "2.1.1", features = ["rand_core"] }
 futures = "0.3.27"
 futures-timer = "3.0.3"
 indexmap = { version = "2.9.0", features = ["std"] }
+ip_network = "0.4"
 libc = "0.2.158"
 mockall = "0.13.1"
 multiaddr = "0.17.0"

src/transport/dummy.rs

@@ -25,7 +25,7 @@ use crate::{
     types::ConnectionId,
 };
 
-use futures::Stream;
+use futures::{future::BoxFuture, Stream};
 use multiaddr::Multiaddr;
 
 use std::{
@@ -71,8 +71,8 @@ impl Transport for DummyTransport {
         Ok(())
     }
 
-    fn accept(&mut self, _: ConnectionId) -> crate::Result<()> {
-        Ok(())
+    fn accept(&mut self, _: ConnectionId) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
+        Ok(Box::pin(async { Ok(()) }))
     }
 
     fn accept_pending(&mut self, _connection_id: ConnectionId) -> crate::Result<()> {

src/transport/manager/address.rs

@@ -20,6 +20,7 @@
 
 use crate::{error::DialError, PeerId};
 
+use ip_network::IpNetwork;
 use multiaddr::{Multiaddr, Protocol};
 use multihash::Multihash;
 
@@ -39,7 +40,13 @@ pub mod scores {
     /// Score for providing an invalid address.
     ///
     /// This address can never be reached.
-    pub const ADDRESS_FAILURE: i32 = i32::MIN;
+    pub const ADDRESS_FAILURE: i32 = -200i32;
+
+    /// Initial score for public/global addresses.
+    ///
+    /// This gives public addresses a slight priority over private addresses
+    /// when all addresses are untested (private addresses start at 0).
+    pub const PUBLIC_ADDRESS_BONUS: i32 = 1i32;
 }
 
 #[allow(clippy::derived_hash_with_manual_eq)]
@@ -70,7 +77,7 @@ impl AddressRecord {
             address
         };
 
-        Self { address, score }
+        Self::from_raw_multiaddr_with_score(address, score)
     }
 
     /// Create `AddressRecord` from `Multiaddr`.
@@ -82,10 +89,7 @@ impl AddressRecord {
             return None;
         }
 
-        Some(AddressRecord {
-            address,
-            score: 0i32,
-        })
+        Some(Self::from_raw_multiaddr_with_score(address, 0))
     }
 
     /// Create `AddressRecord` from `Multiaddr`.
@@ -94,10 +98,7 @@ impl AddressRecord {
     ///
     /// Please consider using [`Self::from_multiaddr`] from the transport manager code.
     pub fn from_raw_multiaddr(address: Multiaddr) -> AddressRecord {
-        AddressRecord {
-            address,
-            score: 0i32,
-        }
+        Self::from_raw_multiaddr_with_score(address, 0)
     }
 
     /// Create `AddressRecord` from `Multiaddr`.
@@ -106,7 +107,14 @@ impl AddressRecord {
     ///
     /// Please consider using [`Self::from_multiaddr`] from the transport manager code.
     pub fn from_raw_multiaddr_with_score(address: Multiaddr, score: i32) -> AddressRecord {
-        AddressRecord { address, score }
+        if is_global_multiaddr(&address) {
+            Self {
+                address,
+                score: score.saturating_add(scores::PUBLIC_ADDRESS_BONUS),
+            }
+        } else {
+            Self { address, score }
+        }
     }
 
     /// Get address score.
@@ -122,10 +130,31 @@ impl AddressRecord {
 
     /// Update score of an address.
     pub fn update_score(&mut self, score: i32) {
-        self.score = self.score.saturating_add(score);
+        self.score = score;
     }
 }
 
+/// Check if a multiaddr represents a global/public address.
+///
+/// DNS addresses are considered potentially public.
+fn is_global_multiaddr(address: &Multiaddr) -> bool {
+    for protocol in address.iter() {
+        match protocol {
+            Protocol::Ip4(ip) => return IpNetwork::from(ip).is_global(),
+            Protocol::Ip6(ip) => return IpNetwork::from(ip).is_global(),
+            // DNS addresses could resolve to public IPs, treat as potentially public.
+            // Ideally we need to resolve DNS to check the actual IPs. However, this
+            // is a more complex operation that requires async DNS resolution in the
+            // transport manager context / transport layer.
+            Protocol::Dns(_) | Protocol::Dns4(_) | Protocol::Dns6(_) => return true,
+            _ => continue,
+        }
+    }
+
+    // Consider the address as non-global if no IP or DNS component is found
+    false
+}
+
 impl PartialEq for AddressRecord {
     fn eq(&self, other: &Self) -> bool {
         self.score.eq(&other.score)
@@ -220,10 +249,17 @@ impl AddressStore {
     /// Insert the address record into [`AddressStore`] with the provided score.
     ///
     /// If the address is not in the store, it will be inserted.
-    /// Otherwise, the score and connection ID will be updated.
+    /// Otherwise, the score will be updated only for significant changes (connection events),
+    /// not for re-adding the same address which would incorrectly accumulate the public address
+    /// bonus.
     pub fn insert(&mut self, record: AddressRecord) {
         if let Entry::Occupied(mut occupied) = self.addresses.entry(record.address.clone()) {
-            occupied.get_mut().update_score(record.score);
+            // Only update score for connection events (score magnitude > PUBLIC_ADDRESS_BONUS).
+            // Re-adding an address has score 0 (private) or 1 (public with bonus), and should
+            // not accumulate the bonus repeatedly.
+            if record.score.abs() > scores::PUBLIC_ADDRESS_BONUS {
+                occupied.get_mut().update_score(record.score);
+            }
             return;
         }
 
@@ -479,12 +515,98 @@ mod tests {
         assert_eq!(store.addresses.len(), 1);
         assert_eq!(store.addresses.get(record.address()).unwrap(), &record);
 
-        // This time the record is updated.
+        // This time the record score is replaced (not accumulated).
         store.insert(record.clone());
 
         assert_eq!(store.addresses.len(), 1);
         let store_record = store.addresses.get(record.address()).unwrap();
-        assert_eq!(store_record.score, record.score * 2);
+        assert_eq!(store_record.score, record.score);
+    }
+
+    #[test]
+    fn insert_record_does_not_accumulate_public_bonus() {
+        let mut store = AddressStore::new();
+        let peer = PeerId::random();
+
+        // Create a public address (8.8.8.8 is global) using from_multiaddr.
+        // This will apply the PUBLIC_ADDRESS_BONUS (+1).
+        let address = Multiaddr::empty()
+            .with(Protocol::Ip4(Ipv4Addr::new(8, 8, 8, 8)))
+            .with(Protocol::Tcp(9999))
+            .with(Protocol::P2p(
+                multihash::Multihash::from_bytes(&peer.to_bytes()).unwrap(),
+            ));
+
+        let record = AddressRecord::from_multiaddr(address.clone()).unwrap();
+        assert_eq!(record.score, scores::PUBLIC_ADDRESS_BONUS);
+
+        store.insert(record.clone());
+        assert_eq!(store.addresses.len(), 1);
+        assert_eq!(
+            store.addresses.get(&address).unwrap().score,
+            scores::PUBLIC_ADDRESS_BONUS
+        );
+
+        // Re-adding the same address should NOT accumulate the bonus.
+        let record2 = AddressRecord::from_multiaddr(address.clone()).unwrap();
+        store.insert(record2);
+
+        assert_eq!(store.addresses.len(), 1);
+        // Score should still be 1, not 2.
+        assert_eq!(
+            store.addresses.get(&address).unwrap().score,
+            scores::PUBLIC_ADDRESS_BONUS
+        );
+
+        // However, connection events should still update (replace) the score.
+        let connection_record =
+            AddressRecord::new(&peer, address.clone(), scores::CONNECTION_ESTABLISHED);
+        store.insert(connection_record);
+
+        assert_eq!(store.addresses.len(), 1);
+        // Score should now be 101 (CONNECTION_ESTABLISHED + PUBLIC_ADDRESS_BONUS).
+        // The score replaces rather than accumulates.
+        assert_eq!(
+            store.addresses.get(&address).unwrap().score,
+            scores::CONNECTION_ESTABLISHED + scores::PUBLIC_ADDRESS_BONUS
+        );
+    }
+
+    #[test]
+    fn rediscovery_does_not_wipe_dial_failure() {
+        let mut store = AddressStore::new();
+        let peer = PeerId::random();
+
+        // Public address (8.8.8.8 is global).
+        let address = Multiaddr::empty()
+            .with(Protocol::Ip4(Ipv4Addr::new(8, 8, 8, 8)))
+            .with(Protocol::Tcp(9999))
+            .with(Protocol::P2p(
+                multihash::Multihash::from_bytes(&peer.to_bytes()).unwrap(),
+            ));
+
+        // First, add the address normally.
+        let record = AddressRecord::from_multiaddr(address.clone()).unwrap();
+        store.insert(record);
+        assert_eq!(
+            store.addresses.get(&address).unwrap().score,
+            scores::PUBLIC_ADDRESS_BONUS
+        );
+
+        // Dial failure occurs.
+        let failure_record = AddressRecord::new(&peer, address.clone(), scores::CONNECTION_FAILURE);
+        store.insert(failure_record);
+        let failure_score = scores::CONNECTION_FAILURE + scores::PUBLIC_ADDRESS_BONUS; // -99
+        assert_eq!(store.addresses.get(&address).unwrap().score, failure_score);
+
+        // Address is rediscovered via Kademlia (creates record with score 0 or 1).
+        // This should NOT wipe out the dial failure score.
+        let rediscovered = AddressRecord::from_multiaddr(address.clone()).unwrap();
+        assert_eq!(rediscovered.score, scores::PUBLIC_ADDRESS_BONUS); // score = 1
+        store.insert(rediscovered);
+
+        // Score should still be -99, not 1.
+        assert_eq!(store.addresses.get(&address).unwrap().score, failure_score);
     }
 
     #[test]

src/transport/manager/mod.rs

@@ -39,7 +39,7 @@ use crate::{
 };
 
 use address::{scores, AddressStore};
-use futures::{Stream, StreamExt};
+use futures::{future::BoxFuture, stream::FuturesUnordered, Stream, StreamExt};
 use indexmap::IndexMap;
 use multiaddr::{Multiaddr, Protocol};
 use multihash::Multihash;
@@ -252,6 +252,10 @@ pub struct TransportManager {
 
     /// Opening connections errors.
     opening_errors: HashMap<ConnectionId, Vec<(Multiaddr, DialError)>>,
+
+    /// Pending accept futures with associated connection information.
+    /// The manager will emit a `TransportEvent::ConnectionEstablished` when complete.
+    pending_accept: FuturesUnordered<BoxFuture<'static, (PeerId, Endpoint, crate::Result<()>)>>,
 }
 
 /// Builder for [`crate::transport::manager::TransportManager`].
@@ -365,6 +369,7 @@ impl TransportManagerBuilder {
             pending_connections: HashMap::new(),
             connection_limits: limits::ConnectionLimits::new(self.connection_limits_config),
             opening_errors: HashMap::new(),
+            pending_accept: FuturesUnordered::new(),
         }
     }
 }
@@ -1091,6 +1096,33 @@ impl TransportManager {
     pub async fn next(&mut self) -> Option<TransportEvent> {
         loop {
             tokio::select! {
+                result = self.pending_accept.next(), if !self.pending_accept.is_empty() => {
+                    let Some((peer, endpoint, result)) = result else {
+                        continue;
+                    };
+
+                    match result {
+                        Ok(()) => {
+                            tracing::trace!(
+                                target: LOG_TARGET,
+                                ?peer,
+                                ?endpoint,
+                                "connection accepted and protocols notified",
+                            );
+
+                            return Some(TransportEvent::ConnectionEstablished { peer, endpoint });
+                        }
+                        Err(error) => {
+                            tracing::error!(
+                                target: LOG_TARGET,
+                                ?peer,
+                                ?endpoint,
+                                ?error,
+                                "failed to notify protocols about connection",
+                            );
+                        }
+                    }
+                }
                 event = self.event_rx.recv() => {
                     let Some(event) = event else {
                         tracing::error!(
@@ -1270,16 +1302,30 @@ impl TransportManager {
                                         "accept connection",
                                     );
 
-                                    let _ = self
+                                    match self
                                         .transports
                                         .get_mut(&transport)
                                         .expect("transport to exist")
-                                        .accept(endpoint.connection_id());
-
-                                    return Some(TransportEvent::ConnectionEstablished {
-                                        peer,
-                                        endpoint,
-                                    });
+                                        .accept(endpoint.connection_id())
+                                    {
+                                        Ok(future) => {
+                                            // A ConnectionEstablished is propagated to the user once
+                                            // all protocols have been notified.
+                                            self.pending_accept.push(Box::pin(async move {
+                                                let result = future.await;
+                                                (peer, endpoint, result)
+                                            }));
+                                        }
+                                        Err(error) => {
+                                            tracing::debug!(
+                                                target: LOG_TARGET,
+                                                ?peer,
+                                                ?endpoint,
+                                                ?error,
+                                                "failed to accept connection",
+                                            );
+                                        }
+                                    }
                                 }
                                 Ok(ConnectionEstablishedResult::Reject) => {
                                     tracing::trace!(
@@ -1474,8 +1520,11 @@ mod tests {
                 Ok(())
             }
 
-            fn accept(&mut self, _connection_id: ConnectionId) -> crate::Result<()> {
-                Ok(())
+            fn accept(
+                &mut self,
+                _connection_id: ConnectionId,
+            ) -> crate::Result<BoxFuture<'static, crate::Result<()>>> {
+                Ok(Box::pin(async { Ok(()) }))
             }
 
             fn accept_pending(&mut self, _connection_id: ConnectionId) -> crate::Result<()> {