Hack in disable_retransmission flag

clients/native/src/websocket/handler.rs

@@ -190,7 +190,14 @@ impl Handler {
         });
 
         // the ack control is now responsible for chunking, etc.
-        let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
+        let disable_retransmissions = false;
+        let input_msg = InputMessage::new_regular(
+            recipient,
+            message,
+            lane,
+            self.packet_type,
+            disable_retransmissions,
+        );
         if let Err(err) = self.msg_input.send(input_msg).await {
             if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send message to the input buffer: {err}");
@@ -222,8 +229,15 @@ impl Handler {
             TransmissionLane::ConnectionId(id)
         });
 
-        let input_msg =
+        let disable_retransmissions = false;
-            InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
+        let input_msg = InputMessage::new_anonymous(
+            recipient,
+            message,
+            reply_surbs,
+            lane,
+            self.packet_type,
+            disable_retransmissions,
+        );
         if let Err(err) = self.msg_input.send(input_msg).await {
             if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send anonymous message to the input buffer: {err}");
@@ -251,7 +265,14 @@ impl Handler {
             TransmissionLane::ConnectionId(id)
         });
 
-        let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
+        let disable_retransmissions = false;
+        let input_msg = InputMessage::new_reply(
+            recipient_tag,
+            message,
+            lane,
+            self.packet_type,
+            disable_retransmissions,
+        );
         if let Err(err) = self.msg_input.send(input_msg).await {
             if !self.task_client.is_shutdown_poll() {
                 error!("Failed to send reply message to the input buffer: {err}");

common/client-core/src/client/inbound_messages.rs

@@ -28,6 +28,7 @@ pub enum InputMessage {
         recipient: Recipient,
         data: Vec<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     },
 
     /// Creates a message used for a duplex anonymous communication where the recipient
@@ -43,6 +44,7 @@ pub enum InputMessage {
         data: Vec<u8>,
         reply_surbs: u32,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     },
 
     /// 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<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     },
 
     MessageWrapper {
@@ -87,11 +90,13 @@ impl InputMessage {
         data: Vec<u8>,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
+        disable_retransmissions: bool,
     ) -> Self {
         let message = InputMessage::Regular {
             recipient,
             data,
             lane,
+            disable_retransmissions,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -106,12 +111,14 @@ impl InputMessage {
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
+        disable_retransmissions: bool,
     ) -> Self {
         let message = InputMessage::Anonymous {
             recipient,
             data,
             reply_surbs,
             lane,
+            disable_retransmissions,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)
@@ -122,36 +129,39 @@ impl InputMessage {
 
     // IMHO `new_anonymous` should take `mix_hops: Option<u8>` 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(
+    //pub fn new_anonymous_with_custom_hops(
-        recipient: Recipient,
+    //    recipient: Recipient,
-        data: Vec<u8>,
+    //    data: Vec<u8>,
-        reply_surbs: u32,
+    //    reply_surbs: u32,
-        lane: TransmissionLane,
+    //    lane: TransmissionLane,
-        packet_type: Option<PacketType>,
+    //    packet_type: Option<PacketType>,
-    ) -> Self {
+    //) -> Self {
-        let message = InputMessage::Anonymous {
+    //    let message = InputMessage::Anonymous {
-            recipient,
+    //        recipient,
-            data,
+    //        data,
-            reply_surbs,
+    //        reply_surbs,
-            lane,
+    //        lane,
-        };
+    //        disable_retransmissions: false,
-        if let Some(packet_type) = packet_type {
+    //    };
-            InputMessage::new_wrapper(message, packet_type)
+    //    if let Some(packet_type) = packet_type {
-        } else {
+    //        InputMessage::new_wrapper(message, packet_type)
-            message
+    //    } else {
-        }
+    //        message
-    }
+    //    }
+    //}
 
     pub fn new_reply(
         recipient_tag: AnonymousSenderTag,
         data: Vec<u8>,
         lane: TransmissionLane,
         packet_type: Option<PacketType>,
+        disable_retransmissions: bool,
     ) -> Self {
         let message = InputMessage::Reply {
             recipient_tag,
             data,
             lane,
+            disable_retransmissions,
         };
         if let Some(packet_type) = packet_type {
             InputMessage::new_wrapper(message, packet_type)

common/client-core/src/client/real_messages_control/acknowledgement_control/input_message_listener.rs

@@ -65,12 +65,15 @@ where
         recipient_tag: AnonymousSenderTag,
         data: Vec<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     ) {
         // offload reply handling to the dedicated task
-        if let Err(err) = self
+        if let Err(err) = self.reply_controller_sender.send_reply(
-            .reply_controller_sender
+            recipient_tag,
-            .send_reply(recipient_tag, data, lane)
+            data,
-        {
+            lane,
+            disable_retransmissions,
+        ) {
             if !self.task_client.is_shutdown_poll() {
                 error!("failed to send a reply - {err}");
             }
@@ -83,10 +86,17 @@ where
         content: Vec<u8>,
         lane: TransmissionLane,
         packet_type: PacketType,
+        disable_retransmissions: bool,
     ) {
         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,
+                disable_retransmissions,
+            )
             .await
         {
             warn!("failed to send a plain message - {err}")
@@ -100,10 +110,18 @@ where
         reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: PacketType,
+        disable_retransmissions: bool,
     ) {
         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,
+                disable_retransmissions,
+            )
             .await
         {
             warn!("failed to send a repliable message - {err}")
@@ -116,25 +134,42 @@ where
                 recipient,
                 data,
                 lane,
+                disable_retransmissions,
             } => {
-                self.handle_plain_message(recipient, data, lane, PacketType::Mix)
+                self.handle_plain_message(
-                    .await
+                    recipient,
+                    data,
+                    lane,
+                    PacketType::Mix,
+                    disable_retransmissions,
+                )
+                .await
             }
             InputMessage::Anonymous {
                 recipient,
                 data,
                 reply_surbs,
                 lane,
+                disable_retransmissions,
             } => {
-                self.handle_repliable_message(recipient, data, reply_surbs, lane, PacketType::Mix)
+                self.handle_repliable_message(
-                    .await
+                    recipient,
+                    data,
+                    reply_surbs,
+                    lane,
+                    PacketType::Mix,
+                    disable_retransmissions,
+                )
+                .await
             }
             InputMessage::Reply {
                 recipient_tag,
                 data,
                 lane,
+                disable_retransmissions,
             } => {
-                self.handle_reply(recipient_tag, data, lane).await;
+                self.handle_reply(recipient_tag, data, lane, disable_retransmissions)
+                    .await;
             }
             InputMessage::Premade { msgs, lane } => self.handle_premade_packets(msgs, lane).await,
             InputMessage::MessageWrapper {
@@ -145,25 +180,42 @@ where
                     recipient,
                     data,
                     lane,
+                    disable_retransmissions,
                 } => {
-                    self.handle_plain_message(recipient, data, lane, packet_type)
+                    self.handle_plain_message(
-                        .await
+                        recipient,
+                        data,
+                        lane,
+                        packet_type,
+                        disable_retransmissions,
+                    )
+                    .await
                 }
                 InputMessage::Anonymous {
                     recipient,
                     data,
                     reply_surbs,
                     lane,
+                    disable_retransmissions,
                 } => {
-                    self.handle_repliable_message(recipient, data, reply_surbs, lane, packet_type)
+                    self.handle_repliable_message(
-                        .await
+                        recipient,
+                        data,
+                        reply_surbs,
+                        lane,
+                        packet_type,
+                        disable_retransmissions,
+                    )
+                    .await
                 }
                 InputMessage::Reply {
                     recipient_tag,
                     data,
                     lane,
+                    disable_retransmissions,
                 } => {
-                    self.handle_reply(recipient_tag, data, lane).await;
+                    self.handle_reply(recipient_tag, data, lane, disable_retransmissions)
+                        .await;
                 }
                 InputMessage::Premade { msgs, lane } => {
                     self.handle_premade_packets(msgs, lane).await

common/client-core/src/client/real_messages_control/message_handler.rs

@@ -350,7 +350,7 @@ where
     pub(crate) async fn try_send_reply_chunks_on_lane(
         &mut self,
         target: AnonymousSenderTag,
-        fragments: Vec<Fragment>,
+        fragments: Vec<(Fragment, bool)>,
         reply_surbs: Vec<ReplySurb>,
         lane: TransmissionLane,
     ) -> Result<(), SurbWrappedPreparationError> {
@@ -367,12 +367,12 @@ where
     pub(crate) async fn try_send_reply_chunks(
         &mut self,
         target: AnonymousSenderTag,
-        fragments: Vec<(TransmissionLane, Fragment)>,
+        fragments: Vec<(TransmissionLane, (Fragment, bool))>,
         reply_surbs: Vec<ReplySurb>,
     ) -> Result<(), SurbWrappedPreparationError> {
         let prepared_fragments = self
             .prepare_reply_chunks_for_sending(
-                fragments.iter().map(|(_, f)| f.clone()).collect(),
+                fragments.iter().map(|(_, f)| f.0.clone()).collect(),
                 reply_surbs,
             )
             .await?;
@@ -382,7 +382,8 @@ where
 
         for (raw, prepared) in fragments.into_iter().zip(prepared_fragments.into_iter()) {
             let lane = raw.0;
-            let fragment = raw.1;
+            let fragment = raw.1 .0;
+            let disable_retransmissions = raw.1 .1;
 
             let real_message =
                 RealMessage::new(prepared.mix_packet, Some(prepared.fragment_identifier));
@@ -391,7 +392,9 @@ where
 
             let entry = to_forward.entry(lane).or_default();
             entry.push(real_message);
-            pending_acks.push(pending_ack);
+            if !disable_retransmissions {
+                pending_acks.push(pending_ack);
+            }
         }
 
         for (lane, real_messages) in to_forward {
@@ -416,10 +419,17 @@ where
         message: Vec<u8>,
         lane: TransmissionLane,
         packet_type: PacketType,
+        disable_retransmissions: bool,
     ) -> Result<(), PreparationError> {
         let message = NymMessage::new_plain(message);
-        self.try_split_and_send_non_reply_message(message, recipient, lane, packet_type)
+        self.try_split_and_send_non_reply_message(
-            .await
+            message,
+            recipient,
+            lane,
+            packet_type,
+            disable_retransmissions,
+        )
+        .await
     }
 
     pub(crate) async fn try_split_and_send_non_reply_message(
@@ -428,6 +438,7 @@ where
         recipient: Recipient,
         lane: TransmissionLane,
         packet_type: PacketType,
+        disable_retransmissions: bool,
     ) -> 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
@@ -473,7 +484,9 @@ where
             pending_acks.push(pending_ack);
         }
 
-        self.insert_pending_acks(pending_acks);
+        if !disable_retransmissions {
+            self.insert_pending_acks(pending_acks);
+        }
         self.forward_messages(real_messages, lane).await;
 
         Ok(())
@@ -495,11 +508,15 @@ where
             reply_surbs,
         ));
 
+        // When sending SURBs we want to retransmit
+        let disable_retransmissions = false;
+
         self.try_split_and_send_non_reply_message(
             message,
             recipient,
             TransmissionLane::AdditionalReplySurbs,
             packet_type,
+            disable_retransmissions,
         )
         .await?;
 
@@ -516,6 +533,7 @@ where
         num_reply_surbs: u32,
         lane: TransmissionLane,
         packet_type: PacketType,
+        disable_retransmissions: bool,
     ) -> 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 +544,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)
+        self.try_split_and_send_non_reply_message(
-            .await?;
+            message,
+            recipient,
+            lane,
+            packet_type,
+            disable_retransmissions,
+        )
+        .await?;
 
         log::trace!("storing {} reply keys", reply_keys.len());
         self.reply_key_storage.insert_multiple(reply_keys);

common/client-core/src/client/replies/reply_controller/mod.rs

@@ -59,7 +59,7 @@ pub struct ReplyController<R> {
     // of surbs required to send the message through
     // expected_reliability: f32,
     request_receiver: ReplyControllerReceiver,
-    pending_replies: HashMap<AnonymousSenderTag, TransmissionBuffer<Fragment>>,
+    pending_replies: HashMap<AnonymousSenderTag, TransmissionBuffer<(Fragment, bool)>>,
 
     /// 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 +96,7 @@ where
         }
     }
 
-    fn insert_pending_replies<I: IntoIterator<Item = Fragment>>(
+    fn insert_pending_replies<I: IntoIterator<Item = (Fragment, bool)>>(
         &mut self,
         recipient: &AnonymousSenderTag,
         fragments: I,
@@ -112,7 +112,7 @@ where
     fn re_insert_pending_replies(
         &mut self,
         recipient: &AnonymousSenderTag,
-        fragments: Vec<(TransmissionLane, Fragment)>,
+        fragments: Vec<(TransmissionLane, (Fragment, bool))>,
     ) {
         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 +205,7 @@ where
         recipient_tag: AnonymousSenderTag,
         data: Vec<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     ) {
         if !self
             .full_reply_storage
@@ -243,6 +244,10 @@ where
 
             if let Some(reply_surbs) = surbs {
                 let to_send = fragments.drain(..max_to_send).collect::<Vec<_>>();
+                let to_send = to_send
+                    .into_iter()
+                    .map(|f| (f, disable_retransmissions))
+                    .collect::<Vec<_>>();
                 if let Err(err) = self
                     .message_handler
                     .try_send_reply_chunks_on_lane(
@@ -276,6 +281,10 @@ where
                 "buffering {no_fragments} fragments for {recipient_tag}",
                 no_fragments = fragments.len()
             );
+            let fragments: Vec<_> = fragments
+                .into_iter()
+                .map(|f| (f, disable_retransmissions))
+                .collect();
             self.insert_pending_replies(&recipient_tag, fragments, lane);
         }
 
@@ -409,7 +418,7 @@ where
         &mut self,
         from: &AnonymousSenderTag,
         amount: usize,
-    ) -> Option<Vec<(TransmissionLane, Fragment)>> {
+    ) -> Option<Vec<(TransmissionLane, (Fragment, bool))>> {
         // 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 +698,11 @@ where
                 recipient,
                 message,
                 lane,
-            } => self.handle_send_reply(recipient, message, lane).await,
+                disable_retransmissions,
+            } => {
+                self.handle_send_reply(recipient, message, lane, disable_retransmissions)
+                    .await
+            }
             ReplyControllerMessage::AdditionalSurbs {
                 sender_tag,
                 reply_surbs,

common/client-core/src/client/replies/reply_controller/requests.rs

@@ -66,12 +66,14 @@ impl ReplyControllerSender {
         recipient: AnonymousSenderTag,
         message: Vec<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     ) -> Result<(), ReplyControllerSenderError> {
         self.0
             .unbounded_send(ReplyControllerMessage::SendReply {
                 recipient,
                 message,
                 lane,
+                disable_retransmissions,
             })
             .map_err(ReplyControllerSenderError::SendReply)
     }
@@ -160,6 +162,7 @@ pub enum ReplyControllerMessage {
         recipient: AnonymousSenderTag,
         message: Vec<u8>,
         lane: TransmissionLane,
+        disable_retransmissions: bool,
     },
 
     AdditionalSurbs {

common/client-core/src/client/statistics_control.rs

@@ -86,11 +86,13 @@ impl StatisticsControl {
     async fn report_stats(&mut self, recipient: Recipient) {
         let stats_report = self.stats.build_report();
 
+        let disable_retransmissions = false;
         let report_message = InputMessage::new_regular(
             recipient,
             stats_report.into(),
             TransmissionLane::General,
             None,
+            disable_retransmissions,
         );
         if let Err(err) = self.report_tx.send(report_message).await {
             log::error!("Failed to report client stats: {:?}", err);

common/socks5-client-core/src/socks/client.rs

@@ -348,12 +348,14 @@ impl SocksClient {
         let msg =
             Socks5ProviderRequest::new_provider_data(self.config.provider_interface_version, req);
 
+        let disable_retransmissions = false;
         let input_message = InputMessage::new_anonymous(
             self.service_provider,
             msg.into_bytes(),
             self.config.connection_start_surbs,
             TransmissionLane::ConnectionId(self.connection_id),
             self.packet_type,
+            disable_retransmissions,
         );
         self.input_sender
             .send(input_message)
@@ -372,11 +374,13 @@ impl SocksClient {
         let msg =
             Socks5ProviderRequest::new_provider_data(self.config.provider_interface_version, req);
 
+        let disable_retransmissions = false;
         let input_message = InputMessage::new_regular(
             self.service_provider,
             msg.into_bytes(),
             TransmissionLane::ConnectionId(self.connection_id),
             self.packet_type,
+            disable_retransmissions,
         );
         self.input_sender
             .send(input_message)
@@ -436,6 +440,7 @@ impl SocksClient {
                 request_version.provider_interface,
                 provider_request,
             );
+            let disable_retransmissions = false;
             if anonymous {
                 InputMessage::new_anonymous(
                     recipient,
@@ -443,6 +448,7 @@ impl SocksClient {
                     per_request_surbs,
                     lane,
                     packet_type,
+                    disable_retransmissions,
                 )
             } else {
                 InputMessage::new_regular(
@@ -450,6 +456,7 @@ impl SocksClient {
                     provider_message.into_bytes(),
                     lane,
                     packet_type,
+                    disable_retransmissions,
                 )
             }
         })

sdk/rust/nym-sdk/src/mixnet/traits.rs

@@ -70,6 +70,7 @@ pub trait MixnetMessageSender {
         M: AsRef<[u8]> + Send,
     {
         let lane = TransmissionLane::General;
+        let disable_retransmissions = false;
         let input_msg = match surbs {
             IncludedSurbs::Amount(surbs) => InputMessage::new_anonymous(
                 address,
@@ -77,12 +78,14 @@ pub trait MixnetMessageSender {
                 surbs,
                 lane,
                 self.packet_type(),
+                disable_retransmissions,
             ),
             IncludedSurbs::ExposeSelfAddress => InputMessage::new_regular(
                 address,
                 message.as_ref().to_vec(),
                 lane,
                 self.packet_type(),
+                disable_retransmissions,
             ),
         };
         self.send(input_msg).await
@@ -108,11 +111,13 @@ pub trait MixnetMessageSender {
         M: AsRef<[u8]> + Send,
     {
         let lane = TransmissionLane::General;
+        let disable_retransmissions = false;
         let input_msg = InputMessage::new_reply(
             recipient_tag,
             message.as_ref().to_vec(),
             lane,
             self.packet_type(),
+            disable_retransmissions,
         );
         self.send(input_msg).await
     }

service-providers/authenticator/src/cli/request.rs

@@ -123,6 +123,7 @@ pub(crate) async fn execute(args: &Request) -> Result<(), AuthenticatorError> {
             AuthenticatorRequest::new_topup_request(*top_up_message)
         }
     };
+    let disabled_retransmissions = false;
     mixnet_client
         .split_sender()
         .send(nym_sdk::mixnet::InputMessage::new_regular(
@@ -130,6 +131,7 @@ pub(crate) async fn execute(args: &Request) -> Result<(), AuthenticatorError> {
             request.to_bytes().unwrap(),
             TransmissionLane::General,
             None,
+            disabled_retransmissions,
         ))
         .await
         .map_err(|source| AuthenticatorError::FailedToSendPacketToMixnet { source })?;

service-providers/authenticator/src/mixnet_listener.rs

@@ -953,6 +953,7 @@ fn create_input_message(
 ) -> Result<InputMessage> {
     let lane = TransmissionLane::General;
     let packet_type = None;
+    let disable_retransmissions = false;
     if let Some(reply_to_tag) = reply_to_tag {
         log::debug!("Creating message using SURB");
         Ok(InputMessage::new_reply(
@@ -960,6 +961,7 @@ fn create_input_message(
             response_packet,
             lane,
             packet_type,
+            disable_retransmissions,
         ))
     } else if let Some(nym_address) = nym_address {
         log::debug!("Creating message using nym_address");
@@ -968,6 +970,7 @@ fn create_input_message(
             response_packet,
             lane,
             packet_type,
+            disable_retransmissions,
         ))
     } else {
         log::error!("No nym-address or sender tag provided");

service-providers/ip-packet-router/src/util/create_message.rs

@@ -9,12 +9,21 @@ pub(crate) fn create_input_message(
 ) -> InputMessage {
     let lane = TransmissionLane::General;
     let packet_type = None;
+    let disable_retransmissions = false;
     match recipient {
-        ConnectedClientId::NymAddress(recipient) => {
+        ConnectedClientId::NymAddress(recipient) => InputMessage::new_regular(
-            InputMessage::new_regular(**recipient, response_packet, lane, packet_type)
+            **recipient,
-        }
+            response_packet,
-        ConnectedClientId::AnonymousSenderTag(tag) => {
+            lane,
-            InputMessage::new_reply(*tag, response_packet, lane, packet_type)
+            packet_type,
-        }
+            disable_retransmissions,
+        ),
+        ConnectedClientId::AnonymousSenderTag(tag) => InputMessage::new_reply(
+            *tag,
+            response_packet,
+            lane,
+            packet_type,
+            disable_retransmissions,
+        ),
     }
 }

service-providers/network-requester/src/reply.rs

@@ -191,6 +191,7 @@ impl MixnetAddress {
                     recipient: *recipient,
                     data: message,
                     lane: TransmissionLane::ConnectionId(connection_id),
+                    disable_retransmissions: false,
                 }),
                 packet_type,
             },
@@ -199,6 +200,7 @@ impl MixnetAddress {
                     recipient_tag: sender_tag,
                     data: message,
                     lane: TransmissionLane::ConnectionId(connection_id),
+                    disable_retransmissions: false,
                 }),
                 packet_type,
             },