Merge 0fa1687 into b76771e

.github/workflows/master.yml

@@ -13,8 +13,6 @@ on:
       - master
 
 env:
-  # Run all cargo commands with --verbose.
-  CARGO_TERM_VERBOSE: true
   RUST_BACKTRACE: 1
 
 jobs:

.github/workflows/pr.yml

@@ -8,8 +8,6 @@ name: PR
 on: pull_request
 
 env:
-  # Run all cargo commands with --verbose.
-  CARGO_TERM_VERBOSE: true
   RUST_BACKTRACE: 1
   # Deny all compiler warnings.
   RUSTFLAGS: "-D warnings -C opt-level=2 -C codegen-units=8"
@@ -93,7 +91,7 @@ jobs:
       - name: rust-tarpaulin code coverage check
         uses: actions-rs/tarpaulin@v0.1
         with:
-          args: '-v --release --out Lcov -- --skip echo_service' # test would timeout on CI
+          args: '-v --release --out Lcov -- --skip echo_service --skip multiple_connections_with_many_larger_concurrent_messages' # tests would timeout on CI
       - name: Push code coverage results to coveralls.io
         uses: coverallsapp/github-action@master
         with:

.gitignore

@@ -3,3 +3,4 @@
 Cargo.lock
 *.*~
 .idea
+*.log

src/api.rs

@@ -10,7 +10,8 @@
 use super::{
     bootstrap_cache::BootstrapCache,
     config::{Config, SerialisableCertificate},
-    endpoint::{DisconnectionEvents, Endpoint, IncomingConnections, IncomingMessages},
+    connections::DisconnectionEvents,
+    endpoint::{Endpoint, IncomingConnections, IncomingMessages},
     error::{Error, Result},
     peer_config::{self, DEFAULT_IDLE_TIMEOUT_MSEC, DEFAULT_KEEP_ALIVE_INTERVAL_MSEC},
 };

src/config.rs

@@ -80,6 +80,9 @@ pub struct Config {
     /// Duration of a UPnP port mapping.
     #[structopt(long)]
     pub upnp_lease_duration: Option<u32>,
+    /// Duration to wait before retrying to resend / reconnect
+    #[structopt(long, default_value = "500")]
+    pub retry_interval: u64,
 }
 
 /// To be used to read and write our certificate and private key to disk esp. as a part of our

src/connection_deduplicator.rs

@@ -73,7 +73,7 @@ impl ConnectionDeduplicator {
 pub(crate) enum Error {
     #[error("Connect error")]
     Connect(#[from] quinn::ConnectError),
-    #[error("Connection error")]
+    #[error("Quinn Connection error during deduplication")]
     Connection(#[from] quinn::ConnectionError),
 }
 

src/connections.rs

@@ -17,9 +17,9 @@ use super::{
 use bytes::Bytes;
 use futures::{future, stream::StreamExt};
 use std::net::SocketAddr;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc::{Receiver, Sender};
 use tokio::time::{timeout, Duration};
-use tracing::{debug, error, trace, warn};
+use tracing::{trace, warn};
 
 /// Connection instance to a node which can be used to send messages to it
 #[derive(Clone)]
@@ -28,6 +28,17 @@ pub(crate) struct Connection {
     remover: ConnectionRemover,
 }
 
+/// Disconnection events, and the result that led to disconnection.
+pub struct DisconnectionEvents(pub Receiver<SocketAddr>);
+
+/// Disconnection
+impl DisconnectionEvents {
+    /// Blocks until there is a disconnection event and returns the address of the disconnected peer
+    pub async fn next(&mut self) -> Option<SocketAddr> {
+        self.0.recv().await
+    }
+}
+
 impl Connection {
     pub(crate) fn new(quic_conn: quinn::Connection, remover: ConnectionRemover) -> Self {
         Self { quic_conn, remover }
@@ -167,10 +178,7 @@ pub(super) fn listen_for_incoming_connections(
                         );
                     }
                     Err(err) => {
-                        error!(
-                            "An incoming connection failed because of an error: {:?}",
-                            err
-                        );
+                        warn!("An incoming connection failed because of: {:?}", err);
                     }
                 },
                 None => {
@@ -193,15 +201,22 @@ pub(super) fn listen_for_incoming_messages(
 ) {
     let src = *remover.remote_addr();
     let _ = tokio::spawn(async move {
-        debug!("qp2p another incoming listerner spawned");
-        let _ = future::join(
+        match future::try_join(
             read_on_uni_streams(&mut uni_streams, src, message_tx.clone()),
             read_on_bi_streams(&mut bi_streams, src, message_tx, &endpoint),
         )
-        .await;
+        .await
+        {
+            Ok(_) => {
+                let _ = disconnection_tx.send(src).await;
+            }
+            Err(error) => {
+                trace!("Issue on stream reading from: {:?} :: {:?}", src, error);
+                let _ = disconnection_tx.send(src).await;
+            }
+        }
 
         trace!("The connection to {:?} has been terminated.", src);
-        let _ = disconnection_tx.send(src).await;
         remover.remove().await;
     });
 }
@@ -211,30 +226,35 @@ async fn read_on_uni_streams(
     uni_streams: &mut quinn::IncomingUniStreams,
     peer_addr: SocketAddr,
     message_tx: Sender<(SocketAddr, Bytes)>,
-) {
+) -> Result<()> {
     while let Some(result) = uni_streams.next().await {
         match result {
-            Err(quinn::ConnectionError::ApplicationClosed { .. }) => {
-                trace!("Connection terminated by peer {:?}.", peer_addr);
-                break;
+            Err(quinn::ConnectionError::ConnectionClosed(_))
+            | Err(quinn::ConnectionError::ApplicationClosed(_)) => {
+                trace!("Connection closed by peer {:?}.", peer_addr);
+                return Err(Error::QuinnConnectionClosed);
             }
             Err(err) => {
                 warn!(
-                    "Failed to read incoming message on uni-stream for peer {:?} with error: {:?}",
+                    "Failed to read incoming message on uni-stream for peer {:?} with: {:?}",
                     peer_addr, err
                 );
-                break;
+                return Err(Error::from(err));
             }
             Ok(mut recv) => loop {
                 match read_bytes(&mut recv).await {
                     Ok(WireMsg::UserMsg(bytes)) => {
+                        trace!("bytes received fine from: {:?} ", peer_addr);
                         let _ = message_tx.send((peer_addr, bytes)).await;
                     }
-                    Ok(msg) => error!("Unexpected message type: {:?}", msg),
-                    Err(Error::StreamRead(quinn::ReadExactError::FinishedEarly)) => break,
+                    Ok(msg) => warn!("Unexpected message type: {:?}", msg),
+                    Err(Error::StreamRead(quinn::ReadExactError::FinishedEarly)) => {
+                        warn!("Stream read finished early");
+                        break;
+                    }
                     Err(err) => {
-                        error!(
-                            "Failed reading from a uni-stream for peer {:?} with error: {:?}",
+                        warn!(
+                            "Failed reading from a uni-stream for peer {:?} with: {:?}",
                             peer_addr, err
                         );
                         break;
@@ -243,6 +263,7 @@ async fn read_on_uni_streams(
             },
         }
     }
+    Ok(())
 }
 
 // Read messages sent by peer in a bidirectional stream.
@@ -251,20 +272,20 @@ async fn read_on_bi_streams(
     peer_addr: SocketAddr,
     message_tx: Sender<(SocketAddr, Bytes)>,
     endpoint: &Endpoint,
-) {
+) -> Result<()> {
     while let Some(result) = bi_streams.next().await {
         match result {
-            Err(quinn::ConnectionError::ApplicationClosed { .. })
-            | Err(quinn::ConnectionError::ConnectionClosed { .. }) => {
-                trace!("Connection terminated by peer {:?}.", peer_addr);
-                break;
+            Err(quinn::ConnectionError::ConnectionClosed(_))
+            | Err(quinn::ConnectionError::ApplicationClosed(_)) => {
+                trace!("Connection closed by peer {:?}.", peer_addr);
+                return Err(Error::QuinnConnectionClosed);
             }
             Err(err) => {
                 warn!(
-                    "Failed to read incoming message on bi-stream for peer {:?} with error: {:?}",
+                    "Failed to read incoming message on bi-stream for peer {:?} with: {:?}",
                     peer_addr, err
                 );
-                break;
+                return Err(Error::from(err));
             }
             Ok((mut send, mut recv)) => loop {
                 match read_bytes(&mut recv).await {
@@ -273,10 +294,12 @@ async fn read_on_bi_streams(
                     }
                     Ok(WireMsg::EndpointEchoReq) => {
                         if let Err(error) = handle_endpoint_echo_req(peer_addr, &mut send).await {
-                            error!(
-                                "Failed to handle Echo Request for peer {:?} with error: {:?}",
+                            warn!(
+                                "Failed to handle Echo Request for peer {:?} with: {:?}",
                                 peer_addr, error
                             );
+
+                            return Err(error);
                         }
                     }
                     Ok(WireMsg::EndpointVerificationReq(address_sent)) => {
@@ -288,19 +311,24 @@ async fn read_on_bi_streams(
                         )
                         .await
                         {
-                            error!("Failed to handle Endpoint verification request for peer {:?} with error: {:?}", peer_addr, error);
+                            warn!("Failed to handle Endpoint verification request for peer {:?} with: {:?}", peer_addr, error);
+
+                            return Err(error);
                         }
                     }
                     Ok(msg) => {
-                        error!(
+                        warn!(
                             "Unexpected message type from peer {:?}: {:?}",
                             peer_addr, msg
                         );
                     }
-                    Err(Error::StreamRead(quinn::ReadExactError::FinishedEarly)) => break,
+                    Err(Error::StreamRead(quinn::ReadExactError::FinishedEarly)) => {
+                        warn!("Stream finished early");
+                        break;
+                    }
                     Err(err) => {
-                        error!(
-                            "Failed reading from a bi-stream for peer {:?} with error: {:?}",
+                        warn!(
+                            "Failed reading from a bi-stream for peer {:?} with: {:?}",
                             peer_addr, err
                         );
                         break;
@@ -309,6 +337,8 @@ async fn read_on_bi_streams(
             },
         }
     }
+
+    Ok(())
 }
 
 async fn handle_endpoint_echo_req(