Bugfix/closing tcp client connections on drop (#167)

* Filtering out early eof errors

* Wrapping ConnectionManager in Abortable

* Decreased pathchecker logging level

* Client aborting all connection futures on drop

* Moved AbortHandle to connection_managers HashMap to couple them closer together

common/clients/multi-tcp-client/src/connection_manager/mod.rs

@@ -1,6 +1,7 @@
 use crate::connection_manager::reconnector::ConnectionReconnector;
 use crate::connection_manager::writer::ConnectionWriter;
 use futures::channel::{mpsc, oneshot};
+use futures::future::{abortable, AbortHandle};
 use futures::task::Poll;
 use futures::{AsyncWriteExt, StreamExt};
 use log::*;
@@ -34,6 +35,12 @@ pub(crate) struct ConnectionManager<'a> {
     state: ConnectionState<'a>,
 }
 
+impl<'a> Drop for ConnectionManager<'a> {
+    fn drop(&mut self) {
+        debug!("Connection manager to {:?} is being dropped", self.address)
+    }
+}
+
 impl<'a> ConnectionManager<'static> {
     pub(crate) async fn new(
         address: SocketAddr,
@@ -68,24 +75,29 @@ impl<'a> ConnectionManager<'static> {
         }
     }
 
-    /// consumes Self and returns channel for communication
-    pub(crate) fn start(mut self, handle: &Handle) -> ConnectionManagerSender {
-        let sender_clone = self.conn_tx.clone();
-        handle.spawn(async move {
-            while let Some(msg) = self.conn_rx.next().await {
-                let (msg_content, res_ch) = msg;
-                let res = self.handle_new_message(msg_content).await;
-                if let Some(res_ch) = res_ch {
-                    if let Err(e) = res_ch.send(res) {
-                        error!(
-                            "failed to send response on the channel to the caller! - {:?}",
-                            e
-                        );
-                    }
+    async fn run(mut self) {
+        while let Some(msg) = self.conn_rx.next().await {
+            let (msg_content, res_ch) = msg;
+            let res = self.handle_new_message(msg_content).await;
+            if let Some(res_ch) = res_ch {
+                if let Err(e) = res_ch.send(res) {
+                    error!(
+                        "failed to send response on the channel to the caller! - {:?}",
+                        e
+                    );
                 }
             }
-        });
-        sender_clone
+        }
+    }
+
+    /// consumes Self and returns channel for communication as well as an `AbortHandle`
+    pub(crate) fn start_abortable(self, handle: &Handle) -> (ConnectionManagerSender, AbortHandle) {
+        let sender_clone = self.conn_tx.clone();
+        let (abort_fut, abort_handle) = abortable(self.run());
+
+        handle.spawn(async move { abort_fut.await });
+
+        (sender_clone, abort_handle)
     }
 
     async fn handle_new_message(&mut self, msg: Vec<u8>) -> io::Result<()> {

common/clients/multi-tcp-client/src/lib.rs

@@ -1,5 +1,6 @@
 use crate::connection_manager::{ConnectionManager, ConnectionManagerSender};
 use futures::channel::oneshot;
+use futures::future::AbortHandle;
 use log::*;
 use std::collections::HashMap;
 use std::io;
@@ -28,7 +29,7 @@ impl Config {
 
 pub struct Client {
     runtime_handle: Handle,
-    connections_managers: HashMap<SocketAddr, ConnectionManagerSender>,
+    connections_managers: HashMap<SocketAddr, (ConnectionManagerSender, AbortHandle)>,
     maximum_reconnection_backoff: Duration,
     initial_reconnection_backoff: Duration,
 }
@@ -46,14 +47,19 @@ impl Client {
         }
     }
 
-    async fn start_new_connection_manager(&self, address: SocketAddr) -> ConnectionManagerSender {
-        ConnectionManager::new(
+    async fn start_new_connection_manager(
+        &mut self,
+        address: SocketAddr,
+    ) -> (ConnectionManagerSender, AbortHandle) {
+        let (sender, abort_handle) = ConnectionManager::new(
             address,
             self.initial_reconnection_backoff,
             self.maximum_reconnection_backoff,
         )
         .await
-        .start(&self.runtime_handle)
+        .start_abortable(&self.runtime_handle);
+
+        (sender, abort_handle)
     }
 
     // if wait_for_response is set to true, we will get information about any possible IO errors
@@ -71,24 +77,33 @@ impl Client {
                 address
             );
 
-            let new_manager_sender = self.start_new_connection_manager(address).await;
+            let (new_manager_sender, abort_handle) =
+                self.start_new_connection_manager(address).await;
             self.connections_managers
-                .insert(address, new_manager_sender);
+                .insert(address, (new_manager_sender, abort_handle));
         }
 
         let manager = self.connections_managers.get_mut(&address).unwrap();
 
         if wait_for_response {
             let (res_tx, res_rx) = oneshot::channel();
-            manager.unbounded_send((message, Some(res_tx))).unwrap();
+            manager.0.unbounded_send((message, Some(res_tx))).unwrap();
             res_rx.await.unwrap()
         } else {
-            manager.unbounded_send((message, None)).unwrap();
+            manager.0.unbounded_send((message, None)).unwrap();
             Ok(())
         }
     }
 }
 
+impl Drop for Client {
+    fn drop(&mut self) {
+        for (_, abort_handle) in self.connections_managers.values() {
+            abort_handle.abort()
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

common/healthcheck/src/path_check.rs

@@ -183,7 +183,7 @@ impl PathChecker {
             return;
         }
 
-        debug!("Checking path: {:?} ({})", path, iteration);
+        trace!("Checking path: {:?} ({})", path, iteration);
         let path_identifier = PathChecker::unique_path_key(path, self.check_id, iteration);
 
         // check if there is even any point in sending the packet

mixnode/src/node/listener.rs

@@ -48,6 +48,8 @@ async fn process_socket_connection(
                 return;
             }
             Ok(n) => {
+                // If I understand it correctly, this if should never be executed as if `read_exact`
+                // does not fill buffer, it will throw UnexpectedEof?
                 if n != sphinx::PACKET_SIZE {
                     warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
                     continue;
@@ -63,10 +65,16 @@ async fn process_socket_connection(
                 ))
             }
             Err(e) => {
-                warn!(
-                    "failed to read from socket. Closing the connection; err = {:?}",
-                    e
-                );
+                if e.kind() == io::ErrorKind::UnexpectedEof {
+                    debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
+                    Also closing the connection on this end.", socket.peer_addr())
+                } else {
+                    warn!(
+                        "failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
+                        socket.peer_addr(),
+                        e
+                    );
+                }
                 return;
             }
         };

sfw-provider/src/provider/client_handling/listener.rs

@@ -65,10 +65,16 @@ async fn process_socket_connection(
             // plus realistically it wouldn't really introduce any speed up
             Ok(n) => process_request(&mut socket, &buf[0..n], &mut request_processor).await,
             Err(e) => {
-                warn!(
-                    "failed to read from socket. Closing the connection; err = {:?}",
-                    e
-                );
+                if e.kind() == io::ErrorKind::UnexpectedEof {
+                    debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
+                    Also closing the connection on this end.", socket.peer_addr())
+                } else {
+                    warn!(
+                        "failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
+                        socket.peer_addr(),
+                        e
+                    );
+                }
                 return;
             }
         };

sfw-provider/src/provider/mix_handling/listener.rs

@@ -36,6 +36,8 @@ async fn process_socket_connection(
                 return;
             }
             Ok(n) => {
+                // If I understand it correctly, this if should never be executed as if `read_exact`
+                // does not fill buffer, it will throw UnexpectedEof?
                 if n != sphinx::PACKET_SIZE {
                     warn!("read data of different length than expected sphinx packet size - {} (expected {})", n, sphinx::PACKET_SIZE);
                     continue;
@@ -45,10 +47,16 @@ async fn process_socket_connection(
                 tokio::spawn(process_received_packet(buf, packet_processor.clone()))
             }
             Err(e) => {
-                warn!(
-                    "failed to read from socket. Closing the connection; err = {:?}",
-                    e
-                );
+                if e.kind() == io::ErrorKind::UnexpectedEof {
+                    debug!("Read buffer was not fully filled. Most likely the client ({:?}) closed the connection.\
+                    Also closing the connection on this end.", socket.peer_addr())
+                } else {
+                    warn!(
+                        "failed to read from socket (source: {:?}). Closing the connection; err = {:?}",
+                        socket.peer_addr(),
+                        e
+                    );
+                }
                 return;
             }
         };