protocol_version: bugfix

The Error result of the version exchange was not being dealt with.
Instead we were only handling when the version exchange timed out.

There was also a bug in the handling of the failure in session/mod.rs
where the error case was similarly ignored. We now explicitly return
with an error in the case that the version exchange fails or times out.

src/net/protocol/protocol_version.rs

@@ -21,9 +21,12 @@ use std::{
     time::{Duration, UNIX_EPOCH},
 };
 
-use futures::future::join_all;
+use futures::{
+    future::{join_all, select, Either},
+    pin_mut,
+};
 use log::{debug, error};
-use smol::Executor;
+use smol::{Executor, Timer};
 
 use super::super::{
     channel::ChannelPtr,
@@ -31,7 +34,7 @@ use super::super::{
     message_subscriber::MessageSubscription,
     settings::SettingsPtr,
 };
-use crate::{system::timeout::timeout, Error, Result};
+use crate::{Error, Result};
 
 /// Implements the protocol version handshake sent out by nodes at
 /// the beginning of a connection.
@@ -62,29 +65,44 @@ impl ProtocolVersion {
     /// version ack.
     pub async fn run(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
         debug!(target: "net::protocol_version::run()", "START => address={}", self.channel.address());
-        // Start timer
-        // Send version, wait for verack
-        // Wait for version, send verack
-        // Fin.
-        let result = timeout(
-            Duration::from_secs(self.settings.channel_handshake_timeout),
-            self.clone().exchange_versions(executor),
-        )
-        .await;
-
-        if let Err(e) = result {
-            error!(
-                target: "net::protocol_version::run()",
-                "[P2P] Version Exchange failed [{}]: {}",
-                self.channel.address(), e,
-            );
-
-            self.channel.stop().await;
-            return Err(Error::ChannelTimeout)
+        let timeout = Timer::after(Duration::from_secs(self.settings.channel_handshake_timeout));
+        let version = self.clone().exchange_versions(executor);
+
+        pin_mut!(timeout);
+        pin_mut!(version);
+
+        // Run timer and version exchange at the same time. Either deal
+        // with the success or failure of the version exchange or
+        // time out.
+        match select(version, timeout).await {
+            Either::Left((Ok(_), _)) => {
+                debug!(target: "net::protocol_version::run()", "END => address={}",
+                self.channel.address());
+
+                Ok(())
+            }
+            Either::Left((Err(e), _)) => {
+                error!(
+                    target: "net::protocol_version::run()",
+                    "[P2P] Version Exchange failed [{}]: {}",
+                    self.channel.address(), e,
+                );
+
+                self.channel.stop().await;
+                Err(e)
+            }
+
+            Either::Right((_, _)) => {
+                error!(
+                    target: "net::protocol_version::run()",
+                    "[P2P] Version Exchange timed out [{}]",
+                    self.channel.address(),
+                );
+
+                self.channel.stop().await;
+                Err(Error::ChannelTimeout)
+            }
         }
-
-        debug!(target: "net::protocol_version::run()", "END => address={}", self.channel.address());
-        Ok(())
     }
 
     /// Send and receive version information

src/net/session/mod.rs

@@ -134,6 +134,8 @@ pub trait Session: Sync {
             Err(e) => {
                 debug!(target: "net::session::register_channel()",
                 "Handshake error {} {}", e, channel.clone().address());
+
+                return Err(e)
             }
         }
 
@@ -162,31 +164,34 @@ pub trait Session: Sync {
         executor: Arc<Executor<'_>>,
     ) -> Result<()> {
         // Perform handshake
-        protocol_version.run(executor.clone()).await?;
-
-        // Upgrade to goldlist if this is a outbound session.
-        if self.type_id() & SESSION_OUTBOUND != 0 {
-            debug!(
-                target: "net::session::perform_handshake_protocols()",
-                "Upgrading {}", channel.address(),
-            );
-
-            let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
-            self.p2p()
-                .hosts()
-                .move_host(channel.address(), last_seen, HostColor::Gold)
-                .await
-                .unwrap();
+        match protocol_version.run(executor.clone()).await {
+            Ok(()) => {
+                // Upgrade to goldlist if this is a outbound session.
+                if self.type_id() & SESSION_OUTBOUND != 0 {
+                    debug!(
+                        target: "net::session::perform_handshake_protocols()",
+                        "Upgrading {}", channel.address(),
+                    );
+
+                    let last_seen = UNIX_EPOCH.elapsed().unwrap().as_secs();
+                    self.p2p()
+                        .hosts()
+                        .move_host(channel.address(), last_seen, HostColor::Gold)
+                        .await
+                        .unwrap();
+                }
+
+                // Attempt to add channel to registry
+                self.p2p().hosts().register_channel(channel.clone()).await;
+
+                // Subscribe to stop, so we can remove from registry
+                executor.spawn(remove_sub_on_stop(self.p2p(), channel, self.type_id())).detach();
+
+                // Channel is ready for use
+                Ok(())
+            }
+            Err(e) => return Err(e),
         }
-
-        // Attempt to add channel to registry
-        self.p2p().hosts().register_channel(channel.clone()).await;
-
-        // Subscribe to stop, so we can remove from registry
-        executor.spawn(remove_sub_on_stop(self.p2p(), channel, self.type_id())).detach();
-
-        // Channel is ready for use
-        Ok(())
     }
 
     /// Returns a pointer to the p2p network interface