feat: collect addresses from connected peers (#421)

* use `HashMap` for `AddrV2Handler.known_peers`

This way from the beginning we only add new peers or update existing ones in instead of adding everything with a valid timestamp and then later in the flow drop all duplicates.

* send `GetAddr` after handshake and process `Addr`

Makes sure we send `GetAddr` after handshake to request `Addr`/`AddrV2` messages and convert `Addr` messages to `AddrV2` if we receive them, which shouldn't really happen since we signal v2 support during the handshake.

* allow a wider range of last-seen timestamps

With the current validation rules we drop most of the addresses while they still might be useful since
it doesn't necessarily mean they are offline. We should still keep more addresses to have at least some addresses
to try to connect to.

Author: xdustinface

dash-spv/src/network/addrv2.rs

@@ -1,7 +1,7 @@
 //! AddrV2 message handling for modern peer exchange protocol
 
 use rand::prelude::*;
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::net::SocketAddr;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
@@ -13,10 +13,23 @@ use dashcore::network::message::NetworkMessage;
 
 use crate::network::constants::{MAX_ADDR_TO_SEND, MAX_ADDR_TO_STORE};
 
+const ONE_WEEK: u32 = 7 * 24 * 60 * 60;
+const TEN_MINUTES: u32 = 600;
+
+/// Evict oldest entries if the map exceeds capacity, keeping the freshest addresses.
+fn evict_if_needed(peers: &mut HashMap<SocketAddr, AddrV2Message>) {
+    if peers.len() > MAX_ADDR_TO_STORE {
+        let mut entries: Vec<_> = peers.drain().collect();
+        entries.sort_by_key(|(_, msg)| std::cmp::Reverse(msg.time));
+        entries.truncate(MAX_ADDR_TO_STORE);
+        peers.extend(entries);
+    }
+}
+
 /// Handler for AddrV2 peer exchange protocol
 pub struct AddrV2Handler {
     /// Known peer addresses from AddrV2 messages
-    known_peers: Arc<RwLock<Vec<AddrV2Message>>>,
+    known_peers: Arc<RwLock<HashMap<SocketAddr, AddrV2Message>>>,
     /// Peers that support AddrV2
     supports_addrv2: Arc<RwLock<HashSet<SocketAddr>>>,
 }
@@ -25,7 +38,7 @@ impl AddrV2Handler {
     /// Create a new AddrV2 handler
     pub fn new() -> Self {
         Self {
-            known_peers: Arc::new(RwLock::new(Vec::new())),
+            known_peers: Arc::new(RwLock::new(HashMap::new())),
             supports_addrv2: Arc::new(RwLock::new(HashSet::new())),
         }
     }
@@ -47,44 +60,38 @@ impl AddrV2Handler {
             })
             .as_secs() as u32;
 
-        let _initial_count = known_peers.len();
+        let received = messages.len();
         let mut added = 0;
+        let mut updated = 0;
 
         for msg in messages {
-            // Validate timestamp
-            // Accept addresses from up to 3 hours ago and up to 10 minutes in the future
-            if msg.time <= now.saturating_sub(10800) || msg.time > now + 600 {
+            // Accept addresses seen within the last week. Older addresses are likely stale.
+            // Also, reject timestamps more than 10 minutes in the future which are invalid.
+            if msg.time < now.saturating_sub(ONE_WEEK) || msg.time > now + TEN_MINUTES {
                 log::trace!("Ignoring AddrV2 with invalid timestamp: {}", msg.time);
                 continue;
             }
 
-            // Only store if we can convert to socket address
-            if msg.socket_addr().is_ok() {
-                known_peers.push(msg);
-                added += 1;
-            }
-        }
-
-        // Sort by timestamp (newest first) and deduplicate
-        known_peers.sort_by_key(|a| std::cmp::Reverse(a.time));
+            let Ok(socket_addr) = msg.socket_addr() else {
+                continue;
+            };
 
-        // Deduplicate by socket address
-        let mut seen = HashSet::new();
-        known_peers.retain(|addr| {
-            if let Ok(socket_addr) = addr.socket_addr() {
-                seen.insert(socket_addr)
-            } else {
-                false
+            // Only update if new or has fresher timestamp
+            match known_peers.get(&socket_addr) {
+                Some(existing) if existing.time >= msg.time => continue,
+                Some(_) => updated += 1,
+                None => added += 1,
             }
-        });
+            known_peers.insert(socket_addr, msg);
+        }
 
-        // Keep only the most recent addresses
-        known_peers.truncate(MAX_ADDR_TO_STORE);
+        evict_if_needed(&mut known_peers);
 
-        let _processed_count = added;
         log::info!(
-            "Processed AddrV2 messages: added {}, total known peers: {}",
+            "Processed AddrV2 messages: received {}, added {}, updated {}, total known peers: {}",
+            received,
             added,
+            updated,
             known_peers.len()
         );
     }
@@ -102,9 +109,8 @@ impl AddrV2Handler {
         let count = count.min(MAX_ADDR_TO_SEND).min(known_peers.len());
 
         let addresses: Vec<AddrV2Message> =
-            known_peers.choose_multiple(&mut rng, count).cloned().collect();
+            known_peers.values().choose_multiple(&mut rng, count).into_iter().cloned().collect();
 
-        log::debug!("Sharing {} addresses with peer", addresses.len());
         addresses
     }
 
@@ -115,7 +121,7 @@ impl AddrV2Handler {
 
     /// Get all known socket addresses
     pub async fn get_known_addresses(&self) -> Vec<AddrV2Message> {
-        self.known_peers.read().await.clone()
+        self.known_peers.read().await.values().cloned().collect()
     }
 
     /// Add a known peer address
@@ -141,13 +147,8 @@ impl AddrV2Handler {
         };
 
         let mut known_peers = self.known_peers.write().await;
-        known_peers.push(addr_msg);
-
-        // Keep size under control
-        if known_peers.len() > MAX_ADDR_TO_STORE {
-            known_peers.sort_by_key(|a| std::cmp::Reverse(a.time));
-            known_peers.truncate(MAX_ADDR_TO_STORE);
-        }
+        known_peers.insert(addr, addr_msg);
+        evict_if_needed(&mut known_peers);
     }
 
     /// Build a GetAddr response message

dash-spv/src/network/manager.rs

@@ -26,7 +26,7 @@ use crate::network::{
 };
 use crate::storage::{PeerStorage, PersistentPeerStorage, PersistentStorage};
 use async_trait::async_trait;
-use dashcore::network::address::AddrV2Message;
+use dashcore::network::address::{AddrV2, AddrV2Message};
 use dashcore::network::constants::ServiceFlags;
 use dashcore::network::message::NetworkMessage;
 use dashcore::network::message_headers2::CompressionState;
@@ -257,6 +257,11 @@ impl PeerNetworkManager {
                         Ok(_) => {
                             log::info!("Successfully connected to {}", addr);
 
+                            // Request addresses from the peer for discovery
+                            if let Err(e) = peer.send_message(NetworkMessage::GetAddr).await {
+                                log::warn!("Failed to send GetAddr to {}: {}", addr, e);
+                            }
+
                             // Record successful connection
                             reputation_manager.record_successful_connection(addr).await;
 
@@ -463,9 +468,32 @@ impl PeerNetworkManager {
                                 );
                                 continue;
                             }
-                            NetworkMessage::Addr(_) => {
-                                // Handle legacy addr messages (convert to AddrV2 if needed)
-                                log::trace!("Received legacy addr message from {}", addr);
+                            NetworkMessage::Addr(addresses) => {
+                                // Convert legacy addr messages to AddrV2 format
+                                let converted: Vec<AddrV2Message> = addresses
+                                    .iter()
+                                    .filter_map(|(time, a)| {
+                                        let socket = a.socket_addr().ok()?;
+                                        let addr_v2 = match socket.ip() {
+                                            std::net::IpAddr::V4(v4) => AddrV2::Ipv4(v4),
+                                            std::net::IpAddr::V6(v6) => AddrV2::Ipv6(v6),
+                                        };
+                                        Some(AddrV2Message {
+                                            time: *time,
+                                            services: a.services,
+                                            addr: addr_v2,
+                                            port: socket.port(),
+                                        })
+                                    })
+                                    .collect();
+                                if !converted.is_empty() {
+                                    log::debug!(
+                                        "Converted {} legacy addr entries from {}",
+                                        converted.len(),
+                                        addr
+                                    );
+                                    addrv2_handler.handle_addrv2(converted).await;
+                                }
                                 continue;
                             }
                             NetworkMessage::Headers(headers) => {