wip

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

@@ -62,6 +62,7 @@ impl MixTrafficController {
 
     async fn on_messages(&mut self, mut mix_packets: Vec<MixPacket>) {
         debug_assert!(!mix_packets.is_empty());
+        info!("MixTrafficController: Sending {} sphinx packets to the gateway", mix_packets.len());
 
         let result = if mix_packets.len() == 1 {
             let mix_packet = mix_packets.pop().unwrap();
@@ -93,24 +94,31 @@ impl MixTrafficController {
         spawn_future(async move {
             debug!("Started MixTrafficController with graceful shutdown support");
 
+            let mut shutdown0 = shutdown.recv_with_delay();
+            tokio::pin!(shutdown0);
+
             loop {
                 tokio::select! {
                     mix_packets = self.mix_rx.recv() => match mix_packets {
                         Some(mix_packets) => {
-                            self.on_messages(mix_packets).await;
+                            let r = tokio::time::timeout(tokio::time::Duration::from_secs(4), self.on_messages(mix_packets)).await;
+                            if r.is_err() {
+                                error!("MixTrafficController: Failed to send mix packets to the gateway");
+                            }
                         },
                         None => {
                             log::trace!("MixTrafficController: Stopping since channel closed");
                             break;
                         }
                     },
-                    _ = shutdown.recv_with_delay() => {
+                    // _ = shutdown.recv_with_delay() => {
+                    _ = &mut shutdown0 => {
                         log::trace!("MixTrafficController: Received shutdown");
                         break;
                     }
                 }
             }
-            shutdown.recv_timeout().await;
+            // shutdown.recv_timeout().await;
             log::debug!("MixTrafficController: Exiting");
         })
     }

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

@@ -40,6 +40,7 @@ pub trait GatewaySender {
         &mut self,
         packets: Vec<MixPacket>,
     ) -> Result<(), ErasedGatewayError> {
+        log::info!("GatewaySender::batch_send_mix_packets - sending {} packets", packets.len());
         // allow for optimisation when sending multiple packets
         for packet in packets {
             self.send_mix_packet(packet).await?;
@@ -78,6 +79,7 @@ impl<G: GatewayTransceiver + ?Sized + Send> GatewayTransceiver for Box<G> {
 impl<G: GatewaySender + ?Sized + Send> GatewaySender for Box<G> {
     #[inline]
     async fn send_mix_packet(&mut self, packet: MixPacket) -> Result<(), ErasedGatewayError> {
+        log::info!("Box<GatewaySender>::send_mix_packet - sending a packet");
         (**self).send_mix_packet(packet).await
     }
 
@@ -130,6 +132,7 @@ where
     St: Send,
 {
     async fn send_mix_packet(&mut self, packet: MixPacket) -> Result<(), ErasedGatewayError> {
+        log::info!("RemoteGateway::send_mix_packet - sending a packet");
         self.gateway_client
             .send_mix_packet(packet)
             .await
@@ -140,6 +143,7 @@ where
         &mut self,
         packets: Vec<MixPacket>,
     ) -> Result<(), ErasedGatewayError> {
+        log::info!("RemoteGateway::batch_send_mix_packets - sending {} packets", packets.len());
         self.gateway_client
             .batch_send_mix_packets(packets)
             .await
@@ -203,6 +207,7 @@ mod nonwasm_sealed {
     #[async_trait]
     impl GatewaySender for LocalGateway {
         async fn send_mix_packet(&mut self, packet: MixPacket) -> Result<(), ErasedGatewayError> {
+            log::info!("LocalGateway::send_mix_packet - sending a packet");
             self.packet_forwarder
                 .unbounded_send(packet)
                 .map_err(|err| err.into_send_error())
@@ -261,6 +266,7 @@ impl GatewayReceiver for MockGateway {
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 impl GatewaySender for MockGateway {
     async fn send_mix_packet(&mut self, packet: MixPacket) -> Result<(), ErasedGatewayError> {
+        log::info!("MockGateway::send_mix_packet - sending a packet");
         self.sent.push(packet);
         Ok(())
     }

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

@@ -358,6 +358,7 @@ impl<C, St> GatewayClient<C, St> {
         &mut self,
         messages: Vec<Message>,
     ) -> Result<(), GatewayClientError> {
+        log::info!("GatewayClient: Sending a batch of messages to the gateway");
         match self.connection {
             SocketState::Available(ref mut conn) => {
                 let stream_messages: Vec<_> = messages.into_iter().map(Ok).collect();
@@ -658,6 +659,7 @@ impl<C, St> GatewayClient<C, St> {
         &mut self,
         packets: Vec<MixPacket>,
     ) -> Result<(), GatewayClientError> {
+        info!("GatewayClient: Sending {} sphinx packets to the gateway", packets.len());
         debug!("Sending {} mix packets", packets.len());
 
         if !self.authenticated {
@@ -688,6 +690,7 @@ impl<C, St> GatewayClient<C, St> {
             .batch_send_websocket_messages_without_response(messages)
             .await
         {
+            log::error!("GatewayClient: Failed to send sphinx packets to the gateway: {err}");
             if err.is_closed_connection() && self.should_reconnect_on_failure {
                 self.attempt_reconnection().await
             } else {
@@ -702,6 +705,7 @@ impl<C, St> GatewayClient<C, St> {
         &mut self,
         msg: Message,
     ) -> Result<(), GatewayClientError> {
+        log::info!("GatewayClient: Sending a message to the gateway");
         if let Err(err) = self.send_websocket_message_without_response(msg).await {
             if err.is_closed_connection() && self.should_reconnect_on_failure {
                 debug!("Going to attempt a reconnection");
@@ -731,6 +735,7 @@ impl<C, St> GatewayClient<C, St> {
         &mut self,
         mix_packet: MixPacket,
     ) -> Result<(), GatewayClientError> {
+        log::info!("GatewayClient: Sending a sphinx packet to the gateway");
         if !self.authenticated {
             return Err(GatewayClientError::NotAuthenticated);
         }

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

@@ -93,6 +93,7 @@ pub enum GatewayClientError {
 
 impl GatewayClientError {
     pub fn is_closed_connection(&self) -> bool {
+        log::info!("GatewayClientError::is_closed_connection");
         match self {
             GatewayClientError::NetworkError(ws_err) => match ws_err {
                 WsError::AlreadyClosed | WsError::ConnectionClosed => true,

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

@@ -160,7 +160,7 @@ impl PartiallyDelegated {
                         log::debug!("GatewayClient listener: Exiting");
                         // The packet router a task client, and as such we need to make
                         // sure it's dropped to not stall the shutdown process.
-                        drop(packet_router);
+                        // drop(packet_router);
                         return;
                     }
                     _ = &mut notify_receiver => {