net: rm SelfHandshake

Previously we would do a version exchange with our own external
address and perform a time stamp when the version exchange occured
successfully. This tuple of (external_addr, last_seen) would be shared
with other nodes in ProtocolAddr and ProtocolSeed.

This commit removes SelfHandshake. Instead, we simply set the last_seen
to now when we send our addrs in ProtocolAddr/Seed.

Rationale:

1. Doing a version exchange with ourselves requires creating two
   channels, an inbound and an outbound channel. This causes unintended
   message broadcasting to our own addrs in p2p.broadcast(), and race
   conditions when one of the two channels disconnects.

2. Being reachable by our own node does not guarantee our node is
   reachable by other nodes, so it's not a reliable method for
   determining whether an address is accessible.

3. As the external addrs ends up on the greylist of other peers, the
   work is essentially redundant, since they will also ping the address
   via their GreylistRefinery.

One notable impact of this change is that setting the timestamp to now
means that external addresses from other nodes will be placed on the top
of our greylist. Hostlists are sorted by last_seen, with the most
recently seen timestamp on the top of the list. However, this does not
impact the refinery ordering because refinery selects peers to refine
randomly.

The one area this could impact is outbound session, since we
search for an address to connect to by looping through the hostlists
(incl. greylist) starting from index 0 (i.e. the most recently seen
node). Therefore, this commit increases the probability of connecting to
a node that has recently sent its address to us (or marking that node as
"Suspend" if we cannot connect to it).

src/net/protocol/protocol_address.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::{sync::Arc, time::UNIX_EPOCH};
 
 use async_trait::async_trait;
 use log::debug;
@@ -64,7 +64,6 @@ pub struct ProtocolAddress {
     hosts: HostsPtr,
     settings: SettingsPtr,
     jobsman: ProtocolJobsManagerPtr,
-    p2p: P2pPtr,
 }
 
 const PROTO_NAME: &str = "ProtocolAddress";
@@ -98,7 +97,6 @@ impl ProtocolAddress {
             hosts,
             jobsman: ProtocolJobsManager::new(PROTO_NAME, channel),
             settings,
-            p2p,
         })
     }
 
@@ -235,14 +233,8 @@ impl ProtocolAddress {
         }
     }
 
-    /// Send our own external addresses over a channel. Get the latest
-    /// last_seen field from RefineSession, and send it along with our
-    /// external address.
-    ///
-    /// If our external address is misconfigured, send an empty vector.
-    /// If we have reached our inbound connection limit, send our external
-    /// address with a `last_seen` field that corresponds to the last time
-    /// we could receive inbound connections.
+    /// Send our own external addresses over a channel. Set the
+    /// last_seen field to now.
     async fn send_my_addrs(self: Arc<Self>) -> Result<()> {
         debug!(
             target: "net::protocol_address::send_my_addrs()",
@@ -263,11 +255,11 @@ impl ProtocolAddress {
         }
 
         let mut addrs = vec![];
-        let refinery = self.p2p.session_refine();
-        for (addr, last_seen) in refinery.self_handshake.addrs.lock().await.iter() {
-            addrs.push((addr.clone(), *last_seen));
-        }
 
+        for addr in self.settings.external_addrs.clone() {
+            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+            addrs.push((addr, last_seen));
+        }
         debug!(target: "net::protocol_address::send_my_addrs()",
         "Broadcasting {} addresses", addrs.len());
         let ext_addr_msg = AddrsMessage { addrs };

src/net/protocol/protocol_seed.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::sync::Arc;
+use std::{sync::Arc, time::UNIX_EPOCH};
 
 use async_trait::async_trait;
 use log::debug;
@@ -41,7 +41,6 @@ pub struct ProtocolSeed {
     hosts: HostsPtr,
     settings: SettingsPtr,
     addr_sub: MessageSubscription<AddrsMessage>,
-    p2p: P2pPtr,
 }
 
 const PROTO_NAME: &str = "ProtocolSeed";
@@ -56,17 +55,11 @@ impl ProtocolSeed {
         let addr_sub =
             channel.subscribe_msg::<AddrsMessage>().await.expect("Missing addr dispatcher!");
 
-        Arc::new(Self { channel, hosts, settings, addr_sub, p2p })
+        Arc::new(Self { channel, hosts, settings, addr_sub })
     }
 
-    /// Send our own external addresses over a channel. Get the latest
-    /// last_seen field from InboundSession, and send it along with our
-    /// external address.
-    ///
-    /// If our external address is misconfigured, send an empty vector.
-    /// If we have reached our inbound connection limit, send our external
-    /// address with a `last_seen` field that corresponds to the last time
-    /// we could receive inbound connections.
+    /// Send our own external addresses over a channel. Set the
+    /// last_seen field to now.
     pub async fn send_my_addrs(&self) -> Result<()> {
         debug!(target: "net::protocol_seed::send_my_addrs()",
         "[START] channel address={}", self.channel.address());
@@ -78,9 +71,10 @@ impl ProtocolSeed {
         }
 
         let mut addrs = vec![];
-        let refinery = self.p2p.session_refine();
-        for (addr, last_seen) in refinery.self_handshake.addrs.lock().await.iter() {
-            addrs.push((addr.clone(), *last_seen));
+
+        for addr in self.settings.external_addrs.clone() {
+            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+            addrs.push((addr, last_seen));
         }
 
         debug!(target: "net::protocol_seed::send_my_addrs()",

src/net/session/refine_session.rs

@@ -16,26 +16,22 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-//! `RefineSession` manages two processes, the `GreylistRefinery`, which
-//! periodically pings entries on the greylist and updates them to whitelist
-//! if active, and `SelfHandshake`, which periodically pings our own external
-//! addresses to ensure they are active before broadcasting to the network.
+//! `RefineSession` manages the `GreylistRefinery`, which randomly selects
+//! entries on the greylist and updates them to whitelist if active,
 //!
-//! Both processes make use of a `RefineSession` method called
+//! `GreylistRefinery` makes use of a `RefineSession` method called
 //! `handshake_node()`, which uses a `Connector` to establish a `Channel` with
 //! a provided address, and then does a version exchange across the channel
 //! (`perform_handshake_protocols`). `handshake_node()` can either succeed,
 //! fail, or timeout.
 
 use std::{
-    collections::HashMap,
     sync::Arc,
     time::{Duration, Instant, UNIX_EPOCH},
 };
 
 use async_trait::async_trait;
 use log::{debug, warn};
-use smol::lock::Mutex;
 use url::Url;
 
 use super::super::p2p::{P2p, P2pPtr};
@@ -59,28 +55,17 @@ pub struct RefineSession {
 
     /// Task that periodically checks entries in the greylist.
     pub(in crate::net) refinery: Arc<GreylistRefinery>,
-
-    /// Task that periodically checks our external addresses.
-    pub(in crate::net) self_handshake: Arc<SelfHandshake>,
 }
 
 impl RefineSession {
     pub fn new() -> RefineSessionPtr {
-        let self_ = Arc::new(Self {
-            p2p: LazyWeak::new(),
-            refinery: GreylistRefinery::new(),
-            self_handshake: SelfHandshake::new(),
-        });
-        self_.self_handshake.session.init(self_.clone());
+        let self_ = Arc::new(Self { p2p: LazyWeak::new(), refinery: GreylistRefinery::new() });
         self_.refinery.session.init(self_.clone());
         self_
     }
 
     /// Start the refinery and self handshake processes.
     pub(crate) async fn start(self: Arc<Self>) {
-        debug!(target: "net::refine_session", "Starting self handshake process");
-        self.self_handshake.clone().start().await;
-
         match self.p2p().hosts().container.load_all(&self.p2p().settings().hostlist).await {
             Ok(()) => {
                 debug!(target: "net::refine_session::start()", "Load hosts successful!");
@@ -105,9 +90,6 @@ impl RefineSession {
 
     /// Stop the refinery and self handshake processes.
     pub(crate) async fn stop(&self) {
-        debug!(target: "net::refine_session", "Stopping self handshake process");
-        self.self_handshake.clone().stop().await;
-
         debug!(target: "net::refine_session", "Stopping refinery process");
         self.refinery.clone().stop().await;
 
@@ -329,105 +311,3 @@ impl GreylistRefinery {
         self.session().p2p()
     }
 }
-
-/// Periodically try to do a version exchange with our own external
-/// addresses. If the version exchange is successful, take a timestamp and
-/// save it along with the external addresses. Each address along with its
-/// timestamp (the `last_seen` data field) is sent in to other nodes in
-/// ProtocolAddr and ProtocolSeed.
-///
-/// On first run, SelfHandshake will immediately conduct a version exchange
-/// with our external addresses, and if successful update the last_seen
-/// field. The process will wait Settings::self_handshake_interval before retrying.
-///
-/// There are two situations in which this can fail:
-///
-///     1. If our external address is misconfigured
-///     2. If we have reached our inbound connection limit.
-///
-/// If our external address is misconfigured, doing a version exchange
-/// with ourselves will not work and so the external addresses will not
-/// be shared with other nodes.
-///
-/// If we have reached our inbound connection limit, the external address
-/// will continue to be broadcast with an older `last_seen` (from before
-/// our inbound connection was reached).
-pub struct SelfHandshake {
-    process: StoppableTaskPtr,
-    session: LazyWeak<RefineSession>,
-    pub(in crate::net) addrs: Mutex<HashMap<Url, u64>>,
-}
-
-impl SelfHandshake {
-    fn new() -> Arc<Self> {
-        Arc::new(Self {
-            process: StoppableTask::new(),
-            session: LazyWeak::new(),
-            addrs: Mutex::new(HashMap::new()),
-        })
-    }
-
-    async fn start(self: Arc<Self>) {
-        let ex = self.session().p2p().executor();
-        self.process.clone().start(
-            async move {
-                self.run().await;
-                unreachable!();
-            },
-            // Ignore stop handler
-            |_| async {},
-            Error::NetworkServiceStopped,
-            ex,
-        );
-    }
-
-    async fn stop(self: Arc<Self>) {
-        self.process.stop().await
-    }
-
-    async fn run(self: Arc<Self>) {
-        let settings = self.session().p2p().settings();
-        let external_addrs = settings.external_addrs.clone();
-        let mut current_attempt = 0;
-
-        loop {
-            if current_attempt >= 1 {
-                sleep(settings.self_handshake_interval).await;
-            }
-
-            // Only proceed if the external address is configured.
-            if external_addrs.is_empty() {
-                current_attempt += 1;
-                continue
-            }
-
-            for addr in external_addrs.iter() {
-                debug!(target: "net::refine_session::self_handshake",
-                "Attempting a version exchange addr={}", addr);
-
-                if self.session().handshake_node(addr.clone(), self.session().p2p()).await {
-                    debug!(target: "net::refine_session::self_handshake",
-                    "Version exchange successful! Updating last seen addr={}", addr);
-                    let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
-                    let mut addrs = self.addrs.lock().await;
-
-                    if addrs.contains_key(addr) {
-                        let val = addrs.get_mut(addr).unwrap();
-                        *val = last_seen;
-                    }
-                    addrs.insert(addr.clone(), last_seen);
-                } else {
-                    // Either our external addr is invalid or our max inbound
-                    // connection count has been reached.
-                    warn!(target: "net::refine_session::self_handshake",
-                    "Version exchange failed! addr={}", addr);
-                }
-            }
-            current_attempt += 1;
-        }
-    }
-
-    fn session(&self) -> RefineSessionPtr {
-        self.session.upgrade()
-    }
-}

src/net/settings.rs

@@ -68,8 +68,6 @@ pub struct Settings {
     pub hostlist: String,
     /// Pause interval within greylist refinery process
     pub greylist_refinery_interval: u64,
-    /// Pause interval before redoing a self-handshake
-    pub self_handshake_interval: u64,
     /// Percent of connections to come from the whitelist
     pub white_connect_count: u32,
     /// Number of goldlist connections
@@ -106,7 +104,6 @@ impl Default for Settings {
             outbound_peer_discovery_attempt_time: 5,
             hostlist: "/dev/null".to_string(),
             greylist_refinery_interval: 15,
-            self_handshake_interval: 600,
             white_connect_count: 90,
             gold_connect_count: 2,
             time_with_no_connections: 30,
@@ -201,10 +198,6 @@ pub struct SettingsOpt {
     #[structopt(skip)]
     pub greylist_refinery_interval: Option<u64>,
 
-    /// Pause interval before redoing a self-handshake
-    #[structopt(skip)]
-    pub self_handshake_interval: Option<u64>,
-
     /// Number of whitelist connections
     #[structopt(skip)]
     pub white_connect_count: Option<u32>,
@@ -260,9 +253,6 @@ impl From<SettingsOpt> for Settings {
             greylist_refinery_interval: opt
                 .greylist_refinery_interval
                 .unwrap_or(def.greylist_refinery_interval),
-            self_handshake_interval: opt
-                .self_handshake_interval
-                .unwrap_or(def.self_handshake_interval),
             white_connect_count: opt.white_connect_count.unwrap_or(def.white_connect_count),
             gold_connect_count: opt.gold_connect_count.unwrap_or(def.gold_connect_count),
             time_with_no_connections: opt