net: BUGFIX: stop duplicate entries in greylist

check if we have the peer and update it's last seen if we do.
this commit also standardizes the format of send_my_addrs across
ProtocolSeed and ProtocolAddr, so we can perhaps extract the
functionality to protocol/mod.rs at some point.

src/error.rs

@@ -190,6 +190,9 @@ pub enum Error {
     #[error("P2P network stopped")]
     P2PNetworkStopped,
 
+    #[error("Invalid hostlist index")]
+    InvalidIndex,
+
     // =============
     // Crypto errors
     // =============

src/net/hosts.rs

@@ -29,7 +29,7 @@ use url::Url;
 use super::{p2p::P2pPtr, settings::SettingsPtr};
 use crate::{
     system::{Subscriber, SubscriberPtr, Subscription},
-    Result,
+    Error, Result,
 };
 
 /// Atomic pointer to hosts object
@@ -158,47 +158,77 @@ impl Hosts {
     // Store the address in the whitelist if we don't have it.
     // Otherwise, update the last_seen field.
     // TODO: test the performance of this method. It might be costly.
-    pub async fn whitelist_store_or_update(&self, addr: &Url, last_seen: u64) {
-        debug!(target: "net::hosts::whitelist_store_or_update()", "hosts::whitelist_store_or_update() [START]");
+    pub async fn whitelist_store_or_update(&self, addr: &Url, last_seen: u64) -> Result<()> {
+        debug!(target: "net::hosts::whitelist_store_or_update()",
+        "hosts::whitelist_store_or_update() [START]");
+
         if !self.whitelist_contains(addr).await {
             self.whitelist_store(addr, last_seen).await;
         } else {
-            let index = self.get_whitelist_index_at_addr(addr).await;
+            let index = self.get_whitelist_index_at_addr(addr).await?;
             self.whitelist_update_last_seen(addr, last_seen, index).await;
         }
+        Ok(())
     }
 
     // Update the last_seen field for a Url on the whitelist.
-    pub async fn whitelist_update(&self, addr: &Url, last_seen: u64) {
-        let index = self.get_whitelist_index_at_addr(addr).await;
+    pub async fn whitelist_update(&self, addr: &Url, last_seen: u64) -> Result<()> {
+        let index = self.get_whitelist_index_at_addr(addr).await?;
         self.whitelist_update_last_seen(addr, last_seen, index).await;
+        Ok(())
+    }
+
+    pub async fn greylist_store_or_update(&self, addrs: &[(Url, u64)]) -> Result<()> {
+        debug!(target: "net::hosts::greylist_store_or_update()",
+        "hosts::greylist_store_or_update() [START]");
+
+        for (addr, last_seen) in addrs {
+            if !self.greylist_contains(addr).await {
+                debug!(target: "net::greylist_store_or_update()", "New greylist candidate found!");
+                // TODO: clean this up: greylist_store one item at a time
+                self.greylist_store(&[(addr.clone(), last_seen.clone())]).await;
+            } else {
+                debug!(target: "net::greylist_store_or_update()",
+                "Existing greylist entry found. Updating last_seen...");
+
+                let index = self.get_greylist_index_at_addr(addr).await?;
+                self.greylist_update_last_seen(addr, last_seen.clone(), index).await;
+            }
+        }
+        Ok(())
     }
 
     // Append host to the greylist. Called on learning of a new peer.
     pub async fn greylist_store(&self, addrs: &[(Url, u64)]) {
         debug!(target: "net::hosts::greylist_store()", "hosts::greylist_store() [START]");
 
+        debug!(target: "net::hosts::greylist_store()", "Filtering addresses...");
         let filtered_addrs = self.filter_addresses(addrs).await;
         let filtered_addrs_len = filtered_addrs.len();
 
+        debug!(target: "net::hosts::greylist_store()", "Filtered addresses.");
         if !filtered_addrs.is_empty() {
+            debug!(target: "net::hosts::greylist_store()", "Starting greylist write...");
             let mut greylist = self.greylist.write().await;
+            debug!(target: "net::hosts::greylist_store()", "Achieved write lock on greylist!");
 
             // Remove oldest element if the greylist reaches max size.
             if greylist.len() == 5000 {
                 let last_entry = greylist.pop().unwrap();
                 debug!(target: "net::hosts::greylist_store()", "Greylist reached max size. Removed {:?}", last_entry);
-            }
-
-            for (addr, last_seen) in filtered_addrs {
-                debug!(target: "net::hosts::greylist_store()", "Inserting {}", addr);
-                greylist.push((addr.clone(), last_seen.clone()))
-            }
+            } else {
+                for (addr, last_seen) in filtered_addrs {
+                    debug!(target: "net::hosts::greylist_store()", "Inserting {}", addr);
+                    greylist.push((addr.clone(), last_seen.clone()))
+                }
 
-            // Sort the list by last_seen.
-            greylist.sort_unstable_by_key(|entry| entry.1);
+                // Sort the list by last_seen.
+                greylist.sort_unstable_by_key(|entry| entry.1);
 
-            debug!(target: "net::hosts::greylist_store()", "Sorted greylist: {:?}", greylist)
+                debug!(target: "net::hosts::greylist_store()", "Sorted greylist: {:?}", greylist)
+            }
+        } else {
+            debug!(target: "net::hosts::greylist_store()", "Empty address message...")
         }
 
         self.store_subscriber.notify(filtered_addrs_len).await;
@@ -236,6 +266,16 @@ impl Hosts {
         whitelist[index] = (addr.clone(), last_seen);
     }
 
+    // Update the last_seen field of a peer on the greylist.
+    pub async fn greylist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
+        debug!(target: "net::hosts::greylist_update_last_seen()", 
+               "hosts::greylist_update_last_seen() [START]");
+
+        let mut greylist = self.greylist.write().await;
+
+        greylist[index] = (addr.clone(), last_seen);
+    }
+
     pub async fn whitelist_downgrade(&self, addr: &Url) {
         // First lookup the entry using its addr.
         let mut entry = vec![];
@@ -438,17 +478,26 @@ impl Hosts {
     }
 
     /// Get the index for a given addr on the whitelist.
-    pub async fn get_whitelist_index_at_addr(&self, addr: &Url) -> usize {
+    pub async fn get_whitelist_index_at_addr(&self, addr: &Url) -> Result<(usize)> {
         let whitelist = self.whitelist.read().await;
         for (i, (url, _time)) in whitelist.iter().enumerate() {
             if url == addr {
-                return i
+                return Ok(i)
             }
         }
-        // TODO: FIXME: This should never happen.
-        return 0
+        return Err(Error::InvalidIndex)
     }
 
+    /// Get the index for a given addr on the greylist.
+    pub async fn get_greylist_index_at_addr(&self, addr: &Url) -> Result<(usize)> {
+        let greylist = self.greylist.read().await;
+        for (i, (url, _time)) in greylist.iter().enumerate() {
+            if url == addr {
+                return Ok(i)
+            }
+        }
+        return Err(Error::InvalidIndex)
+    }
     /// Return all known whitelisted hosts
     pub async fn whitelist_fetch_all(&self) -> Vec<(Url, u64)> {
         self.whitelist.read().await.iter().cloned().collect()

src/net/protocol/protocol_address.rs

@@ -100,7 +100,12 @@ impl ProtocolAddress {
                 "Received {} addrs from {}", addrs_msg.addrs.len(), self.channel.address(),
             );
 
-            self.hosts.greylist_store(&addrs_msg.addrs).await;
+            debug!(
+                target: "net::protocol_address::handle_receive_addrs()",
+                "Appending to greylist...",
+            );
+
+            self.hosts.greylist_store_or_update(&addrs_msg.addrs).await?;
         }
     }
 
@@ -163,18 +168,21 @@ impl ProtocolAddress {
         let type_id = self.channel.session_type_id();
 
         if type_id != SESSION_OUTBOUND {
-            debug!(target: "net::protocol_address::send_my_addrs()", "Externaladdr not configured. Stopping");
+            debug!(target: "net::protocol_address::send_my_addrs()",
+            "Not an outbound session. Stopping");
             return Ok(())
         }
 
         if self.settings.external_addrs.is_empty() {
-            debug!(target: "net::protocol_address::send_my_addrs()", "Externaladdr not configured. Stopping");
+            debug!(target: "net::protocol_address::send_my_addrs()",
+            "External addr not configured. Stopping");
             return Ok(())
         }
 
         // Do nothing if advertise is set to false
         if self.settings.advertise == false {
-            debug!(target: "net::protocol_address::send_my_addrs()", "Advertise is false. Stopping");
+            debug!(target: "net::protocol_address::send_my_addrs()",
+            "Advertise is false. Stopping");
             return Ok(())
         }
 
@@ -183,19 +191,39 @@ impl ProtocolAddress {
             "[START] address={}", self.channel.address(),
         );
 
-        // See if we can do a version exchange with ourself.
-        debug!(target: "net::protocol_address", "Attempting to ping self");
-        if self.session.ping_node(self.channel.address()).await {
-            // We're online. Broadcast our address.
-            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
-            let addrs = vec![(self.channel.address().clone(), last_seen)];
-            let addrs_msg = AddrsMessage { addrs };
-            self.channel.send(&addrs_msg).await?;
-        } else {
-            debug!(target: "net::protocol_address", "Ping self failed");
+        let mut addrs = vec![];
+        for addr in self.settings.external_addrs.clone() {
+            debug!(target: "net::protocol_seed::send_my_addrs()", "Attempting to ping self");
+
+            // See if we can do a version exchange with ourself.
+            if self.session.ping_node(&addr).await {
+                // We're online. Update last_seen and broadcast our address.
+                let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+                addrs.push((addr, last_seen));
+            } else {
+                debug!(target: "net::protocol_seed::send_my_addrs()", "Ping self failed");
+                return Ok(())
+            }
         }
+        //// See if we can do a version exchange with ourself.
+        debug!(target: "net::protocol_address::send_my_addrs()", "Broadcasting address");
+        let ext_addr_msg = AddrsMessage { addrs };
+        self.channel.send(&ext_addr_msg).await?;
+        debug!(target: "net::protocol_address::send_my_addrs()", "[END]");
 
         Ok(())
+        //debug!(target: "net::protocol_address", "Attempting to ping self");
+        //if self.session.ping_node(self.channel.address()).await {
+        //    // We're online. Broadcast our address.
+        //    let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+        //    let addrs = vec![(self.channel.address().clone(), last_seen)];
+        //    let addrs_msg = AddrsMessage { addrs };
+        //    self.channel.send(&addrs_msg).await?;
+        //} else {
+        //    debug!(target: "net::protocol_address", "Ping self failed");
+        //}
+
+        //Ok(())
     }
 }
 

src/net/protocol/protocol_seed.rs

@@ -65,36 +65,41 @@ impl ProtocolSeed {
     /// Sends own external addresses over a channel. Imports own external addresses
     /// from settings, then adds those addresses to an addrs message and sends it
     /// out over the channel.
-    pub async fn send_self_address(&self) -> Result<()> {
-        debug!(target: "net::protocol_seed::send_self_address()", "[START]");
+    pub async fn send_my_addrs(&self) -> Result<()> {
+        debug!(target: "net::protocol_seed::send_my_addrs()", "[START]");
         // Do nothing if external addresses are not configured
         if self.settings.external_addrs.is_empty() {
+            debug!(target: "net::protocol_seed::send_my_addrs()",
+            "Externaladdr not configured. Stopping");
             return Ok(())
         }
 
         // Do nothing if advertise is set to false
-        if self.settings.advertise {
+        if self.settings.advertise == false {
+            debug!(target: "net::protocol_seed::send_my_addrs()",
+            "Advertise is false. Stopping");
             return Ok(())
         }
 
         let mut addrs = vec![];
         for addr in self.settings.external_addrs.clone() {
+            debug!(target: "net::protocol_seed::send_my_addrs()", "Attempting to ping self");
+
             // See if we can do a version exchange with ourself.
             if self.session.ping_node(&addr).await {
                 // We're online. Update last_seen and broadcast our address.
                 let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
                 addrs.push((addr, last_seen));
+            } else {
+                debug!(target: "net::protocol_seed::send_my_addrs()", "Ping self failed");
+                return Ok(())
             }
         }
-
-        debug!(
-            target: "net::protocol_seed::send_self_address()",
-            "ext_addrs={:?}, dest={}", addrs, self.channel.address(),
-        );
-
+        debug!(target: "net::protocol_seed::send_my_addrs()", "Broadcasting address");
         let ext_addr_msg = AddrsMessage { addrs };
         self.channel.send(&ext_addr_msg).await?;
-        debug!(target: "net::protocol_seed::send_self_address()", "[END]");
+        debug!(target: "net::protocol_seed::send_my_addrs()", "[END]");
+
         Ok(())
     }
 }
@@ -107,9 +112,8 @@ impl ProtocolBase for ProtocolSeed {
     async fn start(self: Arc<Self>, _ex: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_seed::start()", "START => address={}", self.channel.address());
 
-        // TODO: only do this if "advertise" is true
         // Send own address to the seed server
-        self.send_self_address().await?;
+        self.send_my_addrs().await?;
 
         // Send get address message
         let get_addr = GetAddrsMessage {
@@ -124,7 +128,11 @@ impl ProtocolBase for ProtocolSeed {
             target: "net::protocol_seed::start()",
             "Received {} addrs from {}", addrs_msg.addrs.len(), self.channel.address(),
         );
-        self.hosts.greylist_store(&addrs_msg.addrs).await;
+        debug!(
+            target: "net::protocol_seed::start()",
+            "Appending to greylist...",
+        );
+        self.hosts.greylist_store_or_update(&addrs_msg.addrs).await?;
 
         debug!(target: "net::protocol_seed::start()", "END => address={}", self.channel.address());
         Ok(())

src/net/session/outbound_session.rs

@@ -87,7 +87,7 @@ impl OutboundSession {
             greylist_refinery: GreylistRefinery::new(),
         });
         self_.peer_discovery.session.init(self_.clone());
-        self_.greylist_refinery.session.init(self_.clone());
+        //self_.greylist_refinery.session.init(self_.clone());
         self_
     }
 
@@ -107,7 +107,7 @@ impl OutboundSession {
         }
 
         self.peer_discovery.clone().start().await;
-        self.greylist_refinery.clone().start().await;
+        //self.greylist_refinery.clone().start().await;
     }
 
     /// Stops the outbound session.
@@ -318,7 +318,8 @@ impl Slot {
             // able to establish a connection to it to it.
             let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
 
-            hosts.whitelist_update(&addr_final, last_seen).await;
+            // TODO: FIXME: unwrap
+            hosts.whitelist_store_or_update(&addr_final, last_seen).await.unwrap();
 
             dnetev!(self, OutboundSlotConnected, {
                 slot: self.slot,

src/net/tests.rs

@@ -16,9 +16,9 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-// cargo +nightly test --release --all-features --lib p2p_propagation-- --include-ignored
+// cargo +nightly test --release --all-features --lib p2p -- --include-ignored
 
-use std::sync::Arc;
+use std::{sync::Arc, time::SystemTime};
 
 use log::info;
 use rand::{prelude::SliceRandom, Rng};
@@ -31,7 +31,7 @@ use crate::{
 };
 
 // Number of nodes to spawn and number of peers each node connects to
-const N_NODES: usize = 2;
+const N_NODES: usize = 5;
 const N_CONNS: usize = 2;
 
 // TODO: test whitelist propagation between peers
@@ -41,23 +41,23 @@ const N_CONNS: usize = 2;
 #[test]
 fn p2p_test() {
     let mut cfg = simplelog::ConfigBuilder::new();
-    cfg.add_filter_ignore("sled".to_string());
-    cfg.add_filter_ignore("net::protocol_ping".to_string());
+    //cfg.add_filter_ignore("sled".to_string());
     cfg.add_filter_ignore("net::channel::subscribe_stop()".to_string());
-    cfg.add_filter_ignore("net::hosts".to_string());
-    cfg.add_filter_ignore("net::session".to_string());
+    //cfg.add_filter_ignore("net::hosts".to_string());
+    //cfg.add_filter_ignore("net::session".to_string());
     cfg.add_filter_ignore("net::message_subscriber".to_string());
-    cfg.add_filter_ignore("net::protocol_version".to_string());
-    cfg.add_filter_ignore("net::protocol_jobs_manager".to_string());
-    cfg.add_filter_ignore("net::protocol_registry".to_string());
-    cfg.add_filter_ignore("net::channel::send()".to_string());
-    cfg.add_filter_ignore("net::channel::start()".to_string());
-    cfg.add_filter_ignore("net::channel::stop()".to_string());
-    cfg.add_filter_ignore("net::channel::handle_stop()".to_string());
-    cfg.add_filter_ignore("net::channel::subscribe_msg()".to_string());
-    cfg.add_filter_ignore("net::channel::main_receive_loop()".to_string());
-    cfg.add_filter_ignore("net::greylist_refinery::run()".to_string());
-    cfg.add_filter_ignore("net::outbound_session::try_connect()".to_string());
+    //cfg.add_filter_ignore("net::protocol_ping".to_string());
+    //cfg.add_filter_ignore("net::protocol_version".to_string());
+    //cfg.add_filter_ignore("net::protocol_jobs_manager".to_string());
+    //cfg.add_filter_ignore("net::protocol_registry".to_string());
+    //cfg.add_filter_ignore("net::channel::send()".to_string());
+    //cfg.add_filter_ignore("net::channel::start()".to_string());
+    //cfg.add_filter_ignore("net::channel::stop()".to_string());
+    //cfg.add_filter_ignore("net::channel::handle_stop()".to_string());
+    //cfg.add_filter_ignore("net::channel::subscribe_msg()".to_string());
+    //cfg.add_filter_ignore("net::channel::main_receive_loop()".to_string());
+    //cfg.add_filter_ignore("net::greylist_refinery::run()".to_string());
+    //cfg.add_filter_ignore("net::outbound_session::try_connect()".to_string());
 
     simplelog::TermLogger::init(
         //simplelog::LevelFilter::Info,
@@ -78,13 +78,13 @@ fn p2p_test() {
         .each(0..N_NODES, |_| future::block_on(ex.run(shutdown.recv())))
         .finish(|| {
             future::block_on(async {
-                p2p_propagation_real(ex_).await;
+                hostlist_propagation(ex_).await;
                 drop(signal);
             })
         });
 }
 
-async fn p2p_propagation_real(ex: Arc<Executor<'static>>) {
+async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
     let seed_addr = Url::parse(&format!("tcp://127.0.0.1:{}", 51505)).unwrap();
 
     let mut p2p_instances = vec![];
@@ -94,31 +94,34 @@ async fn p2p_propagation_real(ex: Arc<Executor<'static>>) {
     let settings = Settings {
         localnet: true,
         inbound_addrs: vec![seed_addr.clone()],
+        external_addrs: vec![seed_addr.clone()],
         outbound_connections: 0,
         outbound_connect_timeout: 2,
         inbound_connections: usize::MAX,
         peers: vec![],
         allowed_transports: vec!["tcp".to_string()],
-        advertise: true,
+        node_id: "seed".to_string(),
+        //advertise: true,
         ..Default::default()
     };
 
     let p2p = P2p::new(settings, ex.clone()).await;
     p2p_instances.push(p2p);
 
-    // Initialize outboun nodes
+    info!("Initializing outbound nodes");
     for i in 0..N_NODES {
         let settings = Settings {
             localnet: true,
             inbound_addrs: vec![Url::parse(&format!("tcp://127.0.0.1:{}", 13200 + i)).unwrap()],
             external_addrs: vec![Url::parse(&format!("tcp://127.0.0.1:{}", 13200 + i)).unwrap()],
             outbound_connections: 2,
-            outbound_connect_timeout: 2,
+            outbound_connect_timeout: 10,
             inbound_connections: usize::MAX,
             seeds: vec![seed_addr.clone()],
             peers: vec![],
             allowed_transports: vec!["tcp".to_string()],
-            advertise: true,
+            node_id: i.to_string(),
+            //advertise: true,
             ..Default::default()
         };
 
@@ -131,8 +134,21 @@ async fn p2p_propagation_real(ex: Arc<Executor<'static>>) {
         p2p.clone().start().await.unwrap();
     }
 
-    info!("Waiting 10s until all peers connect");
-    sleep(10).await;
+    info!("Waiting 30s until all peers connect");
+    sleep(30).await;
+
+    info!("Inspecting peerlists...");
+    for p2p in p2p_instances.iter() {
+        let hosts = p2p.hosts();
+        info!("START peerlist {}", p2p.settings().node_id);
+        assert!(!hosts.is_empty_greylist().await);
+        let greylist = hosts.greylist.read().await;
+        for (url, last_seen) in greylist.iter() {
+            info!("{}", url);
+            info!("{}", last_seen);
+        }
+        info!("END peerlist {}", p2p.settings().node_id);
+    }
 
     // Stop the P2P network
     for p2p in p2p_instances.iter() {