Client packet counters (#4325)

* WIP: put in some packet counters

* ws packet counters

* wip

* Add static counters to client traffic stream

* Tweak status log message

* Add packet statistics control

* fixup! Add static counters to client traffic stream

* tweak log

* Move the packet statistics control one level up

* Redo packet stats control to collect locally

* Switch loop cover traffic report over to new channel mechanism

* Switch packet stats in real message stream to channel report

* Finished switching over to channel reporting

* Fix handle stats event

* Log packets received

* Tidy up

* rustfmt

* Add strongly typed stats reporter

* Count cover packets as well

* Log packet sizes sent

* Also log recieved sizes

common/client-core/src/client/base_client/mod.rs

@@ -1,6 +1,7 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use super::packet_statistics_control::PacketStatisticsReporter;
 use super::received_buffer::ReceivedBufferMessage;
 use super::topology_control::geo_aware_provider::GeoAwareTopologyProvider;
 use crate::client::base_client::storage::gateway_details::GatewayDetailsStore;
@@ -10,6 +11,7 @@ use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputM
 use crate::client::key_manager::persistence::KeyStore;
 use crate::client::mix_traffic::transceiver::{GatewayReceiver, GatewayTransceiver, RemoteGateway};
 use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
+use crate::client::packet_statistics_control::PacketStatisticsControl;
 use crate::client::real_messages_control;
 use crate::client::real_messages_control::RealMessagesController;
 use crate::client::received_buffer::{
@@ -254,6 +256,7 @@ where
         self_address: Recipient,
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
+        stats_tx: PacketStatisticsReporter,
         shutdown: TaskClient,
     ) {
         info!("Starting loop cover traffic stream...");
@@ -266,6 +269,7 @@ where
             topology_accessor,
             debug_config.traffic,
             debug_config.cover_traffic,
+            stats_tx,
         );
 
         stream.start_with_shutdown(shutdown);
@@ -285,6 +289,7 @@ where
         client_connection_rx: ConnectionCommandReceiver,
         shutdown: TaskClient,
         packet_type: PacketType,
+        stats_tx: PacketStatisticsReporter,
     ) {
         info!("Starting real traffic stream...");
 
@@ -299,6 +304,7 @@ where
             reply_controller_receiver,
             lane_queue_lengths,
             client_connection_rx,
+            stats_tx,
         )
         .start_with_shutdown(shutdown, packet_type);
     }
@@ -312,6 +318,7 @@ where
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
         shutdown: TaskClient,
+        packet_statistics_control: PacketStatisticsReporter,
     ) {
         info!("Starting received messages buffer controller...");
         let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
@@ -321,6 +328,7 @@ where
                 mixnet_receiver,
                 reply_key_storage,
                 reply_controller_sender,
+                packet_statistics_control,
             );
         controller.start_with_shutdown(shutdown)
     }
@@ -506,6 +514,13 @@ where
         Ok(())
     }
 
+    fn start_packet_statistics_control(shutdown: TaskClient) -> PacketStatisticsReporter {
+        info!("Starting packet statistics control...");
+        let (packet_statistics_control, packet_stats_reporter) = PacketStatisticsControl::new();
+        packet_statistics_control.start_with_shutdown(shutdown);
+        packet_stats_reporter
+    }
+
     fn start_mix_traffic_controller(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
         shutdown: TaskClient,
@@ -633,6 +648,9 @@ where
         )
         .await?;
 
+        let packet_stats_reporter =
+            Self::start_packet_statistics_control(shutdown.fork("packet_statistics_control"));
+
         let gateway_packet_router = PacketRouter::new(
             ack_sender,
             mixnet_messages_sender,
@@ -662,6 +680,7 @@ where
             reply_storage.key_storage(),
             reply_controller_sender.clone(),
             shutdown.fork("received_messages_buffer"),
+            packet_stats_reporter.clone(),
         );
 
         // The message_sender is the transmitter for any component generating sphinx packets
@@ -700,6 +719,7 @@ where
             client_connection_rx,
             shutdown.fork("real_traffic_controller"),
             self.config.debug.traffic.packet_type,
+            packet_stats_reporter.clone(),
         );
 
         if !self
@@ -714,6 +734,7 @@ where
                 self_address,
                 shared_topology_accessor.clone(),
                 message_sender,
+                packet_stats_reporter,
                 shutdown.fork("cover_traffic_stream"),
             );
         }

common/client-core/src/client/cover_traffic_stream.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::client::mix_traffic::BatchMixMessageSender;
+use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter};
 use crate::client::topology_control::TopologyAccessor;
 use crate::{config, spawn_future};
 use futures::task::{Context, Poll};
@@ -61,6 +62,8 @@ where
     secondary_packet_size: Option<PacketSize>,
 
     packet_type: PacketType,
+
+    stats_tx: PacketStatisticsReporter,
 }
 
 impl<R> Stream for LoopCoverTrafficStream<R>
@@ -97,7 +100,8 @@ where
 // obviously when we finally make shared rng that is on 'higher' level, this should become
 // generic `R`
 impl LoopCoverTrafficStream<OsRng> {
-    pub fn new(
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn new(
         ack_key: Arc<AckKey>,
         average_ack_delay: Duration,
         mix_tx: BatchMixMessageSender,
@@ -105,6 +109,7 @@ impl LoopCoverTrafficStream<OsRng> {
         topology_access: TopologyAccessor,
         traffic_config: config::Traffic,
         cover_config: config::CoverTraffic,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         let rng = OsRng;
 
@@ -122,6 +127,7 @@ impl LoopCoverTrafficStream<OsRng> {
             primary_packet_size: traffic_config.primary_packet_size,
             secondary_packet_size: traffic_config.secondary_packet_size,
             packet_type: traffic_config.packet_type,
+            stats_tx,
         }
     }
 
@@ -191,6 +197,10 @@ impl LoopCoverTrafficStream<OsRng> {
                     log::warn!("Failed to send cover message - channel closed");
                 }
             }
+        } else {
+            self.stats_tx.report(PacketStatisticsEvent::CoverPacketSent(
+                cover_traffic_packet_size.size(),
+            ));
         }
 
         // TODO: I'm not entirely sure whether this is really required, because I'm not 100%

common/client-core/src/client/mod.rs

@@ -7,6 +7,7 @@ pub(crate) mod helpers;
 pub mod inbound_messages;
 pub mod key_manager;
 pub mod mix_traffic;
+pub(crate) mod packet_statistics_control;
 pub mod real_messages_control;
 pub mod received_buffer;
 pub mod replies;

common/client-core/src/client/packet_statistics_control.rs

@@ -0,0 +1,202 @@
+use std::time::Duration;
+
+use crate::spawn_future;
+
+// Time interval between reporting packet statistics
+const PACKET_REPORT_INTERVAL_SECS: u64 = 2;
+
+#[derive(Default, Debug)]
+struct PacketStatistics {
+    // Sent
+    real_packets_sent: u64,
+    real_packets_sent_size: usize,
+    cover_packets_sent: u64,
+    cover_packets_sent_size: usize,
+
+    // Received
+    real_packets_received: u64,
+    real_packets_received_size: usize,
+    cover_packets_received: u64,
+    cover_packets_received_size: usize,
+
+    // Acks
+    total_acks_received: u64,
+    total_acks_received_size: usize,
+    real_acks_received: u64,
+    real_acks_received_size: usize,
+    cover_acks_received: u64,
+    cover_acks_received_size: usize,
+
+    // Types of packets queued
+    // TODO: track the type sent instead
+    real_packets_queued: u64,
+    retransmissions_queued: u64,
+    reply_surbs_queued: u64,
+    additional_reply_surbs_queued: u64,
+}
+
+impl PacketStatistics {
+    fn handle_event(&mut self, event: PacketStatisticsEvent) {
+        match event {
+            PacketStatisticsEvent::RealPacketSent(packet_size) => {
+                self.real_packets_sent += 1;
+                self.real_packets_sent_size += packet_size;
+            }
+            PacketStatisticsEvent::CoverPacketSent(packet_size) => {
+                self.cover_packets_sent += 1;
+                self.cover_packets_sent_size += packet_size;
+            }
+            PacketStatisticsEvent::RealPacketReceived(packet_size) => {
+                self.real_packets_received += 1;
+                self.real_packets_received_size += packet_size;
+            }
+            PacketStatisticsEvent::CoverPacketReceived(packet_size) => {
+                self.cover_packets_received += 1;
+                self.cover_packets_received_size += packet_size;
+            }
+            PacketStatisticsEvent::AckReceived(packet_size) => {
+                self.total_acks_received += 1;
+                self.total_acks_received_size += packet_size;
+            }
+            PacketStatisticsEvent::RealAckReceived(packet_size) => {
+                self.real_acks_received += 1;
+                self.real_acks_received_size += packet_size;
+            }
+            PacketStatisticsEvent::CoverAckReceived(packet_size) => {
+                self.cover_acks_received += 1;
+                self.cover_acks_received_size += packet_size;
+            }
+            PacketStatisticsEvent::RealPacketQueued => {
+                self.real_packets_queued += 1;
+            }
+            PacketStatisticsEvent::RetransmissionQueued => {
+                self.retransmissions_queued += 1;
+            }
+            PacketStatisticsEvent::ReplySurbRequestQueued => {
+                self.reply_surbs_queued += 1;
+            }
+            PacketStatisticsEvent::AdditionalReplySurbRequestQueued => {
+                self.additional_reply_surbs_queued += 1;
+            }
+        }
+    }
+}
+
+pub(crate) enum PacketStatisticsEvent {
+    // The real packets sent. Recall that acks are sent by the gateway, so it's not included here.
+    RealPacketSent(usize),
+    // The cover packets sent
+    CoverPacketSent(usize),
+
+    // Real packets received
+    RealPacketReceived(usize),
+    // Cover packets received
+    CoverPacketReceived(usize),
+
+    // Ack of any type received. This is mostly used as a consistency check, and should be the sum
+    // of real and cover acks received.
+    AckReceived(usize),
+    // Out of the total acks received, this is the subset of those that were real
+    RealAckReceived(usize),
+    // Out of the total acks received, this is the subset of those that were for cover traffic
+    CoverAckReceived(usize),
+
+    // Types of packets queued
+    RealPacketQueued,
+    RetransmissionQueued,
+    ReplySurbRequestQueued,
+    AdditionalReplySurbRequestQueued,
+}
+
+type PacketStatisticsReceiver = tokio::sync::mpsc::UnboundedReceiver<PacketStatisticsEvent>;
+
+#[derive(Clone)]
+pub(crate) struct PacketStatisticsReporter {
+    stats_tx: tokio::sync::mpsc::UnboundedSender<PacketStatisticsEvent>,
+}
+
+impl PacketStatisticsReporter {
+    pub(crate) fn new(stats_tx: tokio::sync::mpsc::UnboundedSender<PacketStatisticsEvent>) -> Self {
+        Self { stats_tx }
+    }
+
+    pub(crate) fn report(&self, event: PacketStatisticsEvent) {
+        self.stats_tx.send(event).unwrap_or_else(|err| {
+            log::error!("Failed to report packet stat: {:?}", err);
+        });
+    }
+}
+
+pub(crate) struct PacketStatisticsControl {
+    // Incoming packet stats events from other tasks
+    stats_rx: PacketStatisticsReceiver,
+    stats: PacketStatistics,
+}
+
+impl PacketStatisticsControl {
+    pub(crate) fn new() -> (Self, PacketStatisticsReporter) {
+        let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel();
+        (
+            Self {
+                stats_rx,
+                stats: PacketStatistics::default(),
+            },
+            PacketStatisticsReporter::new(stats_tx),
+        )
+    }
+
+    fn report_statistics(&self) {
+        log::trace!("packet statistics: {:?}", &self.stats);
+        log::info!(
+            "packets sent: {} (real: {}, cover: {}, retransmissions: {})",
+            self.stats.real_packets_sent + self.stats.cover_packets_sent,
+            self.stats.real_packets_sent,
+            self.stats.cover_packets_sent,
+            self.stats.retransmissions_queued,
+        );
+        log::info!(
+            "packets received: {}, (real: {}, cover: {}, acks: {}, acks for cover: {})",
+            self.stats.real_packets_received + self.stats.cover_packets_received,
+            self.stats.real_packets_received,
+            self.stats.cover_packets_received,
+            self.stats.real_acks_received,
+            self.stats.cover_acks_received,
+        );
+    }
+
+    pub(crate) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) {
+        log::debug!("Started PacketStatisticsControl with graceful shutdown support");
+
+        let interval = Duration::from_secs(PACKET_REPORT_INTERVAL_SECS);
+        let mut interval = tokio::time::interval(interval);
+
+        loop {
+            tokio::select! {
+                stats_event = self.stats_rx.recv() => match stats_event {
+                    Some(stats_event) => {
+                        log::trace!("PacketStatisticsControl: Received stats event");
+                        self.stats.handle_event(stats_event);
+                    },
+                    None => {
+                        log::trace!("PacketStatisticsControl: stopping since stats channel was closed");
+                        break;
+                    }
+                },
+                _ = interval.tick() => {
+                    self.report_statistics();
+                }
+                _ = shutdown.recv_with_delay() => {
+                    log::trace!("PacketStatisticsControl: Received shutdown");
+                    break;
+                },
+            }
+        }
+        log::debug!("PacketStatisticsControl: Exiting");
+    }
+
+    pub(crate) fn start_with_shutdown(mut self, task_client: nym_task::TaskClient) {
+        spawn_future(async move {
+            self.run_with_shutdown(task_client).await;
+        })
+    }
+}

common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs

@@ -1,6 +1,8 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter};
+
 use super::action_controller::{AckActionSender, Action};
 use futures::StreamExt;
 use log::*;
@@ -17,6 +19,7 @@ pub(super) struct AcknowledgementListener {
     ack_key: Arc<AckKey>,
     ack_receiver: AcknowledgementReceiver,
     action_sender: AckActionSender,
+    stats_tx: PacketStatisticsReporter,
 }
 
 impl AcknowledgementListener {
@@ -24,16 +27,21 @@ impl AcknowledgementListener {
         ack_key: Arc<AckKey>,
         ack_receiver: AcknowledgementReceiver,
         action_sender: AckActionSender,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         AcknowledgementListener {
             ack_key,
             ack_receiver,
             action_sender,
+            stats_tx,
         }
     }
 
     async fn on_ack(&mut self, ack_content: Vec<u8>) {
         trace!("Received an ack");
+        self.stats_tx
+            .report(PacketStatisticsEvent::AckReceived(ack_content.len()));
+
         let frag_id = match recover_identifier(&self.ack_key, &ack_content)
             .map(FragmentIdentifier::try_from_bytes)
         {
@@ -48,11 +56,14 @@ impl AcknowledgementListener {
         // because nothing was inserted in the first place
         if frag_id == COVER_FRAG_ID {
             trace!("Received an ack for a cover message - no need to do anything");
+            self.stats_tx
+                .report(PacketStatisticsEvent::CoverAckReceived(ack_content.len()));
             return;
         }
 
         trace!("Received {} from the mix network", frag_id);
-
+        self.stats_tx
+            .report(PacketStatisticsEvent::RealAckReceived(ack_content.len()));
         self.action_sender
             .unbounded_send(Action::new_remove(frag_id))
             .unwrap();

common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs

@@ -8,6 +8,7 @@ use self::{
     sent_notification_listener::SentNotificationListener,
 };
 use crate::client::inbound_messages::InputMessageReceiver;
+use crate::client::packet_statistics_control::PacketStatisticsReporter;
 use crate::client::real_messages_control::message_handler::MessageHandler;
 use crate::client::replies::reply_controller::ReplyControllerSender;
 use crate::spawn_future;
@@ -208,6 +209,7 @@ where
         connectors: AcknowledgementControllerConnectors,
         message_handler: MessageHandler<R>,
         reply_controller_sender: ReplyControllerSender,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
 
@@ -224,6 +226,7 @@ where
             Arc::clone(&ack_key),
             connectors.ack_receiver,
             connectors.ack_action_sender.clone(),
+            stats_tx,
         );
 
         // will listen for any new messages from the client

common/client-core/src/client/real_messages_control/mod.rs

@@ -35,6 +35,8 @@ use crate::client::replies::reply_controller;
 use crate::config;
 pub(crate) use acknowledgement_control::{AckActionSender, Action};
 
+use super::packet_statistics_control::PacketStatisticsReporter;
+
 pub(crate) mod acknowledgement_control;
 pub(crate) mod message_handler;
 pub(crate) mod real_traffic_stream;
@@ -143,6 +145,7 @@ impl RealMessagesController<OsRng> {
         reply_controller_receiver: ReplyControllerReceiver,
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         let rng = OsRng;
 
@@ -181,6 +184,7 @@ impl RealMessagesController<OsRng> {
             ack_controller_connectors,
             message_handler.clone(),
             reply_controller_sender,
+            stats_tx.clone(),
         );
 
         let reply_control = ReplyController::new(
@@ -199,6 +203,7 @@ impl RealMessagesController<OsRng> {
             topology_access,
             lane_queue_lengths,
             client_connection_rx,
+            stats_tx,
         );
 
         RealMessagesController {

common/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -3,6 +3,7 @@
 
 use self::sending_delay_controller::SendingDelayController;
 use crate::client::mix_traffic::BatchMixMessageSender;
+use crate::client::packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter};
 use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender;
 use crate::client::topology_control::TopologyAccessor;
 use crate::client::transmission_buffer::TransmissionBuffer;
@@ -113,6 +114,9 @@ where
 
     /// Report queue lengths so that upstream can backoff sending data, and keep connections open.
     lane_queue_lengths: LaneQueueLengths,
+
+    /// Channel used for sending statistics events to `PacketStatisticsControl`.
+    stats_tx: PacketStatisticsReporter,
 }
 
 #[derive(Debug)]
@@ -171,6 +175,7 @@ where
         topology_access: TopologyAccessor,
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         OutQueueControl {
             config,
@@ -184,6 +189,7 @@ where
             transmission_buffer: TransmissionBuffer::new(),
             client_connection_rx,
             lane_queue_lengths,
+            stats_tx,
         }
     }
 
@@ -214,7 +220,7 @@ where
     async fn on_message(&mut self, next_message: StreamMessage) {
         trace!("created new message");
 
-        let (next_message, fragment_id) = match next_message {
+        let (next_message, fragment_id, packet_size) = match next_message {
             StreamMessage::Cover => {
                 let cover_traffic_packet_size = self.loop_cover_message_size();
                 trace!("the next loop cover message will be put in a {cover_traffic_packet_size} packet");
@@ -250,15 +256,28 @@ where
                         "Somehow failed to generate a loop cover message with a valid topology",
                     ),
                     None,
+                    cover_traffic_packet_size.size(),
                 )
             }
             StreamMessage::Real(real_message) => {
-                (real_message.mix_packet, real_message.fragment_id)
+                let packet_size = real_message.packet_size();
+                (
+                    real_message.mix_packet,
+                    real_message.fragment_id,
+                    packet_size,
+                )
             }
         };
 
         if let Err(err) = self.mix_tx.send(vec![next_message]).await {
             log::error!("Failed to send: {err}");
+        } else {
+            let event = if fragment_id.is_some() {
+                PacketStatisticsEvent::RealPacketSent(packet_size)
+            } else {
+                PacketStatisticsEvent::CoverPacketSent(packet_size)
+            };
+            self.stats_tx.report(event);
         }
 
         // notify ack controller about sending our message only after we actually managed to push it
@@ -340,6 +359,28 @@ where
         let lane_length = self.transmission_buffer.lane_length(&lane);
         self.lane_queue_lengths.set(&lane, lane_length);
 
+        // This is the last step in the pipeline where we know the type of the message, so
+        // lets count the number of retransmissions and reply surb messages sent here.
+        let stat_event = match lane {
+            TransmissionLane::General => None,
+            TransmissionLane::ConnectionId(_) => None,
+            TransmissionLane::ReplySurbRequest => {
+                Some(PacketStatisticsEvent::ReplySurbRequestQueued)
+            }
+            TransmissionLane::AdditionalReplySurbs => {
+                Some(PacketStatisticsEvent::AdditionalReplySurbRequestQueued)
+            }
+            TransmissionLane::Retransmission => Some(PacketStatisticsEvent::RetransmissionQueued),
+        };
+        if let Some(stat_event) = stat_event {
+            self.stats_tx.report(stat_event);
+        }
+        // To avoid comparing apples to oranges when presenting the fraction of packets that are
+        // retransmissions, we also need to keep track to the total number of real messages queued,
+        // even though we also track the actual number of messages sent later in the pipeline.
+        self.stats_tx
+            .report(PacketStatisticsEvent::RealPacketQueued);
+
         Some(real_next)
     }
 
@@ -433,6 +474,13 @@ where
             Poll::Ready(Some((real_messages, conn_id))) => {
                 log::trace!("handling real_messages: size: {}", real_messages.len());
 
+                // This is the last step in the pipeline where we know the type of the message, so
+                // lets count the number of retransmissions here.
+                if conn_id == TransmissionLane::Retransmission {
+                    self.stats_tx
+                        .report(PacketStatisticsEvent::RetransmissionQueued);
+                }
+
                 // First store what we got for the given connection id
                 self.transmission_buffer.store(&conn_id, real_messages);
                 let real_next = self.pop_next_message().expect("we just added one");
@@ -471,10 +519,10 @@ where
         let mult = self.sending_delay_controller.current_multiplier();
         let delay = self.current_average_message_sending_delay().as_millis();
         let status_str = if self.config.traffic.disable_main_poisson_packet_distribution {
-            format!("Status: {lanes} lanes, backlog: {backlog:.2} kiB ({packets}), no delay")
+            format!("Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, no delay")
         } else {
             format!(
-                "Status: {lanes} lanes, backlog: {backlog:.2} kiB ({packets}), avg delay: {delay}ms ({mult})"
+                "Packet backlog: {backlog:.2} kiB ({packets}), {lanes} lanes, avg delay: {delay}ms ({mult})"
             )
         };
         if packets > 1000 {

common/client-core/src/client/received_buffer.rs

@@ -1,8 +1,10 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::client::replies::reply_controller::ReplyControllerSender;
-use crate::client::replies::reply_storage::SentReplyKeys;
+use crate::client::{
+    packet_statistics_control::{PacketStatisticsEvent, PacketStatisticsReporter},
+    replies::{reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys},
+};
 use crate::spawn_future;
 use futures::channel::mpsc;
 use futures::lock::Mutex;
@@ -43,15 +45,26 @@ struct ReceivedMessagesBufferInner<R: MessageReceiver> {
     // but perhaps it should be changed to include timestamps of when the message was reconstructed
     // and every now and then remove ids older than X
     recently_reconstructed: HashSet<i32>,
+
+    stats_tx: PacketStatisticsReporter,
 }
 
 impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
     fn recover_from_fragment(&mut self, fragment_data: &[u8]) -> Option<NymMessage> {
         if nym_sphinx::cover::is_cover(fragment_data) {
             trace!("The message was a loop cover message! Skipping it");
+            self.stats_tx
+                .report(PacketStatisticsEvent::CoverPacketReceived(
+                    fragment_data.len(),
+                ));
             return None;
         }
 
+        self.stats_tx
+            .report(PacketStatisticsEvent::RealPacketReceived(
+                fragment_data.len(),
+            ));
+
         let fragment = match self.message_receiver.recover_fragment(fragment_data) {
             Err(err) => {
                 warn!("failed to recover fragment from raw data: {err}. The whole underlying message might be corrupted and unrecoverable!");
@@ -141,6 +154,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
         local_encryption_keypair: Arc<encryption::KeyPair>,
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
+        stats_tx: PacketStatisticsReporter,
     ) -> Self {
         ReceivedMessagesBuffer {
             inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -149,6 +163,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                 message_receiver: R::new(),
                 message_sender: None,
                 recently_reconstructed: HashSet::new(),
+                stats_tx,
             })),
             reply_key_storage,
             reply_controller_sender,
@@ -353,7 +368,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                 };
 
             if let Some(completed) = completed_message {
-                info!("received {completed}");
+                debug!("received {completed}");
                 completed_messages.push(completed)
             }
         }
@@ -480,11 +495,13 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
         mixnet_packet_receiver: MixnetMessageReceiver,
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
+        packet_statistics_reporter: PacketStatisticsReporter,
     ) -> Self {
         let received_buffer = ReceivedMessagesBuffer::new(
             local_encryption_keypair,
             reply_key_storage,
             reply_controller_sender,
+            packet_statistics_reporter,
         );
 
         ReceivedMessagesBufferController {

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

@@ -354,7 +354,10 @@ impl<C, St> GatewayClient<C, St> {
         msg: Message,
     ) -> Result<(), GatewayClientError> {
         match self.connection {
-            SocketState::Available(ref mut conn) => Ok(conn.send(msg).await?),
+            SocketState::Available(ref mut conn) => {
+                conn.send(msg).await?;
+                Ok(())
+            }
             SocketState::PartiallyDelegated(ref mut partially_delegated) => {
                 if let Err(err) = partially_delegated.send_without_response(msg).await {
                     error!("failed to send message without response - {err}...");