net: enable offline reconnect

This commit introduces the following changes that enable offline reconnect:

1. When we have no connections for longer than the period set in
   settings.time_with_no_connections, disable the refinery. This stops
   the refinery from deleting hosts from the hostlist when we are not
   connected.

2. Introduce a new HostState called Suspend, which replaces
   "quarantine". When we fail to connect to a host in OutboundSession we
   downgrade the peer to greylist and mark its state as "Suspend". This
   prevents us from making connections with this peer before the peer
   passes through the refinery.
   Essentially, setting the state to "Suspend" means the refinery is
   tasked with filtering this peer and ensuring whether it should stay
   on the hostlist or be deleted.

3. When we pause the refinery, we also free up all peers marked as
   "Suspend", so that we can immediately reconnect to such peers when
   we regain connectivity.

At present, time_with_no_connections is set to 30s by default, while the
refinery interval is 15s. This means that we could delete a maximum of 2
healthy hosts before the refinery is disabled. We need to consider this
carefully and test different outcomes.

src/net/hosts/refinery.rs

@@ -18,7 +18,7 @@
 
 use std::{
     sync::Arc,
-    time::{Duration, UNIX_EPOCH},
+    time::{Duration, Instant, UNIX_EPOCH},
 };
 
 use log::{debug, warn};
@@ -93,10 +93,12 @@ impl GreylistRefinery {
     // 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;
+        let mut last_online = Instant::now();
+        let settings = self.p2p().settings();
+        let hosts = self.p2p().hosts();
 
-            let hosts = self.p2p().hosts();
+        loop {
+            sleep(settings.greylist_refinery_interval).await;
 
             if hosts.container.is_empty(HostColor::Grey).await {
                 debug!(target: "net::refinery",
@@ -105,13 +107,35 @@ impl GreylistRefinery {
                 continue
             }
 
+            // Pause the refinery if we've had zero connections for longer than the configured
+            // limit.
+            if hosts.channels().await.is_empty() {
+                let time_offline = Instant::now().duration_since(last_online);
+                let offline_limit = Duration::from_secs(settings.time_with_no_connections);
+
+                if time_offline >= offline_limit {
+                    warn!(target: "net::refinery", "No connections for {}s. Refinery paused.",
+                          time_offline.as_secs());
+
+                    // It is neccessary to clear suspended hosts at this point, otherwise these
+                    // hosts cannot be connected to in Outbound Session. Failure to do this could
+                    // result in the refinery being paused forver (since connections could never be
+                    // made).
+                    let suspended_hosts = hosts.suspended().await;
+                    for host in suspended_hosts {
+                        hosts.unregister(&host).await;
+                    }
+
+                    continue
+                }
+            } else {
+                last_online = Instant::now();
+            }
+
             // Only attempt to refine peers that match our transports.
             match hosts
                 .container
-                .fetch_random_with_schemes(
-                    HostColor::Grey,
-                    &self.p2p().settings().allowed_transports,
-                )
+                .fetch_random_with_schemes(HostColor::Grey, &settings.allowed_transports)
                 .await
             {
                 Some((entry, position)) => {

src/net/hosts/store.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, fmt, fs, fs::File, sync::Arc, time::Instant};
+use std::{collections::HashMap, fmt, fs, fs::File, sync::Arc};
 
 use log::{debug, error, info, trace, warn};
 use rand::{prelude::IteratorRandom, rngs::OsRng, Rng};
@@ -51,19 +51,24 @@ pub type HostRegistry = RwLock<HashMap<Url, HostState>>;
 /// Connected, Disconnected or Refining. The state is `None` when the
 /// corresponding host has been removed from the HostRegistry.
 /// ```
-///                              +--------+
-///                          +-- | refine | ----+
-///                          |   +--------+     |
-///                          |                  |
-///                          v                  v
-///          +---------+    +-----------+    +------+    +--------+
-///          | connect | -> | connected | -> | None | <- | insert |
-///          +---------+    +-----------+    +------+    +--------+
-///               |                             ^
-///               |                             |
-///               |          +------+           |
-///               +--------> | move | ----------+
-///                          +------+
+///                                +--------+                       
+///                                | refine | <------------+
+///                                +--------+              |          
+///                   +---------+    |    |   +--------+   |
+///                   | connect |----+    |   | insert |   |
+///                   +---------+    |    |   +--------+   |
+///                   |              |    |      |         |
+///                   |              |    +------+         |
+///                   |              |           |         |
+///                   |              v           v         |
+///                   |  +-----------+    +------+    +---------+  
+///                   |  | connected | -> | None | <- | suspend |  
+///                   |  +-----------+    +------+    +---------+  
+///                   |                          ^         ^
+///                   |      +------+            |         |
+///                   +----> | move | -----------+---------+
+///                          +------+                   
+///                                               
 /// ```
 #[derive(Clone, Debug)]
 pub enum HostState {
@@ -74,6 +79,8 @@ pub enum HostState {
     Refine,
     /// Hosts that are being connected to in Outbound and Manual Session.
     Connect,
+    /// TODO: documentation
+    Suspend,
     /// Hosts that have been successfully connected to.
     Connected(ChannelPtr),
     /// TODO: doc
@@ -88,6 +95,7 @@ impl HostState {
             HostState::Insert => Err(Error::StateBlocked(self.to_string())),
             HostState::Refine => Err(Error::StateBlocked(self.to_string())),
             HostState::Connect => Err(Error::StateBlocked(self.to_string())),
+            HostState::Suspend => Err(Error::StateBlocked(self.to_string())),
             HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
             HostState::Move => Err(Error::StateBlocked(self.to_string())),
         }
@@ -100,6 +108,7 @@ impl HostState {
             HostState::Insert => Err(Error::StateBlocked(self.to_string())),
             HostState::Refine => Err(Error::StateBlocked(self.to_string())),
             HostState::Connect => Err(Error::StateBlocked(self.to_string())),
+            HostState::Suspend => Ok(HostState::Refine),
             HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
             HostState::Move => Err(Error::StateBlocked(self.to_string())),
         }
@@ -112,6 +121,7 @@ impl HostState {
             HostState::Insert => Err(Error::StateBlocked(self.to_string())),
             HostState::Refine => Err(Error::StateBlocked(self.to_string())),
             HostState::Connect => Err(Error::StateBlocked(self.to_string())),
+            HostState::Suspend => Err(Error::StateBlocked(self.to_string())),
             HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
             HostState::Move => Err(Error::StateBlocked(self.to_string())),
         }
@@ -125,6 +135,7 @@ impl HostState {
             HostState::Insert => Err(Error::StateBlocked(self.to_string())),
             HostState::Refine => Ok(HostState::Connected(channel)),
             HostState::Connect => Ok(HostState::Connected(channel)),
+            HostState::Suspend => Err(Error::StateBlocked(self.to_string())),
             HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
             HostState::Move => Err(Error::StateBlocked(self.to_string())),
         }
@@ -137,10 +148,23 @@ impl HostState {
             HostState::Insert => Err(Error::StateBlocked(self.to_string())),
             HostState::Refine => Err(Error::StateBlocked(self.to_string())),
             HostState::Connect => Ok(HostState::Move),
+            HostState::Suspend => Err(Error::StateBlocked(self.to_string())),
             HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
             HostState::Move => Err(Error::StateBlocked(self.to_string())),
         }
     }
+
+    // TODO
+    fn try_suspend(&self) -> Result<Self> {
+        match self {
+            HostState::Insert => Err(Error::StateBlocked(self.to_string())),
+            HostState::Refine => Err(Error::StateBlocked(self.to_string())),
+            HostState::Connect => Err(Error::StateBlocked(self.to_string())),
+            HostState::Suspend => Err(Error::StateBlocked(self.to_string())),
+            HostState::Connected(_) => Err(Error::StateBlocked(self.to_string())),
+            HostState::Move => Ok(HostState::Suspend),
+        }
+    }
 }
 impl fmt::Display for HostState {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
@@ -663,13 +687,6 @@ pub struct Hosts {
     /// 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
-    /// be helpful in order to self-heal the p2p connections in case we have an
-    /// Internet interrupt (goblins unplugging cables)
-    quarantine: RwLock<HashMap<Url, usize>>,
-
     /// A registry that tracks hosts and their current state.
     registry: HostRegistry,
 
@@ -688,7 +705,6 @@ impl Hosts {
     pub fn new(settings: SettingsPtr) -> HostsPtr {
         Arc::new(Self {
             channel_subscriber: Subscriber::new(),
-            quarantine: RwLock::new(HashMap::new()),
             registry: RwLock::new(HashMap::new()),
             store_subscriber: Subscriber::new(),
             settings,
@@ -699,7 +715,7 @@ impl Hosts {
     /// Safely insert into the HostContainer. Filters the addresses first before storing and
     /// notifies the subscriber. Must be called when first receiving greylist addresses.
     pub async fn insert(&self, color: HostColor, addrs: &[(Url, u64)]) {
-        debug!(target: "net::hosts:insert()", "[START] node={}", self.settings.node_id);
+        trace!(target: "net::hosts:insert()", "[START]");
 
         // First filter these address to ensure this peer doesn't exist in our black, gold or
         // whitelist and apply transport filtering.
@@ -714,17 +730,17 @@ impl Hosts {
         for (i, (addr, last_seen)) in filtered_addrs.iter().enumerate() {
             if self.try_register(addr.clone(), HostState::Insert).await.is_err() {
                 debug!(target: "net::hosts::store_or_update()",
-            "We are already trying to insert {}. Skipping...", addr);
+            "We are already tracking {}. Skipping...", addr);
                 continue
             }
 
             addrs_len += i + 1;
-            self.container.store_or_update(color.clone(), addr.clone(), last_seen.clone()).await;
+            self.container.store_or_update(color.clone(), addr.clone(), *last_seen).await;
             self.unregister(addr).await;
         }
 
         self.store_subscriber.notify(addrs_len).await;
-        debug!(target: "net::hosts:insert()", "[END] node={}", self.settings.node_id);
+        trace!(target: "net::hosts:insert()", "[END]");
     }
 
     /// Try to update the registry. If the host already exists, try to update its state.
@@ -743,6 +759,7 @@ impl Hosts {
                 HostState::Insert => current_state.try_insert(),
                 HostState::Refine => current_state.try_refine(),
                 HostState::Connect => current_state.try_connect(),
+                HostState::Suspend => current_state.try_suspend(),
                 HostState::Connected(c) => current_state.try_connected(c),
                 HostState::Move => current_state.try_move(),
             };
@@ -796,14 +813,27 @@ impl Hosts {
         let registry = self.registry.read().await;
         let mut channels = Vec::new();
 
-        for (_, value) in registry.iter() {
-            if let HostState::Connected(c) = value {
+        for (_, state) in registry.iter() {
+            if let HostState::Connected(c) = state {
                 channels.push(c.clone());
             }
         }
         channels
     }
 
+    /// Returns the list of connected channels.
+    pub async fn suspended(&self) -> Vec<Url> {
+        let registry = self.registry.read().await;
+        let mut addrs = Vec::new();
+
+        for (url, state) in registry.iter() {
+            if let HostState::Suspend = state {
+                addrs.push(url.clone());
+            }
+        }
+        addrs
+    }
+
     /// Retrieve a random connected channel
     pub async fn random_channel(&self) -> ChannelPtr {
         let channels = self.channels().await;
@@ -965,10 +995,15 @@ impl Hosts {
 
         match destination {
             // Downgrade to grey. Remove from white and gold.
+            // TODO: doc
             HostColor::Grey => {
                 self.container.remove_if_exists(HostColor::Gold, addr).await;
                 self.container.remove_if_exists(HostColor::White, addr).await;
                 self.container.store_or_update(HostColor::Grey, addr.clone(), last_seen).await;
+
+                // This should never panic.
+                self.try_register(addr.clone(), HostState::Suspend).await.unwrap();
+                return
             }
 
             // Remove from Greylist, add to Whitelist. Called by the Refinery.
@@ -1007,30 +1042,6 @@ impl Hosts {
         // stuck in the Moving state.
         self.unregister(addr).await;
     }
-
-    /// Quarantine a peer.
-    /// If they've been quarantined for more than a configured limit, move to greylist.
-    pub async fn quarantine(&self, addr: &Url, last_seen: u64) {
-        debug!(target: "net::hosts::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);
-            if *retries == self.settings.hosts_quarantine_limit {
-                debug!(target: "net::hosts::quarantine()",
-                "Reached quarantine limited after {:?}", timer.elapsed());
-                drop(q);
-
-                debug!(target: "net::hosts::quarantine()", "Moving to greylist {}", addr);
-                self.move_host(addr, last_seen, HostColor::Grey).await;
-            }
-        } else {
-            debug!(target: "net::hosts::quarantine()", "Added peer {} to quarantine", addr);
-            q.insert(addr.clone(), 0);
-        }
-    }
 }
 
 #[cfg(test)]

src/net/session/manual_session.rs

@@ -32,7 +32,7 @@
 use std::{sync::Arc, time::UNIX_EPOCH};
 
 use async_trait::async_trait;
-use log::{info, warn};
+use log::{debug, info, warn};
 use smol::lock::Mutex;
 use url::Url;
 
@@ -116,7 +116,7 @@ impl ManualSession {
 
             if let Err(e) = self.p2p().hosts().try_register(addr.clone(), HostState::Connect).await
             {
-                warn!(target: "net::manual_session", "{}", e);
+                debug!(target: "net::manual_session", "{} addr={}", e, addr.clone());
             }
 
             match connector.connect(&addr).await {

src/net/session/outbound_session.rs

@@ -445,7 +445,7 @@ impl Slot {
                 );
 
                 // At this point we failed to connect. We'll downgrade this peer now.
-                self.p2p().hosts().quarantine(&addr, last_seen).await;
+                self.p2p().hosts().move_host(&addr, last_seen, HostColor::Grey).await;
 
                 // Notify that channel processing failed
                 self.session().channel_subscriber.notify(Err(Error::ConnectFailed)).await;

src/net/settings.rs

@@ -62,7 +62,7 @@ pub struct Settings {
     pub channel_heartbeat_interval: u64,
     /// Allow localnet hosts
     pub localnet: bool,
-    /// Delete a peer from hosts if they've been quarantined N times
+    /// Downgrade a peer to greylist if they've been quarantined N times
     pub hosts_quarantine_limit: usize,
     /// Cooling off time for peer discovery when unsuccessful
     pub outbound_peer_discovery_cooloff_time: u64,
@@ -76,6 +76,9 @@ pub struct Settings {
     pub white_connection_percent: usize,
     /// Number of anchorlist connections
     pub anchor_connection_count: usize,
+    /// Number of seconds with no connections after which refinery
+    /// process is paused.
+    pub time_with_no_connections: u64,
 }
 
 impl Default for Settings {
@@ -103,9 +106,10 @@ impl Default for Settings {
             outbound_peer_discovery_cooloff_time: 30,
             outbound_peer_discovery_attempt_time: 5,
             hostlist: "/dev/null".to_string(),
-            greylist_refinery_interval: 5,
+            greylist_refinery_interval: 15,
             white_connection_percent: 90,
             anchor_connection_count: 2,
+            time_with_no_connections: 30,
         }
     }
 }
@@ -183,7 +187,7 @@ pub struct SettingsOpt {
     #[structopt(long)]
     pub localnet: bool,
 
-    /// Delete a peer from hosts if they've been quarantined N times
+    /// Downgrade a peer to greylist if they've been quarantined N times
     #[structopt(skip)]
     pub hosts_quarantine_limit: Option<usize>,
 
@@ -211,6 +215,11 @@ pub struct SettingsOpt {
     /// Number of anchorlist connections
     #[structopt(skip)]
     pub anchor_connection_count: Option<usize>,
+
+    /// Number of seconds with no connections after which refinery
+    /// process is paused.
+    #[structopt(skip)]
+    pub time_with_no_connections: Option<u64>,
 }
 
 impl From<SettingsOpt> for Settings {
@@ -258,6 +267,9 @@ impl From<SettingsOpt> for Settings {
             anchor_connection_count: opt
                 .anchor_connection_count
                 .unwrap_or(def.anchor_connection_count),
+            time_with_no_connections: opt
+                .time_with_no_connections
+                .unwrap_or(def.time_with_no_connections),
         }
     }
 }