diff --git a/common/client-core/src/client/cover_traffic_stream.rs b/common/client-core/src/client/cover_traffic_stream.rs index 1059e22e0f..a7b84cd3a4 100644 --- a/common/client-core/src/client/cover_traffic_stream.rs +++ b/common/client-core/src/client/cover_traffic_stream.rs @@ -175,7 +175,7 @@ impl LoopCoverTrafficStream { } }; - let cover_message = generate_loop_cover_packet( + let cover_message = match generate_loop_cover_packet( &mut self.rng, topology_ref, &self.ack_key, @@ -184,8 +184,15 @@ impl LoopCoverTrafficStream { self.cover_traffic.loop_cover_traffic_average_delay, cover_traffic_packet_size, self.packet_type, - ) - .expect("Somehow failed to generate a loop cover message with a valid topology"); + ) { + Ok(cover_message) => cover_message, + Err(err) => { + warn!( + "Somehow failed to generate a loop cover message with a valid topology: {err}" + ); + return; + } + }; if let Err(err) = self.mix_tx.try_send(vec![cover_message]) { match err { diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs index ed062de74c..abc8c72932 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs @@ -64,9 +64,12 @@ impl AcknowledgementListener { trace!("Received {} from the mix network", frag_id); self.stats_tx .report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into()); - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_remove(frag_id)) - .unwrap(); + { + error!("Failed to send remove action to action controller: {err}"); + } } async fn handle_ack_receiver_item(&mut self, item: Vec>) { diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs index a55a9ac226..cd729e2cf7 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs @@ -79,9 +79,12 @@ where if let Some(limit) = self.maximum_retransmissions { if timed_out_ack.retransmissions >= limit { warn!("reached maximum number of allowed retransmissions for the packet"); - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_remove(frag_id)) - .unwrap(); + { + error!("Failed to send remove action to the controller: {err}"); + } return; } } @@ -114,9 +117,12 @@ where Err(err) => { warn!("Could not retransmit the packet - {err}"); // we NEED to start timer here otherwise we will have this guy permanently stuck in memory - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_start_timer(frag_id)) - .unwrap(); + { + error!("Failed to send start timer action to the controller: {err}"); + } return; } }; @@ -141,9 +147,12 @@ where // is sent to the `OutQueueControl` and has gone through its internal queue // with the additional poisson delay. // And since Actions are executed in order `UpdateTimer` will HAVE TO be executed before `StartTimer` - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_update_pending_ack(frag_id, new_delay)) - .unwrap(); + { + error!("Failed to send update pending ack action to the controller: {err}"); + } // send to `OutQueueControl` to eventually send to the mix network self.message_handler diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs index 33f7139a77..f492a4426a 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs @@ -32,9 +32,12 @@ impl SentNotificationListener { trace!("sent off a cover message - no need to start retransmission timer!"); return; } - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_start_timer(frag_id)) - .unwrap(); + { + error!("Failed to send start timer action to action controller: {err}"); + } } pub(super) async fn run_with_shutdown(&mut self, mut shutdown: nym_task::TaskClient) { diff --git a/common/client-core/src/client/real_messages_control/message_handler.rs b/common/client-core/src/client/real_messages_control/message_handler.rs index 6d1e278c24..8af64bd8d5 100644 --- a/common/client-core/src/client/real_messages_control/message_handler.rs +++ b/common/client-core/src/client/real_messages_control/message_handler.rs @@ -609,15 +609,21 @@ where } pub(crate) fn update_ack_delay(&self, id: FragmentIdentifier, new_delay: Delay) { - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::UpdatePendingAck(id, new_delay)) - .expect("action control task has died") + { + error!("Failed to send update action to the controller: {err}"); + } } pub(crate) fn insert_pending_acks(&self, pending_acks: Vec) { - self.action_sender + if let Err(err) = self + .action_sender .unbounded_send(Action::new_insert(pending_acks)) - .expect("action control task has died") + { + error!("Failed to send insert action to the controller: {err}"); + } } // tells real message sender (with the poisson timer) to send this to the mix network diff --git a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs index 614340ca84..e240340b98 100644 --- a/common/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/common/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -198,7 +198,9 @@ where // queues and client load rather than the required delay. So realistically we can treat // whatever is about to happen as negligible additional delay. trace!("{} is about to get sent to the mixnet", frag_id); - self.sent_notifier.unbounded_send(frag_id).unwrap(); + if let Err(err) = self.sent_notifier.unbounded_send(frag_id) { + error!("Failed to notify about sent message: {err}"); + } } fn loop_cover_message_size(&mut self) -> PacketSize { diff --git a/common/client-core/src/client/replies/reply_controller/requests.rs b/common/client-core/src/client/replies/reply_controller/requests.rs index a215805088..ffb6486023 100644 --- a/common/client-core/src/client/replies/reply_controller/requests.rs +++ b/common/client-core/src/client/replies/reply_controller/requests.rs @@ -31,13 +31,16 @@ impl ReplyControllerSender { timed_out_ack: Weak, extra_surb_request: bool, ) { - self.0 + if let Err(err) = self + .0 .unbounded_send(ReplyControllerMessage::RetransmitReply { recipient, timed_out_ack, extra_surb_request, }) - .expect("ReplyControllerReceiver has died!") + { + error!("Failed to send retransmission data to reply controller: {err}",); + } } pub(crate) fn send_reply( @@ -46,13 +49,13 @@ impl ReplyControllerSender { message: Vec, lane: TransmissionLane, ) { - self.0 - .unbounded_send(ReplyControllerMessage::SendReply { - recipient, - message, - lane, - }) - .expect("ReplyControllerReceiver has died!") + if let Err(err) = self.0.unbounded_send(ReplyControllerMessage::SendReply { + recipient, + message, + lane, + }) { + error!("Failed to send reply to reply controller: {err}",); + } } pub(crate) fn send_additional_surbs( @@ -61,32 +64,41 @@ impl ReplyControllerSender { reply_surbs: Vec, from_surb_request: bool, ) { - self.0 + if let Err(err) = self + .0 .unbounded_send(ReplyControllerMessage::AdditionalSurbs { sender_tag, reply_surbs, from_surb_request, }) - .expect("ReplyControllerReceiver has died!") + { + error!("Failed to send additional surbs to reply controller: {err}",); + } } pub(crate) fn send_additional_surbs_request(&self, recipient: Recipient, amount: u32) { - self.0 + if let Err(err) = self + .0 .unbounded_send(ReplyControllerMessage::AdditionalSurbsRequest { recipient: Box::new(recipient), amount, }) - .expect("ReplyControllerReceiver has died!") + { + error!("Failed to send additional surbs request to reply controller: {err}"); + } } pub async fn get_lane_queue_length(&self, connection_id: ConnectionId) -> usize { let (response_tx, response_rx) = oneshot::channel(); - self.0 + if let Err(err) = self + .0 .unbounded_send(ReplyControllerMessage::LaneQueueLength { connection_id, response_channel: response_tx, }) - .expect("ReplyControllerReceiver has died!"); + { + error!("Failed to send lane queue length request to reply controller: {err}"); + } match response_rx.await { Ok(length) => length,