diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index 57caff1d58..069070c047 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/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, diff --git a/network-monitor/src/monitor/mod.rs b/network-monitor/src/monitor/mod.rs index 991cc1d45d..92720e54c0 100644 --- a/network-monitor/src/monitor/mod.rs +++ b/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); + } + } } } } diff --git a/network-monitor/src/monitor/sender.rs b/network-monitor/src/monitor/sender.rs index a08663640e..8c97b3a993 100644 --- a/network-monitor/src/monitor/sender.rs +++ b/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