Merge branch 'develop' into feature/base58

mixnode/src/mix_peer.rs

@@ -1,4 +1,5 @@
 use addressing;
+use addressing::AddressTypeError;
 use sphinx::route::NodeAddressBytes;
 use std::error::Error;
 use std::net::SocketAddr;
@@ -9,14 +10,27 @@ pub struct MixPeer {
     connection: SocketAddr,
 }
 
+#[derive(Debug)]
+pub enum MixPeerError {
+    InvalidAddressError,
+}
+
+impl From<addressing::AddressTypeError> for MixPeerError {
+    fn from(_: AddressTypeError) -> Self {
+        use MixPeerError::*;
+
+        InvalidAddressError
+    }
+}
+
 impl MixPeer {
     // note that very soon `next_hop_address` will be changed to `next_hop_metadata`
-    pub fn new(next_hop_address: NodeAddressBytes) -> MixPeer {
+    pub fn new(next_hop_address: NodeAddressBytes) -> Result<MixPeer, MixPeerError> {
         let next_hop_socket_address =
-            addressing::socket_address_from_encoded_bytes(next_hop_address.to_bytes());
-        MixPeer {
+            addressing::socket_address_from_encoded_bytes(next_hop_address.to_bytes())?;
+        Ok(MixPeer {
             connection: next_hop_socket_address,
-        }
+        })
     }
 
     pub async fn send(&self, bytes: Vec<u8>) -> Result<(), Box<dyn Error>> {

mixnode/src/node/mod.rs

@@ -40,6 +40,7 @@ pub enum MixProcessingError {
     SphinxRecoveryError,
     ReceivedFinalHopError,
     SphinxProcessingError,
+    InvalidHopAddress,
 }
 
 impl From<sphinx::ProcessingError> for MixProcessingError {
@@ -109,9 +110,14 @@ impl PacketProcessor {
     ) -> Result<ForwardingData, MixProcessingError> {
         // we received something resembling a sphinx packet, report it!
         let processing_data = processing_data.lock().await;
-        let mut received_sender = processing_data.received_metrics_tx.clone();
+        let mut received_metrics_tx = processing_data.received_metrics_tx.clone();
 
-        received_sender.send(()).await.unwrap();
+        // if unwrap failed it means our metrics reporter died, so we should exit application and
+        // force restart
+        if received_metrics_tx.send(()).await.is_err() {
+            error!("failed to send metrics data to the controller - the underlying thread probably died!");
+            std::process::exit(1);
+        }
 
         let packet = SphinxPacket::from_bytes(packet_data.to_vec())?;
         let (next_packet, next_hop_address, delay) =
@@ -126,7 +132,10 @@ impl PacketProcessor {
                 }
             };
 
-        let next_mix = MixPeer::new(next_hop_address);
+        let next_mix = match MixPeer::new(next_hop_address) {
+            Ok(next_mix) => next_mix,
+            Err(_) => return Err(MixProcessingError::InvalidHopAddress),
+        };
 
         let fwd_data = ForwardingData::new(
             next_packet,
@@ -140,11 +149,16 @@ impl PacketProcessor {
     async fn wait_and_forward(mut forwarding_data: ForwardingData) {
         let delay_duration = Duration::from_nanos(forwarding_data.delay.get_value());
         tokio::time::delay_for(delay_duration).await;
-        forwarding_data
+
+        if forwarding_data
             .sent_metrics_tx
             .send(forwarding_data.recipient.to_string())
             .await
-            .unwrap();
+            .is_err()
+        {
+            error!("failed to send metrics data to the controller - the underlying thread probably died!");
+            std::process::exit(1);
+        }
 
         trace!("RECIPIENT: {:?}", forwarding_data.recipient);
         match forwarding_data
@@ -188,7 +202,6 @@ impl MixNode {
         mut socket: tokio::net::TcpStream,
         processing_data: Arc<Mutex<ProcessingData>>,
     ) {
-        // NOTE: processing_data is copied here!!
         let mut buf = [0u8; sphinx::PACKET_SIZE];
 
         // In a loop, read data from the socket and write the data back.
@@ -200,12 +213,18 @@ impl MixNode {
                     return;
                 }
                 Ok(_) => {
-                    let fwd_data = PacketProcessor::process_sphinx_data_packet(
+                    let fwd_data = match PacketProcessor::process_sphinx_data_packet(
                         buf.as_ref(),
                         processing_data.clone(),
                     )
                     .await
-                    .unwrap();
+                    {
+                        Ok(fwd_data) => fwd_data,
+                        Err(e) => {
+                            warn!("failed to process sphinx packet: {:?}", e);
+                            return;
+                        }
+                    };
                     PacketProcessor::wait_and_forward(fwd_data).await;
                 }
                 Err(e) => {