Feature/instant sending (#359)

* Ability to set client in vpn mode

* Connection handler for mixnode

* Initial vpn mode for mixes

* Updated SphinxCodec to contain more metadata

* Renaming

* Removed handle from mixnet client and introduced forwarder

* Mixnode using new forwarder

* Mixnode common module containing shared packet processing

* ibid. incorporated inside mixnode

* New processing for gateway

* Type cleanup

* Wasm fix

* Fixed client config

* Fixed mixnode runtime issues

* Formatting

* Client re-using secret on 'normal' packets

* Using the same key for acks

* WIP

* vpn key manager cleanup

* wasm fix

* VPN_KEY_REUSE_LIMIT moved to config

* Moved AckDelayQueue to separate common crate

* Key cache invalidator

* Updated dashmap used in gateway

* Old typo

* Additional comment

* Cargo fmt

* Fixed tests

* Sphinx update

* cache ttl as config option

* Cargo fmt

mixnode/src/node/listener.rs

@@ -1,112 +0,0 @@
-// Copyright 2020 Nym Technologies SA
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use crate::node::packet_processing::{MixProcessingResult, PacketProcessor};
-use futures::channel::mpsc;
-use log::*;
-use nymsphinx::framing::SphinxCodec;
-use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket};
-use std::io;
-use std::net::SocketAddr;
-use tokio::runtime::Handle;
-use tokio::stream::StreamExt;
-use tokio::task::JoinHandle;
-use tokio_util::codec::Framed;
-
-async fn process_received_packet(
-    sphinx_packet: SphinxPacket,
-    packet_processor: PacketProcessor,
-    forwarding_channel: mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)>,
-) {
-    // all processing incl. delay was done, the only thing left is to forward it
-    match packet_processor.process_sphinx_packet(sphinx_packet).await {
-        Err(e) => debug!("We failed to process received sphinx packet - {:?}", e),
-        Ok(res) => match res {
-            MixProcessingResult::ForwardHop(hop_address, forward_packet) => {
-                // send our data to tcp client for forwarding. If forwarding fails, then it fails,
-                // it's not like we can do anything about it
-                //
-                // in unbounded_send() failed it means that the receiver channel was disconnected
-                // and hence something weird must have happened without a way of recovering
-                forwarding_channel
-                    .unbounded_send((hop_address, forward_packet))
-                    .unwrap();
-                packet_processor.report_sent(hop_address);
-            }
-            MixProcessingResult::LoopMessage => {
-                warn!("Somehow processed a loop cover message that we haven't implemented yet!")
-            }
-        },
-    }
-}
-
-async fn process_socket_connection(
-    socket: tokio::net::TcpStream,
-    packet_processor: PacketProcessor,
-    forwarding_channel: mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)>,
-) {
-    let mut framed = Framed::new(socket, SphinxCodec);
-    while let Some(sphinx_packet) = framed.next().await {
-        match sphinx_packet {
-            Ok(sphinx_packet) => {
-                // we *really* need a worker pool here, because if we receive too many packets,
-                // we will spawn too many tasks and starve CPU due to context switching.
-                // (because presumably tokio has some concept of context switching in its
-                // scheduler)
-                tokio::spawn(process_received_packet(
-                    sphinx_packet,
-                    packet_processor.clone(),
-                    forwarding_channel.clone(),
-                ));
-            }
-            Err(err) => {
-                error!(
-                    "The socket connection got corrupted with error: {:?}. Closing the socket",
-                    err
-                );
-                return;
-            }
-        }
-    }
-    info!(
-        "Closing connection from {:?}",
-        framed.into_inner().peer_addr()
-    );
-}
-
-pub(crate) fn run_socket_listener(
-    handle: &Handle,
-    addr: SocketAddr,
-    packet_processor: PacketProcessor,
-    forwarding_channel: mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)>,
-) -> JoinHandle<io::Result<()>> {
-    let handle_clone = handle.clone();
-    handle.spawn(async move {
-        let mut listener = tokio::net::TcpListener::bind(addr).await?;
-        loop {
-            let (socket, _) = listener.accept().await?;
-
-            let thread_packet_processor = packet_processor.clone();
-            let forwarding_channel_clone = forwarding_channel.clone();
-            handle_clone.spawn(async move {
-                process_socket_connection(
-                    socket,
-                    thread_packet_processor,
-                    forwarding_channel_clone,
-                )
-                .await;
-            });
-        }
-    })
-}

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

@@ -0,0 +1,123 @@
+// Copyright 2020 Nym Technologies SA
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::node::listener::connection_handler::packet_processing::{
+    MixProcessingResult, PacketProcessor,
+};
+use log::*;
+use mixnet_client::forwarder::MixForwardingSender;
+use nymsphinx::forwarding::packet::MixPacket;
+use nymsphinx::framing::codec::SphinxCodec;
+use nymsphinx::framing::packet::FramedSphinxPacket;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::net::TcpStream;
+use tokio::stream::StreamExt;
+use tokio_util::codec::Framed;
+
+pub(crate) mod packet_processing;
+
+pub(crate) struct ConnectionHandler {
+    packet_processor: PacketProcessor,
+    forwarding_channel: MixForwardingSender,
+}
+
+impl ConnectionHandler {
+    pub(crate) fn new(
+        packet_processor: PacketProcessor,
+        forwarding_channel: MixForwardingSender,
+    ) -> Self {
+        ConnectionHandler {
+            packet_processor,
+            forwarding_channel,
+        }
+    }
+
+    pub(crate) fn clone_without_cache(&self) -> Self {
+        ConnectionHandler {
+            packet_processor: self.packet_processor.clone_without_cache(),
+            forwarding_channel: self.forwarding_channel.clone(),
+        }
+    }
+
+    fn forward_packet(&self, mix_packet: MixPacket) {
+        let routing_address = mix_packet.next_hop();
+        // send our data to tcp client for forwarding. If forwarding fails, then it fails,
+        // it's not like we can do anything about it
+        //
+        // in unbounded_send() failed it means that the receiver channel was disconnected
+        // and hence something weird must have happened without a way of recovering
+        self.forwarding_channel.unbounded_send(mix_packet).unwrap();
+        self.packet_processor.report_sent(routing_address);
+    }
+
+    async fn handle_received_packet(self: Arc<Self>, framed_sphinx_packet: FramedSphinxPacket) {
+        //
+        // TODO: here be replay attack detection - it will require similar key cache to the one in
+        // packet processor for vpn packets,
+        // question: can it also be per connection vs global?
+        //
+
+        // all processing including delaying, key caching, etc. was done, the only thing left is to forward it
+        match self
+            .packet_processor
+            .process_received(framed_sphinx_packet)
+            .await
+        {
+            Err(e) => debug!("We failed to process received sphinx packet - {:?}", e),
+            Ok(res) => match res {
+                MixProcessingResult::ForwardHop(forward_packet) => {
+                    self.forward_packet(forward_packet)
+                }
+                MixProcessingResult::FinalHop(..) => {
+                    warn!("Somehow processed a loop cover message that we haven't implemented yet!")
+                }
+            },
+        }
+    }
+
+    pub(crate) async fn handle_connection(self, conn: TcpStream, remote: SocketAddr) {
+        debug!("Starting connection handler for {:?}", remote);
+        let this = Arc::new(self);
+        let mut framed_conn = Framed::new(conn, SphinxCodec);
+        while let Some(framed_sphinx_packet) = framed_conn.next().await {
+            match framed_sphinx_packet {
+                Ok(framed_sphinx_packet) => {
+                    // TODO: benchmark spawning tokio task with full processing vs just processing it
+                    // synchronously (without delaying inside of course,
+                    // delay could be moved to a per-connection DelayQueue. The delay queue future
+                    // could automatically just forward packet that is done being delayed)
+                    // under higher load in single and multi-threaded situation.
+                    //
+                    // My gut feeling is saying that we might get some nice performance boost
+                    // if we introduced the change
+                    let this = Arc::clone(&this);
+                    tokio::spawn(this.handle_received_packet(framed_sphinx_packet));
+                }
+                Err(err) => {
+                    error!(
+                        "The socket connection got corrupted with error: {:?}. Closing the socket",
+                        err
+                    );
+                    return;
+                }
+            }
+        }
+
+        info!(
+            "Closing connection from {:?}",
+            framed_conn.into_inner().peer_addr()
+        );
+    }
+}

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

@@ -0,0 +1,60 @@
+// Copyright 2020 Nym Technologies SA
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::node::metrics;
+use crypto::asymmetric::encryption;
+use mixnode_common::cached_packet_processor::error::MixProcessingError;
+use mixnode_common::cached_packet_processor::processor::CachedPacketProcessor;
+pub use mixnode_common::cached_packet_processor::processor::MixProcessingResult;
+use nymsphinx::addressing::nodes::NymNodeRoutingAddress;
+use nymsphinx::framing::packet::FramedSphinxPacket;
+use tokio::time::Duration;
+
+// PacketProcessor contains all data required to correctly unwrap and forward sphinx packets
+pub struct PacketProcessor {
+    inner_processor: CachedPacketProcessor,
+    metrics_reporter: metrics::MetricsReporter,
+}
+
+impl PacketProcessor {
+    pub(crate) fn new(
+        encryption_key: &encryption::PrivateKey,
+        metrics_reporter: metrics::MetricsReporter,
+        cache_entry_ttl: Duration,
+    ) -> Self {
+        PacketProcessor {
+            inner_processor: CachedPacketProcessor::new(encryption_key.into(), cache_entry_ttl),
+            metrics_reporter,
+        }
+    }
+
+    pub(crate) fn clone_without_cache(&self) -> Self {
+        PacketProcessor {
+            inner_processor: self.inner_processor.clone_without_cache(),
+            metrics_reporter: self.metrics_reporter.clone(),
+        }
+    }
+
+    pub(crate) fn report_sent(&self, address: NymNodeRoutingAddress) {
+        self.metrics_reporter.report_sent(address.to_string())
+    }
+
+    pub(crate) async fn process_received(
+        &self,
+        received: FramedSphinxPacket,
+    ) -> Result<MixProcessingResult, MixProcessingError> {
+        self.metrics_reporter.report_received();
+        self.inner_processor.process_received(received).await
+    }
+}

mixnode/src/node/listener/mod.rs

@@ -0,0 +1,52 @@
+// Copyright 2020 Nym Technologies SA
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::node::listener::connection_handler::ConnectionHandler;
+use log::*;
+use std::net::SocketAddr;
+use tokio::net::TcpListener;
+use tokio::task::JoinHandle;
+
+pub(crate) mod connection_handler;
+
+pub(crate) struct Listener {
+    address: SocketAddr,
+}
+
+impl Listener {
+    pub(crate) fn new(address: SocketAddr) -> Self {
+        Listener { address }
+    }
+
+    async fn run(&mut self, connection_handler: ConnectionHandler) {
+        let mut listener = TcpListener::bind(self.address)
+            .await
+            .expect("Failed to create TCP listener");
+        loop {
+            match listener.accept().await {
+                Ok((socket, remote_addr)) => {
+                    let handler = connection_handler.clone_without_cache();
+                    tokio::spawn(handler.handle_connection(socket, remote_addr));
+                }
+                Err(err) => warn!("Failed to accept incoming connection - {:?}", err),
+            }
+        }
+    }
+
+    pub(crate) fn start(mut self, connection_handler: ConnectionHandler) -> JoinHandle<()> {
+        info!("Running mix listener on {:?}", self.address.to_string());
+
+        tokio::spawn(async move { self.run(connection_handler).await })
+    }
+}

mixnode/src/node/metrics.rs

@@ -21,7 +21,6 @@ use log::*;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime};
-use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
 
 type SentMetricsMap = HashMap<String, u64>;
@@ -88,8 +87,8 @@ impl MetricsReceiver {
         }
     }
 
-    fn start(mut self, handle: &Handle) -> JoinHandle<()> {
-        handle.spawn(async move {
+    fn start(mut self) -> JoinHandle<()> {
+        tokio::spawn(async move {
             while let Some(metrics_data) = self.metrics_rx.next().await {
                 match metrics_data {
                     MetricEvent::Received => self.metrics.increment_received_metrics().await,
@@ -129,8 +128,8 @@ impl MetricsSender {
         }
     }
 
-    fn start(mut self, handle: &Handle) -> JoinHandle<()> {
-        handle.spawn(async move {
+    fn start(mut self) -> JoinHandle<()> {
+        tokio::spawn(async move {
             loop {
                 // set the deadline in the future
                 let sending_delay = tokio::time::delay_for(self.sending_delay);
@@ -283,10 +282,10 @@ impl MetricsController {
     }
 
     // reporter is how node is going to be accessing the metrics data
-    pub(crate) fn start(self, handle: &Handle) -> MetricsReporter {
+    pub(crate) fn start(self) -> MetricsReporter {
         // TODO: should we do anything with JoinHandle(s) returned by start methods?
-        self.receiver.start(handle);
-        self.sender.start(handle);
+        self.receiver.start();
+        self.sender.start();
         self.reporter
     }
 }

mixnode/src/node/mod.rs

@@ -13,33 +13,31 @@
 // limitations under the License.
 
 use crate::config::Config;
-use crate::node::packet_processing::PacketProcessor;
+use crate::node::listener::connection_handler::packet_processing::PacketProcessor;
+use crate::node::listener::connection_handler::ConnectionHandler;
+use crate::node::listener::Listener;
 use crypto::asymmetric::encryption;
 use directory_client::DirectoryClient;
-use futures::channel::mpsc;
 use log::*;
-use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket};
+use mixnet_client::forwarder::{MixForwardingSender, PacketForwarder};
+use std::sync::Arc;
 use tokio::runtime::Runtime;
 
 mod listener;
 mod metrics;
-mod packet_forwarding;
-pub(crate) mod packet_processing;
 mod presence;
 
 // the MixNode will live for whole duration of this program
 pub struct MixNode {
-    runtime: Runtime,
     config: Config,
-    sphinx_keypair: encryption::KeyPair,
+    sphinx_keypair: Arc<encryption::KeyPair>,
 }
 
 impl MixNode {
     pub fn new(config: Config, sphinx_keypair: encryption::KeyPair) -> Self {
         MixNode {
-            runtime: Runtime::new().unwrap(),
             config,
-            sphinx_keypair,
+            sphinx_keypair: Arc::new(sphinx_keypair),
         }
     }
 
@@ -53,7 +51,7 @@ impl MixNode {
             self.config.get_layer(),
             self.config.get_presence_sending_delay(),
         );
-        presence::Notifier::new(notifier_config).start(self.runtime.handle());
+        presence::Notifier::new(notifier_config).start();
     }
 
     fn start_metrics_reporter(&self) -> metrics::MetricsReporter {
@@ -64,51 +62,47 @@ impl MixNode {
             self.config.get_metrics_sending_delay(),
             self.config.get_metrics_running_stats_logging_delay(),
         )
-        .start(self.runtime.handle())
+        .start()
     }
 
     fn start_socket_listener(
         &self,
         metrics_reporter: metrics::MetricsReporter,
-        forwarding_channel: mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)>,
+        forwarding_channel: MixForwardingSender,
     ) {
         info!("Starting socket listener...");
-        // this is the only location where our private key is going to be copied
-        // it will be held in memory owned by `MixNode` and inside an Arc of `PacketProcessor`
-        let packet_processor =
-            PacketProcessor::new(self.sphinx_keypair.private_key().clone(), metrics_reporter);
 
-        listener::run_socket_listener(
-            self.runtime.handle(),
-            self.config.get_listening_address(),
-            packet_processor,
-            forwarding_channel,
+        let packet_processor = PacketProcessor::new(
+            self.sphinx_keypair.private_key(),
+            metrics_reporter,
+            self.config.get_cache_entry_ttl(),
         );
+
+        let connection_handler = ConnectionHandler::new(packet_processor, forwarding_channel);
+
+        let listener = Listener::new(self.config.get_listening_address());
+
+        listener.start(connection_handler);
     }
 
-    fn start_packet_forwarder(
-        &mut self,
-    ) -> mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)> {
+    fn start_packet_forwarder(&mut self) -> MixForwardingSender {
         info!("Starting packet forwarder...");
-        self.runtime
-            .enter(|| {
-                packet_forwarding::PacketForwarder::new(
-                    self.config.get_packet_forwarding_initial_backoff(),
-                    self.config.get_packet_forwarding_maximum_backoff(),
-                    self.config.get_initial_connection_timeout(),
-                )
-            })
-            .start(self.runtime.handle())
+
+        let (mut packet_forwarder, packet_sender) = PacketForwarder::new(
+            self.config.get_packet_forwarding_initial_backoff(),
+            self.config.get_packet_forwarding_maximum_backoff(),
+            self.config.get_initial_connection_timeout(),
+        );
+
+        tokio::spawn(async move { packet_forwarder.run().await });
+        packet_sender
     }
 
-    fn check_if_same_ip_node_exists(&mut self) -> Option<String> {
+    async fn check_if_same_ip_node_exists(&mut self) -> Option<String> {
         let directory_client_config =
             directory_client::Config::new(self.config.get_presence_directory_server());
         let directory_client = directory_client::Client::new(directory_client_config);
-        let topology = self
-            .runtime
-            .block_on(directory_client.get_topology())
-            .ok()?;
+        let topology = directory_client.get_topology().await.ok()?;
         let existing_mixes_presence = topology.mix_nodes;
         existing_mixes_presence
             .iter()
@@ -116,32 +110,38 @@ impl MixNode {
             .map(|node| node.pub_key.clone())
     }
 
-    pub fn run(&mut self) {
-        info!("Starting nym mixnode");
-
-        if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists() {
-            error!(
-                "Our announce-host is identical to an existing node's announce-host! (its key is {:?}",
-                duplicate_node_key
-            );
-            return;
-        }
-        let forwarding_channel = self.start_packet_forwarder();
-        let metrics_reporter = self.start_metrics_reporter();
-        self.start_socket_listener(metrics_reporter, forwarding_channel);
-        self.start_presence_notifier();
-
-        info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
-
-        if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) {
+    async fn wait_for_interrupt(&self) {
+        if let Err(e) = tokio::signal::ctrl_c().await {
             error!(
                 "There was an error while capturing SIGINT - {:?}. We will terminate regardless",
                 e
             );
         }
-
         println!(
             "Received SIGINT - the mixnode will terminate now (threads are not YET nicely stopped)"
         );
     }
+
+    pub fn run(&mut self) {
+        info!("Starting nym mixnode");
+        let mut runtime = Runtime::new().unwrap();
+
+        runtime.block_on(async {
+            if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists().await {
+                error!(
+                    "Our announce-host is identical to an existing node's announce-host! (its key is {:?}",
+                    duplicate_node_key
+                );
+                return;
+            }
+            let forwarding_channel = self.start_packet_forwarder();
+            let metrics_reporter = self.start_metrics_reporter();
+            self.start_socket_listener(metrics_reporter, forwarding_channel);
+            self.start_presence_notifier();
+
+            info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
+
+            self.wait_for_interrupt().await
+        })
+    }
 }

mixnode/src/node/packet_forwarding.rs

@@ -1,65 +0,0 @@
-// Copyright 2020 Nym Technologies SA
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use futures::channel::mpsc;
-use futures::StreamExt;
-use log::*;
-use nymsphinx::{addressing::nodes::NymNodeRoutingAddress, SphinxPacket};
-use std::time::Duration;
-use tokio::runtime::Handle;
-
-pub(crate) struct PacketForwarder {
-    tcp_client: mixnet_client::Client,
-    conn_tx: mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)>,
-    conn_rx: mpsc::UnboundedReceiver<(NymNodeRoutingAddress, SphinxPacket)>,
-}
-
-impl PacketForwarder {
-    pub(crate) fn new(
-        initial_reconnection_backoff: Duration,
-        maximum_reconnection_backoff: Duration,
-        initial_connection_timeout: Duration,
-    ) -> PacketForwarder {
-        let tcp_client_config = mixnet_client::Config::new(
-            initial_reconnection_backoff,
-            maximum_reconnection_backoff,
-            initial_connection_timeout,
-        );
-
-        let (conn_tx, conn_rx) = mpsc::unbounded();
-
-        PacketForwarder {
-            tcp_client: mixnet_client::Client::new(tcp_client_config),
-            conn_tx,
-            conn_rx,
-        }
-    }
-
-    pub(crate) fn start(
-        mut self,
-        handle: &Handle,
-    ) -> mpsc::UnboundedSender<(NymNodeRoutingAddress, SphinxPacket)> {
-        // TODO: what to do with the lost JoinHandle?
-        let sender_channel = self.conn_tx.clone();
-        handle.spawn(async move {
-            while let Some((address, packet)) = self.conn_rx.next().await {
-                trace!("Going to forward packet to {:?}", address);
-                // as a mix node we don't care about responses, we just want to fire packets
-                // as quickly as possible
-                self.tcp_client.send(address, packet, false).await.unwrap(); // if we're not waiting for response, we MUST get an Ok
-            }
-        });
-        sender_channel
-    }
-}

mixnode/src/node/packet_processing.rs

@@ -1,114 +0,0 @@
-// Copyright 2020 Nym Technologies SA
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use crate::node::metrics;
-use crypto::asymmetric::encryption;
-use log::*;
-use nymsphinx::addressing::nodes::{NymNodeRoutingAddress, NymNodeRoutingAddressError};
-use nymsphinx::{
-    Delay as SphinxDelay, Error as SphinxError, NodeAddressBytes, ProcessedPacket, SphinxPacket,
-};
-use std::convert::TryFrom;
-use std::sync::Arc;
-
-#[derive(Debug)]
-pub enum MixProcessingError {
-    ReceivedFinalHopError,
-    SphinxProcessingError(SphinxError),
-    InvalidHopAddress,
-}
-
-pub enum MixProcessingResult {
-    ForwardHop(NymNodeRoutingAddress, SphinxPacket),
-    #[allow(dead_code)]
-    LoopMessage,
-}
-
-impl From<SphinxError> for MixProcessingError {
-    // for time being just have a single error instance for all possible results of SphinxError
-    fn from(err: SphinxError) -> Self {
-        use MixProcessingError::*;
-
-        SphinxProcessingError(err)
-    }
-}
-
-impl From<NymNodeRoutingAddressError> for MixProcessingError {
-    fn from(_: NymNodeRoutingAddressError) -> Self {
-        use MixProcessingError::*;
-
-        InvalidHopAddress
-    }
-}
-
-// PacketProcessor contains all data required to correctly unwrap and forward sphinx packets
-#[derive(Clone)]
-pub struct PacketProcessor {
-    secret_key: Arc<encryption::PrivateKey>,
-    metrics_reporter: metrics::MetricsReporter,
-}
-
-impl PacketProcessor {
-    pub(crate) fn new(
-        secret_key: encryption::PrivateKey,
-        metrics_reporter: metrics::MetricsReporter,
-    ) -> Self {
-        PacketProcessor {
-            secret_key: Arc::new(secret_key),
-            metrics_reporter,
-        }
-    }
-
-    pub(crate) fn report_sent(&self, addr: NymNodeRoutingAddress) {
-        self.metrics_reporter.report_sent(addr.to_string())
-    }
-
-    async fn process_forward_hop(
-        &self,
-        packet: SphinxPacket,
-        forward_address: NodeAddressBytes,
-        delay: SphinxDelay,
-    ) -> Result<MixProcessingResult, MixProcessingError> {
-        let next_hop_address = NymNodeRoutingAddress::try_from(forward_address)?;
-
-        // Delay packet for as long as required
-        tokio::time::delay_for(delay.to_duration()).await;
-
-        Ok(MixProcessingResult::ForwardHop(next_hop_address, packet))
-    }
-
-    pub(crate) async fn process_sphinx_packet(
-        &self,
-        packet: SphinxPacket,
-    ) -> Result<MixProcessingResult, MixProcessingError> {
-        // we received something resembling a sphinx packet, report it!
-        self.metrics_reporter.report_received();
-        match packet.process(&self.secret_key.as_ref().into()) {
-            Ok(ProcessedPacket::ProcessedPacketForwardHop(packet, address, delay)) => {
-                self.process_forward_hop(packet, address, delay).await
-            }
-            Ok(ProcessedPacket::ProcessedPacketFinalHop(_, _, _)) => {
-                warn!("Received a loop cover message that we haven't implemented yet!");
-                Err(MixProcessingError::ReceivedFinalHopError)
-            }
-            Err(e) => {
-                warn!("Failed to unwrap Sphinx packet: {:?}", e);
-                Err(MixProcessingError::SphinxProcessingError(e))
-            }
-        }
-    }
-}
-
-// TODO: the test that definitely needs to be written is as follows:
-// we are stuck trying to write to mix A, can we still forward just fine to mix B?

mixnode/src/node/presence.rs

@@ -17,7 +17,6 @@ use directory_client::presence::mixnodes::MixNodePresence;
 use directory_client::DirectoryClient;
 use log::{error, trace};
 use std::time::Duration;
-use tokio::runtime::Handle;
 use tokio::task::JoinHandle;
 
 pub(crate) struct NotifierConfig {
@@ -87,8 +86,8 @@ impl Notifier {
         }
     }
 
-    pub fn start(self, handle: &Handle) -> JoinHandle<()> {
-        handle.spawn(async move {
+    pub fn start(self) -> JoinHandle<()> {
+        tokio::spawn(async move {
             loop {
                 // set the deadline in the future
                 let sending_delay = tokio::time::delay_for(self.sending_delay);