Introduce event backchannel (#6119)

* Introduce even backchannel

* Rust fmt

* Rename Event to MixnetClientEvent

* Use unbounded_send for events

* Remove unused file

* Remove mut borrow

* Event hierarchy and mixnet client intermediary

* Export MixTrafficEvent in sdk

Author: neacsu

Cargo.lock

@@ -7,11 +7,12 @@ use super::statistics_control::StatisticsControl;
 use crate::client::base_client::storage::helpers::store_client_keys;
 use crate::client::base_client::storage::MixnetClientStorage;
 use crate::client::cover_traffic_stream::LoopCoverTrafficStream;
+use crate::client::event_control::EventControl;
 use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender};
 use crate::client::key_manager::persistence::KeyStore;
 use crate::client::key_manager::ClientKeys;
 use crate::client::mix_traffic::transceiver::{GatewayReceiver, GatewayTransceiver, RemoteGateway};
-use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
+use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController, MixTrafficEvent};
 use crate::client::real_messages_control;
 use crate::client::real_messages_control::RealMessagesController;
 use crate::client::received_buffer::{
@@ -66,7 +67,6 @@ use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::mpsc::Sender;
-use tracing::*;
 use url::Url;
 
 #[cfg(target_arch = "wasm32")]
@@ -83,6 +83,23 @@ pub mod non_wasm_helpers;
 pub mod helpers;
 pub mod storage;
 
+#[derive(Clone, Copy, Debug)]
+pub enum MixnetClientEvent {
+    Traffic(MixTrafficEvent),
+}
+
+pub type EventReceiver = mpsc::UnboundedReceiver<MixnetClientEvent>;
+#[derive(Clone)]
+pub struct EventSender(pub mpsc::UnboundedSender<MixnetClientEvent>);
+
+impl EventSender {
+    pub fn send(&self, event: MixnetClientEvent) {
+        if let Err(err) = self.0.unbounded_send(event) {
+            tracing::warn!("Failed to send error event. The caller event reader was closed: {err}");
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct ClientInput {
     pub connection_command_sender: ConnectionCommandSender,
@@ -199,6 +216,7 @@ pub struct BaseClientBuilder<C, S: MixnetClientStorage> {
     custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
     custom_gateway_transceiver: Option<Box<dyn GatewayTransceiver + Send>>,
     shutdown: Option<ShutdownTracker>,
+    event_tx: Option<EventSender>,
     user_agent: Option<UserAgent>,
     custom_nym_api_client: Option<nym_http_api_client::Client>,
 
@@ -228,6 +246,7 @@ where
             custom_topology_provider: None,
             custom_gateway_transceiver: None,
             shutdown: None,
+            event_tx: None,
             user_agent: None,
             custom_nym_api_client: None,
             setup_method: GatewaySetup::MustLoad { gateway_id: None },
@@ -297,6 +316,12 @@ where
         self
     }
 
+    #[must_use]
+    pub fn with_event_tx(mut self, event_tx: EventSender) -> Self {
+        self.event_tx = Some(event_tx);
+        self
+    }
+
     #[must_use]
     pub fn with_user_agent(mut self, user_agent: UserAgent) -> Self {
         self.user_agent = Some(user_agent);
@@ -327,6 +352,18 @@ where
         details.client_address()
     }
 
+    fn start_event_control(
+        parent_event_tx: Option<EventSender>,
+        children_event_rx: EventReceiver,
+        shutdown_tracker: &ShutdownTracker,
+    ) {
+        let event_control = EventControl::new(parent_event_tx, children_event_rx);
+        shutdown_tracker.try_spawn_named_with_shutdown(
+            async move { event_control.run().await },
+            "EventControl",
+        );
+    }
+
     // future constantly pumping loop cover traffic at some specified average rate
     // the pumped traffic goes to the MixTrafficController
     fn start_cover_traffic_stream(
@@ -338,7 +375,7 @@ where
         stats_tx: ClientStatsSender,
         shutdown_tracker: &ShutdownTracker,
     ) {
-        info!("Starting loop cover traffic stream...");
+        tracing::info!("Starting loop cover traffic stream...");
 
         let mut stream = LoopCoverTrafficStream::new(
             ack_key,
@@ -370,7 +407,7 @@ where
         stats_tx: ClientStatsSender,
         shutdown_tracker: &ShutdownTracker,
     ) {
-        info!("Starting real traffic stream...");
+        tracing::info!("Starting real traffic stream...");
 
         let real_messages_controller = RealMessagesController::new(
             controller_config,
@@ -455,7 +492,7 @@ where
         metrics_reporter: ClientStatsSender,
         shutdown_tracker: &ShutdownTracker,
     ) {
-        info!("Starting received messages buffer controller...");
+        tracing::info!("Starting received messages buffer controller...");
         let controller = ReceivedMessagesBufferController::<SphinxMessageReceiver>::new(
             local_encryption_keypair,
             query_receiver,
@@ -566,7 +603,7 @@ where
             details_store
                 .upgrade_stored_remote_gateway_key(gateway_client.gateway_identity(), &updated_key)
                 .await.map_err(|err| {
-                error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
+                tracing::error!("failed to store upgraded gateway key! this connection might be forever broken now: {err}");
                 ClientCoreError::GatewaysDetailsStoreError { source: Box::new(err) }
             })?
         }
@@ -663,7 +700,7 @@ where
 
         if topology_config.disable_refreshing {
             // if we're not spawning the refresher, don't cause shutdown immediately
-            info!("The background topology refresher is not going to be started");
+            tracing::info!("The background topology refresher is not going to be started");
         }
 
         let mut topology_refresher = TopologyRefresher::new(
@@ -673,7 +710,7 @@ where
         );
         // before returning, block entire runtime to refresh the current network view so that any
         // components depending on topology would see a non-empty view
-        info!("Obtaining initial network topology");
+        tracing::info!("Obtaining initial network topology");
         topology_refresher.try_refresh().await;
 
         if let Err(err) = topology_refresher.ensure_topology_is_routable().await {
@@ -699,21 +736,21 @@ where
                     .wait_for_gateway(local_gateway, waiting_timeout)
                     .await
                 {
-                    error!(
+                    tracing::error!(
                         "the gateway did not come back online within the specified timeout: {err}"
                     );
                     return Err(err.into());
                 }
             } else {
-                error!("the gateway we're supposedly connected to does not exist. We'll not be able to send any packets to ourselves: {err}");
+                tracing::error!("the gateway we're supposedly connected to does not exist. We'll not be able to send any packets to ourselves: {err}");
                 return Err(err.into());
             }
         }
 
         if !topology_config.disable_refreshing {
             // don't spawn the refresher if we don't want to be refreshing the topology.
             // only use the initial values obtained
-            info!("Starting topology refresher...");
+            tracing::info!("Starting topology refresher...");
             shutdown_tracker.try_spawn_named_with_shutdown(
                 async move { topology_refresher.run().await },
                 "TopologyRefresher",
@@ -730,7 +767,7 @@ where
         input_sender: Sender<InputMessage>,
         shutdown_tracker: &ShutdownTracker,
     ) -> ClientStatsSender {
-        info!("Starting statistics control...");
+        tracing::info!("Starting statistics control...");
         StatisticsControl::create_and_start(
             config.debug.stats_reporting,
             user_agent
@@ -745,10 +782,14 @@ where
     fn start_mix_traffic_controller(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
         shutdown_tracker: &ShutdownTracker,
+        event_tx: EventSender,
     ) -> (BatchMixMessageSender, ClientRequestSender) {
-        info!("Starting mix traffic controller...");
-        let mut mix_traffic_controller =
-            MixTrafficController::new(gateway_transceiver, shutdown_tracker.clone_shutdown_token());
+        tracing::info!("Starting mix traffic controller...");
+        let mut mix_traffic_controller = MixTrafficController::new(
+            gateway_transceiver,
+            shutdown_tracker.clone_shutdown_token(),
+            event_tx,
+        );
 
         let mix_tx = mix_traffic_controller.mix_rx();
         let client_tx = mix_traffic_controller.client_tx();
@@ -815,7 +856,7 @@ where
     {
         // if client keys do not exist already, create and persist them
         if key_store.load_keys().await.is_err() {
-            info!("could not find valid client keys - a new set will be generated");
+            tracing::info!("could not find valid client keys - a new set will be generated");
             let mut rng = OsRng;
             let keys = if let Some(derivation_material) = derivation_material {
                 ClientKeys::from_master_key(&mut rng, &derivation_material)
@@ -872,7 +913,7 @@ where
         <S::CredentialStore as CredentialStorage>::StorageError: Send + Sync + 'static,
         <S::GatewaysDetailsStore as GatewaysDetailsStore>::StorageError: Sync + Send,
     {
-        info!("Starting nym client");
+        tracing::info!("Starting nym client");
         #[cfg(debug_assertions)]
         #[cfg(target_arch = "wasm32")]
         {
@@ -906,6 +947,9 @@ where
         // channels responsible for controlling real messages
         let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
 
+        // channels responsible for event management
+        let (event_sender, event_receiver) = mpsc::unbounded();
+
         // channels responsible for controlling ack messages
         let (ack_sender, ack_receiver) = mpsc::unbounded();
         let shared_topology_accessor =
@@ -918,6 +962,8 @@ where
             None => nym_task::get_sdk_shutdown_tracker()?,
         };
 
+        Self::start_event_control(self.event_tx, event_receiver, &shutdown_tracker);
+
         // channels responsible for dealing with reply-related fun
         let (reply_controller_sender, reply_controller_receiver) =
             reply_controller::requests::new_control_channels();
@@ -1012,6 +1058,7 @@ where
         let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
             gateway_transceiver,
             &shutdown_tracker.child_tracker(),
+            EventSender(event_sender),
         );
 
         // Channels that the websocket listener can use to signal downstream to the real traffic
@@ -1061,8 +1108,8 @@ where
             );
         }
 
-        debug!("Core client startup finished!");
-        debug!("The address of this client is: {self_address}");
+        tracing::debug!("Core client startup finished!");
+        tracing::debug!("The address of this client is: {self_address}");
 
         #[cfg(debug_assertions)]
         #[cfg(target_arch = "wasm32")]

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

@@ -0,0 +1,40 @@
+// Copyright 2025 - Nym Technologies SA <[email protected]>
+// SPDX-License-Identifier: Apache-2.0
+
+use futures::StreamExt;
+
+use crate::client::base_client::{EventReceiver, EventSender, MixnetClientEvent};
+
+/// Launches and manages task events, propagating upwards what is not strictly internal.
+pub(crate) struct EventControl {
+    parent_event_tx: Option<EventSender>,
+    children_event_rx: EventReceiver,
+}
+
+impl EventControl {
+    pub(crate) fn new(
+        parent_event_tx: Option<EventSender>,
+        children_event_rx: EventReceiver,
+    ) -> Self {
+        EventControl {
+            parent_event_tx,
+            children_event_rx,
+        }
+    }
+
+    fn is_internal(event: MixnetClientEvent) -> bool {
+        match event {
+            MixnetClientEvent::Traffic(_) => false,
+        }
+    }
+
+    pub(crate) async fn run(mut self) {
+        while let Some(event) = self.children_event_rx.next().await {
+            if let Some(parent_event_tx) = &self.parent_event_tx {
+                if !Self::is_internal(event) {
+                    parent_event_tx.send(event);
+                }
+            }
+        }
+    }
+}

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

@@ -1,7 +1,10 @@
 // Copyright 2021 - Nym Technologies SA <[email protected]>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::client::mix_traffic::transceiver::GatewayTransceiver;
+use crate::client::{
+    base_client::{EventSender, MixnetClientEvent},
+    mix_traffic::transceiver::GatewayTransceiver,
+};
 use nym_gateway_requests::ClientRequest;
 use nym_sphinx::forwarding::packet::MixPacket;
 use nym_task::ShutdownToken;
@@ -22,6 +25,11 @@ const MAX_FAILURE_COUNT: usize = 100;
 // that's also disgusting.
 pub struct Empty;
 
+#[derive(Clone, Copy, Debug)]
+pub enum MixTrafficEvent {
+    FailedSendingSphinx,
+}
+
 pub struct MixTrafficController {
     gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
 
@@ -35,10 +43,15 @@ pub struct MixTrafficController {
     consecutive_gateway_failure_count: usize,
 
     shutdown_token: ShutdownToken,
+    event_tx: EventSender,
 }
 
 impl MixTrafficController {
-    pub fn new<T>(gateway_transceiver: T, shutdown_token: ShutdownToken) -> MixTrafficController
+    pub fn new<T>(
+        gateway_transceiver: T,
+        shutdown_token: ShutdownToken,
+        event_tx: EventSender,
+    ) -> MixTrafficController
     where
         T: GatewayTransceiver + Send + 'static,
     {
@@ -55,14 +68,16 @@ impl MixTrafficController {
             client_tx: client_sender,
             consecutive_gateway_failure_count: 0,
             shutdown_token,
+            event_tx,
         }
     }
 
     pub fn new_dynamic(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
         shutdown_token: ShutdownToken,
+        event_tx: EventSender,
     ) -> MixTrafficController {
-        Self::new(gateway_transceiver, shutdown_token)
+        Self::new(gateway_transceiver, shutdown_token, event_tx)
     }
 
     pub fn client_tx(&self) -> ClientRequestSender {
@@ -140,7 +155,7 @@ impl MixTrafficController {
                             error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
                             // Do we need to handle the embedded mixnet client case
                             // separately?
-                            self.shutdown_token.cancel();
+                            self.event_tx.send(MixnetClientEvent::Traffic(MixTrafficEvent::FailedSendingSphinx));
                             break;
                         }
                     }

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

@@ -3,6 +3,7 @@
 
 pub mod base_client;
 pub mod cover_traffic_stream;
+pub(crate) mod event_control;
 pub(crate) mod helpers;
 pub mod inbound_messages;
 pub mod key_manager;