net: migrate outbound sessions over to new protocol. also replace lilith periodic_purge with periodic_cleanse.

periodic_cleanse doesn't remove any connections, it simply updates the
last_seen field when it is able to establish a connection, per:

bin/lilith/src/main.rs

@@ -21,6 +21,7 @@ use std::{
     path::Path,
     process::exit,
     sync::Arc,
+    time::{Duration, Instant, SystemTime},
 };
 
 use async_trait::async_trait;
@@ -57,7 +58,7 @@ const CONFIG_FILE: &str = "lilith_config.toml";
 const CONFIG_FILE_CONTENTS: &str = include_str!("../lilith_config.toml");
 
 /// Period in which the peer purge happens (in seconds)
-const PURGE_PERIOD: u64 = 60;
+const CLEANSE_PERIOD: u64 = 60;
 /// Amount of hosts to try each purge iteration
 const PROBE_HOSTS_N: u32 = 10;
 
@@ -98,10 +99,10 @@ impl Spawn {
     async fn addresses(&self) -> Vec<JsonValue> {
         self.p2p
             .hosts()
-            .fetch_all()
+            .whitelist_fetch_all()
             .await
             .iter()
-            .map(|addr| JsonValue::String(addr.to_string()))
+            .map(|(addr, url)| JsonValue::String(addr.to_string()))
             .collect()
     }
 
@@ -143,95 +144,52 @@ struct Lilith {
 }
 
 impl Lilith {
-    /// Internal task to run a periodic purge of unreachable hosts
-    /// for a specific P2P network.
-    async fn periodic_purge(name: String, p2p: P2pPtr, ex: Arc<Executor<'_>>) -> Result<()> {
-        info!(target: "lilith", "Starting periodic host purge task for \"{}\"", name);
+    async fn periodic_cleanse(name: String, p2p: P2pPtr, ex: Arc<Executor<'_>>) -> Result<()> {
+        info!(target: "lilith", "Starting periodic host cleanse task for \"{}\"", name);
 
         // Initialize a growable ring buffer(VecDeque) to store known hosts
         let ring_buffer = Arc::new(RwLock::new(VecDeque::<Url>::new()));
         loop {
             // Wait for next purge period
-            sleep(PURGE_PERIOD).await;
-            debug!(target: "lilith", "[{}] The Purge has started...", name);
+            sleep(CLEANSE_PERIOD).await;
+            debug!(target: "lilith", "[{}] The Cleanse has started...", name);
 
             // Check if new hosts exist and add them to the end of the ring buffer
             let mut lock = ring_buffer.write().await;
-            let hosts = p2p.clone().hosts().fetch_all().await;
+            let hosts = p2p.clone().hosts().whitelist_fetch_all().await;
             if hosts.len() != lock.len() {
                 // Since hosts are stored in a HashSet we have to check all of them
-                for host in hosts {
-                    if !lock.contains(&host) {
-                        lock.push_back(host);
+                for (addr, _last_seen) in hosts {
+                    if !lock.contains(&addr) {
+                        lock.push_back(addr);
                     }
                 }
             }
 
             // Pick first up to PROBE_HOSTS_N hosts from the ring buffer
-            let mut purgers = vec![];
+            let mut cleansers = vec![];
             let mut index = 0;
             while index <= PROBE_HOSTS_N {
                 match lock.pop_front() {
-                    Some(host) => purgers.push(host),
+                    Some(host) => cleansers.push(host),
                     None => break,
                 };
                 index += 1;
             }
 
-            // Try to connect to them. If we can't reach them, remove them from our set.
-            let purgers_str: Vec<&str> = purgers.iter().map(|x| x.as_str()).collect();
-            debug!(target: "lilith", "[{}] Got: {:?}", name, purgers_str);
+            // Try to connect to them. If we establish a connection, update the last_seen() field.
+            let cleansers_str: Vec<&str> = cleansers.iter().map(|x| x.as_str()).collect();
+            debug!(target: "lilith", "[{}] Got: {:?}", name, cleansers_str);
 
             let mut tasks = vec![];
 
-            for host in &purgers {
+            for host in &cleansers {
                 let p2p_ = p2p.clone();
                 let ex_ = ex.clone();
                 let ring_buffer_ = ring_buffer.clone();
-                tasks.push(async move {
-                    let session_out = p2p_.session_outbound();
-                    let session_weak = Arc::downgrade(&session_out);
-
-                    let connector = Connector::new(p2p_.settings(), session_weak);
-                    debug!(target: "lilith", "Connecting to {}", host);
-                    match connector.connect(host).await {
-                        Ok((_url, channel)) => {
-                            debug!(target: "lilith", "Connected successfully!");
-                            let proto_ver = ProtocolVersion::new(
-                                channel.clone(),
-                                p2p_.settings().clone(),
-                                //p2p_.hosts().clone(),
-                            )
-                            .await;
-
-                            let handshake_task = session_out.perform_handshake_protocols(
-                                proto_ver,
-                                channel.clone(),
-                                ex_.clone(),
-                            );
-
-                            channel.clone().start(ex_.clone());
-
-                            match handshake_task.await {
-                                Ok(()) => {
-                                    debug!(target: "lilith", "Handshake success! Stopping channel.");
-                                    channel.stop().await;
-                                    // Push host back to the ring buffer
-                                    ring_buffer_.write().await.push_back(host.clone());
-                                }
-                                Err(e) => {
-                                    debug!(target: "lilith", "Handshake failure! {}", e);
-                                    p2p_.hosts().remove(host).await;
-                                }
-                            }
-                        }
 
-                        Err(e) => {
-                            debug!(target: "lilith", "Failed to connect to {}, removing from set ({})", host, e);
-                            // Remove from hosts set
-                            p2p_.hosts().remove(host).await;
-                        }
-                    }
+                tasks.push(async move {
+                    p2p_.hosts().refresh_whitelist(&host, p2p_.clone(), ex_.clone()).await;
                 });
             }
 
@@ -239,6 +197,102 @@ impl Lilith {
         }
     }
 
+    ///// Internal task to run a periodic purge of unreachable hosts
+    ///// for a specific P2P network.
+    //async fn periodic_purge(name: String, p2p: P2pPtr, ex: Arc<Executor<'_>>) -> Result<()> {
+    //    info!(target: "lilith", "Starting periodic host purge task for \"{}\"", name);
+
+    //    // Initialize a growable ring buffer(VecDeque) to store known hosts
+    //    let ring_buffer = Arc::new(RwLock::new(VecDeque::<Url>::new()));
+    //    loop {
+    //        // Wait for next purge period
+    //        sleep(PURGE_PERIOD).await;
+    //        debug!(target: "lilith", "[{}] The Purge has started...", name);
+
+    //        // Check if new hosts exist and add them to the end of the ring buffer
+    //        let mut lock = ring_buffer.write().await;
+    //        let hosts = p2p.clone().hosts().whitelist_fetch_all().await;
+    //        if hosts.len() != lock.len() {
+    //            // Since hosts are stored in a HashSet we have to check all of them
+    //            for host in hosts {
+    //                if !lock.contains(&host) {
+    //                    lock.push_back(host);
+    //                }
+    //            }
+    //        }
+
+    //        // Pick first up to PROBE_HOSTS_N hosts from the ring buffer
+    //        let mut purgers = vec![];
+    //        let mut index = 0;
+    //        while index <= PROBE_HOSTS_N {
+    //            match lock.pop_front() {
+    //                Some(host) => purgers.push(host),
+    //                None => break,
+    //            };
+    //            index += 1;
+    //        }
+
+    //        // Try to connect to them. If we can't reach them, remove them from our set.
+    //        let purgers_str: Vec<&str> = purgers.iter().map(|x| x.as_str()).collect();
+    //        debug!(target: "lilith", "[{}] Got: {:?}", name, purgers_str);
+
+    //        let mut tasks = vec![];
+
+    //        for host in &purgers {
+    //            let p2p_ = p2p.clone();
+    //            let ex_ = ex.clone();
+    //            let ring_buffer_ = ring_buffer.clone();
+    //            tasks.push(async move {
+    //                let session_out = p2p_.session_outbound();
+    //                let session_weak = Arc::downgrade(&session_out);
+
+    //                let connector = Connector::new(p2p_.settings(), session_weak);
+    //                debug!(target: "lilith", "Connecting to {}", host);
+    //                match connector.connect(host).await {
+    //                    Ok((_url, channel)) => {
+    //                        debug!(target: "lilith", "Connected successfully!");
+    //                        let proto_ver = ProtocolVersion::new(
+    //                            channel.clone(),
+    //                            p2p_.settings().clone(),
+    //                            //p2p_.hosts().clone(),
+    //                        )
+    //                        .await;
+
+    //                        let handshake_task = session_out.perform_handshake_protocols(
+    //                            proto_ver,
+    //                            channel.clone(),
+    //                            ex_.clone(),
+    //                        );
+
+    //                        channel.clone().start(ex_.clone());
+
+    //                        match handshake_task.await {
+    //                            Ok(()) => {
+    //                                debug!(target: "lilith", "Handshake success! Stopping channel.");
+    //                                channel.stop().await;
+    //                                // Push host back to the ring buffer
+    //                                ring_buffer_.write().await.push_back(host.clone());
+    //                            }
+    //                            Err(e) => {
+    //                                debug!(target: "lilith", "Handshake failure! {}", e);
+    //                                p2p_.hosts().remove(host).await;
+    //                            }
+    //                        }
+    //                    }
+
+    //                    Err(e) => {
+    //                        debug!(target: "lilith", "Failed to connect to {}, removing from set ({})", host, e);
+    //                        // Remove from hosts set
+    //                        p2p_.hosts().remove(host).await;
+    //                    }
+    //                }
+    //            });
+    //        }
+
+    //        join_all(tasks).await;
+    //    }
+    //}
+
     // RPCAPI:
     // Returns all spawned networks names with their node addresses.
     // --> {"jsonrpc": "2.0", "method": "spawns", "params": [], "id": 42}
@@ -271,41 +325,41 @@ impl RequestHandler for Lilith {
     }
 }
 
-/// Attempt to read existing hosts tsv
-fn load_hosts(path: &Path, networks: &[&str]) -> HashMap<String, HashSet<Url>> {
-    let mut saved_hosts = HashMap::new();
-
-    let contents = load_file(path);
-    if let Err(e) = contents {
-        warn!(target: "lilith", "Failed retrieving saved hosts: {}", e);
-        return saved_hosts
-    }
-
-    for line in contents.unwrap().lines() {
-        let data: Vec<&str> = line.split('\t').collect();
-        if networks.contains(&data[0]) {
-            let mut hosts = match saved_hosts.get(data[0]) {
-                Some(hosts) => hosts.clone(),
-                None => HashSet::new(),
-            };
-
-            let url = match Url::parse(data[1]) {
-                Ok(u) => u,
-                Err(e) => {
-                    warn!(target: "lilith", "Skipping malformed url: {} ({})", data[1], e);
-                    continue
-                }
-            };
-
-            hosts.insert(url);
-            saved_hosts.insert(data[0].to_string(), hosts);
-        }
-    }
-
-    saved_hosts
-}
+///// Attempt to read existing hosts tsv
+//fn load_hosts(path: &Path, networks: &[&str]) -> HashMap<String, HashSet<Url>> {
+//    let mut saved_hosts = HashMap::new();
+//
+//    let contents = load_file(path);
+//    if let Err(e) = contents {
+//        warn!(target: "lilith", "Failed retrieving saved hosts: {}", e);
+//        return saved_hosts
+//    }
+//
+//    for line in contents.unwrap().lines() {
+//        let data: Vec<&str> = line.split('\t').collect();
+//        if networks.contains(&data[0]) {
+//            let mut hosts = match saved_hosts.get(data[0]) {
+//                Some(hosts) => hosts.clone(),
+//                None => HashSet::new(),
+//            };
+//
+//            let url = match Url::parse(data[1]) {
+//                Ok(u) => u,
+//                Err(e) => {
+//                    warn!(target: "lilith", "Skipping malformed url: {} ({})", data[1], e);
+//                    continue
+//                }
+//            };
+//
+//            hosts.insert(url);
+//            saved_hosts.insert(data[0].to_string(), hosts);
+//        }
+//    }
+//
+//    saved_hosts
+//}
 
-fn load_hosts2(path: &Path, networks: &[&str]) -> HashMap<String, Vec<(Url, u64)>> {
+fn load_hosts(path: &Path, networks: &[&str]) -> HashMap<String, Vec<(Url, u64)>> {
     let mut saved_hosts = HashMap::new();
 
     let contents = load_file(path);
@@ -316,7 +370,7 @@ fn load_hosts2(path: &Path, networks: &[&str]) -> HashMap<String, Vec<(Url, u64)
 
     for line in contents.unwrap().lines() {
         let data: Vec<&str> = line.split('\t').collect();
-        debug!(target: "lilith", "::load_hosts2()::data\"{:?}\"", data);
+        debug!(target: "lilith", "::load_hosts()::data\"{:?}\"", data);
         if networks.contains(&data[0]) {
             let mut hosts = match saved_hosts.get(data[0]) {
                 Some(hosts) => hosts.clone(),
@@ -346,24 +400,24 @@ fn load_hosts2(path: &Path, networks: &[&str]) -> HashMap<String, Vec<(Url, u64)
     saved_hosts
 }
 
-async fn save_hosts(path: &Path, networks: &[Spawn]) {
-    let mut tsv = String::new();
-
-    for spawn in networks {
-        for host in spawn.p2p.hosts().fetch_all().await {
-            tsv.push_str(&format!("{}\t{}\n", spawn.name, host.as_str()));
-        }
-    }
-
-    if !tsv.eq("") {
-        info!(target: "lilith", "Saving current hosts of spawned networks to: {:?}", path);
-        if let Err(e) = save_file(path, &tsv) {
-            error!(target: "lilith", "Failed saving hosts: {}", e);
-        }
-    }
-}
+//async fn save_hosts(path: &Path, networks: &[Spawn]) {
+//    let mut tsv = String::new();
+//
+//    for spawn in networks {
+//        for host in spawn.p2p.hosts().fetch_all().await {
+//            tsv.push_str(&format!("{}\t{}\n", spawn.name, host.as_str()));
+//        }
+//    }
+//
+//    if !tsv.eq("") {
+//        info!(target: "lilith", "Saving current hosts of spawned networks to: {:?}", path);
+//        if let Err(e) = save_file(path, &tsv) {
+//            error!(target: "lilith", "Failed saving hosts: {}", e);
+//        }
+//    }
+//}
 
-async fn save_hosts2(path: &Path, networks: &[Spawn]) {
+async fn save_hosts(path: &Path, networks: &[Spawn]) {
     let mut tsv = String::new();
 
     for spawn in networks {
@@ -450,56 +504,56 @@ fn parse_configured_networks(data: &str) -> Result<HashMap<String, NetInfo>> {
     Ok(ret)
 }
 
+//async fn spawn_net(
+//    name: String,
+//    info: &NetInfo,
+//    saved_hosts: Vec<(Url, u64)>,
+//    ex: Arc<Executor<'static>>,
+//) -> Result<Spawn> {
+//    let mut listen_urls = vec![];
+//
+//    // Configure listen addrs for this network
+//    for url in &info.accept_addrs {
+//        listen_urls.push(url.clone());
+//    }
+//
+//    // P2P network settings
+//    let settings = net::Settings {
+//        inbound_addrs: listen_urls.clone(),
+//        seeds: info.seeds.clone(),
+//        peers: info.peers.clone(),
+//        outbound_connections: 0,
+//        outbound_connect_timeout: 30,
+//        inbound_connections: 512,
+//        app_version: info.version.clone(),
+//        localnet: info.localnet,
+//        allowed_transports: vec![
+//            "tcp".to_string(),
+//            "tcp+tls".to_string(),
+//            "tor".to_string(),
+//            "tor+tls".to_string(),
+//            "nym".to_string(),
+//            "nym+tls".to_string(),
+//        ],
+//        ..Default::default()
+//    };
+//
+//    // Create P2P instance
+//    let p2p = P2p::new(settings, ex.clone()).await;
+//
+//    // Fill db with cached hosts
+//    let hosts: Vec<(Url, u64)> = saved_hosts.iter().cloned().collect();
+//    p2p.hosts().greylist_store(&hosts).await;
+//
+//    let addrs_str: Vec<&str> = listen_urls.iter().map(|x| x.as_str()).collect();
+//    info!(target: "lilith", "Starting seed network node for \"{}\" on {:?}", name, addrs_str);
+//    p2p.clone().start().await?;
+//
+//    let spawn = Spawn { name, p2p };
+//    Ok(spawn)
+//}
+//
 async fn spawn_net(
-    name: String,
-    info: &NetInfo,
-    saved_hosts: &HashSet<Url>,
-    ex: Arc<Executor<'static>>,
-) -> Result<Spawn> {
-    let mut listen_urls = vec![];
-
-    // Configure listen addrs for this network
-    for url in &info.accept_addrs {
-        listen_urls.push(url.clone());
-    }
-
-    // P2P network settings
-    let settings = net::Settings {
-        inbound_addrs: listen_urls.clone(),
-        seeds: info.seeds.clone(),
-        peers: info.peers.clone(),
-        outbound_connections: 0,
-        outbound_connect_timeout: 30,
-        inbound_connections: 512,
-        app_version: info.version.clone(),
-        localnet: info.localnet,
-        allowed_transports: vec![
-            "tcp".to_string(),
-            "tcp+tls".to_string(),
-            "tor".to_string(),
-            "tor+tls".to_string(),
-            "nym".to_string(),
-            "nym+tls".to_string(),
-        ],
-        ..Default::default()
-    };
-
-    // Create P2P instance
-    let p2p = P2p::new(settings, ex.clone()).await;
-
-    // Fill db with cached hosts
-    let hosts: Vec<Url> = saved_hosts.iter().cloned().collect();
-    p2p.hosts().store(&hosts).await;
-
-    let addrs_str: Vec<&str> = listen_urls.iter().map(|x| x.as_str()).collect();
-    info!(target: "lilith", "Starting seed network node for \"{}\" on {:?}", name, addrs_str);
-    p2p.clone().start().await?;
-
-    let spawn = Spawn { name, p2p };
-    Ok(spawn)
-}
-
-async fn spawn_net2(
     name: String,
     info: &NetInfo,
     saved_hosts: &Vec<(Url, u64)>,
@@ -646,8 +700,8 @@ async fn spawn_net2(
 //    Ok(())
 //}
 
-async_daemonize!(realmain2);
-async fn realmain2(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
+async_daemonize!(realmain);
+async fn realmain(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     // Pick up network settings from the TOML config
     let cfg_path = get_config_path(args.config, CONFIG_FILE)?;
     let toml_contents = std::fs::read_to_string(cfg_path)?;
@@ -660,7 +714,7 @@ async fn realmain2(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
 
     // Retrieve any saved hosts for configured networks
     let net_names: Vec<&str> = configured_nets.keys().map(|x| x.as_str()).collect();
-    let saved_hosts = load_hosts2(&expand_path(&args.hosts_file)?, &net_names);
+    let saved_hosts = load_hosts(&expand_path(&args.hosts_file)?, &net_names);
 
     // Spawn configured networks
     let mut networks = vec![];
@@ -669,7 +723,7 @@ async fn realmain2(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         // e.g. p2p_v3, p2p_v4, etc. Therefore we can spawn multiple networks
         // and they would all be version-checked, so we avoid mismatches when
         // seeding peers.
-        match spawn_net2(
+        match spawn_net(
             name.to_string(),
             info,
             saved_hosts.get(name).unwrap_or(&Vec::new()),
@@ -692,7 +746,7 @@ async fn realmain2(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
         let name = network.name.clone();
         let task = StoppableTask::new();
         task.clone().start(
-            Lilith::periodic_purge(name.clone(), network.p2p.clone(), ex.clone()),
+            Lilith::periodic_cleanse(name.clone(), network.p2p.clone(), ex.clone()),
             |res| async move {
                 match res {
                     Ok(()) | Err(Error::DetachedTaskStopped) => { /* Do nothing */ }
@@ -727,7 +781,7 @@ async fn realmain2(args: Args, ex: Arc<Executor<'static>>) -> Result<()> {
     info!(target: "lilith", "Caught termination signal, cleaning up and exiting...");
 
     // Save in-memory hosts to tsv file
-    save_hosts2(&expand_path(&args.hosts_file)?, &lilith.networks).await;
+    save_hosts(&expand_path(&args.hosts_file)?, &lilith.networks).await;
 
     info!(target: "lilith", "Stopping JSON-RPC server...");
     rpc_task.stop().await;

src/net/hosts.rs

@@ -53,11 +53,10 @@ pub struct Hosts {
     greylist: RwLock<Vec<(Url, u64)>>,
 
     // Recently seen nodes.
-    whitelist: RwLock<Vec<(Url, u64)>>,
-
-    /// Set of stored addresses
-    addrs: RwLock<HashSet<Url>>,
+    pub whitelist: RwLock<Vec<(Url, u64)>>,
 
+    ///// Set of stored addresses
+    //addrs: RwLock<HashSet<Url>>,
     /// Set of stored addresses that are quarantined.
     /// We quarantine peers we've been unable to connect to, but we keep them
     /// around so we can potentially try them again, up to n tries. This should
@@ -81,7 +80,7 @@ impl Hosts {
         Arc::new(Self {
             whitelist: RwLock::new(Vec::new()),
             greylist: RwLock::new(Vec::new()),
-            addrs: RwLock::new(HashSet::new()),
+            //addrs: RwLock::new(HashSet::new()),
             quarantine: RwLock::new(HashMap::new()),
             rejected: RwLock::new(HashSet::new()),
             store_subscriber: Subscriber::new(),
@@ -89,24 +88,24 @@ impl Hosts {
         })
     }
 
-    /// Append given addrs to the known set.
-    pub async fn store(&self, addrs: &[Url]) {
-        debug!(target: "net::hosts::store()", "hosts::store() [START]");
+    ///// Append given addrs to the known set.
+    //pub async fn store(&self, addrs: &[Url]) {
+    //    debug!(target: "net::hosts::store()", "hosts::store() [START]");
 
-        let filtered_addrs = self.filter_addresses(addrs).await;
-        let filtered_addrs_len = filtered_addrs.len();
+    //    let filtered_addrs = self.filter_addresses(addrs).await;
+    //    let filtered_addrs_len = filtered_addrs.len();
 
-        if !filtered_addrs.is_empty() {
-            let mut addrs_map = self.addrs.write().await;
-            for addr in filtered_addrs {
-                debug!(target: "net::hosts::store()", "Inserting {}", addr);
-                addrs_map.insert(addr);
-            }
-        }
+    //    if !filtered_addrs.is_empty() {
+    //        let mut addrs_map = self.addrs.write().await;
+    //        for addr in filtered_addrs {
+    //            debug!(target: "net::hosts::store()", "Inserting {}", addr);
+    //            addrs_map.insert(addr);
+    //        }
+    //    }
 
-        self.store_subscriber.notify(filtered_addrs_len).await;
-        debug!(target: "net::hosts::store()", "hosts::store() [END]");
-    }
+    //    self.store_subscriber.notify(filtered_addrs_len).await;
+    //    debug!(target: "net::hosts::store()", "hosts::store() [END]");
+    //}
 
     // Gets addresses from the whitelist.
     pub async fn whitelist_fetch_address_with_lock(
@@ -519,7 +518,34 @@ impl Hosts {
         }
     }
 
-    async fn probe_node(&self, host: &Url, p2p: P2pPtr, ex: Arc<Executor<'_>>) -> bool {
+    pub async fn refresh_whitelist(&self, url: &Url, p2p: P2pPtr, ex: Arc<Executor<'_>>) {
+        let mut whitelist = self.whitelist.write().await;
+
+        // Probe node to see if it's active.
+        let online: bool = self.probe_node(url, p2p.clone(), ex.clone()).await;
+
+        if online {
+            // Peer is responsive. Update last_seen and add it to the whitelist.
+            let last_seen =
+                SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
+
+            // Remove oldest element if the whitelist reaches max size.
+            if whitelist.len() == 1000 {
+                // Last element in vector should have the oldest timestamp.
+                // This should never crash as only returns None when whitelist len() == 0.
+                let entry = whitelist.pop().unwrap();
+                debug!(target: "net::hosts::refresh_whitelist()", "Whitelist reached max size. Removed host {}", entry.0);
+            }
+            // Append to the whitelist.
+            debug!(target: "net::hosts::refresh_whitelist()", "Adding peer {} to whitelist", url);
+            whitelist.push((url.clone(), last_seen));
+
+            // Sort whitelist by last_seen.
+            whitelist.sort_unstable_by_key(|entry| entry.1);
+        }
+    }
+
+    pub async fn probe_node(&self, host: &Url, p2p: P2pPtr, ex: Arc<Executor<'_>>) -> bool {
         let p2p_ = p2p.clone();
         let ex_ = ex.clone();
         let session_out = p2p_.session_outbound();
@@ -559,33 +585,33 @@ impl Hosts {
             }
         }
     }
-    pub async fn remove(&self, url: &Url) {
-        debug!(target: "net::hosts::remove()", "Removing peer {}", url);
-        self.addrs.write().await.remove(url);
-        self.quarantine.write().await.remove(url);
-    }
-
-    /// Quarantine a peer.
-    /// If they've been quarantined for more than a configured limit, forget them.
-    pub async fn quarantine(&self, url: &Url) {
-        debug!(target: "net::hosts::remove()", "Quarantining peer {}", url);
-        // Remove from main hosts set
-        self.addrs.write().await.remove(url);
-
-        let mut q = self.quarantine.write().await;
-        if let Some(retries) = q.get_mut(url) {
-            *retries += 1;
-            debug!(target: "net::hosts::quarantine()", "Peer {} quarantined {} times", url, retries);
-            if *retries == self.settings.hosts_quarantine_limit {
-                debug!(target: "net::hosts::quarantine()", "Banning peer {}", url);
-                q.remove(url);
-                self.mark_rejected(url).await;
-            }
-        } else {
-            debug!(target: "net::hosts::remove()", "Added peer {} to quarantine", url);
-            q.insert(url.clone(), 0);
-        }
-    }
+    //pub async fn remove(&self, url: &Url) {
+    //    debug!(target: "net::hosts::remove()", "Removing peer {}", url);
+    //    self.addrs.write().await.remove(url);
+    //    self.quarantine.write().await.remove(url);
+    //}
+
+    ///// Quarantine a peer.
+    ///// If they've been quarantined for more than a configured limit, forget them.
+    //pub async fn quarantine(&self, url: &Url) {
+    //    debug!(target: "net::hosts::remove()", "Quarantining peer {}", url);
+    //    // Remove from main hosts set
+    //    self.addrs.write().await.remove(url);
+
+    //    let mut q = self.quarantine.write().await;
+    //    if let Some(retries) = q.get_mut(url) {
+    //        *retries += 1;
+    //        debug!(target: "net::hosts::quarantine()", "Peer {} quarantined {} times", url, retries);
+    //        if *retries == self.settings.hosts_quarantine_limit {
+    //            debug!(target: "net::hosts::quarantine()", "Banning peer {}", url);
+    //            q.remove(url);
+    //            self.mark_rejected(url).await;
+    //        }
+    //    } else {
+    //        debug!(target: "net::hosts::remove()", "Added peer {} to quarantine", url);
+    //        q.insert(url.clone(), 0);
+    //    }
+    //}
 
     /// Check if a given peer (URL) is in the set of rejected hosts
     pub async fn is_rejected(&self, peer: &Url) -> bool {
@@ -622,10 +648,10 @@ impl Hosts {
         }
     }
 
-    /// Check if the host list is empty.
-    pub async fn is_empty(&self) -> bool {
-        self.addrs.read().await.is_empty()
-    }
+    ///// Check if the host list is empty.
+    //pub async fn is_empty(&self) -> bool {
+    //    self.addrs.read().await.is_empty()
+    //}
 
     // Check if the greylist is empty.
     pub async fn is_empty_greylist(&self) -> bool {
@@ -667,49 +693,49 @@ impl Hosts {
         return 0
     }
 
-    /// Check if host is already in the set
-    pub async fn contains(&self, addr: &Url) -> bool {
-        self.addrs.read().await.contains(addr)
-    }
+    ///// Check if host is already in the set
+    //pub async fn contains(&self, addr: &Url) -> bool {
+    //    self.addrs.read().await.contains(addr)
+    //}
 
-    /// Return all known hosts
-    pub async fn fetch_all(&self) -> Vec<Url> {
-        self.addrs.read().await.iter().cloned().collect()
-    }
+    ///// Return all known hosts
+    //pub async fn fetch_all(&self) -> Vec<Url> {
+    //    self.addrs.read().await.iter().cloned().collect()
+    //}
 
     /// Return all known whitelisted hosts
     pub async fn whitelist_fetch_all(&self) -> Vec<(Url, u64)> {
         self.whitelist.read().await.iter().cloned().collect()
     }
 
-    /// Get up to n random peers from the hosts set.
-    pub async fn fetch_n_random(&self, n: u32) -> Vec<Url> {
-        let n = n as usize;
-        if n == 0 {
-            return vec![]
-        }
-        let addrs = self.addrs.read().await;
-        let urls = addrs.iter().choose_multiple(&mut OsRng, n.min(addrs.len()));
-        urls.iter().map(|&url| url.clone()).collect()
-    }
-
-    /// Get up to n random peers that match the given transport schemes from the hosts set.
-    pub async fn fetch_n_random_with_schemes(&self, schemes: &[String], n: u32) -> Vec<Url> {
-        let n = n as usize;
-        if n == 0 {
-            return vec![]
-        }
-
-        // Retrieve all peers corresponding to that transport schemes
-        let hosts = self.fetch_with_schemes(schemes, None).await;
-        if hosts.is_empty() {
-            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 peers from the hosts set.
+    //pub async fn fetch_n_random(&self, n: u32) -> Vec<Url> {
+    //    let n = n as usize;
+    //    if n == 0 {
+    //        return vec![]
+    //    }
+    //    let addrs = self.addrs.read().await;
+    //    let urls = addrs.iter().choose_multiple(&mut OsRng, n.min(addrs.len()));
+    //    urls.iter().map(|&url| url.clone()).collect()
+    //}
+
+    ///// Get up to n random peers that match the given transport schemes from the hosts set.
+    //pub async fn fetch_n_random_with_schemes(&self, schemes: &[String], n: u32) -> Vec<Url> {
+    //    let n = n as usize;
+    //    if n == 0 {
+    //        return vec![]
+    //    }
+
+    //    // Retrieve all peers corresponding to that transport schemes
+    //    let hosts = self.fetch_with_schemes(schemes, None).await;
+    //    if hosts.is_empty() {
+    //        return hosts
+    //    }
+
+    //    // Grab random ones
+    //    let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
+    //    urls.iter().map(|&url| url.clone()).collect()
+    //}
 
     pub async fn whitelist_fetch_n_random_with_schemes(
         &self,
@@ -732,23 +758,23 @@ impl Hosts {
         urls.iter().map(|&url| url.clone()).collect()
     }
 
-    /// Get up to n random peers that don't match the given transport schemes from the hosts set.
-    pub async fn fetch_n_random_excluding_schemes(&self, schemes: &[String], n: u32) -> Vec<Url> {
-        let n = n as usize;
-        if n == 0 {
-            return vec![]
-        }
+    ///// Get up to n random peers that don't match the given transport schemes from the hosts set.
+    //pub async fn fetch_n_random_excluding_schemes(&self, schemes: &[String], n: u32) -> Vec<Url> {
+    //    let n = n as usize;
+    //    if n == 0 {
+    //        return vec![]
+    //    }
 
-        // Retrieve all peers not corresponding to that transport schemes
-        let hosts = self.fetch_exluding_schemes(schemes, None).await;
-        if hosts.is_empty() {
-            return hosts
-        }
+    //    // Retrieve all peers not corresponding to that transport schemes
+    //    let hosts = self.fetch_exluding_schemes(schemes, None).await;
+    //    if hosts.is_empty() {
+    //        return hosts
+    //    }
 
-        // Grab random ones
-        let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
-        urls.iter().map(|&url| url.clone()).collect()
-    }
+    //    // Grab random ones
+    //    let urls = hosts.iter().choose_multiple(&mut OsRng, n.min(hosts.len()));
+    //    urls.iter().map(|&url| url.clone()).collect()
+    //}
 
     pub async fn whitelist_fetch_n_random_excluding_schemes(
         &self,
@@ -771,45 +797,45 @@ impl Hosts {
         urls.iter().map(|&url| url.clone()).collect()
     }
 
-    /// Get up to limit peers that match the given transport schemes from the hosts set.
-    /// If limit was not provided, return all matching peers.
-    pub async fn fetch_with_schemes(&self, schemes: &[String], limit: Option<usize>) -> Vec<Url> {
-        let addrs = self.addrs.read().await;
-        let mut limit = match limit {
-            Some(l) => l.min(addrs.len()),
-            None => addrs.len(),
-        };
-        let mut ret = vec![];
-
-        if limit == 0 {
-            return ret
-        }
-
-        for addr in addrs.iter() {
-            if schemes.contains(&addr.scheme().to_string()) {
-                ret.push(addr.clone());
-                limit -= 1;
-                if limit == 0 {
-                    return ret
-                }
-            }
-        }
-
-        // If we didn't find any, pick some from the quarantine zone
-        if ret.is_empty() {
-            for addr in self.quarantine.read().await.keys() {
-                if schemes.contains(&addr.scheme().to_string()) {
-                    ret.push(addr.clone());
-                    limit -= 1;
-                    if limit == 0 {
-                        break
-                    }
-                }
-            }
-        }
-
-        ret
-    }
+    ///// Get up to limit peers that match the given transport schemes from the hosts set.
+    ///// If limit was not provided, return all matching peers.
+    //pub async fn fetch_with_schemes(&self, schemes: &[String], limit: Option<usize>) -> Vec<Url> {
+    //    let addrs = self.addrs.read().await;
+    //    let mut limit = match limit {
+    //        Some(l) => l.min(addrs.len()),
+    //        None => addrs.len(),
+    //    };
+    //    let mut ret = vec![];
+
+    //    if limit == 0 {
+    //        return ret
+    //    }
+
+    //    for addr in addrs.iter() {
+    //        if schemes.contains(&addr.scheme().to_string()) {
+    //            ret.push(addr.clone());
+    //            limit -= 1;
+    //            if limit == 0 {
+    //                return ret
+    //            }
+    //        }
+    //    }
+
+    //    // If we didn't find any, pick some from the quarantine zone
+    //    if ret.is_empty() {
+    //        for addr in self.quarantine.read().await.keys() {
+    //            if schemes.contains(&addr.scheme().to_string()) {
+    //                ret.push(addr.clone());
+    //                limit -= 1;
+    //                if limit == 0 {
+    //                    break
+    //                }
+    //            }
+    //        }
+    //    }
+
+    //    ret
+    //}
 
     pub async fn whitelist_fetch_with_schemes(
         &self,
@@ -853,49 +879,49 @@ impl Hosts {
         ret
     }
 
-    /// Get up to limit peers that don't match the given transport schemes from the hosts set.
-    /// If limit was not provided, return all matching peers.
-    pub async fn fetch_exluding_schemes(
-        &self,
-        schemes: &[String],
-        limit: Option<usize>,
-    ) -> Vec<Url> {
-        let addrs = self.addrs.read().await;
-        let mut limit = match limit {
-            Some(l) => l.min(addrs.len()),
-            None => addrs.len(),
-        };
-        let mut ret = vec![];
-
-        if limit == 0 {
-            return ret
-        }
-
-        for addr in addrs.iter() {
-            if !schemes.contains(&addr.scheme().to_string()) {
-                ret.push(addr.clone());
-                limit -= 1;
-                if limit == 0 {
-                    return ret
-                }
-            }
-        }
-
-        // If we didn't find any, pick some from the quarantine zone
-        if ret.is_empty() {
-            for addr in self.quarantine.read().await.keys() {
-                if !schemes.contains(&addr.scheme().to_string()) {
-                    ret.push(addr.clone());
-                    limit -= 1;
-                    if limit == 0 {
-                        break
-                    }
-                }
-            }
-        }
-
-        ret
-    }
+    ///// Get up to limit peers that don't match the given transport schemes from the hosts set.
+    ///// If limit was not provided, return all matching peers.
+    //pub async fn fetch_exluding_schemes(
+    //    &self,
+    //    schemes: &[String],
+    //    limit: Option<usize>,
+    //) -> Vec<Url> {
+    //    let addrs = self.addrs.read().await;
+    //    let mut limit = match limit {
+    //        Some(l) => l.min(addrs.len()),
+    //        None => addrs.len(),
+    //    };
+    //    let mut ret = vec![];
+
+    //    if limit == 0 {
+    //        return ret
+    //    }
+
+    //    for addr in addrs.iter() {
+    //        if !schemes.contains(&addr.scheme().to_string()) {
+    //            ret.push(addr.clone());
+    //            limit -= 1;
+    //            if limit == 0 {
+    //                return ret
+    //            }
+    //        }
+    //    }
+
+    //    // If we didn't find any, pick some from the quarantine zone
+    //    if ret.is_empty() {
+    //        for addr in self.quarantine.read().await.keys() {
+    //            if !schemes.contains(&addr.scheme().to_string()) {
+    //                ret.push(addr.clone());
+    //                limit -= 1;
+    //                if limit == 0 {
+    //                    break
+    //                }
+    //            }
+    //        }
+    //    }
+
+    //    ret
+    //}
 
     pub async fn whitelist_fetch_excluding_schemes(
         &self,
@@ -945,86 +971,86 @@ mod tests {
     use super::{super::settings::Settings, *};
     use std::time::SystemTime;
 
-    #[test]
-    fn test_store_localnet() {
-        smol::block_on(async {
-            let settings = Settings {
-                localnet: true,
-                external_addrs: vec![
-                    Url::parse("tcp://foo.bar:123").unwrap(),
-                    Url::parse("tcp://lol.cat:321").unwrap(),
-                ],
-                ..Default::default()
-            };
-
-            let hosts = Hosts::new(Arc::new(settings.clone()));
-            hosts.store(&settings.external_addrs).await;
-            for i in settings.external_addrs {
-                assert!(hosts.contains(&i).await);
-            }
-
-            let local_hosts = vec![
-                Url::parse("tcp://localhost:3921").unwrap(),
-                Url::parse("tcp://127.0.0.1:23957").unwrap(),
-                Url::parse("tcp://[::1]:21481").unwrap(),
-                Url::parse("tcp://192.168.10.65:311").unwrap(),
-                Url::parse("tcp://0.0.0.0:2312").unwrap(),
-                Url::parse("tcp://255.255.255.255:2131").unwrap(),
-            ];
-            hosts.store(&local_hosts).await;
-            for i in local_hosts {
-                assert!(hosts.contains(&i).await);
-            }
-
-            let remote_hosts = vec![
-                Url::parse("tcp://dark.fi:80").unwrap(),
-                Url::parse("tcp://top.kek:111").unwrap(),
-                Url::parse("tcp://http.cat:401").unwrap(),
-            ];
-            hosts.store(&remote_hosts).await;
-            for i in remote_hosts {
-                assert!(hosts.contains(&i).await);
-            }
-        });
-    }
-
-    #[test]
-    fn test_store() {
-        smol::block_on(async {
-            let settings = Settings {
-                localnet: false,
-                external_addrs: vec![
-                    Url::parse("tcp://foo.bar:123").unwrap(),
-                    Url::parse("tcp://lol.cat:321").unwrap(),
-                ],
-                ..Default::default()
-            };
-
-            let hosts = Hosts::new(Arc::new(settings.clone()));
-            hosts.store(&settings.external_addrs).await;
-            assert!(hosts.is_empty().await);
-
-            let local_hosts = vec![
-                Url::parse("tcp://localhost:3921").unwrap(),
-                Url::parse("tor://[::1]:21481").unwrap(),
-                Url::parse("tcp://192.168.10.65:311").unwrap(),
-                Url::parse("tcp+tls://0.0.0.0:2312").unwrap(),
-                Url::parse("tcp://255.255.255.255:2131").unwrap(),
-            ];
-            hosts.store(&local_hosts).await;
-            assert!(hosts.is_empty().await);
-
-            let remote_hosts = vec![
-                Url::parse("tcp://dark.fi:80").unwrap(),
-                Url::parse("tcp://http.cat:401").unwrap(),
-                Url::parse("tcp://foo.bar:111").unwrap(),
-            ];
-            hosts.store(&remote_hosts).await;
-            assert!(hosts.contains(&remote_hosts[0]).await);
-            assert!(hosts.contains(&remote_hosts[1]).await);
-            assert!(!hosts.contains(&remote_hosts[2]).await);
-        });
-    }
+    //#[test]
+    //fn test_store_localnet() {
+    //    smol::block_on(async {
+    //        let settings = Settings {
+    //            localnet: true,
+    //            external_addrs: vec![
+    //                Url::parse("tcp://foo.bar:123").unwrap(),
+    //                Url::parse("tcp://lol.cat:321").unwrap(),
+    //            ],
+    //            ..Default::default()
+    //        };
+
+    //        let hosts = Hosts::new(Arc::new(settings.clone()));
+    //        hosts.store(&settings.external_addrs).await;
+    //        for i in settings.external_addrs {
+    //            assert!(hosts.contains(&i).await);
+    //        }
+
+    //        let local_hosts = vec![
+    //            Url::parse("tcp://localhost:3921").unwrap(),
+    //            Url::parse("tcp://127.0.0.1:23957").unwrap(),
+    //            Url::parse("tcp://[::1]:21481").unwrap(),
+    //            Url::parse("tcp://192.168.10.65:311").unwrap(),
+    //            Url::parse("tcp://0.0.0.0:2312").unwrap(),
+    //            Url::parse("tcp://255.255.255.255:2131").unwrap(),
+    //        ];
+    //        hosts.store(&local_hosts).await;
+    //        for i in local_hosts {
+    //            assert!(hosts.contains(&i).await);
+    //        }
+
+    //        let remote_hosts = vec![
+    //            Url::parse("tcp://dark.fi:80").unwrap(),
+    //            Url::parse("tcp://top.kek:111").unwrap(),
+    //            Url::parse("tcp://http.cat:401").unwrap(),
+    //        ];
+    //        hosts.store(&remote_hosts).await;
+    //        for i in remote_hosts {
+    //            assert!(hosts.contains(&i).await);
+    //        }
+    //    });
+    //}
+
+    //#[test]
+    //fn test_store() {
+    //    smol::block_on(async {
+    //        let settings = Settings {
+    //            localnet: false,
+    //            external_addrs: vec![
+    //                Url::parse("tcp://foo.bar:123").unwrap(),
+    //                Url::parse("tcp://lol.cat:321").unwrap(),
+    //            ],
+    //            ..Default::default()
+    //        };
+
+    //        let hosts = Hosts::new(Arc::new(settings.clone()));
+    //        hosts.store(&settings.external_addrs).await;
+    //        assert!(hosts.is_empty().await);
+
+    //        let local_hosts = vec![
+    //            Url::parse("tcp://localhost:3921").unwrap(),
+    //            Url::parse("tor://[::1]:21481").unwrap(),
+    //            Url::parse("tcp://192.168.10.65:311").unwrap(),
+    //            Url::parse("tcp+tls://0.0.0.0:2312").unwrap(),
+    //            Url::parse("tcp://255.255.255.255:2131").unwrap(),
+    //        ];
+    //        hosts.store(&local_hosts).await;
+    //        assert!(hosts.is_empty().await);
+
+    //        let remote_hosts = vec![
+    //            Url::parse("tcp://dark.fi:80").unwrap(),
+    //            Url::parse("tcp://http.cat:401").unwrap(),
+    //            Url::parse("tcp://foo.bar:111").unwrap(),
+    //        ];
+    //        hosts.store(&remote_hosts).await;
+    //        assert!(hosts.contains(&remote_hosts[0]).await);
+    //        assert!(hosts.contains(&remote_hosts[1]).await);
+    //        assert!(!hosts.contains(&remote_hosts[2]).await);
+    //    });
+    //}
 
     #[test]
     fn test_is_local_host() {

src/net/protocol/protocol_address.rs

@@ -37,139 +37,139 @@ use super::{
 };
 use crate::{system::sleep, Result};
 
-/// Defines address and get-address messages
-pub struct ProtocolAddress {
-    channel: ChannelPtr,
-    addrs_sub: MessageSubscription<AddrsMessage>,
-    get_addrs_sub: MessageSubscription<GetAddrsMessage>,
-    hosts: HostsPtr,
-    settings: SettingsPtr,
-    jobsman: ProtocolJobsManagerPtr,
-}
-
-const PROTO_NAME: &str = "ProtocolAddress";
-
-impl ProtocolAddress {
-    /// Creates a new address protocol. Makes an address, an external address
-    /// and a get-address subscription and adds them to the address protocol
-    /// instance.
-    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
-        let settings = p2p.settings();
-        let hosts = p2p.hosts();
-
-        // Creates a subscription to address message
-        let addrs_sub =
-            channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addrs dispatcher!");
-
-        // Creates a subscription to get-address message
-        let get_addrs_sub =
-            channel.subscribe_msg::<GetAddrsMessage>().await.expect("Missing getaddrs dispatcher!");
-
-        Arc::new(Self {
-            channel: channel.clone(),
-            addrs_sub,
-            get_addrs_sub,
-            hosts,
-            jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
-            settings,
-        })
-    }
-
-    /// Handles receiving the address message. Loops to continually receive
-    /// address messages on the address subscription. Validates and adds the
-    /// received addresses to the hosts set.
-    async fn handle_receive_addrs(self: Arc<Self>) -> Result<()> {
-        debug!(
-            target: "net::protocol_address::handle_receive_addrs()",
-            "[START] address={}", self.channel.address(),
-        );
-
-        loop {
-            let addrs_msg = self.addrs_sub.receive().await?;
-            debug!(
-                target: "net::protocol_address::handle_receive_addrs()",
-                "Received {} addrs from {}", addrs_msg.addrs.len(), self.channel.address(),
-            );
-
-            // TODO: We might want to close the channel here if we're getting
-            // corrupted addresses.
-            self.hosts.store(&addrs_msg.addrs).await;
-        }
-    }
-
-    /// Handles receiving the get-address message. Continually receives get-address
-    /// messages on the get-address subscription. Then replies with an address message.
-    async fn handle_receive_get_addrs(self: Arc<Self>) -> Result<()> {
-        debug!(
-            target: "net::protocol_address::handle_receive_get_addrs()",
-            "[START] address={}", self.channel.address(),
-        );
-
-        loop {
-            let get_addrs_msg = self.get_addrs_sub.receive().await?;
-
-            debug!(
-                target: "net::protocol_address::handle_receive_get_addrs()",
-                "Received GetAddrs({}) message from {}", get_addrs_msg.max, self.channel.address(),
-            );
-
-            // Validate transports length
-            // TODO: Verify this limit. It should be the max number of all our allowed transports,
-            //       plus their mixing.
-            if get_addrs_msg.transports.len() > 20 {
-                // TODO: Should this error out, effectively ending the connection?
-                let addrs_msg = AddrsMessage { addrs: vec![] };
-                self.channel.send(&addrs_msg).await?;
-                continue
-            }
-
-            // First we grab address with the requested transports
-            let mut addrs = self
-                .hosts
-                .fetch_n_random_with_schemes(&get_addrs_msg.transports, get_addrs_msg.max)
-                .await;
-
-            // Then we grab addresses without the requested transports
-            // to fill a 2 * max length vector.
-            let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
-            addrs.append(
-                &mut self
-                    .hosts
-                    .fetch_n_random_excluding_schemes(&get_addrs_msg.transports, remain)
-                    .await,
-            );
-
-            debug!(
-                target: "net::protocol_address::handle_receive_get_addrs()",
-                "Sending {} addresses to {}", addrs.len(), self.channel.address(),
-            );
-
-            let addrs_msg = AddrsMessage { addrs };
-            self.channel.send(&addrs_msg).await?;
-        }
-    }
-
-    /// Periodically send our external addresses through the channel.
-    async fn send_my_addrs(self: Arc<Self>) -> Result<()> {
-        debug!(
-            target: "net::protocol_address::send_my_addrs()",
-            "[START] address={}", self.channel.address(),
-        );
-
-        // FIXME: Revisit this. Why do we keep sending it?
-        loop {
-            let ext_addr_msg = AddrsMessage { addrs: self.settings.external_addrs.clone() };
-            self.channel.send(&ext_addr_msg).await?;
-            sleep(900).await;
-        }
-    }
-}
+///// Defines address and get-address messages
+//pub struct ProtocolAddress {
+//    channel: ChannelPtr,
+//    addrs_sub: MessageSubscription<AddrsMessage>,
+//    get_addrs_sub: MessageSubscription<GetAddrsMessage>,
+//    hosts: HostsPtr,
+//    settings: SettingsPtr,
+//    jobsman: ProtocolJobsManagerPtr,
+//}
+//
+//const PROTO_NAME: &str = "ProtocolAddress";
+//
+//impl ProtocolAddress {
+//    /// Creates a new address protocol. Makes an address, an external address
+//    /// and a get-address subscription and adds them to the address protocol
+//    /// instance.
+//    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
+//        let settings = p2p.settings();
+//        let hosts = p2p.hosts();
+//
+//        // Creates a subscription to address message
+//        let addrs_sub =
+//            channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addrs dispatcher!");
+//
+//        // Creates a subscription to get-address message
+//        let get_addrs_sub =
+//            channel.subscribe_msg::<GetAddrsMessage>().await.expect("Missing getaddrs dispatcher!");
+//
+//        Arc::new(Self {
+//            channel: channel.clone(),
+//            addrs_sub,
+//            get_addrs_sub,
+//            hosts,
+//            jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
+//            settings,
+//        })
+//    }
+//
+//    /// Handles receiving the address message. Loops to continually receive
+//    /// address messages on the address subscription. Validates and adds the
+//    /// received addresses to the hosts set.
+//    async fn handle_receive_addrs(self: Arc<Self>) -> Result<()> {
+//        debug!(
+//            target: "net::protocol_address::handle_receive_addrs()",
+//            "[START] address={}", self.channel.address(),
+//        );
+//
+//        loop {
+//            let addrs_msg = self.addrs_sub.receive().await?;
+//            debug!(
+//                target: "net::protocol_address::handle_receive_addrs()",
+//                "Received {} addrs from {}", addrs_msg.addrs.len(), self.channel.address(),
+//            );
+//
+//            // TODO: We might want to close the channel here if we're getting
+//            // corrupted addresses.
+//            self.hosts.store(&addrs_msg.addrs).await;
+//        }
+//    }
+//
+//    /// Handles receiving the get-address message. Continually receives get-address
+//    /// messages on the get-address subscription. Then replies with an address message.
+//    async fn handle_receive_get_addrs(self: Arc<Self>) -> Result<()> {
+//        debug!(
+//            target: "net::protocol_address::handle_receive_get_addrs()",
+//            "[START] address={}", self.channel.address(),
+//        );
+//
+//        loop {
+//            let get_addrs_msg = self.get_addrs_sub.receive().await?;
+//
+//            debug!(
+//                target: "net::protocol_address::handle_receive_get_addrs()",
+//                "Received GetAddrs({}) message from {}", get_addrs_msg.max, self.channel.address(),
+//            );
+//
+//            // Validate transports length
+//            // TODO: Verify this limit. It should be the max number of all our allowed transports,
+//            //       plus their mixing.
+//            if get_addrs_msg.transports.len() > 20 {
+//                // TODO: Should this error out, effectively ending the connection?
+//                let addrs_msg = AddrsMessage { addrs: vec![] };
+//                self.channel.send(&addrs_msg).await?;
+//                continue
+//            }
+//
+//            // First we grab address with the requested transports
+//            let mut addrs = self
+//                .hosts
+//                .fetch_n_random_with_schemes(&get_addrs_msg.transports, get_addrs_msg.max)
+//                .await;
+//
+//            // Then we grab addresses without the requested transports
+//            // to fill a 2 * max length vector.
+//            let remain = 2 * get_addrs_msg.max - addrs.len() as u32;
+//            addrs.append(
+//                &mut self
+//                    .hosts
+//                    .fetch_n_random_excluding_schemes(&get_addrs_msg.transports, remain)
+//                    .await,
+//            );
+//
+//            debug!(
+//                target: "net::protocol_address::handle_receive_get_addrs()",
+//                "Sending {} addresses to {}", addrs.len(), self.channel.address(),
+//            );
+//
+//            let addrs_msg = AddrsMessage { addrs };
+//            self.channel.send(&addrs_msg).await?;
+//        }
+//    }
+//
+//    /// Periodically send our external addresses through the channel.
+//    async fn send_my_addrs(self: Arc<Self>) -> Result<()> {
+//        debug!(
+//            target: "net::protocol_address::send_my_addrs()",
+//            "[START] address={}", self.channel.address(),
+//        );
+//
+//        // FIXME: Revisit this. Why do we keep sending it?
+//        loop {
+//            let ext_addr_msg = AddrsMessage { addrs: self.settings.external_addrs.clone() };
+//            self.channel.send(&ext_addr_msg).await?;
+//            sleep(900).await;
+//        }
+//    }
+//}
 
 // New protocol that sends and receives whitelist info instead of Vec<Url>.
 // AddrMessage is of the format Vec<(Url, u64)>. On receiving GetAddr, nodes send AddrMessage
 // with whitelisted nodes. On receiving an AddrMessage, nodes enter the info into their greylists.
 // The format of GetAddrMessage remains the same.
-pub struct ProtocolAddress2 {
+pub struct ProtocolAddress {
     channel: ChannelPtr,
     addrs_sub: MessageSubscription<AddrsMessage2>,
     get_addrs_sub: MessageSubscription<GetAddrsMessage>,
@@ -178,10 +178,10 @@ pub struct ProtocolAddress2 {
     jobsman: ProtocolJobsManagerPtr,
 }
 
-const PROTO_NAME2: &str = "ProtocolAddress2";
+const PROTO_NAME: &str = "ProtocolAddress";
 
-impl ProtocolAddress2 {
-    pub async fn init2(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
+impl ProtocolAddress {
+    pub async fn init(channel: ChannelPtr, p2p: P2pPtr) -> ProtocolBasePtr {
         let settings = p2p.settings();
         let hosts = p2p.hosts();
 
@@ -198,13 +198,13 @@ impl ProtocolAddress2 {
             addrs_sub,
             get_addrs_sub,
             hosts,
-            jobsman: ProtocolJobsManager::new(PROTO_NAME2, channel),
+            jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
             settings,
         })
     }
 
     // When we learn of a new address, append it to the greylist.
-    async fn handle_receive_addrs2(self: Arc<Self>) -> Result<()> {
+    async fn handle_receive_addrs(self: Arc<Self>) -> Result<()> {
         debug!(
             target: "net::protocol_address::handle_receive_addrs2()",
             "[START] address={}", self.channel.address(),
@@ -221,7 +221,7 @@ impl ProtocolAddress2 {
         }
     }
 
-    async fn handle_receive_get_addrs2(self: Arc<Self>) -> Result<()> {
+    async fn handle_receive_get_addrs(self: Arc<Self>) -> Result<()> {
         debug!(
             target: "net::protocol_address::handle_receive_get_addrs()",
             "[START] address={}", self.channel.address(),
@@ -271,7 +271,7 @@ impl ProtocolAddress2 {
         }
     }
 
-    async fn send_my_addrs2(self: Arc<Self>) -> Result<()> {
+    async fn send_my_addrs(self: Arc<Self>) -> Result<()> {
         debug!(
             target: "net::protocol_address::send_my_addrs()",
             "[START] address={}", self.channel.address(),
@@ -320,36 +320,36 @@ impl ProtocolBase for ProtocolAddress {
     }
 }
 
-#[async_trait]
-impl ProtocolBase for ProtocolAddress2 {
-    // TODO
-    async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
-        debug!(target: "net::protocol_address::start()", "START => address={}", self.channel.address());
-
-        let type_id = self.channel.session_type_id();
-
-        self.jobsman.clone().start(ex.clone());
-
-        // If it's an outbound session + has an extern_addr, send our address.
-        if type_id == SESSION_OUTBOUND && !self.settings.external_addrs.is_empty() {
-            self.jobsman.clone().spawn(self.clone().send_my_addrs2(), ex.clone()).await;
-        }
-
-        self.jobsman.clone().spawn(self.clone().handle_receive_addrs2(), ex.clone()).await;
-        self.jobsman.spawn(self.clone().handle_receive_get_addrs2(), ex).await;
-
-        // Send get_address message.
-        let get_addrs = GetAddrsMessage {
-            max: self.settings.outbound_connections as u32,
-            transports: self.settings.allowed_transports.clone(),
-        };
-        self.channel.send(&get_addrs).await?;
-
-        debug!(target: "net::protocol_address::start()", "END => address={}", self.channel.address());
-        Ok(())
-    }
-
-    fn name(&self) -> &'static str {
-        PROTO_NAME2
-    }
-}
+//#[async_trait]
+//impl ProtocolBase for ProtocolAddress {
+//    // TODO
+//    async fn start(self: Arc<Self>, ex: Arc<Executor<'_>>) -> Result<()> {
+//        debug!(target: "net::protocol_address::start()", "START => address={}", self.channel.address());
+//
+//        let type_id = self.channel.session_type_id();
+//
+//        self.jobsman.clone().start(ex.clone());
+//
+//        // If it's an outbound session + has an extern_addr, send our address.
+//        if type_id == SESSION_OUTBOUND && !self.settings.external_addrs.is_empty() {
+//            self.jobsman.clone().spawn(self.clone().send_my_addrs2(), ex.clone()).await;
+//        }
+//
+//        self.jobsman.clone().spawn(self.clone().handle_receive_addrs2(), ex.clone()).await;
+//        self.jobsman.spawn(self.clone().handle_receive_get_addrs2(), ex).await;
+//
+//        // Send get_address message.
+//        let get_addrs = GetAddrsMessage {
+//            max: self.settings.outbound_connections as u32,
+//            transports: self.settings.allowed_transports.clone(),
+//        };
+//        self.channel.send(&get_addrs).await?;
+//
+//        debug!(target: "net::protocol_address::start()", "END => address={}", self.channel.address());
+//        Ok(())
+//    }
+//
+//    fn name(&self) -> &'static str {
+//        PROTO_NAME2
+//    }
+//}

src/net/session/outbound_session.rs

@@ -172,7 +172,7 @@ impl Slot {
 
         self.process.clone().start(
             async move {
-                self.run2().await;
+                self.run().await;
                 unreachable!();
             },
             // Ignore stop handler
@@ -185,116 +185,116 @@ impl Slot {
         self.process.stop().await
     }
 
-    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.
-        loop {
-            // Activate the slot
-            debug!(
-                target: "net::outbound_session::try_connect()",
-                "[P2P] Finding a host to connect to for outbound slot #{}",
-                self.slot,
-            );
-
-            // Retrieve whitelisted outbound transports
-            let transports = &self.p2p().settings().allowed_transports;
-
-            // Find an address to connect to. We also do peer discovery here if needed.
-            let addr = if let Some(addr) = self.fetch_address_with_lock(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);
-        }
-    }
+    //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.
+    //    loop {
+    //        // Activate the slot
+    //        debug!(
+    //            target: "net::outbound_session::try_connect()",
+    //            "[P2P] Finding a host to connect to for outbound slot #{}",
+    //            self.slot,
+    //        );
+
+    //        // Retrieve whitelisted outbound transports
+    //        let transports = &self.p2p().settings().allowed_transports;
+
+    //        // Find an address to connect to. We also do peer discovery here if needed.
+    //        let addr = if let Some(addr) = self.fetch_address_with_lock(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);
+    //    }
+    //}
 
     // Looks up whitelisted addresses. Tries to connect to them.
     // On success, updates the whitelist last_seen field.
-    async fn run2(self: Arc<Self>) {
+    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
@@ -348,7 +348,7 @@ impl Slot {
             });
 
             let (addr_final, channel) =
-                match self.try_connect2(addr.clone(), last_seen.clone()).await {
+                match self.try_connect(addr.clone(), last_seen.clone()).await {
                     Ok(connect_info) => connect_info,
                     Err(err) => {
                         error!(
@@ -420,42 +420,42 @@ impl Slot {
         }
     }
 
-    /// 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
-    /// to is found. Once connected, registers the channel, removes it from
-    /// the list of pending channels, and starts sending messages across the
-    /// channel. In case of any failures, a network error is returned and the
-    /// main connect loop (parent of this function) will iterate again.
-    async fn try_connect(&self, addr: Url) -> Result<(Url, ChannelPtr)> {
-        let parent = Arc::downgrade(&self.session());
-        let connector = Connector::new(self.p2p().settings(), parent);
-
-        match connector.connect(&addr).await {
-            Ok((addr_final, channel)) => Ok((addr_final, channel)),
-
-            Err(e) => {
-                error!(
-                    target: "net::outbound_session::try_connect()",
-                    "[P2P] Unable to connect outbound slot #{} [{}]: {}",
-                    self.slot, addr, e
-                );
-
-                // At this point we failed to connect. We'll quarantine this peer now.
-                self.p2p().hosts().quarantine(&addr).await;
-
-                // Remove connection from pending
-                self.p2p().remove_pending(&addr).await;
-
-                // Notify that channel processing failed
-                self.session().channel_subscriber.notify(Err(Error::ConnectFailed)).await;
-
-                Err(Error::ConnectFailed)
-            }
-        }
-    }
-
-    async fn try_connect2(&self, addr: Url, last_seen: u64) -> Result<(Url, ChannelPtr)> {
+    ///// 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
+    ///// to is found. Once connected, registers the channel, removes it from
+    ///// the list of pending channels, and starts sending messages across the
+    ///// channel. In case of any failures, a network error is returned and the
+    ///// main connect loop (parent of this function) will iterate again.
+    //async fn try_connect(&self, addr: Url) -> Result<(Url, ChannelPtr)> {
+    //    let parent = Arc::downgrade(&self.session());
+    //    let connector = Connector::new(self.p2p().settings(), parent);
+
+    //    match connector.connect(&addr).await {
+    //        Ok((addr_final, channel)) => Ok((addr_final, channel)),
+
+    //        Err(e) => {
+    //            error!(
+    //                target: "net::outbound_session::try_connect()",
+    //                "[P2P] Unable to connect outbound slot #{} [{}]: {}",
+    //                self.slot, addr, e
+    //            );
+
+    //            // At this point we failed to connect. We'll quarantine this peer now.
+    //            self.p2p().hosts().quarantine(&addr).await;
+
+    //            // Remove connection from pending
+    //            self.p2p().remove_pending(&addr).await;
+
+    //            // Notify that channel processing failed
+    //            self.session().channel_subscriber.notify(Err(Error::ConnectFailed)).await;
+
+    //            Err(Error::ConnectFailed)
+    //        }
+    //    }
+    //}
+
+    async fn try_connect(&self, addr: Url, last_seen: u64) -> Result<(Url, ChannelPtr)> {
         let parent = Arc::downgrade(&self.session());
         let connector = Connector::new(self.p2p().settings(), parent);
 
@@ -498,89 +498,89 @@ impl Slot {
         Ok(())
     }
 
-    /// Loops through host addresses to find an outbound address that we can
-    /// connect to. Check whether the address is valid by making sure it isn't
-    /// our own inbound address, then checks whether it is already connected
-    /// (exists) or connecting (pending).
-    /// Lastly adds matching address to the pending list.
-    /// TODO: this method should go in hosts
-    async fn fetch_address_with_lock(&self, transports: &[String]) -> Option<Url> {
-        let p2p = self.p2p();
-
-        // Collect hosts
-        let mut hosts = vec![];
-
-        // If transport mixing is enabled, then for example we're allowed to
-        // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
-        // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
-        let transport_mixing = p2p.settings().transport_mixing;
-        macro_rules! mix_transport {
-            ($a:expr, $b:expr) => {
-                if transports.contains(&$a.to_string()) && transport_mixing {
-                    let mut a_to_b = p2p.hosts().fetch_with_schemes(&[$b.to_string()], None).await;
-                    for addr in a_to_b.iter_mut() {
-                        addr.set_scheme($a).unwrap();
-                        hosts.push(addr.clone());
-                    }
-                }
-            };
-        }
-        mix_transport!("tor", "tcp");
-        mix_transport!("tor+tls", "tcp+tls");
-        mix_transport!("nym", "tcp");
-        mix_transport!("nym+tls", "tcp+tls");
-
-        // And now the actual requested transports
-        for addr in p2p.hosts().fetch_with_schemes(transports, None).await {
-            hosts.push(addr);
-        }
-
-        // Randomize hosts list. Do not try to connect in a deterministic order.
-        // This is healthier for multiple slots to not compete for the same addrs.
-        hosts.shuffle(&mut OsRng);
-
-        // Try to find an unused host in the set.
-        for host in hosts.iter() {
-            // Check if we already have this connection established
-            if p2p.exists(host).await {
-                trace!(
-                    target: "net::outbound_session::fetch_address_with_lock()",
-                    "Host '{}' exists so skipping",
-                    host
-                );
-                continue
-            }
-
-            // Check if we already have this configured as a manual peer
-            if p2p.settings().peers.contains(host) {
-                trace!(
-                    target: "net::outbound_session::fetch_address_with_lock()",
-                    "Host '{}' configured as manual peer so skipping",
-                    host
-                );
-                continue
-            }
-
-            // Obtain a lock on this address to prevent duplicate connection
-            if !p2p.add_pending(host).await {
-                trace!(
-                    target: "net::outbound_session::fetch_address_with_lock()",
-                    "Host '{}' pending so skipping",
-                    host
-                );
-                continue
-            }
-
-            trace!(
-                target: "net::outbound_session::fetch_address_with_lock()",
-                "Found valid host '{}",
-                host
-            );
-            return Some(host.clone())
-        }
-
-        None
-    }
+    ///// Loops through host addresses to find an outbound address that we can
+    ///// connect to. Check whether the address is valid by making sure it isn't
+    ///// our own inbound address, then checks whether it is already connected
+    ///// (exists) or connecting (pending).
+    ///// Lastly adds matching address to the pending list.
+    ///// TODO: this method should go in hosts
+    //async fn fetch_address_with_lock(&self, transports: &[String]) -> Option<Url> {
+    //    let p2p = self.p2p();
+
+    //    // Collect hosts
+    //    let mut hosts = vec![];
+
+    //    // If transport mixing is enabled, then for example we're allowed to
+    //    // use tor:// to connect to tcp:// and tor+tls:// to connect to tcp+tls://.
+    //    // However, **do not** mix tor:// and tcp+tls://, nor tor+tls:// and tcp://.
+    //    let transport_mixing = p2p.settings().transport_mixing;
+    //    macro_rules! mix_transport {
+    //        ($a:expr, $b:expr) => {
+    //            if transports.contains(&$a.to_string()) && transport_mixing {
+    //                let mut a_to_b = p2p.hosts().whitelist_fetch_with_schemes(&[$b.to_string()], None).await;
+    //                for (addr, last_seen) in a_to_b.iter_mut() {
+    //                    addr.set_scheme($a).unwrap();
+    //                    hosts.push(addr.clone());
+    //                }
+    //            }
+    //        };
+    //    }
+    //    mix_transport!("tor", "tcp");
+    //    mix_transport!("tor+tls", "tcp+tls");
+    //    mix_transport!("nym", "tcp");
+    //    mix_transport!("nym+tls", "tcp+tls");
+
+    //    // And now the actual requested transports
+    //    for (addr, last_seen) in p2p.hosts().whitelist_fetch_with_schemes(transports, None).await {
+    //        hosts.push(addr);
+    //    }
+
+    //    // Randomize hosts list. Do not try to connect in a deterministic order.
+    //    // This is healthier for multiple slots to not compete for the same addrs.
+    //    hosts.shuffle(&mut OsRng);
+
+    //    // Try to find an unused host in the set.
+    //    for host in hosts.iter() {
+    //        // Check if we already have this connection established
+    //        if p2p.exists(host).await {
+    //            trace!(
+    //                target: "net::outbound_session::fetch_address_with_lock()",
+    //                "Host '{}' exists so skipping",
+    //                host
+    //            );
+    //            continue
+    //        }
+
+    //        // Check if we already have this configured as a manual peer
+    //        if p2p.settings().peers.contains(host) {
+    //            trace!(
+    //                target: "net::outbound_session::fetch_address_with_lock()",
+    //                "Host '{}' configured as manual peer so skipping",
+    //                host
+    //            );
+    //            continue
+    //        }
+
+    //        // Obtain a lock on this address to prevent duplicate connection
+    //        if !p2p.add_pending(host).await {
+    //            trace!(
+    //                target: "net::outbound_session::fetch_address_with_lock()",
+    //                "Host '{}' pending so skipping",
+    //                host
+    //            );
+    //            continue
+    //        }
+
+    //        trace!(
+    //            target: "net::outbound_session::fetch_address_with_lock()",
+    //            "Found valid host '{}",
+    //            host
+    //        );
+    //        return Some(host.clone())
+    //    }
+
+    //    None
+    //}
 
     fn notify(&self) {
         self.wakeup_self.notify()

src/net/session/seedsync_session.rs

@@ -115,7 +115,7 @@ impl SeedSyncSession {
         }
 
         // Seed process complete
-        if self.p2p().hosts().is_empty().await {
+        if self.p2p().hosts().is_empty_greylist().await {
             warn!(target: "net::session::seedsync_session", "[P2P] Hosts pool empty after seeding");
         }