GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove a bunch of unwraps from client-core

Browse Source
This commit is contained in:
Jon Häggblad
2025-02-10 23:42:26 +01:00
parent 4d486abfef
commit cb8b4c56af
7 changed files with 75 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/client-core/src/client/cover_traffic_stream.rs
+10 -3
View File
@@ -175,7 +175,7 @@ impl LoopCoverTrafficStream<OsRng> {
}
};
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<OsRng> {
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 {
common/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs
+5 -2
View File
@@ -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<Vec<u8>>) {
common/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs
+15 -6
View File
@@ -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
common/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs
+5 -2
View File
@@ -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) {
common/client-core/src/client/real_messages_control/message_handler.rs
+10 -4
View File
@@ -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<PendingAcknowledgement>) {
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
common/client-core/src/client/real_messages_control/real_traffic_stream.rs
+3 -1
View File
@@ -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 {
common/client-core/src/client/replies/reply_controller/requests.rs
+27 -15
View File
@@ -31,13 +31,16 @@ impl ReplyControllerSender {
timed_out_ack: Weak<PendingAcknowledgement>,
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<u8>,
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<ReplySurb>,
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,