net: remove connection from anchorlist when it disconnects and cleanup.

also remove perform_local_handshake function as it's made redudant by
the anchorlist removal change.

src/net/hosts/refinery.rs

@@ -132,7 +132,7 @@ pub async fn ping_node(addr: &Url, p2p: P2pPtr) -> bool {
             debug!(target: "net::refinery::ping_node()", "Connected successfully!");
             let proto_ver = ProtocolVersion::new(channel.clone(), p2p.settings()).await;
 
-            let handshake_task = session_outbound.perform_local_handshake(
+            let handshake_task = session_outbound.perform_handshake_protocols(
                 proto_ver,
                 channel.clone(),
                 p2p.executor(),

src/net/hosts/store.rs

@@ -53,8 +53,7 @@ const WHITELIST_MAX_LEN: usize = 5000;
 const GREYLIST_MAX_LEN: usize = 2000;
 
 /// Manages a store of network addresses
-// TODO:
-//       * Test the performance overhead of using vectors for white/grey/anchor lists.
+// TODO: Test the performance overhead of using vectors for white/grey/anchor lists.
 //       * Check whether anchorlist has a max size in Monero.
 pub struct Hosts {
     // Intermediary node list that is periodically probed and updated to whitelist.
@@ -509,6 +508,16 @@ impl Hosts {
         greylist.sort_by_key(|entry| entry.1);
     }
 
+    pub async fn anchorlist_remove(&self, addr: &Url, position: usize) {
+        debug!(target: "net::refinery::run()", "Removing disconnected peer {} from anchorlist", addr);
+        let mut anchorlist = self.anchorlist.write().await;
+
+        anchorlist.remove(position);
+
+        // Sort the list by last_seen.
+        anchorlist.sort_by_key(|entry| entry.1);
+    }
+
     pub async fn subscribe_store(&self) -> Result<Subscription<usize>> {
         let sub = self.store_subscriber.clone().subscribe().await;
         Ok(sub)

src/net/session/manual_session.rs

@@ -32,7 +32,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use log::{info, warn};
+use log::{debug, info, warn};
 use smol::lock::Mutex;
 use url::Url;
 

src/net/session/mod.rs

@@ -22,7 +22,7 @@ use std::{
 };
 
 use async_trait::async_trait;
-use log::debug;
+use log::{debug, error};
 use smol::Executor;
 
 use super::{channel::ChannelPtr, p2p::P2pPtr, protocol::ProtocolVersion};
@@ -49,7 +49,7 @@ pub type SessionWeakPtr = Weak<dyn Session + Send + Sync + 'static>;
 
 /// Removes channel from the list of connected channels when a stop signal
 /// is received.
-pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
+pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr, type_id: SessionBitFlag) {
     debug!(target: "net::session::remove_sub_on_stop()", "[START]");
     // Subscribe to stop events
     let stop_sub = channel.clone().subscribe_stop().await;
@@ -64,6 +64,16 @@ pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
         "Received stop event. Removing channel {}", channel.address(),
     );
 
+    if type_id != SESSION_INBOUND {
+        match p2p.hosts().get_anchorlist_index_at_addr(channel.address()).await {
+            Ok(index) => p2p.hosts().anchorlist_remove(channel.address(), index).await,
+            Err(e) => {
+                error!(target: "net::session::remove_sub_on_stop()",
+                "Can't remove anchor connection {}", e)
+            }
+        }
+    }
+
     // Remove channel from p2p
     p2p.remove(channel).await;
     debug!(target: "net::session::remove_sub_on_stop()", "[END]");
@@ -151,25 +161,7 @@ pub trait Session: Sync {
         self.p2p().store(channel.clone()).await;
 
         // Subscribe to stop, so we can remove from p2p
-        executor.spawn(remove_sub_on_stop(self.p2p(), channel)).detach();
-
-        // Channel is ready for use
-        Ok(())
-    }
-
-    async fn perform_local_handshake(
-        &self,
-        protocol_version: Arc<ProtocolVersion>,
-        channel: ChannelPtr,
-        executor: Arc<Executor<'_>>,
-    ) -> Result<()> {
-        // Perform handshake
-        protocol_version.run(executor.clone()).await?;
-
-        // Channel is now initialized
-
-        // Subscribe to stop, so we can remove from p2p
-        executor.spawn(remove_sub_on_stop(self.p2p(), channel)).detach();
+        executor.spawn(remove_sub_on_stop(self.p2p(), channel, self.type_id())).detach();
 
         // Channel is ready for use
         Ok(())

src/net/session/outbound_session.rs

@@ -374,117 +374,6 @@ impl Slot {
         Ok(())
     }
 
-    //async fn run(self: Arc<Self>) {
-    //    // This is the main outbound connection loop where we try to establish
-    //    // a connection in the slot. The `try_connect` function will block in
-    //    // case the connection was sucessfully established. If it fails, then
-    //    // we will wait for a defined number of seconds and try to fill the
-    //    // slot again. This function should never exit during the lifetime of
-    //    // the P2P network, as it is supposed to represent an outbound slot we
-    //    // want to fill.
-    //    // The actual connection logic and peer selection is in `try_connect`.
-    //    // If the connection is successful, `try_connect` will wait for a stop
-    //    // signal and then exit. Once it exits, we'll run `try_connect` again
-    //    // and attempt to fill the slot with another peer.
-    //    let hosts = self.p2p().hosts();
-    //    loop {
-    //        // Activate the slot
-    //        debug!(
-    //            target: "net::outbound_session::try_connect()",
-    //            "[P2P] Finding a host to connect to for outbound slot #{}",
-    //            self.slot,
-    //        );
-
-    //        // Retrieve outbound transports
-    //        let transports = &self.p2p().settings().allowed_transports;
-
-    //        // Find a whitelisted address to connect to. We also do peer discovery here if needed.
-    //        // Start by looking up an anchorlist address we can connect to.
-    //        let (addr, _last_seen) = if let Some(addr) =
-    //            hosts.anchorlist_fetch_address_with_lock(self.p2p(), transports).await
-    //        {
-    //            addr
-    //        } else {
-    //            dnetev!(self, OutboundSlotSleeping, {
-    //                slot: self.slot,
-    //            });
-
-    //            self.wakeup_self.reset();
-    //            // Peer discovery
-    //            self.session().wakeup_peer_discovery();
-    //            // Wait to be woken up by peer discovery
-    //            self.wakeup_self.wait().await;
-    //            continue
-    //        };
-
-    //        info!(
-    //            target: "net::outbound_session::try_connect()",
-    //            "[P2P] Connecting outbound slot #{} [{}]",
-    //            self.slot, addr,
-    //        );
-
-    //        dnetev!(self, OutboundSlotConnecting, {
-    //            slot: self.slot,
-    //            addr: addr.clone(),
-    //        });
-
-    //        let (addr_final, channel) = match self.try_connect(addr.clone()).await {
-    //            Ok(connect_info) => connect_info,
-    //            Err(err) => {
-    //                error!(
-    //                    target: "net::outbound_session",
-    //                    "[P2P] Outbound slot #{} connection failed: {}",
-    //                    self.slot, err,
-    //                );
-
-    //                dnetev!(self, OutboundSlotDisconnected, {
-    //                    slot: self.slot,
-    //                    err: err.to_string()
-    //                });
-
-    //                self.channel_id.store(0, Ordering::Relaxed);
-    //                continue
-    //            }
-    //        };
-
-    //        info!(
-    //            target: "net::outbound_session::try_connect()",
-    //            "[P2P] Outbound slot #{} connected [{}]",
-    //            self.slot, addr_final
-    //        );
-
-    //        dnetev!(self, OutboundSlotConnected, {
-    //            slot: self.slot,
-    //            addr: addr_final.clone(),
-    //            channel_id: channel.info.id
-    //        });
-
-    //        let stop_sub = channel.subscribe_stop().await.expect("Channel should not be stopped");
-    //        // Setup new channel
-    //        if let Err(err) = self.setup_channel(addr, channel.clone()).await {
-    //            info!(
-    //                target: "net::outbound_session",
-    //                "[P2P] Outbound slot #{} disconnected: {}",
-    //                self.slot, err
-    //            );
-
-    //            dnetev!(self, OutboundSlotDisconnected, {
-    //                slot: self.slot,
-    //                err: err.to_string()
-    //            });
-
-    //            self.channel_id.store(0, Ordering::Relaxed);
-    //            continue
-    //        }
-
-    //        self.channel_id.store(channel.info.id, Ordering::Relaxed);
-
-    //        // Wait for channel to close
-    //        stop_sub.receive().await;
-    //        self.channel_id.store(0, Ordering::Relaxed);
-    //    }
-    //}
-
     /// Start making an outbound connection, using provided [`Connector`].
     /// Tries to find a valid address to connect to, otherwise does peer
     /// discovery. The peer discovery loops until some peer we can connect

src/net/tests.rs

@@ -88,7 +88,7 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
             outbound_connect_timeout: 10,
             inbound_connections: usize::MAX,
             seeds: vec![seed_addr.clone()],
-            hostlist: String::from(".config/darkfi/hosts.tsv"),
+            hostlist: String::from(format!(".config/darkfi/hosts{}.tsv", i)),
             peers,
             allowed_transports: vec!["tcp".to_string()],
             node_id: i.to_string(),
@@ -106,7 +106,7 @@ async fn hostlist_propagation(ex: Arc<Executor<'static>>) {
     }
 
     info!("Waiting until all peers connect");
-    sleep(10).await;
+    sleep(60).await;
 
     info!("Inspecting hostlists...");
     for p2p in p2p_instances.iter() {