Feature/gateway graceful shutdown (#2834)

* task dependency

* unifying some startup code and passing TaskClient around

* graceful shutdown handling for mix socket listener

* graceful shutdown handling for clients listener

* graceful shutdown handling for packet forwarding

* unified waiting for interrupt across binaries

* made 'validate_bech32_address_or_exit' into a function that returns proper Result

* printing formatted message on main error

* fixed failing test

* removed duplicate code that should have been gone ages ago

* ibid

* removed biased selection for authenticated handler

* minor refactoring to 'ensure_config_version_compatibility'

mixnode/src/node/listener/connection_handler/mod.rs

@@ -80,6 +80,10 @@ impl ConnectionHandler {
         let mut framed_conn = Framed::new(conn, SphinxCodec);
         while !shutdown.is_shutdown() {
             tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("ConnectionHandler: received shutdown");
+                }
                 Some(framed_sphinx_packet) = framed_conn.next() => {
                     match framed_sphinx_packet {
                         Ok(framed_sphinx_packet) => {
@@ -95,16 +99,12 @@ impl ConnectionHandler {
                         }
                         Err(err) => {
                             error!(
-                                "The socket connection got corrupted with error: {:?}. Closing the socket",
-                                err
+                                "The socket connection got corrupted with error: {err}. Closing the socket",
                             );
                             return;
                         }
                     }
                 },
-                _ = shutdown.recv() => {
-                    log::trace!("ConnectionHandler: received shutdown");
-                }
             }
         }
 

mixnode/src/node/listener/mod.rs

@@ -34,6 +34,10 @@ impl Listener {
 
         while !self.shutdown.is_shutdown() {
             tokio::select! {
+                biased;
+                _ = self.shutdown.recv() => {
+                    log::trace!("Listener: Received shutdown");
+                }
                 connection = listener.accept() => {
                     match connection {
                         Ok((socket, remote_addr)) => {
@@ -43,9 +47,6 @@ impl Listener {
                         Err(err) => warn!("Failed to accept incoming connection - {err}"),
                     }
                 },
-                _ = self.shutdown.recv() => {
-                    log::trace!("Listener: Received shutdown");
-                }
             };
         }
         log::trace!("Listener: Exiting");

mixnode/src/node/mod.rs

@@ -28,7 +28,7 @@ use rand::thread_rng;
 use std::net::SocketAddr;
 use std::process;
 use std::sync::Arc;
-use task::{wait_for_signal, TaskClient, TaskManager};
+use task::{TaskClient, TaskManager};
 use version_checker::parse_version;
 
 mod http;
@@ -292,15 +292,8 @@ impl MixNode {
             .map(|node| node.bond_information.mix_node.identity_key.clone())
     }
 
-    async fn wait_for_interrupt(&self, mut shutdown: TaskManager) {
-        wait_for_signal().await;
-
-        log::info!("Sending shutdown");
-        shutdown.signal_shutdown().ok();
-
-        log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
-        shutdown.wait_for_shutdown().await;
-
+    async fn wait_for_interrupt(&self, shutdown: TaskManager) {
+        let _res = shutdown.catch_interrupt().await;
         log::info!("Stopping nym mixnode");
     }