outbound_session: create run2() method that changes run() behavior to new whitelist protocol.

get an address from the whitelist using whitelist_fetch_address_with_lock().

try to establish a connection to it. on success, update the last_seen field for the address on the whitelist.

TODO: we're still using try_connect here which quarantines a peer on failure to connect. In the greylist update we will simply drop this connection and move on.

src/net/hosts.rs

@@ -104,45 +104,73 @@ impl Hosts {
         debug!(target: "net::hosts::store()", "hosts::store() [END]");
     }
 
+    // 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]");
+        if !self.whitelist_contains(addr).await {
+            self.whitelist_store(addr, last_seen).await;
+        } else {
+            let index = self.get_whitelist_index_at_addr(addr).await;
+            self.whitelist_update_last_seen(addr, last_seen, index).await;
+        }
+    }
+
+    // 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;
+        self.whitelist_update_last_seen(addr, last_seen, index).await;
+    }
+
     // Append host to the greylist. Called on learning of a new peer.
-    pub async fn store_greylist(&self, addr: &Url, last_seen: u64) {
-        debug!(target: "net::hosts::store_greylist()", "hosts::store_greylist() [START]");
+    pub async fn greylist_store(&self, addr: &Url, last_seen: u64) {
+        debug!(target: "net::hosts::greylist_store()", "hosts::greylist_store() [START]");
 
         let mut greylist = self.greylist.write().await;
 
-        debug!(target: "net::hosts::store_greylist()", "Inserting {}. Last seen {:?}", addr, last_seen);
+        debug!(target: "net::hosts::greylist_store()", "Inserting {}. Last seen {:?}", addr, last_seen);
 
         // Remove oldest element if the greylist reaches max size.
         if greylist.len() == 5000 {
             let last_entry = greylist.pop().unwrap();
-            debug!(target: "net::hosts::store_greylist()", "Greylist reached max size. Removed {:?}", last_entry);
+            debug!(target: "net::hosts::greylist_store()", "Greylist reached max size. Removed {:?}", last_entry);
         }
         greylist.push((addr.clone(), last_seen));
 
         // Sort the list by last_seen.
         greylist.sort_unstable_by_key(|entry| entry.1);
 
-        debug!(target: "net::hosts::store_greylist()", "hosts::store_greylist() [END]");
+        debug!(target: "net::hosts::greylist_store()", "hosts::greylist_store() [END]");
     }
 
     // Append host to the whitelist. Called after a successful interaction with an online peer.
-    pub async fn store_whitelist(&self, addr: &Url, last_seen: u64) {
-        debug!(target: "net::hosts::store_whitelist()", "hosts::store_whitelist() [START]");
+    pub async fn whitelist_store(&self, addr: &Url, last_seen: u64) {
+        debug!(target: "net::hosts::whitelist_store()", "hosts::whitelist_store() [START]");
 
         let mut whitelist = self.whitelist.write().await;
 
-        debug!(target: "net::hosts::store_whitelist()", "Inserting {}. Last seen {:?}", addr, last_seen);
+        debug!(target: "net::hosts::whitelist_store()", "Inserting {}. Last seen {:?}", addr, last_seen);
 
         // Remove oldest element if the whitelist reaches max size.
         if whitelist.len() == 1000 {
             let last_entry = whitelist.pop().unwrap();
-            debug!(target: "net::hosts::store_whitelist()", "Whitelist reached max size. Removed {:?}", last_entry);
+            debug!(target: "net::hosts::whitelist_store()", "Whitelist reached max size. Removed {:?}", last_entry);
         }
         whitelist.push((addr.clone(), last_seen));
 
         // Sort the list by last_seen.
         whitelist.sort_unstable_by_key(|entry| entry.1);
-        debug!(target: "net::hosts::store_whitelist()", "hosts::store_greylist() [END]");
+        debug!(target: "net::hosts::whitelist_store()", "hosts::greylist_store() [END]");
+    }
+
+    // Update the last_seen field of a peer on the whitelist.
+    pub async fn whitelist_update_last_seen(&self, addr: &Url, last_seen: u64, index: usize) {
+        debug!(target: "net::hosts::update_last_seen()", "hosts::update_last_seen() [START]");
+
+        let mut whitelist = self.whitelist.write().await;
+
+        whitelist[index] = (addr.clone(), last_seen);
     }
 
     pub async fn subscribe_store(&self) -> Result<Subscription<usize>> {
@@ -452,6 +480,18 @@ impl Hosts {
         return false
     }
 
+    // Get the index for a given addr on the whitelist.
+    pub async fn get_whitelist_index_at_addr(&self, addr: &Url) -> usize {
+        let whitelist = self.whitelist.read().await;
+        for (i, (url, _time)) in whitelist.iter().enumerate() {
+            if url == addr {
+                return i
+            }
+        }
+        // TODO: FIXME: This should never happen.
+        return 0
+    }
+
     /// Check if host is already in the set
     pub async fn contains(&self, addr: &Url) -> bool {
         self.addrs.read().await.contains(addr)
@@ -760,7 +800,7 @@ mod tests {
     }
 
     #[test]
-    fn test_store_greylist() {
+    fn test_greylist_store() {
         smol::block_on(async {
             let settings = Settings {
                 localnet: false,
@@ -778,7 +818,7 @@ mod tests {
             let last_seen =
                 SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
 
-            hosts.store_greylist(&url, last_seen).await;
+            hosts.greylist_store(&url, last_seen).await;
 
             assert!(!hosts.is_empty_greylist().await);
             assert!(hosts.greylist_contains(&url).await);
@@ -786,7 +826,7 @@ mod tests {
     }
 
     #[test]
-    fn test_store_whitelist() {
+    fn test_whitelist_store() {
         smol::block_on(async {
             let settings = Settings {
                 localnet: false,
@@ -804,7 +844,7 @@ mod tests {
             let last_seen =
                 SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
 
-            hosts.store_whitelist(&url, last_seen).await;
+            hosts.whitelist_store(&url, last_seen).await;
 
             assert!(!hosts.is_empty_whitelist().await);
             assert!(hosts.whitelist_contains(&url).await);

src/net/session/outbound_session.rs

@@ -31,7 +31,7 @@ use std::{
         atomic::{AtomicU32, Ordering},
         Arc, Weak,
     },
-    time::{Duration, Instant},
+    time::{Duration, Instant, SystemTime},
 };
 
 use async_trait::async_trait;
@@ -292,6 +292,113 @@ impl Slot {
         }
     }
 
+    // Looks up whitelisted addresses. Tries to connect to them.
+    // On success, updates the whitelist last_seen field.
+    async fn run2(self: Arc<Self>) {
+        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.
+            let addr = if let Some(addr) = self.whitelist_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
+            );
+
+            let hosts = self.p2p().hosts();
+            let last_seen =
+                SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_secs();
+
+            // Update the last_seen field for this whitelisted peer.
+            // TODO: This peer should also be flagged as an "anchor" because we have been
+            // able to establish a connection to it to it.
+            hosts.whitelist_update(&addr_final, last_seen).await;
+
+            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
@@ -425,6 +532,86 @@ impl Slot {
         None
     }
 
+    // Gets addresses from the whitelist.
+    async fn whitelist_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 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().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::whitelist_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::whitelist_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::whitelist_fetch_address_with_lock()",
+                    "Host '{}' pending so skipping",
+                    host
+                );
+                continue
+            }
+
+            trace!(
+                target: "net::outbound_session::whitelist_fetch_address_with_lock()",
+                "Found valid host '{}",
+                host
+            );
+            return Some(host.clone())
+        }
+
+        None
+    }
+
     fn notify(&self) {
         self.wakeup_self.notify()
     }