Feature/network monitor gateway pings (#608)

* Ability to send Ping frame to gateway

* Sending ping frame to gateways every 1min

* Increased monitor run interval back to 15min

* Whitespace

common/client-libs/gateway-client/src/client.rs

@@ -485,6 +485,27 @@ impl GatewayClient {
         }
     }
 
+    pub async fn send_ping_message(&mut self) -> Result<(), GatewayClientError> {
+        if !self.connection.is_established() {
+            return Err(GatewayClientError::ConnectionNotEstablished);
+        }
+
+        // as per RFC6455 section 5.5.2, `Ping frame MAY include "Application data".`
+        // so we don't need to include any here.
+        let msg = Message::Ping(Vec::new());
+
+        if let Err(err) = self.send_websocket_message_without_response(msg).await {
+            if err.is_closed_connection() && self.should_reconnect_on_failure {
+                info!("Going to attempt a reconnection");
+                self.attempt_reconnection().await
+            } else {
+                Err(err)
+            }
+        } else {
+            Ok(())
+        }
+    }
+
     // TODO: possibly make responses optional
     pub async fn send_mix_packet(
         &mut self,

network-monitor/src/monitor/mod.rs

@@ -10,7 +10,7 @@ use crate::node_status_api::models::{BatchGatewayStatus, BatchMixStatus};
 use crate::test_packet::NodeType;
 use crate::tested_network::TestedNetwork;
 use log::*;
-use tokio::time::{sleep, Duration};
+use tokio::time::{sleep, Duration, Instant};
 
 pub(crate) mod preparer;
 pub(crate) mod processor;
@@ -19,7 +19,8 @@ pub(crate) mod sender;
 pub(crate) mod summary_producer;
 
 const PACKET_DELIVERY_TIMEOUT: Duration = Duration::from_secs(20);
-const MONITOR_RUN_INTERVAL: Duration = Duration::from_secs(2 * 60);
+const MONITOR_RUN_INTERVAL: Duration = Duration::from_secs(15 * 60);
+const GATEWAY_PING_INTERVAL: Duration = Duration::from_secs(60);
 
 pub(super) struct Monitor {
     nonce: u64,
@@ -182,11 +183,38 @@ impl Monitor {
         self.nonce += 1;
     }
 
+    async fn ping_all_gateways(&mut self) {
+        self.packet_sender.ping_all_active_gateways().await;
+    }
+
     pub(crate) async fn run(&mut self) {
+        // start from 0 to run test immediately on startup
+        let test_delay = sleep(Duration::from_secs(0));
+        tokio::pin!(test_delay);
+
+        let ping_delay = sleep(GATEWAY_PING_INTERVAL);
+        tokio::pin!(ping_delay);
+
         loop {
-            self.test_run().await;
-            info!(target: "Monitor", "Next test run will happen in {:?}", MONITOR_RUN_INTERVAL);
-            sleep(MONITOR_RUN_INTERVAL).await;
+            tokio::select! {
+                _ = &mut test_delay => {
+                    self.test_run().await;
+                    info!(target: "Monitor", "Next test run will happen in {:?}", MONITOR_RUN_INTERVAL);
+
+                    let now = Instant::now();
+                    test_delay.as_mut().reset(now + MONITOR_RUN_INTERVAL);
+                    // since we just sent packets through gateways, there's no need to ping them
+                    ping_delay.as_mut().reset(now + GATEWAY_PING_INTERVAL);
+
+                }
+                _ = &mut ping_delay => {
+                    info!(target: "Monitor", "Pinging all active gateways");
+                    self.ping_all_gateways().await;
+
+                    let now = Instant::now();
+                    ping_delay.as_mut().reset(now + GATEWAY_PING_INTERVAL);
+                }
+            }
         }
     }
 }

network-monitor/src/monitor/sender.rs

@@ -20,6 +20,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 use std::task::Poll;
 use std::time::Duration;
+use tokio::time::Instant;
 
 pub(crate) struct GatewayPackets {
     /// Network address of the target gateway if wanted to be accessed by the client.
@@ -304,6 +305,52 @@ impl PacketSender {
             }
         })
     }
+
+    pub(super) async fn ping_all_active_gateways(&mut self) {
+        if self.active_gateway_clients.is_empty() {
+            info!(target: "Monitor", "no gateways to ping");
+            return;
+        }
+
+        let ping_start = Instant::now();
+
+        let mut clients_to_purge = Vec::new();
+
+        // since we don't need to wait for response, we can just ping all gateways sequentially
+        // if it becomes problem later on, we can adjust it.
+        for (gateway_id, active_client) in self.active_gateway_clients.iter_mut() {
+            if let Err(err) = active_client.send_ping_message().await {
+                warn!(
+                    target: "Monitor",
+                    "failed to send ping message to gateway {} - {} - assuming the connection is dead.",
+                    active_client.gateway_identity().to_base58_string(),
+                    err,
+                );
+                clients_to_purge.push(*gateway_id);
+            }
+        }
+
+        // purge all dead connections
+        for gateway_id in clients_to_purge.into_iter() {
+            // if this unwrap failed it means something extremely weird is going on
+            // and we got some solar flare bitflip type of corruption
+            let gateway_key = identity::PublicKey::from_bytes(&gateway_id)
+                .expect("failed to recover gateways public key from valid bytes");
+
+            // remove the gateway listener channels
+            self.fresh_gateway_client_data
+                .gateways_status_updater
+                .unbounded_send(GatewayClientUpdate::Failure(gateway_key))
+                .expect("packet receiver seems to have died!");
+
+            // and remove it from our cache
+            self.active_gateway_clients.remove(&gateway_id);
+        }
+
+        let ping_end = Instant::now();
+        let time_taken = ping_end.duration_since(ping_start);
+        debug!(target: "Monitor", "pinging all active gateways took {:?}", time_taken);
+    }
 }
 
 // A slightly modified and less generic version of the futures' ForEachConcurrent that allows the futures to return