Merge branch 'release/1.0.0-rc.1' into develop

Conflicts:
	validator-api/Cargo.toml

validator-api/src/config/mod.rs

@@ -34,7 +34,8 @@ const DEFAULT_PER_NODE_TEST_PACKETS: usize = 3;
 
 const DEFAULT_CACHE_INTERVAL: Duration = Duration::from_secs(30);
 const DEFAULT_MONITOR_THRESHOLD: u8 = 60;
-const DEFAULT_MIN_RELIABILITY: u8 = 50;
+const DEFAULT_MIN_MIXNODE_RELIABILITY: u8 = 50;
+const DEFAULT_MIN_GATEWAY_RELIABILITY: u8 = 20;
 
 #[derive(Debug, Default, Deserialize, PartialEq, Serialize)]
 pub struct Config {
@@ -109,8 +110,8 @@ impl Default for Base {
 #[serde(default)]
 pub struct NetworkMonitor {
     //  Mixnodes and gateways with relialability lower the this get blacklisted by network monitor, get no traffic and cannot be selected into a rewarded set.
-    min_reliability: u8,
-
+    min_mixnode_reliability: u8, // defaults to 50
+    min_gateway_reliability: u8, // defaults to 20
     /// Specifies whether network monitoring service is enabled in this process.
     enabled: bool,
 
@@ -192,7 +193,8 @@ impl NetworkMonitor {
 impl Default for NetworkMonitor {
     fn default() -> Self {
         NetworkMonitor {
-            min_reliability: DEFAULT_MIN_RELIABILITY,
+            min_mixnode_reliability: DEFAULT_MIN_MIXNODE_RELIABILITY,
+            min_gateway_reliability: DEFAULT_MIN_GATEWAY_RELIABILITY,
             enabled: false,
             testnet_mode: false,
             all_validator_apis: default_api_endpoints(),
@@ -334,6 +336,16 @@ impl Config {
         self
     }
 
+    pub fn with_min_mixnode_reliability(mut self, min_mixnode_reliability: u8) -> Self {
+        self.network_monitor.min_mixnode_reliability = min_mixnode_reliability;
+        self
+    }
+
+    pub fn with_min_gateway_reliability(mut self, min_gateway_reliability: u8) -> Self {
+        self.network_monitor.min_gateway_reliability = min_gateway_reliability;
+        self
+    }
+
     #[cfg(not(feature = "coconut"))]
     pub fn with_eth_private_key(mut self, eth_private_key: String) -> Self {
         self.network_monitor.eth_private_key = eth_private_key;
@@ -423,8 +435,12 @@ impl Config {
         self.network_monitor.minimum_test_routes
     }
 
-    pub fn get_min_reliability(&self) -> u8 {
-        self.network_monitor.min_reliability
+    pub fn get_min_mixnode_reliability(&self) -> u8 {
+        self.network_monitor.min_mixnode_reliability
+    }
+
+    pub fn get_min_gateway_reliability(&self) -> u8 {
+        self.network_monitor.min_gateway_reliability
     }
 
     pub fn get_route_test_packets(&self) -> usize {

validator-api/src/config/template.rs

@@ -21,7 +21,8 @@ mixnet_contract_address = '{{ base.mixnet_contract_address }}'
 [network_monitor]
 
 # Mixnodes and gateways with relialability lower the this get blacklisted by network monitor, get no traffic and cannot be selected into a rewarded set.
-min_reliability = {{ network_monitor.min_reliability }}
+min_mixnode_reliability = {{ network_monitor.min_mixnode_reliability }} # deafults to 50
+min_gateway_reliability = {{ network_monitor.min_gateway_reliability }} # defaults to 20
 
 # Specifies whether network monitoring service is enabled in this process.
 enabled = {{ network_monitor.enabled }}

validator-api/src/main.rs

@@ -67,6 +67,9 @@ const ETH_PRIVATE_KEY: &str = "eth_private_key";
 
 const REWARDING_MONITOR_THRESHOLD_ARG: &str = "monitor-threshold";
 
+const MIN_MIXNODE_RELIABILITY_ARG: &str = "min_mixnode_reliability";
+const MIN_GATEWAY_RELIABILITY_ARG: &str = "min_gateway_reliability";
+
 fn parse_validators(raw: &str) -> Vec<Url> {
     raw.split(',')
         .map(|raw_validator| {
@@ -284,6 +287,24 @@ fn override_config(mut config: Config, matches: &ArgMatches<'_>) -> Config {
         config = config.with_minimum_interval_monitor_threshold(monitor_threshold)
     }
 
+    if let Some(reliability) = matches
+        .value_of(MIN_MIXNODE_RELIABILITY_ARG)
+        .map(|t| t.parse::<u8>())
+    {
+        config = config.with_min_mixnode_reliability(
+            reliability.expect("Provided reliability is not a u8 number!"),
+        )
+    }
+
+    if let Some(reliability) = matches
+        .value_of(MIN_GATEWAY_RELIABILITY_ARG)
+        .map(|t| t.parse::<u8>())
+    {
+        config = config.with_min_gateway_reliability(
+            reliability.expect("Provided reliability is not a u8 number!"),
+        )
+    }
+
     #[cfg(feature = "coconut")]
     if let Some(keypair_path) = matches.value_of(KEYPAIR_ARG) {
         let keypair_bs58 = std::fs::read_to_string(keypair_path)

validator-api/src/network_monitor/gateways_reader.rs

@@ -2,181 +2,41 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crypto::asymmetric::identity;
-use futures::stream::Stream;
-use futures::task::Context;
 use gateway_client::{AcknowledgementReceiver, MixnetMessageReceiver};
-use std::pin::Pin;
-use std::task::{Poll, Waker};
+use tokio_stream::StreamMap;
 
-/// Constant used to determine maximum number of times the GatewayReader can poll. It basically
-/// tries to solve the same problem that `FuturesUnordered` has: https://github.com/rust-lang/futures-rs/issues/2047
-const YIELD_EVERY: usize = 32;
-
-// TODO: Originally I set it to (identity::PublicKey, Vec<Vec<u8>>) and I definitely
-// had a reason for doing so, but right now I can't remember what that was...
 pub(crate) type GatewayMessages = Vec<Vec<u8>>;
 
-pub(crate) struct GatewayChannel {
-    id: identity::PublicKey,
-    message_receiver: MixnetMessageReceiver,
-    ack_receiver: AcknowledgementReceiver,
-    is_closed: bool,
-}
-
-impl GatewayChannel {
-    pub(crate) fn new(
-        id: identity::PublicKey,
-        message_receiver: MixnetMessageReceiver,
-        ack_receiver: AcknowledgementReceiver,
-    ) -> Self {
-        GatewayChannel {
-            id,
-            message_receiver,
-            ack_receiver,
-            is_closed: false,
-        }
-    }
-}
-
-impl Stream for GatewayChannel {
-    type Item = Vec<Vec<u8>>;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        if self.is_closed {
-            return Poll::Ready(None);
-        }
-
-        let mut polled = 0;
-
-        // empty the ack channel if anything is on it (we don't care about the content at all at the
-        // moment)
-        while let Poll::Ready(_ack) = Pin::new(&mut self.ack_receiver).poll_next(cx) {
-            polled += 1;
-        }
-
-        let item = futures::ready!(Pin::new(&mut self.message_receiver).poll_next(cx));
-
-        match item {
-            None => Poll::Ready(None),
-            // if we managed to get an item, try to also read additional ones
-            Some(mut messages) => {
-                while let Poll::Ready(new_item) = Pin::new(&mut self.message_receiver).poll_next(cx)
-                {
-                    polled += 1;
-                    match new_item {
-                        None => {
-                            self.is_closed = true;
-                            cx.waker().wake_by_ref();
-                            return Poll::Ready(Some(messages));
-                        }
-                        Some(mut additional_messages) => {
-                            messages.append(&mut additional_messages);
-
-                            // it is fine enough to use the same constant here as in the main GatewayReader
-                            // as only a single channel, i.e. the main gateway will be capable
-                            // of returning more than 2 values
-                            if polled >= YIELD_EVERY {
-                                cx.waker().wake_by_ref();
-                                break;
-                            }
-                        }
-                    }
-                }
-                Poll::Ready(Some(messages))
-            }
-        }
-    }
-}
-
 pub(crate) struct GatewaysReader {
-    latest_read: usize,
-    // channels: FuturesUnordered<GatewayChannel>,
-    channels: Vec<GatewayChannel>,
-    waker: Option<Waker>,
+    ack_map: StreamMap<String, AcknowledgementReceiver>,
+    stream_map: StreamMap<String, MixnetMessageReceiver>,
 }
 
 impl GatewaysReader {
     pub(crate) fn new() -> Self {
         GatewaysReader {
-            latest_read: 0,
-            channels: Vec::new(),
-            waker: None,
+            ack_map: StreamMap::new(),
+            stream_map: StreamMap::new(),
         }
     }
 
-    fn remove_nth(&mut self, i: usize) {
-        self.channels.remove(i);
+    pub fn stream_map(&mut self) -> &mut StreamMap<String, MixnetMessageReceiver> {
+        &mut self.stream_map
     }
 
-    // todo: if we find that this method is called frequently, perhaps the vector should get
-    // replaced with different data structure
-    pub(crate) fn remove_by_key(&mut self, key: identity::PublicKey) {
-        match self.channels.iter().position(|item| item.id == key) {
-            Some(i) => {
-                self.channels.remove(i);
-            }
-            // this shouldn't ever get thrown, so perhaps a panic would be more in order?
-            None => error!(
-                "tried to remove gateway reader {} but it doesn't exist!",
-                key.to_base58_string()
-            ),
-        }
-    }
-
-    fn poll_nth(
+    pub fn add_recievers(
         &mut self,
-        cx: &mut Context<'_>,
-        i: usize,
-    ) -> Option<Poll<Option<GatewayMessages>>> {
-        if let Poll::Ready(item) = Pin::new(&mut self.channels[i]).poll_next(cx) {
-            self.latest_read = i;
-            match item {
-                // Some(messages) => return Some(Poll::Ready(Some((self.channels[i].id, messages)))),
-                Some(messages) => return Some(Poll::Ready(Some(messages))),
-                // remove dead channel
-                None => self.remove_nth(i),
-            }
-        }
-        None
+        id: identity::PublicKey,
+        message_receiver: MixnetMessageReceiver,
+        ack_receiver: AcknowledgementReceiver,
+    ) {
+        let channel_id = id.to_string();
+        self.stream_map.insert(channel_id.clone(), message_receiver);
+        self.ack_map.insert(channel_id, ack_receiver);
     }
 
-    pub(crate) fn insert_channel(&mut self, channel: GatewayChannel) {
-        self.channels.push(channel);
-        if let Some(waker) = self.waker.take() {
-            waker.wake()
-        }
-    }
-}
-
-// TODO: not sure if this will scale well, but I don't know what would be a good alternative,
-// perhaps try to somehow incorporate FuturesUnordered?
-// also, perhaps reading should be done in parallel?
-impl Stream for GatewaysReader {
-    // item represents gateway that returned messages alongside the actual messages
-    type Item = GatewayMessages;
-
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        if self.latest_read >= self.channels.len() {
-            self.latest_read = 0;
-        }
-
-        // don't start reading from beginning each time to at least slightly help with the bias
-        for i in self.latest_read..self.channels.len() {
-            if let Some(item) = self.poll_nth(cx, i) {
-                return item;
-            }
-        }
-
-        for i in 0..self.latest_read {
-            if let Some(item) = self.poll_nth(cx, i) {
-                return item;
-            }
-        }
-
-        // if we have no channels available, store the waker to be woken when a new one is pushed
-        if self.channels.is_empty() {
-            self.waker = Some(cx.waker().clone())
-        }
-        Poll::Pending
+    pub fn remove_recievers(&mut self, id: &str) {
+        self.stream_map.remove(id);
+        self.ack_map.remove(id);
     }
 }

validator-api/src/network_monitor/monitor/gateways_pinger.rs

@@ -44,14 +44,17 @@ impl GatewayPinger {
     }
 
     async fn ping_and_cleanup_all_gateways(&self) {
-        info!(target: "GatewayPinger", "Pinging all active gateways");
+        info!("Pinging all active gateways");
 
         let lock_acquire_start = Instant::now();
         let active_gateway_clients_guard = self.gateway_clients.lock().await;
-        trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start));
+        trace!(
+            "Acquiring lock took {:?}",
+            Instant::now().duration_since(lock_acquire_start)
+        );
 
         if active_gateway_clients_guard.is_empty() {
-            debug!(target: "GatewayPinger", "no gateways to ping");
+            debug!("no gateways to ping");
             return;
         }
 
@@ -89,7 +92,6 @@ impl GatewayPinger {
                     {
                         Err(_timeout) => {
                             warn!(
-                                target: "GatewayPinger",
                                 "we timed out trying to ping {} - assuming the connection is dead.",
                                 active_client.gateway_identity().to_base58_string(),
                             );
@@ -97,7 +99,6 @@ impl GatewayPinger {
                         }
                         Ok(Err(err)) => {
                             warn!(
-                                target: "GatewayPinger",
                                 "failed to send ping message to gateway {} - {} - assuming the connection is dead.",
                                 active_client.gateway_identity().to_base58_string(),
                                 err,
@@ -112,25 +113,34 @@ impl GatewayPinger {
             }
         }
 
+        info!(
+            "Purging {} gateways, acquiring lock",
+            clients_to_purge.len()
+        );
         // purge all dead connections
         // reacquire the guard
         let lock_acquire_start = Instant::now();
         let mut active_gateway_clients_guard = self.gateway_clients.lock().await;
-        trace!(target: "GatewayPinger", "Acquiring lock took {:?}", Instant::now().duration_since(lock_acquire_start));
+        info!(
+            "Acquiring lock took {:?}",
+            Instant::now().duration_since(lock_acquire_start)
+        );
 
         for gateway_id in clients_to_purge.into_iter() {
             if let Some(removed_handle) = active_gateway_clients_guard.remove(&gateway_id) {
                 if !removed_handle.is_invalid().await {
+                    info!("Handle is invalid, purging");
                     // it was not invalidated by the packet sender meaning it probably was some unbonded node
                     // that was never cleared
                     self.notify_connection_failure(gateway_id);
                 }
+                info!("Handle is not invalid, not purged")
             }
         }
 
         let ping_end = Instant::now();
         let time_taken = ping_end.duration_since(ping_start);
-        debug!(target: "GatewayPinger", "Pinging all active gateways took {:?}", time_taken);
+        info!("Pinging all active gateways took {:?}", time_taken);
     }
 
     pub(crate) async fn run(&self) {

validator-api/src/network_monitor/monitor/mod.rs

@@ -42,7 +42,8 @@ pub(super) struct Monitor {
     /// The minimum number of test routes that need to be constructed (and working) in order for
     /// a monitor test run to be valid.
     minimum_test_routes: usize,
-    min_reliability: u8,
+    min_mixnode_reliability: u8,
+    min_gateway_reliability: u8,
 }
 
 impl Monitor {
@@ -67,7 +68,8 @@ impl Monitor {
             route_test_packets: config.get_route_test_packets(),
             test_routes: config.get_test_routes(),
             minimum_test_routes: config.get_minimum_test_routes(),
-            min_reliability: config.get_min_reliability(),
+            min_mixnode_reliability: config.get_min_mixnode_reliability(),
+            min_gateway_reliability: config.get_min_gateway_reliability(),
         }
     }
 
@@ -78,7 +80,7 @@ impl Monitor {
         // uptime calculations
 
         for result in test_summary.mixnode_results.iter() {
-            if result.reliability < self.min_reliability {
+            if result.reliability < self.min_mixnode_reliability {
                 self.packet_preparer
                     .validator_cache()
                     .insert_mixnodes_blacklist(result.identity.clone())
@@ -92,7 +94,7 @@ impl Monitor {
         }
 
         for result in test_summary.gateway_results.iter() {
-            if result.reliability < self.min_reliability {
+            if result.reliability < self.min_gateway_reliability {
                 self.packet_preparer
                     .validator_cache()
                     .insert_gateways_blacklist(result.identity.clone())
@@ -175,9 +177,9 @@ impl Monitor {
 
         for entry in results.iter() {
             if *entry.1 == self.route_test_packets {
-                debug!("✔️ {} succeeded", entry.0)
+                info!("✔️ {} succeeded", entry.0)
             } else {
-                debug!(
+                info!(
                     "❌️ {} failed ({}/{} received)",
                     entry.0, entry.1, self.route_test_packets
                 )
@@ -198,7 +200,7 @@ impl Monitor {
     }
 
     async fn prepare_test_routes(&mut self) -> Option<Vec<TestRoute>> {
-        info!(target: "Monitor", "Generating test routes...");
+        info!("Generating test routes...");
 
         // keep track of nodes that should not be used for route construction
         let mut blacklist = HashSet::new();
@@ -207,7 +209,7 @@ impl Monitor {
         let mut current_attempt = 0;
 
         // todo: tweak this to something more appropriate
-        let max_attempts = self.test_routes * 2;
+        let max_attempts = self.test_routes * 10;
 
         'outer: loop {
             if current_attempt >= max_attempts {
@@ -275,13 +277,12 @@ impl Monitor {
             .set_new_test_nonce(self.test_nonce)
             .await;
 
-        debug!("Sending packets to all gateways...");
+        info!("Sending packets to all gateways...");
         self.packet_sender
             .send_packets(prepared_packets.packets)
             .await;
 
-        debug!(
-            target: "Monitor",
+        info!(
             "Sending is over, waiting for {:?} before checking what we received",
             self.packet_delivery_timeout
         );
@@ -291,6 +292,8 @@ impl Monitor {
 
         let received = self.received_processor.return_received().await;
         let total_received = received.len();
+        info!("Test routes: {:?}", routes);
+        info!("Received {}/{} packets", total_received, total_sent);
 
         let summary = self.summary_producer.produce_summary(
             prepared_packets.tested_mixnodes,
@@ -308,11 +311,14 @@ impl Monitor {
     }
 
     async fn test_run(&mut self) {
-        info!(target: "Monitor", "Starting test run no. {}", self.test_nonce);
+        info!("Starting test run no. {}", self.test_nonce);
         let start = Instant::now();
 
         if let Some(test_routes) = self.prepare_test_routes().await {
-            debug!(target: "Monitor", "Determined reliable routes to test all other nodes against. : {:?}", test_routes);
+            info!(
+                "Determined reliable routes to test all other nodes against. : {:?}",
+                test_routes
+            );
             self.test_network_against(&test_routes).await;
         } else {
             error!("We failed to construct sufficient number of test routes to test the network against")

validator-api/src/network_monitor/monitor/preparer.rs

@@ -25,6 +25,7 @@ type Id = String;
 type Owner = Addr;
 
 #[derive(Clone)]
+#[allow(dead_code)]
 pub(crate) enum InvalidNode {
     Outdated(Id, Owner, Version),
     Malformed(Id, Owner),
@@ -188,8 +189,8 @@ impl PacketPreparer {
         info!("Waiting for minimal topology to be online");
         let initialisation_backoff = Duration::from_secs(30);
         loop {
-            let gateways = self.validator_cache.gateways().await;
-            let mixnodes = self.validator_cache.rewarded_set().await;
+            let gateways = self.validator_cache.gateways_all().await;
+            let mixnodes = self.validator_cache.mixnodes_all().await;
 
             if gateways.len() < minimum_full_routes {
                 info!(
@@ -201,7 +202,7 @@ impl PacketPreparer {
             }
 
             let mut layered_mixes = HashMap::new();
-            for mix in mixnodes.into_inner() {
+            for mix in mixnodes {
                 let layer = mix.layer;
                 let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
                 mixes.push(mix)
@@ -229,7 +230,7 @@ impl PacketPreparer {
     }
 
     async fn all_mixnodes_and_gateways(&self) -> (Vec<MixNodeBond>, Vec<GatewayBond>) {
-        info!(target: "Monitor", "Obtaining network topology...");
+        info!("Obtaining network topology...");
 
         let mixnodes = self.validator_cache.mixnodes_all().await;
         let gateways = self.validator_cache.gateways_all().await;
@@ -239,9 +240,6 @@ impl PacketPreparer {
 
     pub(crate) fn try_parse_mix_bond(&self, mix: &MixNodeBond) -> Result<mix::Node, String> {
         let identity = mix.mix_node.identity_key.clone();
-        if !self.check_version_compatibility(&mix.mix_node.version) {
-            return Err(identity);
-        }
         mix.try_into().map_err(|_| identity)
     }
 
@@ -250,9 +248,6 @@ impl PacketPreparer {
         gateway: &GatewayBond,
     ) -> Result<gateway::Node, String> {
         let identity = gateway.gateway.identity_key.clone();
-        if !self.check_version_compatibility(&gateway.gateway.version) {
-            return Err(identity);
-        }
         gateway.try_into().map_err(|_| identity)
     }
 
@@ -270,19 +265,10 @@ impl PacketPreparer {
         // separate mixes into layers for easier selection
         let mut layered_mixes = HashMap::new();
         for mix in mixnodes {
-            // filter out mixes on the blacklist
-            if blacklist.contains(&mix.mix_node.identity_key) {
-                continue;
-            }
             let layer = mix.layer;
             let mixes = layered_mixes.entry(layer).or_insert_with(Vec::new);
             mixes.push(mix)
         }
-        // filter out gateways on the blacklist
-        let gateways = gateways
-            .into_iter()
-            .filter(|gateway| !blacklist.contains(&gateway.gateway.identity_key))
-            .collect::<Vec<_>>();
 
         // get all nodes from each layer...
         let l1 = layered_mixes.get(&Layer::One)?;
@@ -297,9 +283,6 @@ impl PacketPreparer {
         let rand_l3 = l3.choose_multiple(&mut rng, n).collect::<Vec<_>>();
         let rand_gateways = gateways.choose_multiple(&mut rng, n).collect::<Vec<_>>();
 
-        // let rand_gateways = gateways.iter().filter(|g| g.gateway.host ==  "109.74.196.254".to_string()).collect::<Vec<_>>();
-        // let rand_gateways = gateways.iter().filter(|g| g.gateway.host ==  "185.3.94.33".to_string()).collect::<Vec<_>>();
-
         // the unwrap on `min()` is fine as we know the iterator is not empty
         let most_available = *[
             rand_l1.len(),
@@ -312,9 +295,10 @@ impl PacketPreparer {
         .unwrap();
 
         if most_available == 0 {
-            // it's impossible to generate a single route
+            error!("Cannot construct test routes. No nodes or gateways available");
             None
         } else {
+            trace!("Generating test routes...");
             let mut routes = Vec::new();
             for i in 0..most_available {
                 let node_1 = match self.try_parse_mix_bond(rand_l1[i]) {
@@ -358,14 +342,11 @@ impl PacketPreparer {
                     gateway,
                 ))
             }
+            info!("{:?}", routes);
             Some(routes)
         }
     }
 
-    fn check_version_compatibility(&self, node_version: &str) -> bool {
-        version_checker::is_minor_version_compatible(node_version, &self.system_version)
-    }
-
     fn create_packet_sender(&self, gateway: &gateway::Node) -> Recipient {
         Recipient::new(
             self.self_public_identity,
@@ -404,14 +385,6 @@ impl PacketPreparer {
         let mut parsed_nodes = Vec::new();
         let mut invalid_nodes = Vec::new();
         for mixnode in nodes {
-            if !self.check_version_compatibility(&mixnode.mix_node.version) {
-                invalid_nodes.push(InvalidNode::Outdated(
-                    mixnode.mix_node.identity_key,
-                    mixnode.owner,
-                    mixnode.mix_node.version,
-                ));
-                continue;
-            }
             if let Ok(parsed_node) = (&mixnode).try_into() {
                 parsed_nodes.push(parsed_node)
             } else {
@@ -431,14 +404,6 @@ impl PacketPreparer {
         let mut parsed_nodes = Vec::new();
         let mut invalid_nodes = Vec::new();
         for gateway in nodes {
-            if !self.check_version_compatibility(&gateway.gateway.version) {
-                invalid_nodes.push(InvalidNode::Outdated(
-                    gateway.gateway.identity_key,
-                    gateway.owner,
-                    gateway.gateway.version,
-                ));
-                continue;
-            }
             if let Ok(parsed_node) = (&gateway).try_into() {
                 parsed_nodes.push(parsed_node)
             } else {
@@ -460,18 +425,17 @@ impl PacketPreparer {
         // (remember that "idle" nodes are still part of that set)
         // we don't care about other nodes, i.e. nodes that are bonded but will not get
         // any reward during the current rewarding interval
-        let (rewarded_set, all_gateways) = self.all_mixnodes_and_gateways().await;
+        let (mixnodes, gateways) = self.all_mixnodes_and_gateways().await;
 
-        let (mixes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(rewarded_set);
-        let (gateways, invalid_gateways) =
-            self.filter_outdated_and_malformed_gateways(all_gateways);
+        let (mixnodes, invalid_mixnodes) = self.filter_outdated_and_malformed_mixnodes(mixnodes);
+        let (gateways, invalid_gateways) = self.filter_outdated_and_malformed_gateways(gateways);
 
-        let tested_mixnodes = mixes.iter().map(|node| node.into()).collect::<Vec<_>>();
+        let tested_mixnodes = mixnodes.iter().map(|node| node.into()).collect::<Vec<_>>();
         let tested_gateways = gateways.iter().map(|node| node.into()).collect::<Vec<_>>();
 
         let packets_to_create = (test_routes.len() * self.per_node_test_packets)
             * (tested_mixnodes.len() + tested_gateways.len());
-        info!(target: "TestPreparer", "Need to create {} mix packets", packets_to_create);
+        info!("Need to create {} mix packets", packets_to_create);
 
         let mut all_gateway_packets = HashMap::new();
 
@@ -483,10 +447,10 @@ impl PacketPreparer {
             let gateway_owner = test_route.gateway_owner();
 
             // it's actually going to be a tiny bit more due to gateway testing, but it's a good enough approximation
-            let mut mix_packets = Vec::with_capacity(mixes.len() * self.per_node_test_packets);
+            let mut mix_packets = Vec::with_capacity(mixnodes.len() * self.per_node_test_packets);
 
             // and for each mixnode...
-            for mixnode in &mixes {
+            for mixnode in &mixnodes {
                 let test_packet = TestPacket::from_mixnode(mixnode, test_route.id(), test_nonce);
                 let topology = test_route.substitute_mix(mixnode);
                 // produce n mix packets

validator-api/src/network_monitor/monitor/receiver.rs

@@ -1,7 +1,7 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::network_monitor::gateways_reader::{GatewayChannel, GatewayMessages, GatewaysReader};
+use crate::network_monitor::gateways_reader::{GatewayMessages, GatewaysReader};
 use crate::network_monitor::monitor::processor::ReceivedProcessorSender;
 use crypto::asymmetric::identity;
 use futures::channel::mpsc;
@@ -40,10 +40,12 @@ impl PacketReceiver {
     fn process_gateway_update(&mut self, update: GatewayClientUpdate) {
         match update {
             GatewayClientUpdate::New(id, (message_receiver, ack_receiver)) => {
-                let channel = GatewayChannel::new(id, message_receiver, ack_receiver);
-                self.gateways_reader.insert_channel(channel);
+                self.gateways_reader
+                    .add_recievers(id, message_receiver, ack_receiver);
+            }
+            GatewayClientUpdate::Failure(id) => {
+                self.gateways_reader.remove_recievers(&id.to_string());
             }
-            GatewayClientUpdate::Failure(id) => self.gateways_reader.remove_by_key(id),
         }
     }
 
@@ -62,7 +64,9 @@ impl PacketReceiver {
                 // similarly gateway reader will never return a `None` as it's implemented
                 // as an infinite stream that returns Poll::Pending if it doesn't have anything
                 // to return
-                messages = self.gateways_reader.next() => self.process_gateway_messages(messages.unwrap()),
+                Some((_gateway_id, message)) = self.gateways_reader.stream_map().next() => {
+                        self.process_gateway_messages(message)
+                }
             }
         }
     }

validator-api/src/network_monitor/monitor/sender.rs

@@ -232,14 +232,13 @@ impl PacketSender {
     ) -> Result<(), GatewayClientError> {
         let gateway_id = client.gateway_identity().to_base58_string();
         info!(
-            target: "MessageSender",
             "Got {} packets to send to gateway {}",
             mix_packets.len(),
             gateway_id
         );
 
         if mix_packets.len() <= max_sending_rate {
-            debug!(target: "MessageSender","Everything is going to get sent as one.");
+            debug!("Everything is going to get sent as one.");
             client.batch_send_mix_packets(mix_packets).await?;
         } else {
             let packets_per_time_chunk =
@@ -248,7 +247,6 @@ impl PacketSender {
             let total_expected_time =
                 Duration::from_secs_f64(mix_packets.len() as f64 / max_sending_rate as f64);
             info!(
-                target: "MessageSender",
                 "With our rate of {} packets/s it should take around {:?} to send it all to {} ...",
                 max_sending_rate, total_expected_time, gateway_id
             );
@@ -268,7 +266,7 @@ impl PacketSender {
             // this way we won't have to do reallocations in here as they're unavoidable when
             // splitting a vector into multiple vectors
             while let Some(retained) = split_off_vec(&mut mix_packets, packets_per_time_chunk) {
-                trace!(target: "MessageSender","Sending {} packets...", mix_packets.len());
+                trace!("Sending {} packets...", mix_packets.len());
 
                 if mix_packets.len() == 1 {
                     client.send_mix_packet(mix_packets.pop().unwrap()).await?;
@@ -280,7 +278,7 @@ impl PacketSender {
 
                 mix_packets = retained;
             }
-            debug!(target: "MessageSender", "Done sending");
+            debug!("Done sending");
         }
 
         Ok(())