net: make fetch address logic less nested + fix bug

Previously when we called whitelist_fetch_[...](), it would call
greylist_fetch_[...]() inside that method if insufficient whitelist
entries were found.

This was confusing and non-intuitive.

This commit removes this nesting. Now we call
whitelist_fetch_[...] followed by greylist_fetch_[...] in ProtocolAddr
and OutboundSession explicitly.

This commit also fixes a bug in the refinery.

src/net/hosts/refinery.rs

@@ -94,33 +94,41 @@ impl GreylistRefinery {
             }
 
             // Only attempt to refine peers that match our transports.
-            let (entry, position) = hosts.greylist_fetch_random_with_schemes().await;
-            let url = &entry.0;
-
-            // Skip this node if it's being migrated currently.
-            if hosts.is_migrating(url).await {
-                continue
-            }
-
-            let mut greylist = hosts.greylist.write().await;
-            if !ping_node(url, self.p2p().clone()).await {
-                greylist.remove(position);
-                debug!(
-                    target: "net::refinery",
-                    "Peer {} is non-responsive. Removed from greylist", url,
-                );
+            match hosts.greylist_fetch_random_with_schemes().await {
+                Some((entry, position)) => {
+                    let url = &entry.0;
+
+                    // Skip this node if it's being migrated currently.
+                    if hosts.is_migrating(url).await {
+                        continue
+                    }
+
+                    let mut greylist = hosts.greylist.write().await;
+                    if !ping_node(url, self.p2p().clone()).await {
+                        greylist.remove(position);
+                        debug!(
+                            target: "net::refinery",
+                            "Peer {} is non-responsive. Removed from greylist", url,
+                        );
+
+                        continue
+                    }
+                    drop(greylist);
+
+                    let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+
+                    // Append to the whitelist.
+                    hosts.whitelist_store_or_update(&[(url.clone(), last_seen)]).await;
+
+                    // Remove whitelisted peer from the greylist.
+                    hosts.greylist_remove(url, position).await;
+                }
+                None => {
+                    debug!(target: "net::refinery", "No matching greylist entries found. Cannot proceed with refinery");
 
-                continue
+                    continue
+                }
             }
-            drop(greylist);
-
-            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
-
-            // Append to the whitelist.
-            hosts.whitelist_store_or_update(&[(url.clone(), last_seen)]).await;
-
-            // Remove whitelisted peer from the greylist.
-            hosts.greylist_remove(url, position).await;
         }
     }
 

src/net/hosts/store.rs

@@ -213,6 +213,10 @@ impl Hosts {
         hosts
     }
 
+    /// Check whether:
+    /// *   We already have this connection established
+    /// *   We already have this configured as a manual peer
+    /// *   This address is already pending a connection
     pub async fn check_address_with_lock(
         &self,
         p2p: P2pPtr,
@@ -322,6 +326,11 @@ impl Hosts {
         // Filter addresses before writing to the greylist.
         let filtered_addrs = self.filter_addresses(addrs).await;
         let filtered_addrs_len = filtered_addrs.len();
+
+        if filtered_addrs.is_empty() {
+            debug!(target: "store::greylist_store_or_update()", "Filtered out all received addresses");
+        }
+
         for (addr, last_seen) in filtered_addrs {
             if !self.greylist_contains(&addr).await {
                 debug!(target: "store::greylist_store_or_update()",
@@ -810,16 +819,70 @@ impl Hosts {
     }
 
     /// Get a random greylist peer that matches the given transport schemes.
-    pub async fn greylist_fetch_random_with_schemes(&self) -> ((Url, u64), usize) {
+    pub async fn greylist_fetch_random_with_schemes(&self) -> Option<((Url, u64), usize)> {
         trace!(target: "store::greylist_fetch_random_with_schemes", "[START]");
 
         // Retrieve all peers corresponding to that transport schemes
         let schemes = &self.settings.allowed_transports;
         let greylist = self.greylist_fetch_with_schemes(schemes, None).await;
 
+        if greylist.is_empty() {
+            return None
+        }
+
         let position = rand::thread_rng().gen_range(0..greylist.len());
         let entry = &greylist[position];
-        (entry.clone(), position)
+        Some((entry.clone(), position))
+    }
+
+    /// Get up to n random greylist peers. Schemes are not taken into account.
+    pub async fn greylist_fetch_n_random(&self, n: u32) -> Vec<(Url, u64)> {
+        trace!(target: "store::greylist_fetch_n_random", "[START]");
+        let n = n as usize;
+        if n == 0 {
+            return vec![]
+        }
+        let mut hosts = vec![];
+
+        let greylist = self.greylist.read().await;
+
+        for (addr, last_seen) in greylist.iter() {
+            hosts.push((addr.clone(), *last_seen));
+        }
+
+        if hosts.is_empty() {
+            debug!(target: "store::greylist_fetch_n_random", "No greylist entries found!");
+            return hosts
+        }
+
+        // Grab random ones
+        let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
+        urls.iter().map(|&url| url.clone()).collect()
+    }
+
+    /// Get up to n random greylist peers that match the given transport schemes.
+    pub async fn greylist_fetch_n_random_with_schemes(
+        &self,
+        schemes: &[String],
+        n: u32,
+    ) -> Vec<(Url, u64)> {
+        let n = n as usize;
+        if n == 0 {
+            return vec![]
+        }
+        trace!(target: "store::greylist_fetch_n_random_with_schemes", "[START]");
+
+        // Retrieve all peers corresponding to that transport schemes
+        let hosts = self.greylist_fetch_with_schemes(schemes, None).await;
+        if hosts.is_empty() {
+            debug!(target: "store::greylist_fetch_n_random_with_schemes",
+                  "No such schemes found on greylist!");
+            return hosts
+        }
+
+        // Grab random ones
+        let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
+        urls.iter().map(|&url| url.clone()).collect()
     }
 
     /// Get up to n random whitelist peers that match the given transport schemes.
@@ -837,8 +900,8 @@ impl Hosts {
         // Retrieve all peers corresponding to that transport schemes
         let hosts = self.whitelist_fetch_with_schemes(schemes, None).await;
         if hosts.is_empty() {
-            trace!(target: "store::whitelist_fetch_n_random_with_schemes",
-                  "Whitelist is empty {:?}! Exiting...", hosts);
+            debug!(target: "store::whitelist_fetch_n_random_with_schemes",
+                  "No such schemes found on whitelist!");
             return hosts
         }
 
@@ -847,6 +910,42 @@ impl Hosts {
         urls.iter().map(|&url| url.clone()).collect()
     }
 
+    /// Get up to limit peers that don't match the given transport schemes from the greylist.
+    /// If limit was not provided, return all matching peers.
+    pub async fn greylist_fetch_excluding_schemes(
+        &self,
+        schemes: &[String],
+        limit: Option<usize>,
+    ) -> Vec<(Url, u64)> {
+        let greylist = self.greylist.read().await;
+        let mut limit = match limit {
+            Some(l) => l.min(greylist.len()),
+            None => greylist.len(),
+        };
+        let mut ret = vec![];
+
+        if limit == 0 {
+            return ret
+        }
+
+        for (addr, last_seen) in greylist.iter() {
+            if !schemes.contains(&addr.scheme().to_string()) {
+                ret.push((addr.clone(), *last_seen));
+                limit -= 1;
+                if limit == 0 {
+                    return ret
+                }
+            }
+        }
+
+        if ret.is_empty() {
+            debug!(target: "store::greylist_fetch_excluding_schemes",
+                  "No such schemes found on greylist!")
+        }
+
+        ret
+    }
+
     /// Get up to limit peers that don't match the given transport schemes from the whitelist.
     /// If limit was not provided, return all matching peers.
     pub async fn whitelist_fetch_excluding_schemes(
@@ -875,17 +974,9 @@ impl Hosts {
             }
         }
 
-        // If we didn't find any, pick some from the greylist
         if ret.is_empty() {
-            for (addr, last_seen) in self.greylist.read().await.iter() {
-                if !schemes.contains(&addr.scheme().to_string()) {
-                    ret.push((addr.clone(), *last_seen));
-                    limit -= 1;
-                    if limit == 0 {
-                        break
-                    }
-                }
-            }
+            debug!(target: "store::whiteist_fetch_excluding_schemes",
+                  "No such schemes found on whitelist!")
         }
 
         ret
@@ -909,7 +1000,7 @@ impl Hosts {
 
         if hosts.is_empty() {
             debug!(target: "store::whitelist_fetch_n_random_excluding_schemes",
-                  "No address without schemes found! Exiting...");
+                  "No such schemes found on whitelist!");
             return hosts
         }
 
@@ -925,7 +1016,7 @@ impl Hosts {
         schemes: &[String],
         limit: Option<usize>,
     ) -> Vec<(Url, u64)> {
-        debug!(target: "store::greylist_fetch_with_schemes", "[START]");
+        trace!(target: "store::greylist_fetch_with_schemes", "[START]");
         let greylist = self.greylist.read().await;
 
         let mut limit = match limit {
@@ -943,14 +1034,20 @@ impl Hosts {
                 ret.push((addr.clone(), *last_seen));
                 limit -= 1;
                 if limit == 0 {
-                    debug!(target: "store::greylist_fetch_with_schemes", "Found matching greylist entry, returning");
+                    debug!(target: "store::greylist_fetch_with_schemes",
+                        "Found matching scheme, returning {} grey addresses",
+                        ret.len());
                     return ret
                 }
             }
         }
 
-        trace!(target: "store::greylist_fetch_with_schemes", "END");
+        if ret.is_empty() {
+            debug!(target: "store::greylist_fetch_with_schemes",
+                  "No such schemes found on greylist!")
+        }
 
+        trace!(target: "store::greylist_fetch_with_schemes", "END");
         ret
     }
 
@@ -961,41 +1058,38 @@ impl Hosts {
         schemes: &[String],
         limit: Option<usize>,
     ) -> Vec<(Url, u64)> {
-        debug!(target: "store::whitelist_fetch_with_schemes", "[START]");
-        let mut ret = vec![];
+        trace!(target: "store::whitelist_fetch_with_schemes", "[START]");
+        let whitelist = self.whitelist.read().await;
 
-        if !self.is_empty_whitelist().await {
-            let whitelist = self.whitelist.read().await;
+        let mut limit = match limit {
+            Some(l) => l.min(whitelist.len()),
+            None => whitelist.len(),
+        };
+        let mut ret = vec![];
 
-            let mut parsed_limit = match limit {
-                Some(l) => l.min(whitelist.len()),
-                None => whitelist.len(),
-            };
+        if limit == 0 {
+            return ret
+        }
 
-            for (addr, last_seen) in whitelist.iter() {
-                if schemes.contains(&addr.scheme().to_string()) {
-                    ret.push((addr.clone(), *last_seen));
-                    parsed_limit -= 1;
-                    if parsed_limit == 0 {
-                        trace!(target: "store::whitelist_fetch_with_schemes",
-                           "Found matching white scheme, returning {:?}", ret);
-                        return ret
-                    }
-                } else {
-                    warn!(target: "store::whitelist_fetch_with_schemes",
-                          "No matching schemes! Trying greylist...");
-                    return self.greylist_fetch_with_schemes(schemes, limit).await
+        for (addr, last_seen) in whitelist.iter() {
+            if schemes.contains(&addr.scheme().to_string()) {
+                ret.push((addr.clone(), *last_seen));
+                limit -= 1;
+                if limit == 0 {
+                    debug!(target: "store::whitelist_fetch_with_schemes",
+                           "Found matching scheme, returning {} white addresses",
+                           ret.len());
+                    return ret
                 }
             }
         }
-        // Whitelist is empty!
-        if !self.is_empty_greylist().await {
-            // Select from the greylist providing it's not empty.
-            return self.greylist_fetch_with_schemes(schemes, limit).await
+
+        if ret.is_empty() {
+            debug!(target: "store::whitelist_fetch_with_schemes",
+                  "No such schemes found on whitelist!")
         }
 
         trace!(target: "store::whitelist_fetch_with_schemes", "END");
-
         ret
     }
 
@@ -1006,47 +1100,38 @@ impl Hosts {
         schemes: &[String],
         limit: Option<usize>,
     ) -> Vec<(Url, u64)> {
-        trace!(target: "store::anchorlist_fetch_with_schemes", "[START]");
-        let mut ret = vec![];
+        //trace!(target: "store::anchorlist_fetch_with_schemes", "[START]");
+        let anchorlist = self.anchorlist.read().await;
 
-        // Select from the anchorlist providing it's not empty.
-        if !self.is_empty_anchorlist().await {
-            let anchorlist = self.anchorlist.read().await;
+        let mut limit = match limit {
+            Some(l) => l.min(anchorlist.len()),
+            None => anchorlist.len(),
+        };
+        let mut ret = vec![];
 
-            let mut parsed_limit = match limit {
-                Some(l) => l.min(anchorlist.len()),
-                None => anchorlist.len(),
-            };
+        if limit == 0 {
+            return ret
+        }
 
-            for (addr, last_seen) in anchorlist.iter() {
-                if schemes.contains(&addr.scheme().to_string()) {
-                    ret.push((addr.clone(), *last_seen));
-                    parsed_limit -= 1;
-                    if parsed_limit == 0 {
-                        trace!(target: "store::anchorlist_fetch_with_schemes",
-                           "Found matching anchor scheme, returning {:?}", ret);
-                        return ret
-                    }
-                } else {
-                    warn!(target: "store::anchorlist_fetch_with_schemes",
-                          "No matching schemes! Trying whitelist...");
-                    return self.whitelist_fetch_with_schemes(schemes, limit).await
+        for (addr, last_seen) in anchorlist.iter() {
+            if schemes.contains(&addr.scheme().to_string()) {
+                ret.push((addr.clone(), *last_seen));
+                limit -= 1;
+                if limit == 0 {
+                    debug!(target: "store::anchorlist_fetch_with_schemes",
+                           "Found matching scheme, returning {} anchor addresses",
+                           ret.len());
+                    return ret
                 }
             }
         }
 
-        // Select from the whitelist providing it's not empty.
-        if !self.is_empty_whitelist().await {
-            return self.whitelist_fetch_with_schemes(schemes, limit).await
-        }
-
-        // Select from the greyist providing it's not empty.
-        if !self.is_empty_greylist().await {
-            return self.greylist_fetch_with_schemes(schemes, limit).await
+        if ret.is_empty() {
+            warn!(target: "store::anchorlist_fetch_with_schemes",
+                  "No matching schemes found on anchorlist")
         }
 
         trace!(target: "store::anchorlist_fetch_with_schemes", "END");
-
         ret
     }
 

src/net/protocol/protocol_address.rs

@@ -153,6 +153,13 @@ impl ProtocolAddress {
                     .await,
             );
 
+            // If there's still space available, take from the greylist.
+            // Schemes are not taken into account.
+            debug!(target: "net::protocol_address::handle_receive_get_addrs()",
+            "Fetching greylist entries");
+            let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
+            addrs.append(&mut self.hosts.greylist_fetch_n_random(remain).await);
+
             debug!(
                 target: "net::protocol_address::handle_receive_get_addrs()",
                 "Sending {} addresses to {}", addrs.len(), self.channel.address(),
@@ -186,6 +193,7 @@ impl ProtocolAddress {
 
         let mut addrs = vec![];
         for addr in self.settings.external_addrs.clone() {
+            //addrs.push((addr, 0));
             debug!(target: "net::protocol_address::send_my_addrs()", "Attempting to ping self");
 
             // See if we can do a version exchange with ourself.

src/net/protocol/protocol_seed.rs

@@ -73,6 +73,7 @@ impl ProtocolSeed {
 
         let mut addrs = vec![];
         for addr in self.settings.external_addrs.clone() {
+            //addrs.push((addr, 0));
             debug!(target: "net::protocol_seed::send_my_addrs()", "Attempting to ping self");
 
             // See if we can do a version exchange with ourself.

src/net/session/outbound_session.rs

@@ -190,44 +190,61 @@ impl Slot {
         let connects = self.p2p().settings().outbound_connections;
         let white_count = connects * self.p2p().settings().white_connection_percent / 100;
 
-        let addrs = {
-            // Up to anchor_connection_count connections:
-            //
+        if slot_count < self.p2p().settings().anchor_connection_count {
+            //  Up to anchor_connection_count connections:
             //  Select from the anchorlist
             //  If the anchorlist is empty, select from the whitelist
             //  If the whitelist is empty, select from the greylist
-            //  If the greylist is empty, do peer discovery
-            if slot_count < self.p2p().settings().anchor_connection_count {
-                debug!(target: "net::outbound_session::fetch_address()",
-                "First two connections- prefer anchor connections");
-                hosts.anchorlist_fetch_address(transports).await
+            //  If the greylist is empty, return None and do peer discovery
+            if !hosts.anchorlist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.anchorlist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
+            }
+
+            if !hosts.whitelist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.whitelist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
             }
+
+            if !hosts.greylist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.greylist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
+            }
+
+            return None
+        } else if slot_count < white_count {
             // Up to white_connection_percent connections:
-            //
             //  Select from the whitelist
             //  If the whitelist is empty, select from the greylist
-            //  If the greylist is empty, do peer discovery
-            else if slot_count < white_count {
-                debug!(target: "net::outbound_session::fetch_address()",
-                "Next N connections- prefer white connections");
-                hosts.whitelist_fetch_address(transports).await
+            //  If the greylist is empty, return None and do peer discovery
+            if !hosts.whitelist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.whitelist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
+            }
+
+            if !hosts.greylist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.greylist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
             }
+
+            return None
+        } else {
             // All other connections:
-            //
             //  Select from the greylist
             //  If the greylist is empty, do peer discovery
-            else {
-                debug!(target: "net::outbound_session::fetch_address()",
-                "All other connections- get grey connections");
-                hosts.greylist_fetch_address(transports).await
+            if !hosts.greylist_fetch_address(transports).await.is_empty() {
+                let addrs = hosts.greylist_fetch_address(transports).await;
+
+                return hosts.check_address_with_lock(self.p2p(), addrs).await
             }
-        };
 
-        // Check whether:
-        // * we already have this connection established
-        // * we already have this configured as a manual peer
-        // * address is already pending a connection
-        hosts.check_address_with_lock(self.p2p(), addrs).await
+            return None
+        }
     }
 
     // We first try to make connections to the addresses on our anchor list. We then find some