diff --git a/common/client-core/src/client/mix_traffic/mod.rs b/common/client-core/src/client/mix_traffic/mod.rs index 91c652efba..a08ebfbedc 100644 --- a/common/client-core/src/client/mix_traffic/mod.rs +++ b/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) { 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"); }) } diff --git a/common/client-core/src/client/mix_traffic/transceiver.rs b/common/client-core/src/client/mix_traffic/transceiver.rs index b6197fb490..0a8bdc6a2b 100644 --- a/common/client-core/src/client/mix_traffic/transceiver.rs +++ b/common/client-core/src/client/mix_traffic/transceiver.rs @@ -40,6 +40,7 @@ pub trait GatewaySender { &mut self, packets: Vec, ) -> 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 GatewayTransceiver for Box { impl GatewaySender for Box { #[inline] async fn send_mix_packet(&mut self, packet: MixPacket) -> Result<(), ErasedGatewayError> { + log::info!("Box::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, ) -> 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(()) } diff --git a/common/client-libs/gateway-client/src/client.rs b/common/client-libs/gateway-client/src/client.rs index 633832889e..425a3e37bc 100644 --- a/common/client-libs/gateway-client/src/client.rs +++ b/common/client-libs/gateway-client/src/client.rs @@ -358,6 +358,7 @@ impl GatewayClient { &mut self, messages: Vec, ) -> 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 GatewayClient { &mut self, packets: Vec, ) -> 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 GatewayClient { .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 GatewayClient { &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 GatewayClient { &mut self, mix_packet: MixPacket, ) -> Result<(), GatewayClientError> { + log::info!("GatewayClient: Sending a sphinx packet to the gateway"); if !self.authenticated { return Err(GatewayClientError::NotAuthenticated); } diff --git a/common/client-libs/gateway-client/src/error.rs b/common/client-libs/gateway-client/src/error.rs index 4bcaf4cfd1..e2d9aae77c 100644 --- a/common/client-libs/gateway-client/src/error.rs +++ b/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, diff --git a/common/client-libs/gateway-client/src/socket_state.rs b/common/client-libs/gateway-client/src/socket_state.rs index 2e3f096b3f..54c02f8998 100644 --- a/common/client-libs/gateway-client/src/socket_state.rs +++ b/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 => {