net: introduce HostState state machine to safely manage host activity

This commit removes excessive use of RwLocks in the net code, replacing
them with a state machine that:

* Protects against race conditions where multiple threads are trying to
  modify the same resource (i.e. the hostlists) by retricting state
  transitions to a fixed set of safe steps.

* Minimizes risk of deadlocks by aggregating is_migrating(), is_pending(),
  is_connected() etc checks to a single locked HashMap, `HostRegistry`.

This commit also simplifies the `p2p.rs` interface by migrating
connected channel utilities into `hosts/store.rs`.

src/error.rs

@@ -185,6 +185,9 @@ pub enum Error {
     #[error("No matching hostlist entry")]
     HostDoesNotExist,
 
+    #[error("Attempted state change was blocked: {0}")]
+    StateBlocked(String),
+
     // =============
     // Crypto errors
     // =============

src/event_graph/mod.rs

@@ -204,7 +204,7 @@ impl EventGraph {
         //   from the beginning
 
         // Get references to all our peers.
-        let channels = self.p2p.channels().await;
+        let channels = self.p2p.hosts().channels().await;
         let mut communicated_peers = channels.len();
         info!(
             target: "event_graph::dag_sync()",

src/net/hosts/refinery.rs

@@ -26,7 +26,9 @@ use url::Url;
 
 use super::super::p2p::{P2p, P2pPtr};
 use crate::{
-    net::{connector::Connector, protocol::ProtocolVersion, session::Session},
+    net::{
+        connector::Connector, hosts::store::HostState, protocol::ProtocolVersion, session::Session,
+    },
     system::{
         run_until_completion, sleep, timeout::timeout, LazyWeak, StoppableTask, StoppableTaskPtr,
     },
@@ -85,6 +87,8 @@ impl GreylistRefinery {
     }
 
     // Randomly select a peer on the greylist and probe it.
+    // This method will remove from the greylist and store on the whitelist
+    // providing the peer is responsive.
     async fn run(self: Arc<Self>) {
         loop {
             sleep(self.p2p().settings().greylist_refinery_interval).await;
@@ -103,33 +107,25 @@ impl GreylistRefinery {
                 Some((entry, position)) => {
                     let url = &entry.0;
 
-                    // Skip this node if it's being migrated currently.
-                    if hosts.is_migrating(url).await {
+                    if let Err(_) =
+                        hosts.try_update_registry(url.clone(), HostState::Refining).await
+                    {
                         continue
                     }
-
-                    // Don't refine nodes that we are already connected to.
-                    if self.p2p().exists(url).await {
-                        continue
-                    }
-
-                    // Don't refine nodes that we are trying to connect to.
-                    if self.p2p().is_pending(url).await {
-                        continue
-                    }
-
-                    let mut greylist = hosts.greylist.write().await;
                     if !ping_node(url.clone(), self.p2p().clone()).await {
-                        greylist.remove(position);
+                        hosts.greylist_remove(url, position).await;
 
                         debug!(
                             target: "net::refinery",
                             "Peer {} is non-responsive. Removed from greylist", url,
                         );
 
+                        // Remove this entry from HostRegistry to avoid this host getting
+                        // stuck in the Refining state.
+                        hosts.remove_refining(url).await;
+
                         continue
                     }
-                    drop(greylist);
 
                     let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
 

src/net/hosts/store.rs

@@ -18,7 +18,7 @@
 
 use std::{
     collections::{HashMap, HashSet},
-    fs,
+    fmt, fs,
     fs::File,
     sync::Arc,
     time::{Instant, UNIX_EPOCH},
@@ -29,23 +29,113 @@ use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
 use smol::lock::RwLock;
 use url::Url;
 
-use super::super::{p2p::P2pPtr, settings::SettingsPtr};
+use super::super::{settings::SettingsPtr, ChannelPtr};
 use crate::{
     system::{Subscriber, SubscriberPtr, Subscription},
     util::{
         file::{load_file, save_file},
         path::expand_path,
     },
-    Result,
+    Error, Result,
 };
 
 /// Atomic pointer to hosts object
 pub type HostsPtr = Arc<Hosts>;
 
+/// Keeps track of hosts and their current state. Prevents race conditions
+/// where multiple threads are simultaenously trying to change the state of
+/// a given host.
+pub type HostRegistry = RwLock<HashMap<Url, HostState>>;
+
+/// HostState is a set of mutually exclusive states that can be Pending,
+/// Connected, Disconnected or Refining. The state is `None` when the
+/// corresponding host has been removed from the HostRegistry.
+///
+///                              +----------+
+///                          +-- | refining | --+
+///                          |   +----------+   |
+///                          |                  |
+///                          v                  v
+///          +---------+    +-----------+    +------+
+///          | pending | -> | connected | -> | None |
+///          +---------+    +-----------+    +------+
+///               |                             ^
+///               |                             |
+///               |       +-------------+       |
+///               +-----> | downgrading | ------+
+///                       +-------------+
+///
+#[derive(Clone, Debug)]
+pub enum HostState {
+    /// Hosts that are being connected to in Outbound and Manual Session.
+    Pending,
+    /// Hosts that have been successfully connected to.
+    Connected(ChannelPtr),
+    /// Hosts that we have repeatedly failed to connect to, and that are being
+    /// removed from the anchorlist and whitelist and added to the greylist.
+    Downgrading,
+    /// Hosts that are migrating from the greylist to the whitelist or being
+    /// removed from the greylist, as defined in `refinery.rs`.
+    Refining,
+}
+
+impl HostState {
+    // Try to change state to Downgrading. Only possible if this
+    // connection is pending i.e. if we are trying to connect to this
+    // host.
+    fn try_downgrade(&self) -> Result<Self> {
+        match self {
+            HostState::Pending => Ok(HostState::Downgrading),
+            HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
+            HostState::Downgrading => Err(Error::StateBlocked(self.to_string())),
+            HostState::Refining => Err(Error::StateBlocked(self.to_string())),
+        }
+    }
+
+    // Try to change state to Refining. Only possible if we are not yet
+    // tracking this host in the HostRegistry.
+    fn try_refine(&self) -> Result<Self> {
+        match self {
+            HostState::Pending => Err(Error::StateBlocked(self.to_string())),
+            HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
+            HostState::Downgrading => Err(Error::StateBlocked(self.to_string())),
+            HostState::Refining => Err(Error::StateBlocked(self.to_string())),
+        }
+    }
+
+    // Try to change state to Connected. Possible if this peer is
+    // currently Pending or being Refined. The latter is necessary since
+    // the refinery process requires us to establish a connection to
+    // a peer.
+    fn try_connect(&self, channel: ChannelPtr) -> Result<Self> {
+        match self {
+            HostState::Pending => Ok(HostState::Connected(channel)),
+            HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
+            HostState::Downgrading => Err(Error::StateBlocked(self.to_string())),
+            HostState::Refining => Ok(HostState::Connected(channel)),
+        }
+    }
+
+    // Try to change state to Pending. Only possible if we are not yet
+    // tracking this host in the HostRegistry.
+    fn try_pending(&self) -> Result<Self> {
+        match self {
+            HostState::Pending => Err(Error::StateBlocked(self.to_string())),
+            HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
+            HostState::Downgrading => Err(Error::StateBlocked(self.to_string())),
+            HostState::Refining => Err(Error::StateBlocked(self.to_string())),
+        }
+    }
+}
+impl fmt::Display for HostState {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        fmt::Debug::fmt(self, f)
+    }
+}
+
 // An array containing all possible local host strings
 // TODO: This could perhaps be more exhaustive?
 pub const LOCAL_HOST_STRS: [&str; 2] = ["localhost", "localhost.localdomain"];
-
 const WHITELIST_MAX_LEN: usize = 5000;
 const GREYLIST_MAX_LEN: usize = 2000;
 
@@ -63,6 +153,9 @@ pub struct Hosts {
     /// Nodes to which we have already been able to establish a connection.
     pub anchorlist: RwLock<Vec<(Url, u64)>>,
 
+    /// Subscriber for notifications of new channels
+    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
+
     /// 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
@@ -70,13 +163,13 @@ pub struct Hosts {
     /// Internet interrupt (goblins unplugging cables)
     quarantine: RwLock<HashMap<Url, usize>>,
 
+    /// A registry that tracks hosts and their current state.
+    registry: HostRegistry,
+
     /// Peers on the blacklist are considered hostile and can neither be connected to
     /// nor establish connections to us for the duration of the program.
     blacklist: RwLock<HashSet<String>>,
 
-    /// Peers that are currently being removed from the hostlist
-    migrating: RwLock<HashSet<Url>>,
-
     /// Subscriber listening for store updates
     store_subscriber: SubscriberPtr<usize>,
 
@@ -91,14 +184,150 @@ impl Hosts {
             greylist: RwLock::new(Vec::new()),
             whitelist: RwLock::new(Vec::new()),
             anchorlist: RwLock::new(Vec::new()),
+            channel_subscriber: Subscriber::new(),
             quarantine: RwLock::new(HashMap::new()),
+            registry: RwLock::new(HashMap::new()),
             blacklist: RwLock::new(HashSet::new()),
-            migrating: RwLock::new(HashSet::new()),
             store_subscriber: Subscriber::new(),
             settings,
         })
     }
 
+    /// Try to update the registry. If the host already exists, try to update its state.
+    /// Otherwise add the host to the registry along with its state.
+    pub async fn try_update_registry(&self, addr: Url, new_state: HostState) -> Result<HostState> {
+        let mut registry = self.registry.write().await;
+
+        if registry.contains_key(&addr) {
+            let current_state = registry.get(&addr).unwrap().clone();
+
+            debug!(target: "store::try_update_registry()",
+            "Attempting to update addr={} current_state={}, new_state={}",
+            addr, current_state, new_state.to_string());
+
+            let result: Result<HostState> = match new_state {
+                HostState::Pending => current_state.try_pending(),
+                HostState::Connected(c) => current_state.try_connect(c),
+                HostState::Downgrading => current_state.try_downgrade(),
+                HostState::Refining => current_state.try_refine(),
+            };
+
+            if let Ok(state) = &result {
+                registry.insert(addr.clone(), state.clone());
+            }
+
+            result
+        } else {
+            // We don't know this peer. We can safely update the state.
+            registry.insert(addr.clone(), new_state.clone());
+
+            Ok(new_state)
+        }
+    }
+
+    pub async fn check_address(&self, hosts: Vec<(Url, u64)>) -> Option<(Url, u64)> {
+        // Try to find an unused host in the set.
+        for (host, last_seen) in hosts {
+            debug!(target: "store::check_address()", "Starting checks");
+
+            if let Err(_) = self.try_update_registry(host.clone(), HostState::Pending).await {
+                continue
+            }
+
+            debug!(
+                target: "store::check_address()",
+                "Found valid host {}",
+                host
+            );
+            return Some((host.clone(), last_seen))
+        }
+
+        None
+    }
+
+    /// Remove a channel from the list of connected channels. Called
+    /// when a channel disconnects.
+    pub async fn remove_connected(&self, channel: ChannelPtr) {
+        debug!(target: "store", "remove_connected() Removing {}", channel.address());
+        let mut registry = self.registry.write().await;
+        let addr = channel.address();
+
+        let state = registry.get(addr).unwrap();
+
+        if let HostState::Connected(_) = state {
+            debug!(target: "store", "remove_connected() Removed {}", channel.address());
+            registry.remove(addr);
+        }
+    }
+
+    /// Remove a host from the list of downgrading hosts. Must be called
+    /// when downgrade process is finished to avoid the host getting stuck
+    /// in the Downgrading state.
+    pub async fn remove_downgrading(&self, addr: &Url) {
+        debug!(target: "store", "remove_downgrading() Removing {}", addr);
+        let mut registry = self.registry.write().await;
+
+        let state = registry.get(addr).unwrap();
+
+        if let HostState::Downgrading = state {
+            debug!(target: "store", "remove_downgrading() Removed {}", addr);
+            registry.remove(addr);
+        }
+    }
+
+    /// Remove a host from the list of refining hosts. Must be called
+    /// when refinery process fails to avoid the host getting stuck
+    /// in the Refining state.
+    ///
+    /// It is not necessary to call this when the refinery passes, since the
+    /// host state will be changed to Connected (and then the host will be removed
+    /// from the HostRegistry with remove_connected()).
+    pub async fn remove_refining(&self, addr: &Url) {
+        debug!(target: "store", "remove_refining() Removing {}", addr);
+        let mut registry = self.registry.write().await;
+
+        let state = registry.get(addr).unwrap();
+
+        if let HostState::Refining = state {
+            debug!(target: "store", "remove_refining() Removed {}", addr);
+            registry.remove(addr);
+        }
+    }
+
+    /// Returns the list of connected channels.
+    pub async fn channels(&self) -> Vec<ChannelPtr> {
+        let registry = self.registry.read().await;
+        let mut channels = Vec::new();
+
+        for (_, value) in registry.iter() {
+            if let HostState::Connected(c) = value {
+                channels.push(c.clone());
+            }
+        }
+        channels
+    }
+
+    /// Retrieve a random connected channel
+    pub async fn random_channel(&self) -> ChannelPtr {
+        let channels = self.channels().await;
+        let position = rand::thread_rng().gen_range(0..channels.len());
+        channels[position].clone()
+    }
+
+    /// Add a channel to the set of connected channels
+    pub async fn store(&self, channel: ChannelPtr) -> Result<()> {
+        let address = channel.address().clone();
+
+        if let Err(e) =
+            self.try_update_registry(address.clone(), HostState::Connected(channel.clone())).await
+        {
+            return Err(e)
+        }
+
+        self.channel_subscriber.notify(Ok(channel)).await;
+        Ok(())
+    }
+
     /// Loops through greylist 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
@@ -222,71 +451,6 @@ impl Hosts {
         hosts
     }
 
-    /// Check whether:
-    /// *   We already have this connection established
-    /// *   We already have this configured as a manual peer
-    /// *   This address is already pending a connection
-    /// *   This peer is migrating between hostlists
-    pub async fn check_address_with_lock(
-        &self,
-        p2p: P2pPtr,
-        hosts: Vec<(Url, u64)>,
-    ) -> Option<(Url, u64)> {
-        // Try to find an unused host in the set.
-        for (host, last_seen) in hosts {
-            debug!(target: "store::check_address_with_lock()",
-            "Starting checks");
-            // Check if we already have this connection established
-            if p2p.exists(&host).await {
-                debug!(
-                    target: "store::check_address_with_lock()",
-                    "Host '{}' exists so skipping",
-                    host
-                );
-                continue
-            }
-
-            // Check if we already have this configured as a manual peer
-            if self.settings.peers.contains(&host) {
-                debug!(
-                    target: "store::check_address_with_lock()",
-                    "Host '{}' configured as manual peer so skipping",
-                    host
-                );
-                continue
-            }
-
-            // Check this peer isn't currently being migrated from hostlists
-            if self.is_migrating(&host).await {
-                debug!(
-                    target: "store::check_address_with_lock()",
-                    "Host '{}' is migrating so skipping",
-                    host
-                );
-                continue
-            }
-
-            // Obtain a lock on this address to prevent duplicate connection
-            if !p2p.add_pending(&host).await {
-                debug!(
-                    target: "store::check_address_with_lock()",
-                    "Host '{}' pending so skipping",
-                    host
-                );
-                continue
-            }
-
-            debug!(
-                target: "store::check_address_with_lock()",
-                "Found valid host {}",
-                host
-            );
-            return Some((host.clone(), last_seen))
-        }
-
-        None
-    }
-
     /// Upgrade a connection to the anchorlist. Called after a connection has been successfully
     /// established in Outbound and Manual sessions.
     pub async fn upgrade_host(&self, addr: &Url) {
@@ -294,13 +458,19 @@ impl Hosts {
         self.anchorlist_store_or_update(&[(addr.clone(), last_seen)]).await;
     }
 
-    /// Downgrade a host to greylist. Called after we have failed to connect to a host
-    /// outbound_connect_limit times in quarantine()
+    /// Downgrade a host to greylist. If the host is on the anchorlist or whitelist, remove it.
+    /// If it's already on the greylist we can't do anything here.
     pub async fn downgrade_host(&self, addr: &Url, last_seen: u64) {
+        if let Err(_) = self.try_update_registry(addr.clone(), HostState::Downgrading).await {
+            return
+        }
+
         debug!(target: "store::downgrade_host", "Downgrading host {}", addr);
-        self.mark_migrating(addr).await;
+        if self.greylist_contains(addr).await {
+            warn!(target: "store::downgrade_host",
+                  "Cannot downgrade a host that is already on the greylist! {}", addr);
+        }
 
-        // Remove channel from anchorlist
         if self.anchorlist_contains(addr).await {
             debug!(target: "store::downgrade_host", "Removing from anchorlist {}", addr);
 
@@ -310,9 +480,9 @@ impl Hosts {
                 .expect("Expected anchorlist index to exist");
 
             self.anchorlist_remove(addr, index).await;
+            self.greylist_store_or_update(&[(addr.clone(), last_seen)]).await;
         }
 
-        // Remove channel from whitelist
         if self.whitelist_contains(addr).await {
             debug!(target: "store::downgrade_host", "Removing from whitelist {}", addr);
 
@@ -322,11 +492,12 @@ impl Hosts {
                 .expect("Expected whitelist index to exist");
 
             self.whitelist_remove(addr, index).await;
+            self.greylist_store_or_update(&[(addr.clone(), last_seen)]).await;
         }
 
-        self.greylist_store_or_update(&[(addr.clone(), last_seen)]).await;
-
-        self.unmark_migrating(addr).await;
+        // Remove this entry from HostRegistry to avoid this host getting
+        // stuck in the Downgrading state.
+        self.remove_downgrading(&addr).await;
     }
 
     /// Stores an address on the greylist or updates its last_seen field if we already
@@ -338,7 +509,7 @@ impl Hosts {
         let filtered_addrs_len = filtered_addrs.len();
 
         if filtered_addrs.is_empty() {
-            debug!(target: "store::greylist_store_or_update()", "Filtered out all received addresses");
+            debug!(target: "store::greylist_store_or_update()", "Filtered out all addresses");
         }
 
         for (addr, last_seen) in filtered_addrs {
@@ -657,15 +828,15 @@ impl Hosts {
     /// Quarantine a peer.
     /// If they've been quarantined for more than a configured limit, downgrade to greylist.
     pub async fn quarantine(&self, addr: &Url, last_seen: u64) {
-        debug!(target: "store::remove()", "Quarantining peer {}", addr);
+        debug!(target: "store::quarantine()", "Quarantining peer {}", addr);
         let timer = Instant::now();
         let mut q = self.quarantine.write().await;
         if let Some(retries) = q.get_mut(addr) {
             *retries += 1;
-            debug!(target: "net::hosts::quarantine()", "Peer {} quarantined {} times", addr, retries);
+            debug!(target: "store::quarantine()", "Peer {} quarantined {} times", addr, retries);
             if *retries == self.settings.hosts_quarantine_limit {
-                debug!(target: "net::hosts::quarantine()", "Reached quarantine limited after {:?}", timer.elapsed());
-                debug!(target: "net::hosts::quarantine()", "Removing from hostlist {}", addr);
+                debug!(target: "store::quarantine()", "Reached quarantine limited after {:?}", timer.elapsed());
+                debug!(target: "store::quarantine()", "Removing from hostlist {}", addr);
                 drop(q);
                 self.downgrade_host(addr, last_seen).await;
             }
@@ -710,21 +881,6 @@ impl Hosts {
         }
     }
 
-    /// Peer that is currently being removed from hostlists.
-    pub async fn is_migrating(&self, peer: &Url) -> bool {
-        self.migrating.read().await.contains(peer)
-    }
-
-    /// Mark a peer as currently migrating.
-    pub async fn mark_migrating(&self, peer: &Url) {
-        self.migrating.write().await.insert(peer.clone());
-    }
-
-    /// Unmark a migrating peer.
-    pub async fn unmark_migrating(&self, peer: &Url) {
-        self.migrating.write().await.remove(peer);
-    }
-
     /// Check if the greylist is empty.
     pub async fn is_empty_greylist(&self) -> bool {
         self.greylist.read().await.is_empty()

src/net/p2p.rs

@@ -23,7 +23,6 @@ use std::{
 
 use futures::{stream::FuturesUnordered, TryFutureExt};
 use log::{debug, error, info, warn};
-use rand::{prelude::IteratorRandom, rngs::OsRng};
 use smol::{lock::Mutex, stream::StreamExt};
 use url::Url;
 
@@ -58,12 +57,6 @@ pub type P2pPtr = Arc<P2p>;
 pub struct P2p {
     /// Global multithreaded executor reference
     executor: ExecutorPtr,
-    /// Channels pending connection
-    pending: PendingChannels,
-    /// Connected channels
-    channels: ConnectedChannels,
-    /// Subscriber for notifications of new channels
-    channel_subscriber: SubscriberPtr<Result<ChannelPtr>>,
     /// Known hosts (peers)
     hosts: HostsPtr,
     /// Protocol registry
@@ -85,7 +78,7 @@ pub struct P2p {
     /// The subscriber for which we can give dnet info over
     dnet_subscriber: SubscriberPtr<DnetEvent>,
 
-    // Greylist refinery process
+    /// Greylist refinery process
     greylist_refinery: Arc<GreylistRefinery>,
 }
 
@@ -103,9 +96,6 @@ impl P2p {
 
         let self_ = Arc::new(Self {
             executor,
-            pending: Mutex::new(HashSet::new()),
-            channels: Mutex::new(HashMap::new()),
-            channel_subscriber: Subscriber::new(),
             hosts: Hosts::new(settings.clone()),
             protocol_registry: ProtocolRegistry::new(),
             settings,
@@ -192,7 +182,7 @@ impl P2p {
     /// the ones provided in `exclude_list`.
     pub async fn broadcast_with_exclude<M: Message>(&self, message: &M, exclude_list: &[Url]) {
         let mut channels = Vec::new();
-        for channel in self.channels().await {
+        for channel in self.hosts().channels().await {
             if exclude_list.contains(channel.address()) {
                 continue
             }
@@ -226,51 +216,8 @@ impl P2p {
         let _results: Vec<_> = futures.collect().await;
     }
 
-    /// Check whether we're connected to a given address
-    pub async fn exists(&self, addr: &Url) -> bool {
-        self.channels.lock().await.contains_key(addr)
-    }
-
-    /// Add a channel to the set of connected channels
-    pub(super) async fn store(&self, channel: ChannelPtr) {
-        self.channels.lock().await.insert(channel.address().clone(), channel.clone());
-
-        self.channel_subscriber.notify(Ok(channel)).await;
-    }
-
-    /// Remove a channel from the set of connected channels
-    pub(super) async fn remove(&self, channel: ChannelPtr) {
-        self.channels.lock().await.remove(channel.address());
-    }
-
-    /// Add an address to the list of pending channels.
-    pub(super) async fn add_pending(&self, addr: &Url) -> bool {
-        self.pending.lock().await.insert(addr.clone())
-    }
-
-    /// Check whether a connection is currently pending.
-    pub(super) async fn is_pending(&self, addr: &Url) -> bool {
-        self.pending.lock().await.contains(addr)
-    }
-
-    /// Remove a channel from the list of pending channels.
-    pub(super) async fn remove_pending(&self, addr: &Url) {
-        self.pending.lock().await.remove(addr);
-    }
-
-    /// Return all connected channels
-    pub async fn channels(&self) -> Vec<ChannelPtr> {
-        self.channels.lock().await.values().cloned().collect()
-    }
-
-    /// Retrieve a random connected channel from the
-    pub async fn random_channel(&self) -> Option<ChannelPtr> {
-        let channels = self.channels.lock().await;
-        channels.values().choose(&mut OsRng).cloned()
-    }
-
     pub async fn is_connected(&self) -> bool {
-        !self.channels.lock().await.is_empty()
+        !self.hosts().channels().await.is_empty()
     }
 
     /// Return an atomic pointer to the set network settings

src/net/session/inbound_session.rs

@@ -199,7 +199,7 @@ impl InboundSession {
 
         stop_sub.receive().await;
 
-        self.p2p().remove(channel.clone()).await;
+        self.p2p().hosts().remove_connected(channel.clone()).await;
 
         debug!(
             target: "net::inbound_session::setup_channel()",

src/net/session/manual_session.rs

@@ -45,6 +45,7 @@ use super::{
     Session, SessionBitFlag, SESSION_MANUAL,
 };
 use crate::{
+    net::hosts::store::HostState,
     system::{sleep, LazyWeak, StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr},
     Error, Result,
 };
@@ -103,9 +104,6 @@ impl ManualSession {
         let attempts = settings.manual_attempt_limit;
         let mut remaining = attempts;
 
-        // Add the peer to list of pending channels
-        self.p2p().add_pending(&addr).await;
-
         // Loop forever if attempts==0, otherwise loop attempts number of times.
         let mut tried_attempts = 0;
         loop {
@@ -115,6 +113,13 @@ impl ManualSession {
                 "[P2P] Connecting to manual outbound [{}] (attempt #{})",
                 addr, tried_attempts,
             );
+
+            if let Err(_) =
+                self.p2p().hosts().try_update_registry(addr.clone(), HostState::Pending).await
+            {
+                continue
+            }
+
             match connector.connect(&addr).await {
                 Ok((url, channel)) => {
                     info!(
@@ -130,9 +135,6 @@ impl ManualSession {
                     // Register the new channel
                     self.register_channel(channel.clone(), ex.clone()).await?;
 
-                    // Remove pending lock since register_channel will add the channel to p2p
-                    self.p2p().remove_pending(&addr).await;
-
                     // Add this connection to the anchorlist
                     self.p2p().hosts().upgrade_host(&addr).await;
 
@@ -182,8 +184,6 @@ impl ManualSession {
             addr, attempts,
         );
 
-        self.p2p().remove_pending(&addr).await;
-
         Ok(())
     }
 }

src/net/session/mod.rs

@@ -19,7 +19,7 @@
 use std::sync::{Arc, Weak};
 
 use async_trait::async_trait;
-use log::debug;
+use log::{debug, warn};
 use smol::Executor;
 
 use super::{channel::ChannelPtr, p2p::P2pPtr, protocol::ProtocolVersion};
@@ -62,7 +62,7 @@ pub async fn remove_sub_on_stop(p2p: P2pPtr, channel: ChannelPtr) {
     );
 
     // Remove channel from p2p
-    p2p.remove(channel).await;
+    p2p.hosts().remove_connected(channel).await;
     debug!(target: "net::session::remove_sub_on_stop()", "[END]");
 }
 
@@ -144,10 +144,14 @@ pub trait Session: Sync {
         // Perform handshake
         protocol_version.run(executor.clone()).await?;
 
-        // Add channel to p2p
-        self.p2p().store(channel.clone()).await;
+        // Attempt to add channel to registry
+        if let Err(e) = self.p2p().hosts().store(channel.clone()).await {
+            warn!(target: "net::session::perform_handshake_protocols()",
+            "Couldn't add channel {} to registry!! {}", channel.address(), e);
+            return Err(e)
+        }
 
-        // Subscribe to stop, so we can remove from p2p
+        // Subscribe to stop, so we can remove from registry
         executor.spawn(remove_sub_on_stop(self.p2p(), channel)).detach();
 
         // Channel is ready for use

src/net/session/outbound_session.rs

@@ -185,6 +185,7 @@ impl Slot {
         self.process.stop().await
     }
 
+    // TODO: rethink this logic.
     async fn fetch_address(&self, slot_count: usize, transports: &[String]) -> Option<(Url, u64)> {
         let hosts = self.p2p().hosts();
         let connects = self.p2p().settings().outbound_connections;
@@ -199,19 +200,19 @@ impl Slot {
             if !hosts.anchorlist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.anchorlist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
 
             if !hosts.whitelist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.whitelist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
 
             if !hosts.greylist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.greylist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
         } else if slot_count < white_count {
             // Up to white_connection_percent connections:
@@ -221,13 +222,13 @@ impl Slot {
             if !hosts.whitelist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.whitelist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
 
             if !hosts.greylist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.greylist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
         } else {
             // All other connections:
@@ -236,7 +237,7 @@ impl Slot {
             if !hosts.greylist_fetch_address(transports).await.is_empty() {
                 let addrs = hosts.greylist_fetch_address(transports).await;
 
-                return hosts.check_address_with_lock(self.p2p(), addrs).await
+                return hosts.check_address(addrs).await
             }
         }
 
@@ -345,7 +346,7 @@ impl Slot {
 
             let stop_sub = channel.subscribe_stop().await.expect("Channel should not be stopped");
             // Setup new channel
-            if let Err(err) = self.setup_channel(host.clone(), channel.clone()).await {
+            if let Err(err) = self.setup_channel(channel.clone()).await {
                 info!(
                     target: "net::outbound_session",
                     "[P2P] Outbound slot #{} disconnected: {}",
@@ -393,12 +394,9 @@ impl Slot {
                     self.slot, addr, e
                 );
 
-                // At this point we failed to connect. We'll quarantine this peer now.
+                // At this point we failed to connect. We'll downgrade this peer now.
                 self.p2p().hosts().quarantine(&addr, last_seen).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;
 
@@ -407,16 +405,12 @@ impl Slot {
         }
     }
 
-    async fn setup_channel(&self, addr: Url, channel: ChannelPtr) -> Result<()> {
+    async fn setup_channel(&self, channel: ChannelPtr) -> Result<()> {
         // Register the new channel
         debug!(target: "net::outbound_session::setup_channel", "register_channel {}", channel.clone().address());
         self.session().register_channel(channel.clone(), self.p2p().executor()).await?;
 
         // Channel is now connected but not yet setup
-        // Remove pending lock since register_channel will add the channel to p2p
-        debug!(target: "net::outbound_session::setup_channel", "removing channel...");
-        self.p2p().remove_pending(&addr).await;
-        debug!(target: "net::outbound_session::setup_channel", "channel removed!");
 
         // Notify that channel processing has been finished
         self.session().channel_subscriber.notify(Ok(channel)).await;

src/rpc/p2p_method.rs

@@ -28,7 +28,7 @@ use crate::net;
 pub trait HandlerP2p: Sync + Send {
     async fn p2p_get_info(&self, id: u16, _params: JsonValue) -> JsonResult {
         let mut channels = Vec::new();
-        for channel in self.p2p().channels().await {
+        for channel in self.p2p().hosts().channels().await {
             let session = match channel.session_type_id() {
                 net::session::SESSION_INBOUND => "inbound",
                 net::session::SESSION_OUTBOUND => "outbound",