From 1a37e60483ffcd51c19d38341317c1edf5fe7be0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jon=20H=C3=A4ggblad?= Date: Thu, 20 Mar 2025 19:54:07 +0100 Subject: [PATCH] Add max_retransmissions flag on each message (#5642) --- .github/workflows/ci-docs.yml | 2 + .../src/client/inbound_messages.rs | 58 +++++++++------ .../action_controller.rs | 3 - .../input_message_listener.rs | 74 +++++++++++++++---- .../acknowledgement_control/mod.rs | 17 +++++ .../retransmission_request_listener.rs | 18 ++--- .../real_messages_control/message_handler.rs | 66 ++++++++++++++--- .../client/replies/reply_controller/mod.rs | 40 ++++++++-- .../replies/reply_controller/requests.rs | 3 + .../src/clients/connected_client_handler.rs | 9 ++- .../network-requester/src/reply.rs | 2 + 11 files changed, 218 insertions(+), 74 deletions(-) diff --git a/.github/workflows/ci-docs.yml b/.github/workflows/ci-docs.yml index 307df47733..36f2191f24 100644 --- a/.github/workflows/ci-docs.yml +++ b/.github/workflows/ci-docs.yml @@ -11,6 +11,8 @@ on: jobs: build: runs-on: arc-ubuntu-20.04 + env: + RUSTUP_PERMIT_COPY_RENAME: 1 defaults: run: working-directory: documentation/docs diff --git a/common/client-core/src/client/inbound_messages.rs b/common/client-core/src/client/inbound_messages.rs index b14a4d7ed5..ad5e4b0196 100644 --- a/common/client-core/src/client/inbound_messages.rs +++ b/common/client-core/src/client/inbound_messages.rs @@ -28,6 +28,7 @@ pub enum InputMessage { recipient: Recipient, data: Vec, lane: TransmissionLane, + max_retransmissions: Option, }, /// Creates a message used for a duplex anonymous communication where the recipient @@ -43,6 +44,7 @@ pub enum InputMessage { data: Vec, reply_surbs: u32, lane: TransmissionLane, + max_retransmissions: Option, }, /// Attempt to use our internally received and stored `ReplySurb` to send the message back @@ -53,6 +55,7 @@ pub enum InputMessage { recipient_tag: AnonymousSenderTag, data: Vec, lane: TransmissionLane, + max_retransmissions: Option, }, MessageWrapper { @@ -92,6 +95,7 @@ impl InputMessage { recipient, data, lane, + max_retransmissions: None, }; if let Some(packet_type) = packet_type { InputMessage::new_wrapper(message, packet_type) @@ -112,28 +116,7 @@ impl InputMessage { data, reply_surbs, lane, - }; - if let Some(packet_type) = packet_type { - InputMessage::new_wrapper(message, packet_type) - } else { - message - } - } - - // IMHO `new_anonymous` should take `mix_hops: Option` as an argument instead of creating - // this function, but that would potentially break backwards compatibility with the current API - pub fn new_anonymous_with_custom_hops( - recipient: Recipient, - data: Vec, - reply_surbs: u32, - lane: TransmissionLane, - packet_type: Option, - ) -> Self { - let message = InputMessage::Anonymous { - recipient, - data, - reply_surbs, - lane, + max_retransmissions: None, }; if let Some(packet_type) = packet_type { InputMessage::new_wrapper(message, packet_type) @@ -152,6 +135,7 @@ impl InputMessage { recipient_tag, data, lane, + max_retransmissions: None, }; if let Some(packet_type) = packet_type { InputMessage::new_wrapper(message, packet_type) @@ -169,4 +153,34 @@ impl InputMessage { InputMessage::MessageWrapper { message, .. } => message.lane(), } } + + pub fn set_max_retransmissions(&mut self, max_retransmissions: u32) -> &mut Self { + match self { + InputMessage::Regular { + max_retransmissions: m, + .. + } + | InputMessage::Anonymous { + max_retransmissions: m, + .. + } + | InputMessage::Reply { + max_retransmissions: m, + .. + } => { + *m = Some(max_retransmissions); + } + InputMessage::Premade { .. } => {} + InputMessage::MessageWrapper { message, .. } => { + message.set_max_retransmissions(max_retransmissions); + } + } + + self + } + + pub fn with_max_retransmissions(mut self, max_retransmissions: u32) -> Self { + self.set_max_retransmissions(max_retransmissions); + self + } } diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs index 2aca3d58c7..6235b7c477 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs @@ -222,9 +222,6 @@ impl ActionController { // note: when the entry expires it's automatically removed from pending_acks_timers fn handle_expired_ack_timer(&mut self, expired_ack: Expired) { - // I'm honestly not sure how to handle it, because getting it means other things in our - // system are already misbehaving. If we ever see this panic, then I guess we should worry - // about it. Perhaps just reschedule it at later point? let frag_id = expired_ack.into_inner(); trace!("{frag_id} has expired"); diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs index 566fe87402..49da4bce9c 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs @@ -65,11 +65,12 @@ where recipient_tag: AnonymousSenderTag, data: Vec, lane: TransmissionLane, + max_retransmissions: Option, ) { // offload reply handling to the dedicated task - if let Err(err) = self - .reply_controller_sender - .send_reply(recipient_tag, data, lane) + if let Err(err) = + self.reply_controller_sender + .send_reply(recipient_tag, data, lane, max_retransmissions) { if !self.task_client.is_shutdown_poll() { error!("failed to send a reply - {err}"); @@ -83,10 +84,11 @@ where content: Vec, lane: TransmissionLane, packet_type: PacketType, + max_retransmissions: Option, ) { if let Err(err) = self .message_handler - .try_send_plain_message(recipient, content, lane, packet_type) + .try_send_plain_message(recipient, content, lane, packet_type, max_retransmissions) .await { warn!("failed to send a plain message - {err}") @@ -100,10 +102,18 @@ where reply_surbs: u32, lane: TransmissionLane, packet_type: PacketType, + max_retransmissions: Option, ) { if let Err(err) = self .message_handler - .try_send_message_with_reply_surbs(recipient, content, reply_surbs, lane, packet_type) + .try_send_message_with_reply_surbs( + recipient, + content, + reply_surbs, + lane, + packet_type, + max_retransmissions, + ) .await { warn!("failed to send a repliable message - {err}") @@ -116,25 +126,42 @@ where recipient, data, lane, + max_retransmissions, } => { - self.handle_plain_message(recipient, data, lane, PacketType::Mix) - .await + self.handle_plain_message( + recipient, + data, + lane, + PacketType::Mix, + max_retransmissions, + ) + .await } InputMessage::Anonymous { recipient, data, reply_surbs, lane, + max_retransmissions, } => { - self.handle_repliable_message(recipient, data, reply_surbs, lane, PacketType::Mix) - .await + self.handle_repliable_message( + recipient, + data, + reply_surbs, + lane, + PacketType::Mix, + max_retransmissions, + ) + .await } InputMessage::Reply { recipient_tag, data, lane, + max_retransmissions, } => { - self.handle_reply(recipient_tag, data, lane).await; + self.handle_reply(recipient_tag, data, lane, max_retransmissions) + .await; } InputMessage::Premade { msgs, lane } => self.handle_premade_packets(msgs, lane).await, InputMessage::MessageWrapper { @@ -145,25 +172,42 @@ where recipient, data, lane, + max_retransmissions, } => { - self.handle_plain_message(recipient, data, lane, packet_type) - .await + self.handle_plain_message( + recipient, + data, + lane, + packet_type, + max_retransmissions, + ) + .await } InputMessage::Anonymous { recipient, data, reply_surbs, lane, + max_retransmissions, } => { - self.handle_repliable_message(recipient, data, reply_surbs, lane, packet_type) - .await + self.handle_repliable_message( + recipient, + data, + reply_surbs, + lane, + packet_type, + max_retransmissions, + ) + .await } InputMessage::Reply { recipient_tag, data, lane, + max_retransmissions, } => { - self.handle_reply(recipient_tag, data, lane).await; + self.handle_reply(recipient_tag, data, lane, max_retransmissions) + .await; } InputMessage::Premade { msgs, lane } => { self.handle_premade_packets(msgs, lane).await diff --git a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs index 080b0966d5..c3f63b07ad 100644 --- a/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs +++ b/common/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs @@ -72,6 +72,7 @@ pub struct PendingAcknowledgement { delay: SphinxDelay, destination: PacketDestination, retransmissions: u32, + max_retransmissions: Option, } impl PendingAcknowledgement { @@ -80,12 +81,14 @@ impl PendingAcknowledgement { message_chunk: Fragment, delay: SphinxDelay, recipient: Recipient, + max_retransmissions: Option, ) -> Self { PendingAcknowledgement { message_chunk, delay, destination: PacketDestination::KnownRecipient(recipient.into()), retransmissions: 0, + max_retransmissions, } } @@ -94,6 +97,7 @@ impl PendingAcknowledgement { delay: SphinxDelay, recipient_tag: AnonymousSenderTag, extra_surb_request: bool, + max_retransmissions: Option, ) -> Self { PendingAcknowledgement { message_chunk, @@ -103,6 +107,7 @@ impl PendingAcknowledgement { extra_surb_request, }, retransmissions: 0, + max_retransmissions, } } @@ -118,6 +123,18 @@ impl PendingAcknowledgement { self.delay = new_delay; self.retransmissions += 1; } + + pub(crate) fn reached_max_retransmissions( + &self, + global_max_retransmissions: Option, + ) -> bool { + let reached_local_max = self + .max_retransmissions + .is_some_and(|limit| self.retransmissions >= limit); + let reached_global_max = + global_max_retransmissions.is_some_and(|limit| self.retransmissions >= limit); + reached_local_max || reached_global_max + } } /// AcknowledgementControllerConnectors represents set of channels for communication with 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 1c6feabf52..2f7b5d976e 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,17 +79,15 @@ where let frag_id = timed_out_ack.message_chunk.fragment_identifier(); - if let Some(limit) = self.maximum_retransmissions { - if timed_out_ack.retransmissions >= limit { - warn!("reached maximum number of allowed retransmissions for the packet"); - if let Err(err) = self - .action_sender - .unbounded_send(Action::new_remove(frag_id)) - { - error!("Failed to send remove action to the controller: {err}"); - } - return; + if timed_out_ack.reached_max_retransmissions(self.maximum_retransmissions) { + debug!("reached maximum number of allowed retransmissions for the packet"); + if let Err(err) = self + .action_sender + .unbounded_send(Action::new_remove(frag_id)) + { + error!("Failed to send remove action to the controller: {err}"); } + return; } let maybe_prepared_fragment = match &timed_out_ack.destination { 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 222fa79cc0..fc74610b60 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 @@ -6,6 +6,7 @@ use crate::client::real_messages_control::real_traffic_stream::{ BatchRealMessageSender, RealMessage, }; use crate::client::real_messages_control::{AckActionSender, Action}; +use crate::client::replies::reply_controller::MaxRetransmissions; use crate::client::replies::reply_storage::{ReceivedReplySurbsMap, SentReplyKeys, UsedSenderTags}; use crate::client::topology_control::{TopologyAccessor, TopologyReadPermit}; use log::{debug, error, info, trace, warn}; @@ -142,6 +143,12 @@ impl Config { } } +#[derive(Clone)] +pub(crate) struct FragmentWithMaxRetransmissions { + pub(crate) fragment: Fragment, + pub(crate) max_retransmissions: MaxRetransmissions, +} + #[derive(Clone)] pub(crate) struct MessageHandler { config: Config, @@ -294,8 +301,14 @@ where Some(chunk.fragment_identifier()), ); let delay = prepared_fragment.total_delay; - let pending_ack = - PendingAcknowledgement::new_anonymous(chunk, delay, target, is_extra_surb_request); + let max_retransmissions = None; + let pending_ack = PendingAcknowledgement::new_anonymous( + chunk, + delay, + target, + is_extra_surb_request, + max_retransmissions, + ); let lane = if is_extra_surb_request { TransmissionLane::ReplySurbRequest @@ -350,7 +363,7 @@ where pub(crate) async fn try_send_reply_chunks_on_lane( &mut self, target: AnonymousSenderTag, - fragments: Vec, + fragments: Vec, reply_surbs: Vec, lane: TransmissionLane, ) -> Result<(), SurbWrappedPreparationError> { @@ -367,12 +380,12 @@ where pub(crate) async fn try_send_reply_chunks( &mut self, target: AnonymousSenderTag, - fragments: Vec<(TransmissionLane, Fragment)>, + fragments: Vec<(TransmissionLane, FragmentWithMaxRetransmissions)>, reply_surbs: Vec, ) -> Result<(), SurbWrappedPreparationError> { let prepared_fragments = self .prepare_reply_chunks_for_sending( - fragments.iter().map(|(_, f)| f.clone()).collect(), + fragments.iter().map(|(_, f)| f.fragment.clone()).collect(), reply_surbs, ) .await?; @@ -382,12 +395,21 @@ where for (raw, prepared) in fragments.into_iter().zip(prepared_fragments.into_iter()) { let lane = raw.0; - let fragment = raw.1; + let FragmentWithMaxRetransmissions { + fragment, + max_retransmissions, + } = raw.1; let real_message = RealMessage::new(prepared.mix_packet, Some(prepared.fragment_identifier)); let delay = prepared.total_delay; - let pending_ack = PendingAcknowledgement::new_anonymous(fragment, delay, target, false); + let pending_ack = PendingAcknowledgement::new_anonymous( + fragment, + delay, + target, + false, + max_retransmissions, + ); let entry = to_forward.entry(lane).or_default(); entry.push(real_message); @@ -416,10 +438,17 @@ where message: Vec, lane: TransmissionLane, packet_type: PacketType, + max_retransmissions: Option, ) -> Result<(), PreparationError> { let message = NymMessage::new_plain(message); - self.try_split_and_send_non_reply_message(message, recipient, lane, packet_type) - .await + self.try_split_and_send_non_reply_message( + message, + recipient, + lane, + packet_type, + max_retransmissions, + ) + .await } pub(crate) async fn try_split_and_send_non_reply_message( @@ -428,6 +457,7 @@ where recipient: Recipient, lane: TransmissionLane, packet_type: PacketType, + max_retransmissions: Option, ) -> Result<(), PreparationError> { debug!("Sending non-reply message with packet type {packet_type}"); // TODO: I really dislike existence of this assertion, it implies code has to be re-organised @@ -467,7 +497,8 @@ where Some(fragment.fragment_identifier()), ); let delay = prepared_fragment.total_delay; - let pending_ack = PendingAcknowledgement::new_known(fragment, delay, recipient); + let pending_ack = + PendingAcknowledgement::new_known(fragment, delay, recipient, max_retransmissions); real_messages.push(real_message); pending_acks.push(pending_ack); @@ -495,11 +526,15 @@ where reply_surbs, )); + // When sending SURBs we want to retransmit + let max_retransmissions = None; + self.try_split_and_send_non_reply_message( message, recipient, TransmissionLane::AdditionalReplySurbs, packet_type, + max_retransmissions, ) .await?; @@ -516,6 +551,7 @@ where num_reply_surbs: u32, lane: TransmissionLane, packet_type: PacketType, + max_retransmissions: Option, ) -> Result<(), SurbWrappedPreparationError> { debug!("Sending message with reply SURBs with packet type {packet_type}"); let sender_tag = self.get_or_create_sender_tag(&recipient); @@ -526,8 +562,14 @@ where let message = NymMessage::new_repliable(RepliableMessage::new_data(message, sender_tag, reply_surbs)); - self.try_split_and_send_non_reply_message(message, recipient, lane, packet_type) - .await?; + self.try_split_and_send_non_reply_message( + message, + recipient, + lane, + packet_type, + max_retransmissions, + ) + .await?; log::trace!("storing {} reply keys", reply_keys.len()); self.reply_key_storage.insert_multiple(reply_keys); diff --git a/common/client-core/src/client/replies/reply_controller/mod.rs b/common/client-core/src/client/replies/reply_controller/mod.rs index 7b59359407..2cf7ac9eaa 100644 --- a/common/client-core/src/client/replies/reply_controller/mod.rs +++ b/common/client-core/src/client/replies/reply_controller/mod.rs @@ -2,7 +2,9 @@ // SPDX-License-Identifier: Apache-2.0 use crate::client::real_messages_control::acknowledgement_control::PendingAcknowledgement; -use crate::client::real_messages_control::message_handler::{MessageHandler, PreparationError}; +use crate::client::real_messages_control::message_handler::{ + FragmentWithMaxRetransmissions, MessageHandler, PreparationError, +}; use crate::client::replies::reply_storage::CombinedReplyStorage; use futures::channel::oneshot; use futures::StreamExt; @@ -10,7 +12,7 @@ use log::{debug, error, info, trace, warn}; use nym_sphinx::addressing::clients::Recipient; use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag; use nym_sphinx::anonymous_replies::ReplySurb; -use nym_sphinx::chunking::fragment::{Fragment, FragmentIdentifier}; +use nym_sphinx::chunking::fragment::FragmentIdentifier; use nym_task::connections::{ConnectionId, TransmissionLane}; use nym_task::TaskClient; use rand::{CryptoRng, Rng}; @@ -49,6 +51,8 @@ impl Config { // - replies to "give additional surbs" requests // - will reply to future heartbeats +pub type MaxRetransmissions = Option; + // TODO: this should be split into ingress and egress controllers // because currently its trying to perform two distinct jobs pub struct ReplyController { @@ -59,7 +63,8 @@ pub struct ReplyController { // of surbs required to send the message through // expected_reliability: f32, request_receiver: ReplyControllerReceiver, - pending_replies: HashMap>, + pending_replies: + HashMap>, /// Retransmission packets that have already timed out and are waiting for additional reply SURBs /// so that they could be sent back to the network. Once we receive more SURBs, we should send them ASAP. @@ -96,7 +101,7 @@ where } } - fn insert_pending_replies>( + fn insert_pending_replies>( &mut self, recipient: &AnonymousSenderTag, fragments: I, @@ -112,7 +117,7 @@ where fn re_insert_pending_replies( &mut self, recipient: &AnonymousSenderTag, - fragments: Vec<(TransmissionLane, Fragment)>, + fragments: Vec<(TransmissionLane, FragmentWithMaxRetransmissions)>, ) { trace!("re-inserting pending replies for {recipient}"); // the buffer should ALWAYS exist at this point, if it doesn't, it's a bug... @@ -205,6 +210,7 @@ where recipient_tag: AnonymousSenderTag, data: Vec, lane: TransmissionLane, + max_retransmissions: Option, ) { if !self .full_reply_storage @@ -242,7 +248,14 @@ where .get_reply_surbs(&recipient_tag, max_to_send); if let Some(reply_surbs) = surbs { - let to_send = fragments.drain(..max_to_send).collect::>(); + let to_send = fragments + .drain(..max_to_send) + .map(|f| FragmentWithMaxRetransmissions { + fragment: f, + max_retransmissions, + }) + .collect::>(); + if let Err(err) = self .message_handler .try_send_reply_chunks_on_lane( @@ -276,6 +289,13 @@ where "buffering {no_fragments} fragments for {recipient_tag}", no_fragments = fragments.len() ); + let fragments: Vec<_> = fragments + .into_iter() + .map(|fragment| FragmentWithMaxRetransmissions { + fragment, + max_retransmissions, + }) + .collect(); self.insert_pending_replies(&recipient_tag, fragments, lane); } @@ -409,7 +429,7 @@ where &mut self, from: &AnonymousSenderTag, amount: usize, - ) -> Option> { + ) -> Option> { // if possible, pop all pending replies, if not, pop only entries for which we'd have a reply surb let total = self.pending_replies.get(from)?.total_size(); trace!("pending queue has {total} elements"); @@ -689,7 +709,11 @@ where recipient, message, lane, - } => self.handle_send_reply(recipient, message, lane).await, + max_retransmissions, + } => { + self.handle_send_reply(recipient, message, lane, max_retransmissions) + .await + } ReplyControllerMessage::AdditionalSurbs { sender_tag, reply_surbs, 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 a1555329e5..30afc2585a 100644 --- a/common/client-core/src/client/replies/reply_controller/requests.rs +++ b/common/client-core/src/client/replies/reply_controller/requests.rs @@ -66,12 +66,14 @@ impl ReplyControllerSender { recipient: AnonymousSenderTag, message: Vec, lane: TransmissionLane, + max_retransmissions: Option, ) -> Result<(), ReplyControllerSenderError> { self.0 .unbounded_send(ReplyControllerMessage::SendReply { recipient, message, lane, + max_retransmissions, }) .map_err(ReplyControllerSenderError::SendReply) } @@ -160,6 +162,7 @@ pub enum ReplyControllerMessage { recipient: AnonymousSenderTag, message: Vec, lane: TransmissionLane, + max_retransmissions: Option, }, AdditionalSurbs { diff --git a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs index c9a4364067..3d7c81133b 100644 --- a/service-providers/ip-packet-router/src/clients/connected_client_handler.rs +++ b/service-providers/ip-packet-router/src/clients/connected_client_handler.rs @@ -182,10 +182,11 @@ impl MixnetMessageSinkTranslator for ToIprDataResponse { let response_packet = create_ip_packet_response(bundled_ip_packets, self.client_version)?; // Wrap the response packet in a mixnet input message - Ok(crate::util::create_message::create_input_message( - &self.send_to, - response_packet, - )) + let input_message = + crate::util::create_message::create_input_message(&self.send_to, response_packet) + .with_max_retransmissions(0); + + Ok(input_message) } } diff --git a/service-providers/network-requester/src/reply.rs b/service-providers/network-requester/src/reply.rs index 8bfd67e070..9dd171b444 100644 --- a/service-providers/network-requester/src/reply.rs +++ b/service-providers/network-requester/src/reply.rs @@ -191,6 +191,7 @@ impl MixnetAddress { recipient: *recipient, data: message, lane: TransmissionLane::ConnectionId(connection_id), + max_retransmissions: None, }), packet_type, }, @@ -199,6 +200,7 @@ impl MixnetAddress { recipient_tag: sender_tag, data: message, lane: TransmissionLane::ConnectionId(connection_id), + max_retransmissions: None, }), packet_type, },