Merge #904: fix: another bundle of small fixes for how we select peers

857a80b fix: use AddrMan::enough_addresses to decide whether to use hardcoded peers (Davidson Souza)
b108023 chore: increase timeout to two minutes (Davidson Souza)
ed87428 fix: don't drop inflight requests before completion (JoseSK999)
9caa04f fix: fix service bits for some hardcoded addresses (Davidson Souza)
497a59c fix: update excess peer's data (Davidson Souza)
af42edc fix(sync_ctx): make it more aggressive at oppening utreexo connections (Davidson Souza)
04ba097 test: add a new test for adding fixed peers on addr_man (Davidson Souza)
2efbbec fix: update update_set_service_flag to require NETWORK_LIMITED (Davidson Souza)
43b5574 fix(addr_man): don't return connected addresses in get_address_by_service (Davidson Souza)

Pull request description:

  ### Description and Notes

  This PR bundles a couple of small changes intended to make our node more effective at finding Utreexo peers. See each commit message for more details about what they are changing and why.

  Changelog:

  ```
  fix(addr_man: don't return connected addresses in get_address_by_service
  fix: update update_set_service_flag to require NETWORK_LIMITED
  test: add a new test for adding fixed peers on addr_man
  fix(sync_ctx): make it more aggressive at oppening utreexo connections
  fix: update excess peer's data
  fix: fix service bits for some hardcoded addresses
  ```

ACKs for top commit:
  JoseSK999:
    ACK 857a80b; works well on signet
  luisschwab:
    ACK 857a80b

Tree-SHA512: ec18bb8ef87cf6df8856292666a34c61b6fbde2583ff536749e84958263d557779cf5cfe0c6fd6e488628c9b29b02a85b7555263ea44407d5bbbf6d1d24b9886

Author: Davidson-Souza

crates/floresta-wire/src/p2p_wire/address_man.rs

@@ -21,6 +21,7 @@ use bitcoin::p2p::ServiceFlags;
 use bitcoin::Network;
 use floresta_chain::DnsSeed;
 use floresta_common::service_flags;
+use rand::seq::IteratorRandom;
 use serde::Deserialize;
 use serde::Serialize;
 use tracing::debug;
@@ -742,15 +743,16 @@ impl AddressMan {
     }
 
     fn get_address_by_service(&self, service: ServiceFlags) -> Option<(usize, LocalAddress)> {
-        let peers = self.good_peers_by_service.get(&service)?;
-        if peers.is_empty() {
-            return None;
-        }
-
-        let idx = rand::random::<usize>() % peers.len();
-        let utreexo_peer = peers.get(idx)?;
+        let candidates = self.good_peers_by_service.get(&service)?;
 
-        Some((*utreexo_peer, self.addresses.get(utreexo_peer)?.to_owned()))
+        candidates
+            .iter()
+            .filter_map(|id| {
+                let addr = self.addresses.get(id)?;
+                (addr.state != AddressState::Connected).then_some((id, addr))
+            })
+            .choose(&mut rand::thread_rng())
+            .map(|(id, addr)| (*id, addr.to_owned()))
     }
 
     pub fn start_addr_man(&mut self, datadir: String) -> Vec<LocalAddress> {
@@ -969,7 +971,7 @@ impl AddressMan {
     /// Updates the service flags after we receive a version message
     pub fn update_set_service_flag(&mut self, idx: usize, flags: ServiceFlags) -> &mut Self {
         // if this peer turns out to not have the minimum required services, we remove it
-        if !flags.has(ServiceFlags::NETWORK) || !flags.has(ServiceFlags::WITNESS) {
+        if !flags.has(ServiceFlags::NETWORK_LIMITED) || !flags.has(ServiceFlags::WITNESS) {
             self.addresses.remove(&idx);
             for peers in self.peers_by_service.values_mut() {
                 peers.retain(|&x| x != idx);
@@ -1321,6 +1323,41 @@ mod test {
         }
     }
 
+    #[test]
+    fn test_adding_fixed_peer() {
+        let signet_addresses =
+            load_addresses_from_json("./src/p2p_wire/seeds/signet_seeds.json").unwrap();
+
+        let mut addr_man =
+            AddressMan::new(None, &[ReachableNetworks::IPv4, ReachableNetworks::IPv6]);
+        addr_man.add_fixed_addresses(Network::Signet);
+
+        assert_eq!(addr_man.good_addresses.len(), signet_addresses.len());
+
+        let utreexo_addresses = signet_addresses
+            .iter()
+            .filter(|address| address.services.has(service_flags::UTREEXO.into()))
+            .collect::<Vec<_>>();
+
+        assert_eq!(
+            addr_man
+                .good_peers_by_service
+                .get(&service_flags::UTREEXO.into())
+                .unwrap()
+                .len(),
+            utreexo_addresses.len()
+        );
+
+        assert_eq!(
+            addr_man
+                .peers_by_service
+                .get(&service_flags::UTREEXO.into())
+                .unwrap()
+                .len(),
+            utreexo_addresses.len()
+        );
+    }
+
     #[test]
     fn test_parse() {
         let signet_address =

crates/floresta-wire/src/p2p_wire/node/chain_selector_ctx.rs

@@ -1005,7 +1005,7 @@ where
             }
 
             PeerMessages::Ready(version) => {
-                self.handle_peer_ready(peer, &version)?;
+                self.handle_peer_ready(peer, version)?;
                 if matches!(self.context.state, ChainSelectorState::LookingForForks(_)) {
                     let locator = self.chain.get_block_locator().unwrap();
                     self.send_to_peer(peer, NodeRequest::GetHeaders(locator))?;

crates/floresta-wire/src/p2p_wire/node/conn.rs

@@ -549,7 +549,7 @@ where
     /// This is only done if we don't have any peers for a long time, or we
     /// can't find a Utreexo peer in a context we need them. This function
     /// won't do anything if `--connect` was used
-    fn maybe_use_hardcoded_addresses(&mut self, needs_utreexo: bool) {
+    fn maybe_use_hardcoded_addresses(&mut self) {
         if self.fixed_peer.is_some() {
             return;
         }
@@ -558,18 +558,11 @@ where
             return;
         }
 
-        let has_peers = !self.peers.is_empty();
-        // Return if we have peers and utreexo isn't needed OR we have utreexo peers
-        if has_peers && (!needs_utreexo || self.has_utreexo_peers()) {
+        if self.address_man.enough_addresses() {
             return;
         }
 
-        let mut wait = HARDCODED_ADDRESSES_GRACE_PERIOD;
-        if needs_utreexo {
-            // This gives some extra time for the node to try connections after chain selection
-            wait += Duration::from_secs(60);
-        }
-
+        let wait = HARDCODED_ADDRESSES_GRACE_PERIOD;
         if self.startup_time.elapsed() < wait {
             return;
         }
@@ -629,8 +622,7 @@ where
         // If we've tried getting some connections, but the addresses we have are not
         // working. Try getting some more addresses from DNS
         self.maybe_ask_dns_seed_for_addresses();
-        let needs_utreexo = required_service.has(service_flags::UTREEXO.into());
-        self.maybe_use_hardcoded_addresses(needs_utreexo);
+        self.maybe_use_hardcoded_addresses();
 
         for _ in 0..T::NEW_CONNECTIONS_BATCH_SIZE {
             // Ignore the error so we don't break out of the loop

crates/floresta-wire/src/p2p_wire/node/peer_man.rs

@@ -31,6 +31,7 @@ use crate::address_man::AddressState;
 use crate::address_man::LocalAddress;
 use crate::block_proof::Bitmap;
 use crate::node::running_ctx::RunningNode;
+use crate::node::try_and_log;
 use crate::node_context::NodeContext;
 use crate::node_context::PeerId;
 use crate::node_interface::NodeResponse;
@@ -212,7 +213,7 @@ where
     pub(crate) fn handle_peer_ready(
         &mut self,
         peer: u32,
-        version: &Version,
+        mut version: Version,
     ) -> Result<(), WireError> {
         self.inflight.remove(&InflightRequests::Connect(peer));
 
@@ -226,6 +227,30 @@ where
         self.inflight
             .insert(InflightRequests::GetAddresses, (peer, Instant::now()));
 
+        let good_peers_count = self.connected_peers();
+        if good_peers_count > T::MAX_OUTGOING_PEERS {
+            // We allow utreexo, extra and manual peers to bypass our connection limits
+            let is_utreexo_peer = matches!(version.kind, ConnectionKind::Regular(services) if services.has(service_flags::UTREEXO.into()));
+            let is_manual_peer = version.kind == ConnectionKind::Manual;
+            let is_extra = version.kind == ConnectionKind::Extra;
+
+            if !(is_utreexo_peer || is_manual_peer || is_extra) {
+                debug!(
+                    "Already have {} peers, disconnecting peer to avoid blowing up our max of {}",
+                    good_peers_count,
+                    T::MAX_OUTGOING_PEERS
+                );
+
+                // If a peer exceeds our max, just turn them into a feeler so we can receive their
+                // AddrV2 message and then disconnect.
+                self.peers.entry(peer).and_modify(|p| {
+                    p.kind = ConnectionKind::Feeler;
+                });
+
+                version.kind = ConnectionKind::Feeler;
+            }
+        }
+
         if version.kind == ConnectionKind::Feeler {
             let now = SystemTime::now()
                 .duration_since(UNIX_EPOCH)
@@ -249,27 +274,6 @@ where
             return Ok(());
         }
 
-        let good_peers_count = self.connected_peers();
-        if good_peers_count > T::MAX_OUTGOING_PEERS {
-            // Don't allow our node to have more than T::MAX_OUTGOING_PEERS, unless this is a
-            // manual peer, those can exceed our quota.
-            if version.kind != ConnectionKind::Manual {
-                debug!(
-                    "Already have {} peers, disconnecting peer to avoid blowing up our max of {}",
-                    good_peers_count,
-                    T::MAX_OUTGOING_PEERS
-                );
-
-                // If a peer exceeds our max, just turn them into a feeler so we can receive their
-                // AddrV2 message and then disconnect.
-                self.peers.entry(peer).and_modify(|p| {
-                    p.kind = ConnectionKind::Feeler;
-                });
-
-                return Ok(());
-            }
-        }
-
         info!(
             "New peer id={} version={} blocks={} services={}",
             version.id, version.user_agent, version.blocks, version.services
@@ -479,7 +483,12 @@ where
 
         for req in inflight {
             self.inflight.remove(&req.0);
-            self.redo_inflight_request(req.0.clone())?;
+
+            if let Err(e) = self.redo_inflight_request(&req.0) {
+                // CRITICAL: never drop the request, so we retry it later
+                self.inflight.insert(req.0, req.1);
+                return Err(e);
+            }
         }
 
         #[cfg(feature = "metrics")]
@@ -565,7 +574,7 @@ where
             .collect::<Vec<_>>();
 
         for req in timed_out {
-            let Some((peer, _)) = self.inflight.remove(&req) else {
+            let Some((peer, time)) = self.inflight.remove(&req) else {
                 continue;
             };
 
@@ -587,8 +596,14 @@ where
             }
 
             debug!("Request timed out: {req:?}");
-            self.increase_banscore(peer, 1)?;
-            self.redo_inflight_request(req)?;
+            // Increase the banscore and try banning the peer if needed, then re-request
+            try_and_log!(self.increase_banscore(peer, 1));
+
+            if let Err(e) = self.redo_inflight_request(&req) {
+                // CRITICAL: never drop the request, so we retry it later
+                self.inflight.insert(req, (peer, time));
+                return Err(e);
+            }
         }
 
         Ok(())
@@ -617,39 +632,39 @@ where
         Ok(())
     }
 
-    pub(crate) fn redo_inflight_request(&mut self, req: InflightRequests) -> Result<(), WireError> {
+    pub(crate) fn redo_inflight_request(
+        &mut self,
+        req: &InflightRequests,
+    ) -> Result<(), WireError> {
         match req {
             InflightRequests::UtreexoProof(block_hash) => {
                 if !self.has_utreexo_peers() {
                     return Ok(());
                 }
 
-                if !self.blocks.contains_key(&block_hash) {
+                if !self.blocks.contains_key(block_hash) {
                     // If we don't have the block anymore, we can't ask for the proof
                     return Ok(());
                 }
 
-                if self
-                    .inflight
-                    .contains_key(&InflightRequests::UtreexoProof(block_hash))
-                {
+                if self.inflight.contains_key(req) {
                     // If we already have an inflight request for this block, we don't need to redo it
                     return Ok(());
                 }
 
                 let peer = self.send_to_fast_peer(
-                    NodeRequest::GetBlockProof((block_hash, Bitmap::new(), Bitmap::new())),
+                    NodeRequest::GetBlockProof((*block_hash, Bitmap::new(), Bitmap::new())),
                     service_flags::UTREEXO.into(),
                 )?;
 
                 self.inflight.insert(
-                    InflightRequests::UtreexoProof(block_hash),
+                    InflightRequests::UtreexoProof(*block_hash),
                     (peer, Instant::now()),
                 );
             }
 
             InflightRequests::Blocks(block) => {
-                self.request_blocks(vec![block])?;
+                self.request_blocks(vec![*block])?;
             }
 
             InflightRequests::Headers => {

crates/floresta-wire/src/p2p_wire/node/running_ctx.rs

@@ -732,7 +732,7 @@ where
                             "handshake with peer={peer} succeeded feeler={:?}",
                             version.kind
                         );
-                        self.handle_peer_ready(peer, &version)?;
+                        self.handle_peer_ready(peer, version)?;
                     }
 
                     PeerMessages::Disconnected(idx) => {

crates/floresta-wire/src/p2p_wire/node/sync_ctx.rs

@@ -48,8 +48,7 @@ impl NodeContext for SyncNode {
         ServiceFlags::NETWORK | ServiceFlags::WITNESS | service_flags::UTREEXO_ARCHIVE.into()
     }
 
-    const TRY_NEW_CONNECTION: u64 = 30; // 30 seconds
-    const REQUEST_TIMEOUT: u64 = 60; // 1 minute
+    const REQUEST_TIMEOUT: u64 = 60 * 2; // 2 minutes
     const MAX_INFLIGHT_REQUESTS: usize = 100; // double the default
 
     // A more conservative value than the default of 1 second, since we'll have many peer messages
@@ -321,7 +320,7 @@ where
                     }
 
                     PeerMessages::Ready(version) => {
-                        try_and_log!(self.handle_peer_ready(peer, &version));
+                        try_and_log!(self.handle_peer_ready(peer, version));
                     }
 
                     PeerMessages::Disconnected(idx) => {

crates/floresta-wire/src/p2p_wire/seeds/signet_seeds.json

@@ -106,7 +106,7 @@
         "state": {
             "Tried": 0
         },
-        "services": 1037,
+        "services": 1033,
         "port": 38333
     },
     {
@@ -139,7 +139,7 @@
         "state": {
             "Tried": 0
         },
-        "services": 0,
+        "services": 1033,
         "port": 38333
     },
     {
@@ -150,7 +150,7 @@
         "state": {
             "Tried": 0
         },
-        "services": 0,
+        "services": 1033,
         "port": 38333
     },
     {
@@ -161,7 +161,7 @@
         "state": {
             "Tried": 0
         },
-        "services": 0,
+        "services": 1033,
         "port": 38333
     }
 ]