net: fix death loop

previously we would downgrade disconnected hosts by putting them in the
greylist. this commit changes downgrade_host() to remove_host() and removes the host
from all known lists, including greylist, when it disconnects.

otherwise on a small network the following situation could happen:

    node B stores node A on greylist
    connects to node A, upgrades to anchorlist

    hostlist is currently: greylist: node A
                           anchorlist: node A

    node A disconnects
    hostlist is currently: greylist: node A

    node B tries to find another addr to connect to
    it's only known host is node A
        tries to reconnect
        connection fails
        goes to greylist, retries
        loops forever

removing from all hostlists is "safe" because if the node A goes back
online again it will once again be added to the greylist of node B
through the address propagation protocols.

src/net/hosts/refinery.rs

@@ -87,7 +87,7 @@ impl GreylistRefinery {
             let hosts = self.p2p().hosts();
 
             if hosts.is_empty_greylist().await {
-                warn!(target: "net::refinery::run()",
+                warn!(target: "net::refinery",
                 "Greylist is empty! Cannot start refinery process");
 
                 continue
@@ -96,10 +96,16 @@ impl GreylistRefinery {
             let (entry, position) = hosts.greylist_fetch_random().await;
             let url = &entry.0;
 
+            // Skip this node if it's being migrated currently.
+            if hosts.is_migrating(url).await {
+                continue
+            }
+
             if !ping_node(url, self.p2p().clone()).await {
                 let mut greylist = hosts.greylist.write().await;
                 greylist.remove(position);
-                debug!(target: "net::refinery::run()", "Peer {} is non-responsive. Removed from greylist", url);
+                debug!(target: "net::refinery",
+                       "Peer {} is non-responsive. Removed from greylist", url);
 
                 continue
             }

src/net/hosts/store.rs

@@ -66,6 +66,9 @@ pub struct Hosts {
     /// Peers we reject from connecting to
     rejected: 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>,
 
@@ -81,6 +84,7 @@ impl Hosts {
             whitelist: RwLock::new(Vec::new()),
             anchorlist: RwLock::new(Vec::new()),
             rejected: RwLock::new(HashSet::new()),
+            migrating: RwLock::new(HashSet::new()),
             store_subscriber: Subscriber::new(),
             settings,
         })
@@ -266,18 +270,15 @@ impl Hosts {
         self.anchorlist_store_or_update(&[(addr.clone(), last_seen)]).await;
     }
 
-    /// Downgrade a connection. If it's on the anchorlist or the whitelist, remove it and add it
-    /// to the greylist. Called when we cannot establish a connection to a host or when a
-    /// pre-existing connection disconnects.
-    pub async fn downgrade_host(&self, addr: &Url) {
-        // Remove channel from anchorlist and add it to greylist
-        if self.anchorlist_contains(addr).await {
-            let (url, last_seen) = self
-                .get_anchorlist_entry_at_addr(addr)
-                .await
-                .expect("Expected anchorlist entry to exist");
+    /// Remove an entry from the hostlist. Called when we cannot establish a connection to a host or 
+    /// when a pre-existing connection disconnects.
+    pub async fn remove_host(&self, addr: &Url) {
+        debug!(target: "store::downgrade_host", "Removing host {}", addr);
+        self.mark_migrating(addr).await;
 
-            self.greylist_store_or_update(&[(url, last_seen)]).await;
+        // Remove channel from anchorlist 
+        if self.anchorlist_contains(addr).await {
+            debug!(target: "store::downgrade_host", "Removing from anchorlist {}", addr);
 
             let index = self
                 .get_anchorlist_index_at_addr(addr.clone())
@@ -287,14 +288,9 @@ impl Hosts {
             self.anchorlist_remove(addr, index).await;
         }
 
-        // Remove channel from whitelist and add to greylist
+        // Remove channel from whitelist
         if self.whitelist_contains(addr).await {
-            let (url, last_seen) = self
-                .get_whitelist_entry_at_addr(addr)
-                .await
-                .expect("Expected whitelist entry to exist");
-
-            self.greylist_store_or_update(&[(url, last_seen)]).await;
+            debug!(target: "store::downgrade_host", "Removing from whitelist {}", addr);
 
             let index = self
                 .get_whitelist_index_at_addr(addr.clone())
@@ -303,6 +299,20 @@ impl Hosts {
 
             self.whitelist_remove(addr, index).await;
         }
+
+        // Remove channel the greylist
+        if self.greylist_contains(addr).await {
+            debug!(target: "store::downgrade_host", "Removing from greylist {}", addr);
+
+            let index = self
+                .get_greylist_index_at_addr(addr.clone())
+                .await
+                .expect("Expected greylist index to exist");
+
+            self.greylist_remove(addr, index).await;
+        }
+
+        self.unmark_migrating(addr).await;
     }
 
     /// Stores an address on the greylist or updates its last_seen field if we already
@@ -480,19 +490,19 @@ impl Hosts {
 
     /// Remove an entry from the greylist.
     pub async fn greylist_remove(&self, addr: &Url, index: usize) {
-        debug!(target: "net::refinery::run()", "Removing peer {} from greylist", addr);
+        debug!(target: "store::greylist_remove", "Removing peer {} from greylist", addr);
         self.greylist.write().await.remove(index);
     }
 
     /// Remove an entry from the whitelist.
     pub async fn whitelist_remove(&self, addr: &Url, index: usize) {
-        debug!(target: "net::refinery::run()", "Removing peer {} from whitelist", addr);
+        debug!(target: "store::whitelist_remove", "Removing peer {} from whitelist", addr);
         self.whitelist.write().await.remove(index);
     }
 
     /// Remove an entry from the anchorlist.
     pub async fn anchorlist_remove(&self, addr: &Url, index: usize) {
-        debug!(target: "net::refinery::run()", "Removing peer {} from anchorlist", addr);
+        debug!(target: "store::anchorlist_remove", "Removing peer {} from anchorlist", addr);
         self.anchorlist.write().await.remove(index);
     }
 
@@ -649,6 +659,21 @@ 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/session/manual_session.rs

@@ -146,8 +146,9 @@ impl ManualSession {
                         target: "net::manual_session",
                         "[P2P] Manual outbound disconnected [{}]", url,
                     );
-                    // Downgrade this host to greylist if it's on the whitelist or anchorlist.
-                    self.p2p().hosts().downgrade_host(&addr).await;
+
+                    // Remove this host from the hostlist.
+                    self.p2p().hosts().remove_host(&addr).await;
 
                     // DEV NOTE: Here we can choose to attempt reconnection again
                     return Ok(())

src/net/session/outbound_session.rs

@@ -310,8 +310,8 @@ impl Slot {
                         err: err.to_string()
                     });
 
-                    // Downgrade this host to greylist if it's on the whitelist or anchorlist.
-                    hosts.downgrade_host(&host).await;
+                    // Remove this host from the hostlist.
+                    hosts.remove_host(&host).await;
 
                     self.channel_id.store(0, Ordering::Relaxed);
                     continue
@@ -359,8 +359,8 @@ impl Slot {
             stop_sub.receive().await;
             self.channel_id.store(0, Ordering::Relaxed);
 
-            // Downgrade this host to greylist if it's on the whitelist or anchorlist.
-            hosts.downgrade_host(&addr).await;
+            // Remove this host from the hostlist.
+            hosts.remove_host(&addr).await;
         }
     }