feat: NMv3: forwarding of final hop network monitor packets

nym-node/Cargo.toml

@@ -151,6 +151,8 @@ cargo_metadata = { workspace = true }
 nym-lp = { workspace = true, default-features = true, features = ["mock"] }
 criterion = { workspace = true, features = ["async_tokio"] }
 nym-test-utils = { workspace = true }
+# for standing up an in-memory gateway storage in final-hop tests
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite"] }
 strum = { workspace = true }
 strum_macros = { workspace = true }
 

nym-node/nym-node-metrics/src/mixnet.rs

@@ -168,6 +168,12 @@ pub struct EgressRecipientStats {
 pub struct EgressMixingStats {
     disk_persisted_packets: AtomicUsize,
 
+    // final hop packets from an authorised network monitor agent, split by whether the recipient
+    // session was live. monitor packets are never persisted, so these two account for all of them
+    monitor_final_hop_packets_delivered: AtomicUsize,
+
+    monitor_final_hop_packets_dropped: AtomicUsize,
+
     // final hop packets dropped because the recipient has never registered with this gateway,
     // so the payload could never have been retrieved
     unknown_recipient_dropped_packets: AtomicUsize,
@@ -191,6 +197,26 @@ impl EgressMixingStats {
         self.disk_persisted_packets.load(Ordering::Relaxed)
     }
 
+    pub fn add_monitor_final_hop_packet_delivered(&self) {
+        self.monitor_final_hop_packets_delivered
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn monitor_final_hop_packets_delivered(&self) -> usize {
+        self.monitor_final_hop_packets_delivered
+            .load(Ordering::Relaxed)
+    }
+
+    pub fn add_monitor_final_hop_packet_dropped(&self) {
+        self.monitor_final_hop_packets_dropped
+            .fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn monitor_final_hop_packets_dropped(&self) -> usize {
+        self.monitor_final_hop_packets_dropped
+            .load(Ordering::Relaxed)
+    }
+
     pub fn add_unknown_recipient_dropped_packet(&self) {
         self.unknown_recipient_dropped_packets
             .fetch_add(1, Ordering::Relaxed);

nym-node/nym-node-metrics/src/prometheus_wrapper.rs

@@ -97,6 +97,16 @@ pub enum PrometheusMetric {
     ))]
     MixnetEgressStoredOnDiskFinalHopPackets,
 
+    #[strum(props(
+        help = "The number of network monitor final hop packets delivered to a live client session"
+    ))]
+    MixnetEgressMonitorFinalHopPacketsDelivered,
+
+    #[strum(props(
+        help = "The number of network monitor final hop packets dropped for want of a live client session (they are never stored on disk)"
+    ))]
+    MixnetEgressMonitorFinalHopPacketsDropped,
+
     #[strum(props(
         help = "The number of unwrapped final hop packets dropped because their recipient has never registered with this gateway"
     ))]
@@ -326,6 +336,12 @@ impl PrometheusMetric {
             PrometheusMetric::MixnetEgressStoredOnDiskFinalHopPackets => {
                 Metric::new_int_gauge(&name, help)
             }
+            PrometheusMetric::MixnetEgressMonitorFinalHopPacketsDelivered => {
+                Metric::new_int_gauge(&name, help)
+            }
+            PrometheusMetric::MixnetEgressMonitorFinalHopPacketsDropped => {
+                Metric::new_int_gauge(&name, help)
+            }
             PrometheusMetric::MixnetEgressUnknownRecipientDroppedFinalHopPackets => {
                 Metric::new_int_gauge(&name, help)
             }
@@ -499,7 +515,7 @@ mod tests {
         // a sanity check for anyone adding new metrics. if this test fails,
         // make sure any methods on `PrometheusMetric` enum don't need updating
         // or require custom Display impl
-        assert_eq!(54, PrometheusMetric::COUNT)
+        assert_eq!(56, PrometheusMetric::COUNT)
     }
 
     #[test]

nym-node/src/node/metrics/handler/global_prometheus_updater/mod.rs

@@ -90,6 +90,20 @@ impl OnUpdateMetricsHandler for PrometheusGlobalNodeMetricsRegistryUpdater {
             MixnetEgressStoredOnDiskFinalHopPackets,
             self.metrics.mixnet.egress.disk_persisted_packets() as i64,
         );
+        self.prometheus_wrapper.set(
+            MixnetEgressMonitorFinalHopPacketsDelivered,
+            self.metrics
+                .mixnet
+                .egress
+                .monitor_final_hop_packets_delivered() as i64,
+        );
+        self.prometheus_wrapper.set(
+            MixnetEgressMonitorFinalHopPacketsDropped,
+            self.metrics
+                .mixnet
+                .egress
+                .monitor_final_hop_packets_dropped() as i64,
+        );
         self.prometheus_wrapper.set(
             MixnetEgressUnknownRecipientDroppedFinalHopPackets,
             self.metrics

nym-node/src/node/mixnet/handler.rs

@@ -3,6 +3,7 @@
 
 use crate::node::key_rotation::active_keys::SphinxKeyGuard;
 use crate::node::mixnet::shared::SharedData;
+use crate::node::mixnet::shared::final_hop::FinalHopResult;
 use futures::StreamExt;
 use nym_mixnet_client::metrics::{MixnetMetric, PacketTrace, Traced};
 use nym_noise::connection::Connection;
@@ -267,65 +268,64 @@ impl ConnectionHandler {
             return;
         }
 
-        if network_monitor_packet {
-            warn!(
-                event = "packet.dropped.network_monitor_final_hop",
-                remote_addr = %self.remote_address,
-                "dropping packet: unsupported network monitor final hop packets"
-            );
-            self.shared
-                .dropped_final_hop_packet(self.remote_address.ip());
-            return;
-        }
-
         let client = final_hop_data.destination;
         let message = final_hop_data.message;
         let has_ack = final_hop_data.forward_ack.is_some();
 
-        // if possible attempt to push message directly to the client
-        match self.shared.try_push_message_to_client(client, message) {
-            Err(unsent_plaintext) => {
-                // if that failed, store it on disk
-                Span::current().record("client_online", false);
-                match self
-                    .shared
-                    .store_processed_packet_payload(client, unsent_plaintext)
-                    .await
-                {
-                    Err(err) => error!("Failed to store client data - {err}"),
-                    Ok(true) => {
-                        Span::current().record("disk_fallback", true);
-                        self.shared
-                            .metrics
-                            .mixnet
-                            .egress
-                            .add_disk_persisted_packet();
-                        trace!("Stored packet for {client}")
-                    }
-                    // nothing was stored: the recipient has never registered with this gateway
-                    Ok(false) => {
-                        debug!(
-                            event = "packet.dropped.unknown_recipient",
-                            remote_addr = %self.remote_address,
-                            "dropping packet: {client} has never registered with this gateway, so nothing could ever retrieve it"
-                        );
-                        self.shared
-                            .metrics
-                            .mixnet
-                            .egress
-                            .add_unknown_recipient_dropped_packet();
-                        self.shared
-                            .dropped_final_hop_packet(self.remote_address.ip());
-
-                        // the ack is still forwarded below: withholding it would tell the sender
-                        // whether the recipient is registered with this gateway
-                    }
-                }
-            }
-            Ok(_) => {
+        // if possible push the message directly to the client, otherwise fall back to disk -
+        // unless it came from a monitor, whose packets are never persisted
+        match self
+            .shared
+            .deliver_final_hop(client, message, network_monitor_packet)
+            .await
+        {
+            FinalHopResult::Delivered => {
                 Span::current().record("client_online", true);
+                if network_monitor_packet {
+                    self.shared
+                        .metrics
+                        .mixnet
+                        .egress
+                        .add_monitor_final_hop_packet_delivered();
+                }
                 trace!("Pushed received packet to {client}");
             }
+            FinalHopResult::Stored => {
+                Span::current().record("client_online", false);
+                Span::current().record("disk_fallback", true);
+                self.shared
+                    .metrics
+                    .mixnet
+                    .egress
+                    .add_disk_persisted_packet();
+                trace!("Stored packet for {client}")
+            }
+            FinalHopResult::StoreFailed(err) => {
+                Span::current().record("client_online", false);
+                error!("Failed to store client data - {err}");
+            }
+            FinalHopResult::DroppedNoSession => {
+                Span::current().record("client_online", false);
+                debug!(
+                    event = "packet.dropped.unknown_recipient",
+                    remote_addr = %self.remote_address,
+                    "dropping packet: no live session for {client}"
+                );
+                if !network_monitor_packet {
+                    self.shared
+                        .metrics
+                        .mixnet
+                        .egress
+                        .add_unknown_recipient_dropped_packet()
+                }
+                self.shared
+                    .metrics
+                    .mixnet
+                    .egress
+                    .add_monitor_final_hop_packet_dropped();
+                self.shared
+                    .dropped_final_hop_packet(self.remote_address.ip());
+            }
         }
 
         // forward the ack regardless of what happened to the payload (pushed, stored, or dropped);

nym-node/src/node/mixnet/shared/final_hop.rs

@@ -8,6 +8,24 @@ use nym_sphinx_types::DestinationAddressBytes;
 use tokio::time::Instant;
 use tracing::{debug, warn};
 
+/// What happened to a final-hop payload.
+pub(crate) enum FinalHopResult {
+    /// Pushed straight into the recipient's live session.
+    Delivered,
+
+    /// No live session, so it went to the recipient's on-disk inbox. The inbox holds no reference
+    /// to `shared_keys`, so this accepts a recipient that never registered here too, whose row is
+    /// then never collected.
+    // NOTE: this will be eventually removed
+    Stored,
+
+    /// No live session, and the store rejected it.
+    StoreFailed(GatewayStorageError),
+
+    /// Came from a client with no live session, so it was neither delivered nor persisted.
+    DroppedNoSession,
+}
+
 #[derive(Clone)]
 pub(crate) struct SharedFinalHopData {
     active_clients: ActiveClientsStore,
@@ -22,6 +40,42 @@ impl SharedFinalHopData {
         }
     }
 
+    /// Push a final-hop payload into the recipient's live session, falling back to their on-disk
+    /// inbox - except for a network monitor's packet, which is dropped instead of persisted.
+    ///
+    /// The monitor's agent scores a probe on what arrived on its socket, so a packet that missed
+    /// the session must be definitively undelivered rather than waiting in an inbox nobody reads,
+    /// and monitor traffic must not accrue undeliverable rows on every gateway in the network.
+    pub(crate) async fn deliver_final_hop(
+        &self,
+        client_address: DestinationAddressBytes,
+        message: Vec<u8>,
+        network_monitor_packet: bool,
+    ) -> FinalHopResult {
+        let unsent = match self.try_push_message_to_client(client_address, message) {
+            Ok(()) => return FinalHopResult::Delivered,
+            Err(unsent) => unsent,
+        };
+
+        if network_monitor_packet {
+            return FinalHopResult::DroppedNoSession;
+        }
+
+        match self
+            .store_processed_packet_payload(client_address, unsent)
+            .await
+        {
+            Ok(stored) => {
+                if stored {
+                    FinalHopResult::Stored
+                } else {
+                    FinalHopResult::DroppedNoSession
+                }
+            }
+            Err(err) => FinalHopResult::StoreFailed(err),
+        }
+    }
+
     pub(crate) fn try_push_message_to_client(
         &self,
         client_address: DestinationAddressBytes,
@@ -94,3 +148,63 @@ impl SharedFinalHopData {
         result
     }
 }
+
+#[cfg(test)]
+#[allow(clippy::unwrap_used, clippy::expect_used)]
+mod tests {
+    use super::*;
+    use nym_sphinx_types::DESTINATION_ADDRESS_LENGTH;
+
+    fn recipient() -> DestinationAddressBytes {
+        DestinationAddressBytes::from_bytes([42u8; DESTINATION_ADDRESS_LENGTH])
+    }
+
+    /// Final hop data over an in-memory store whose active-clients store is empty, so every push
+    /// fails and the fallback decision is the thing under test.
+    async fn no_live_sessions() -> SharedFinalHopData {
+        let pool = sqlx::SqlitePool::connect("sqlite::memory:")
+            .await
+            .expect("failed to create in-memory SQLite pool");
+        let storage = GatewayStorage::from_connection_pool(pool, 100)
+            .await
+            .expect("failed to initialise gateway storage");
+
+        SharedFinalHopData::new(ActiveClientsStore::new(), storage)
+    }
+
+    async fn inbox_of(final_hop: &SharedFinalHopData) -> Vec<Vec<u8>> {
+        final_hop
+            .storage
+            .retrieve_messages(recipient(), None)
+            .await
+            .unwrap()
+            .0
+            .into_iter()
+            .map(|stored| stored.content)
+            .collect()
+    }
+
+    #[tokio::test]
+    async fn monitor_packet_with_no_session_is_dropped_without_touching_the_store() {
+        let final_hop = no_live_sessions().await;
+
+        let result = final_hop
+            .deliver_final_hop(recipient(), b"probe".to_vec(), true)
+            .await;
+
+        assert!(matches!(result, FinalHopResult::DroppedNoSession));
+        assert!(inbox_of(&final_hop).await.is_empty());
+    }
+
+    #[tokio::test]
+    async fn ordinary_packet_with_no_session_is_dropped_without_touching_the_store() {
+        let final_hop = no_live_sessions().await;
+
+        let result = final_hop
+            .deliver_final_hop(recipient(), b"payload".to_vec(), false)
+            .await;
+
+        assert!(matches!(result, FinalHopResult::DroppedNoSession));
+        assert!(inbox_of(&final_hop).await.is_empty());
+    }
+}