Merge pull request #3535 from nymtech/feature/socks5-message-ordering

Feature/socks5 message ordering

Cargo.lock

@@ -3949,7 +3949,6 @@ dependencies = [
  "nym-credential-storage",
  "nym-crypto",
  "nym-network-defaults",
- "nym-ordered-buffer",
  "nym-sdk",
  "nym-service-providers-common",
  "nym-socks5-proxy-helpers",

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

@@ -428,18 +428,14 @@ impl SocksClient {
             Some(self.lane_queue_lengths.clone()),
             self.shutdown_listener.clone(),
         )
-        .run(move |conn_id, read_data, socket_closed| {
-            let provider_request = Socks5Request::new_send(
-                request_version.provider_protocol,
-                conn_id,
-                read_data,
-                socket_closed,
-            );
+        .run(move |socket_data| {
+            let lane = TransmissionLane::ConnectionId(socket_data.header.connection_id);
+            let provider_request =
+                Socks5Request::new_send(request_version.provider_protocol, socket_data);
             let provider_message = Socks5ProviderRequest::new_provider_data(
                 request_version.provider_interface,
                 provider_request,
             );
-            let lane = TransmissionLane::ConnectionId(conn_id);
             if anonymous {
                 InputMessage::new_anonymous(
                     recipient,

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

@@ -1,19 +1,20 @@
+// Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::Socks5ClientCoreError;
 use futures::channel::mpsc;
 use futures::StreamExt;
 use log::*;
-
 use nym_client_core::client::received_buffer::ReconstructedMessagesReceiver;
 use nym_client_core::client::received_buffer::{
     ReceivedBufferMessage, ReceivedBufferRequestSender,
 };
 use nym_service_providers_common::interface::{ControlResponse, ResponseContent};
-use nym_socks5_proxy_helpers::connection_controller::ControllerSender;
+use nym_socks5_proxy_helpers::connection_controller::{ControllerCommand, ControllerSender};
 use nym_socks5_requests::{Socks5ProviderResponse, Socks5Response, Socks5ResponseContent};
 use nym_sphinx::receiver::ReconstructedMessage;
 use nym_task::TaskClient;
 
-use crate::error::Socks5ClientCoreError;
-
 pub(crate) struct MixnetResponseListener {
     buffer_requester: ReceivedBufferRequestSender,
     mix_response_receiver: ReconstructedMessagesReceiver,
@@ -79,9 +80,9 @@ impl MixnetResponseListener {
                 );
                 Err(err_response.into())
             }
-            Socks5ResponseContent::NetworkData(response) => {
+            Socks5ResponseContent::NetworkData { content } => {
                 self.controller_sender
-                    .unbounded_send(response.into())
+                    .unbounded_send(ControllerCommand::new_send(content))
                     .unwrap();
                 Ok(())
             }

common/socks5/ordered-buffer/src/buffer.rs

@@ -1,6 +1,21 @@
-use crate::message::OrderedMessage;
+// Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
 use log::*;
-use std::collections::HashMap;
+use std::collections::BTreeMap;
+use thiserror::Error;
+
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum OrderedMessageError {
+    #[error("received message with sequence number {received}, which is way higher than our current {current}")]
+    MessageSequenceTooLarge { current: u64, received: u64 },
+
+    #[error("received message with sequence number {received}, while we're already at {current}!")]
+    MessageAlreadyReconstructed { current: u64, received: u64 },
+
+    #[error("attempted to overwrite message at sequence {received}")]
+    AttemptedToOverwriteSequence { received: u64 },
+}
 
 /// Stores messages and emits them in order.
 ///
@@ -9,36 +24,58 @@ use std::collections::HashMap;
 /// to fill up with the full sequence.
 #[derive(Debug)]
 pub struct OrderedMessageBuffer {
-    next_index: u64,
-    messages: HashMap<u64, OrderedMessage>,
+    next_sequence: u64,
+    messages: BTreeMap<u64, Vec<u8>>,
 }
 
 /// Data returned from `OrderedMessageBuffer` on a successful read of gapless ordered data.
 #[derive(Debug, PartialEq, Eq)]
 pub struct ReadContiguousData {
     pub data: Vec<u8>,
-    pub last_index: u64,
+    pub last_sequence: u64,
 }
 
+const MAX_REASONABLE_OFFSET: u64 = 1000;
+
 impl OrderedMessageBuffer {
     pub fn new() -> OrderedMessageBuffer {
         OrderedMessageBuffer {
-            next_index: 0,
-            messages: HashMap::new(),
+            next_sequence: 0,
+            messages: BTreeMap::new(),
         }
     }
 
     /// Writes a message to the buffer. messages are sort on insertion, so
     /// that later on multiple reads for incomplete sequences don't result in
     /// useless sort work.
-    pub fn write(&mut self, message: OrderedMessage) {
+    pub fn write(&mut self, sequence: u64, data: Vec<u8>) -> Result<(), OrderedMessageError> {
+        // reject messages that have clearly malformed sequence
+        if sequence > self.next_sequence + MAX_REASONABLE_OFFSET {
+            return Err(OrderedMessageError::MessageSequenceTooLarge {
+                current: self.next_sequence,
+                received: sequence,
+            });
+        }
+
+        if self.messages.contains_key(&sequence) {
+            return Err(OrderedMessageError::AttemptedToOverwriteSequence { received: sequence });
+        }
+
+        if sequence < self.next_sequence {
+            return Err(OrderedMessageError::MessageAlreadyReconstructed {
+                current: self.next_sequence,
+                received: sequence,
+            });
+        }
+
         trace!(
-            "Writing message index: {} length {:?} to OrderedMessageBuffer.",
-            message.index,
-            message.data.len()
+            "Writing message index: {} length {} to OrderedMessageBuffer.",
+            sequence,
+            data.len()
         );
 
-        self.messages.insert(message.index, message);
+        self.messages.insert(sequence, data);
+        Ok(())
     }
 
     /// Returns `Option<Vec<u8>>` where it's `Some(bytes)` if there is gapless
@@ -49,33 +86,31 @@ impl OrderedMessageBuffer {
     /// a read will return the bytes of messages 0, 1, 2. Subsequent reads will
     /// return `None` until message 3 comes in, at which point 3, 4, and any
     /// further contiguous messages which have arrived will be returned.
+    #[must_use]
     pub fn read(&mut self) -> Option<ReadContiguousData> {
-        if !self.messages.contains_key(&self.next_index) {
+        if !self.messages.contains_key(&self.next_sequence) {
             return None;
         }
 
         let mut contiguous_messages = Vec::new();
-        let mut index = self.next_index;
+        let mut seq = self.next_sequence;
 
-        while let Some(ordered_message) = self.messages.remove(&index) {
-            contiguous_messages.push(ordered_message);
-            index += 1;
+        while let Some(mut data) = self.messages.remove(&seq) {
+            contiguous_messages.append(&mut data);
+            seq += 1;
         }
 
-        let high_water = index;
-        self.next_index = high_water;
-        trace!("Next high water mark is: {}", high_water);
+        let high_water = seq;
+        self.next_sequence = high_water;
+        trace!("Next high water mark is: {high_water}");
 
-        // dig out the bytes from inside the struct
-        let data: Vec<u8> = contiguous_messages
-            .into_iter()
-            .flat_map(|message| message.data)
-            .collect();
-
-        trace!("Returning {} bytes from ordered message buffer", data.len());
+        trace!(
+            "Returning {} bytes from ordered message buffer",
+            contiguous_messages.len()
+        );
         Some(ReadContiguousData {
-            data,
-            last_index: index,
+            data: contiguous_messages,
+            last_sequence: seq,
         })
     }
 }
@@ -90,6 +125,64 @@ impl Default for OrderedMessageBuffer {
 mod test_chunking_and_reassembling {
     use super::*;
 
+    #[test]
+    fn trying_to_write_unreasonable_high_sequence() {
+        let mut buffer = OrderedMessageBuffer::new();
+        let first_message = vec![1, 2, 3, 4];
+        let second_message = vec![5, 6, 7, 8];
+
+        buffer.write(0, first_message).unwrap();
+        buffer.write(1, second_message).unwrap();
+
+        assert_eq!(
+            Err(OrderedMessageError::MessageSequenceTooLarge {
+                current: 0,
+                received: 12345678
+            }),
+            buffer.write(12345678, b"foomp".to_vec())
+        )
+    }
+
+    #[test]
+    fn trying_to_overwrite_sequence() {
+        let mut buffer = OrderedMessageBuffer::new();
+        let message = vec![1, 2, 3, 4];
+
+        buffer.write(0, message.clone()).unwrap();
+        buffer.write(1, message.clone()).unwrap();
+        buffer.write(2, message.clone()).unwrap();
+        buffer.write(3, message.clone()).unwrap();
+
+        for seq in 0..=3 {
+            assert_eq!(
+                Err(OrderedMessageError::AttemptedToOverwriteSequence { received: seq }),
+                buffer.write(seq, message.clone())
+            )
+        }
+    }
+
+    #[test]
+    fn writing_past_data() {
+        let mut buffer = OrderedMessageBuffer::new();
+        let message = vec![1, 2, 3, 4];
+
+        buffer.write(0, message.clone()).unwrap();
+        buffer.write(1, message.clone()).unwrap();
+        buffer.write(2, message.clone()).unwrap();
+        buffer.write(3, message.clone()).unwrap();
+        let _ = buffer.read().unwrap();
+
+        for seq in 0..=3 {
+            assert_eq!(
+                Err(OrderedMessageError::MessageAlreadyReconstructed {
+                    current: 4,
+                    received: seq
+                }),
+                buffer.write(seq, message.clone())
+            )
+        }
+    }
+
     #[cfg(test)]
     mod reading_from_and_writing_to_the_buffer {
         use super::*;
@@ -102,20 +195,14 @@ mod test_chunking_and_reassembling {
             fn read_returns_ordered_bytes_and_resets_buffer() {
                 let mut buffer = OrderedMessageBuffer::new();
 
-                let first_message = OrderedMessage {
-                    data: vec![1, 2, 3, 4],
-                    index: 0,
-                };
-                let second_message = OrderedMessage {
-                    data: vec![5, 6, 7, 8],
-                    index: 1,
-                };
+                let first_message = vec![1, 2, 3, 4];
+                let second_message = vec![5, 6, 7, 8];
 
-                buffer.write(first_message);
+                buffer.write(0, first_message).unwrap();
                 let first_read = buffer.read().unwrap().data;
                 assert_eq!(vec![1, 2, 3, 4], first_read);
 
-                buffer.write(second_message);
+                buffer.write(1, second_message).unwrap();
                 let second_read = buffer.read().unwrap().data;
                 assert_eq!(vec![5, 6, 7, 8], second_read);
 
@@ -126,17 +213,11 @@ mod test_chunking_and_reassembling {
             fn test_multiple_adds_stacks_up_bytes_in_the_buffer() {
                 let mut buffer = OrderedMessageBuffer::new();
 
-                let first_message = OrderedMessage {
-                    data: vec![1, 2, 3, 4],
-                    index: 0,
-                };
-                let second_message = OrderedMessage {
-                    data: vec![5, 6, 7, 8],
-                    index: 1,
-                };
+                let first_message = vec![1, 2, 3, 4];
+                let second_message = vec![5, 6, 7, 8];
 
-                buffer.write(first_message);
-                buffer.write(second_message);
+                buffer.write(0, first_message).unwrap();
+                buffer.write(1, second_message).unwrap();
                 let second_read = buffer.read();
                 assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], second_read.unwrap().data);
                 assert_eq!(None, buffer.read()); // second read on fully ordered result set is empty
@@ -146,17 +227,11 @@ mod test_chunking_and_reassembling {
             fn out_of_order_adds_results_in_ordered_byte_vector() {
                 let mut buffer = OrderedMessageBuffer::new();
 
-                let first_message = OrderedMessage {
-                    data: vec![1, 2, 3, 4],
-                    index: 0,
-                };
-                let second_message = OrderedMessage {
-                    data: vec![5, 6, 7, 8],
-                    index: 1,
-                };
+                let first_message = vec![1, 2, 3, 4];
+                let second_message = vec![5, 6, 7, 8];
 
-                buffer.write(second_message);
-                buffer.write(first_message);
+                buffer.write(1, second_message).unwrap();
+                buffer.write(0, first_message).unwrap();
                 let read = buffer.read().unwrap().data;
                 assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], read);
                 assert_eq!(None, buffer.read()); // second read on fully ordered result set is empty
@@ -170,23 +245,13 @@ mod test_chunking_and_reassembling {
             fn setup() -> OrderedMessageBuffer {
                 let mut buffer = OrderedMessageBuffer::new();
 
-                let zero_message = OrderedMessage {
-                    data: vec![0, 0, 0, 0],
-                    index: 0,
-                };
-                let one_message = OrderedMessage {
-                    data: vec![1, 1, 1, 1],
-                    index: 1,
-                };
+                let zero_message = vec![0, 0, 0, 0];
+                let one_message = vec![1, 1, 1, 1];
+                let three_message = vec![3, 3, 3, 3];
 
-                let three_message = OrderedMessage {
-                    data: vec![3, 3, 3, 3],
-                    index: 3,
-                };
-
-                buffer.write(zero_message);
-                buffer.write(one_message);
-                buffer.write(three_message);
+                buffer.write(0, zero_message).unwrap();
+                buffer.write(1, one_message).unwrap();
+                buffer.write(3, three_message).unwrap();
                 buffer
             }
             #[test]
@@ -199,43 +264,31 @@ mod test_chunking_and_reassembling {
                 assert_eq!(None, buffer.read());
 
                 // let's add another message, leaving a gap in place at index 2
-                let five_message = OrderedMessage {
-                    data: vec![5, 5, 5, 5],
-                    index: 5,
-                };
-                buffer.write(five_message);
+                let five_message = vec![5, 5, 5, 5];
+                buffer.write(5, five_message).unwrap();
                 assert_eq!(None, buffer.read());
             }
 
             #[test]
             fn filling_the_gap_allows_us_to_get_everything() {
                 let mut buffer = setup();
-                buffer.read(); // that burns the first two. We still have a gap before the 3s.
+                let _ = buffer.read(); // that burns the first two. We still have a gap before the 3s.
 
-                let two_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 2,
-                };
-                buffer.write(two_message);
+                let two_message = vec![2, 2, 2, 2];
+                buffer.write(2, two_message).unwrap();
 
                 let more_ordered_bytes = buffer.read().unwrap().data;
                 assert_eq!([2, 2, 2, 2, 3, 3, 3, 3].to_vec(), more_ordered_bytes);
 
                 // let's add another message
-                let five_message = OrderedMessage {
-                    data: vec![5, 5, 5, 5],
-                    index: 5,
-                };
-                buffer.write(five_message);
+                let five_message = vec![5, 5, 5, 5];
+                buffer.write(5, five_message).unwrap();
 
                 assert_eq!(None, buffer.read());
 
                 // let's fill in the gap of 4s now and read again
-                let four_message = OrderedMessage {
-                    data: vec![4, 4, 4, 4],
-                    index: 4,
-                };
-                buffer.write(four_message);
+                let four_message = vec![4, 4, 4, 4];
+                buffer.write(4, four_message).unwrap();
 
                 assert_eq!(
                     [4, 4, 4, 4, 5, 5, 5, 5].to_vec(),
@@ -249,70 +302,47 @@ mod test_chunking_and_reassembling {
             #[test]
             fn filling_the_gap_allows_us_to_get_everything_when_last_element_is_empty() {
                 let mut buffer = OrderedMessageBuffer::new();
-                let zero_message = OrderedMessage {
-                    data: vec![0, 0, 0, 0],
-                    index: 0,
-                };
-                let one_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 1,
-                };
-                let two_message = OrderedMessage {
-                    data: vec![],
-                    index: 2,
-                };
+                let zero_message = vec![0, 0, 0, 0];
+                let one_message = vec![2, 2, 2, 2];
+                let two_message = vec![];
 
-                buffer.write(zero_message);
+                buffer.write(0, zero_message).unwrap();
                 assert!(buffer.read().is_some()); // burn the buffer
 
-                buffer.write(two_message);
-                buffer.write(one_message);
+                buffer.write(2, two_message).unwrap();
+                buffer.write(1, one_message).unwrap();
                 assert!(buffer.read().is_some());
-                assert_eq!(buffer.next_index, 3);
+                assert_eq!(buffer.next_sequence, 3);
             }
 
             #[test]
             fn works_with_gaps_bigger_than_one() {
                 let mut buffer = OrderedMessageBuffer::new();
-                let zero_message = OrderedMessage {
-                    data: vec![0, 0, 0, 0],
-                    index: 0,
-                };
-                let one_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 1,
-                };
-                let two_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 2,
-                };
-                let three_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 3,
-                };
-                let four_message = OrderedMessage {
-                    data: vec![2, 2, 2, 2],
-                    index: 4,
-                };
-                buffer.write(zero_message);
+                let zero_message = vec![0, 0, 0, 0];
+                let one_message = vec![2, 2, 2, 2];
+                let two_message = vec![2, 2, 2, 2];
+                let three_message = vec![2, 2, 2, 2];
+                let four_message = vec![2, 2, 2, 2];
+
+                buffer.write(0, zero_message).unwrap();
                 assert!(buffer.read().is_some());
-                assert_eq!(buffer.next_index, 1);
+                assert_eq!(buffer.next_sequence, 1);
 
-                buffer.write(four_message);
+                buffer.write(4, four_message).unwrap();
                 assert!(buffer.read().is_none());
-                assert_eq!(buffer.next_index, 1);
+                assert_eq!(buffer.next_sequence, 1);
 
-                buffer.write(three_message);
+                buffer.write(3, three_message).unwrap();
                 assert!(buffer.read().is_none());
-                assert_eq!(buffer.next_index, 1);
+                assert_eq!(buffer.next_sequence, 1);
 
-                buffer.write(two_message);
+                buffer.write(2, two_message).unwrap();
                 assert!(buffer.read().is_none());
-                assert_eq!(buffer.next_index, 1);
+                assert_eq!(buffer.next_sequence, 1);
 
-                buffer.write(one_message);
+                buffer.write(1, one_message).unwrap();
                 assert!(buffer.read().is_some());
-                assert_eq!(buffer.next_index, 5)
+                assert_eq!(buffer.next_sequence, 5)
             }
         }
     }

common/socks5/ordered-buffer/src/lib.rs

@@ -1,8 +1,3 @@
 mod buffer;
-mod message;
-mod sender;
 
-pub use buffer::{OrderedMessageBuffer, ReadContiguousData};
-pub use message::MessageError;
-pub use message::OrderedMessage;
-pub use sender::OrderedMessageSender;
+pub use buffer::{OrderedMessageBuffer, OrderedMessageError, ReadContiguousData};

common/socks5/ordered-buffer/src/message.rs

@@ -1,143 +0,0 @@
-// Copyright 2020-2022 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use std::cmp::Ordering;
-use thiserror::Error;
-
-#[derive(Error, Debug, PartialEq, Eq)]
-pub enum MessageError {
-    #[error("the received message was empty")]
-    NoData,
-
-    #[error("could not extract message index. Received {received} bytes, but expected {expected}")]
-    IndexTooShort { received: usize, expected: usize },
-}
-
-#[derive(Clone, Debug, Eq, PartialEq)]
-pub struct OrderedMessage {
-    pub data: Vec<u8>,
-    pub index: u64,
-}
-
-impl OrderedMessage {
-    /// Serializes an `OrderedMessage` into bytes.
-    /// The output format is:
-    /// | 8 bytes index | data... |
-    pub fn into_bytes(self) -> Vec<u8> {
-        self.index
-            .to_be_bytes()
-            .iter()
-            .cloned()
-            .chain(self.data.into_iter())
-            .collect()
-    }
-
-    /// Attempts to deserialize an `OrderedMessage` from bytes.
-    pub fn try_from_bytes(data: Vec<u8>) -> Result<OrderedMessage, MessageError> {
-        if data.is_empty() {
-            return Err(MessageError::NoData);
-        }
-
-        if data.len() < 8 {
-            return Err(MessageError::IndexTooShort {
-                received: data.len(),
-                expected: 8,
-            });
-        }
-        let index = u64::from_be_bytes([
-            data[0], data[1], data[2], data[3], data[4], data[5], data[6], data[7],
-        ]);
-        Ok(OrderedMessage {
-            data: data[8..].to_vec(),
-            index,
-        })
-    }
-}
-
-/// Order messages by their index only, ignoring their data
-impl PartialOrd for OrderedMessage {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some((self.index).cmp(&(other.index)))
-    }
-}
-
-#[cfg(test)]
-mod ordered_message_to_bytes {
-    use super::*;
-
-    #[test]
-    fn works() {
-        let message = OrderedMessage {
-            data: vec![123],
-            index: 1,
-        };
-        let bytes = message.into_bytes();
-
-        let expected = vec![0, 0, 0, 0, 0, 0, 0, 1, 123];
-        assert_eq!(expected, bytes);
-    }
-}
-
-#[cfg(test)]
-mod ordered_message_from_bytes {
-    use super::*;
-
-    #[test]
-    fn fails_when_there_is_no_data() {
-        let result = OrderedMessage::try_from_bytes(Vec::new());
-        assert_eq!(Err(MessageError::NoData), result);
-    }
-
-    #[test]
-    fn fails_when_data_is_too_short() {
-        let result = OrderedMessage::try_from_bytes(vec![1, 2, 3]);
-        assert_eq!(
-            Err(MessageError::IndexTooShort {
-                received: 3,
-                expected: 8
-            }),
-            result
-        );
-    }
-
-    #[test]
-    fn works_when_there_is_enough_to_make_a_sequence_number_but_no_message_data() {
-        let expected = OrderedMessage {
-            data: Vec::new(),
-            index: 1,
-        };
-        let result = OrderedMessage::try_from_bytes(vec![0, 0, 0, 0, 0, 0, 0, 1]).unwrap();
-        assert_eq!(expected, result);
-    }
-
-    #[test]
-    fn works_when_there_is_seq_number_and_data() {
-        let expected = OrderedMessage {
-            data: vec![255, 255, 255],
-            index: 1,
-        };
-        let result =
-            OrderedMessage::try_from_bytes(vec![0, 0, 0, 0, 0, 0, 0, 1, 255, 255, 255]).unwrap();
-        assert_eq!(expected, result);
-    }
-}
-
-#[test]
-fn empty_message_does_not_affect_ordering() {
-    let mut msg1 = OrderedMessage {
-        data: vec![255, 255, 255],
-        index: 1,
-    };
-
-    let mut msg2 = OrderedMessage {
-        data: vec![],
-        index: 2,
-    };
-
-    assert!(msg1 < msg2);
-
-    msg1.index = 2;
-    msg2.index = 1;
-
-    assert!(msg1 > msg2);
-}

common/socks5/ordered-buffer/src/sender.rs

@@ -1,57 +0,0 @@
-use crate::message::OrderedMessage;
-
-/// Assigns sequence numbers to outbound byte vectors. These messages can then
-/// be reassembled into an ordered sequence by the `OrderedMessageSender`.
-#[derive(Debug)]
-pub struct OrderedMessageSender {
-    next_index: u64,
-}
-
-impl OrderedMessageSender {
-    pub fn new() -> OrderedMessageSender {
-        OrderedMessageSender { next_index: 0 }
-    }
-
-    /// Turns raw bytes into an OrderedMessage containing the original bytes
-    /// and a sequence number;
-    pub fn wrap_message(&mut self, input: Vec<u8>) -> OrderedMessage {
-        let message = OrderedMessage {
-            data: input.to_vec(),
-            index: self.next_index,
-        };
-        self.next_index += 1;
-        message
-    }
-}
-
-impl Default for OrderedMessageSender {
-    fn default() -> Self {
-        OrderedMessageSender::new()
-    }
-}
-
-#[cfg(test)]
-mod ordered_message_sender {
-    use super::*;
-
-    mod when_input_bytes_are_empty {}
-
-    #[cfg(test)]
-    mod sequence_index_numbers {
-        use super::*;
-
-        #[test]
-        fn increase_as_messages_are_sent() {
-            let mut sender = OrderedMessageSender::new();
-            let first_bytes = vec![1, 2, 3, 4];
-            let second_bytes = vec![5, 6, 7, 8];
-
-            let first_message = sender.wrap_message(first_bytes);
-
-            assert_eq!(first_message.index, 0);
-
-            let second_message = sender.wrap_message(second_bytes);
-            assert_eq!(second_message.index, 1);
-        }
-    }
-}

common/socks5/proxy-helpers/src/connection_controller.rs

@@ -1,11 +1,11 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
 use futures::channel::mpsc;
 use futures::StreamExt;
 use log::*;
-use nym_ordered_buffer::{OrderedMessage, OrderedMessageBuffer, ReadContiguousData};
-use nym_socks5_requests::{ConnectionId, NetworkData, SendRequest};
+use nym_ordered_buffer::{OrderedMessageBuffer, ReadContiguousData};
+use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::{ConnectionCommand, ConnectionCommandSender};
 use nym_task::TaskClient;
 use std::collections::{HashMap, HashSet};
@@ -40,29 +40,13 @@ pub enum ControllerCommand {
         connection_id: ConnectionId,
     },
     Send {
-        connection_id: ConnectionId,
-        data: Vec<u8>,
-        is_closed: bool,
+        data: SocketData,
     },
 }
 
-impl From<NetworkData> for ControllerCommand {
-    fn from(value: NetworkData) -> Self {
-        ControllerCommand::Send {
-            connection_id: value.connection_id,
-            data: value.data,
-            is_closed: value.is_closed,
-        }
-    }
-}
-
-impl From<SendRequest> for ControllerCommand {
-    fn from(value: SendRequest) -> Self {
-        ControllerCommand::Send {
-            connection_id: value.conn_id,
-            data: value.data,
-            is_closed: value.local_closed,
-        }
+impl ControllerCommand {
+    pub fn new_send(data: SocketData) -> Self {
+        ControllerCommand::Send { data }
     }
 }
 
@@ -74,18 +58,13 @@ struct ActiveConnection {
 }
 
 impl ActiveConnection {
-    fn write_to_buf(&mut self, payload: Vec<u8>, is_closed: bool) {
-        let ordered_message = match OrderedMessage::try_from_bytes(payload) {
-            Ok(msg) => msg,
-            Err(err) => {
-                error!("Malformed ordered message - {err}");
-                return;
-            }
-        };
+    fn write_to_buf(&mut self, seq: u64, payload: Vec<u8>, is_closed: bool) {
         if is_closed {
-            self.closed_at_index = Some(ordered_message.index);
+            self.closed_at_index = Some(seq);
+        }
+        if let Err(err) = self.ordered_buffer.write(seq, payload) {
+            error!("failed to write to the buffer: {err}")
         }
-        self.ordered_buffer.write(ordered_message);
     }
 
     fn read_from_buf(&mut self) -> Option<ReadContiguousData> {
@@ -117,7 +96,7 @@ pub struct Controller {
 
     // buffer for messages received before connection was established due to mixnet being able to
     // un-order messages. Note we don't ever expect to have more than 1-2 messages per connection here
-    pending_messages: HashMap<ConnectionId, Vec<(Vec<u8>, bool)>>,
+    pending_messages: HashMap<ConnectionId, Vec<SocketData>>,
 
     shutdown: TaskClient,
 }
@@ -154,8 +133,8 @@ impl Controller {
             // check if there were any pending messages
             if let Some(pending) = self.pending_messages.remove(&conn_id) {
                 debug!("There were some pending messages for {}", conn_id);
-                for (payload, is_closed) in pending {
-                    self.send_to_connection(conn_id, payload, is_closed)
+                for data in pending {
+                    self.send_to_connection(data)
                 }
             }
         }
@@ -184,20 +163,25 @@ impl Controller {
         }
     }
 
-    fn send_to_connection(&mut self, conn_id: ConnectionId, payload: Vec<u8>, is_closed: bool) {
-        if let Some(active_connection) = self.active_connections.get_mut(&conn_id) {
-            if !payload.is_empty() {
-                active_connection.write_to_buf(payload, is_closed);
-            } else if !is_closed {
-                error!("Tried to write an empty message to a not-closing connection. Please let us know if you see this message");
-            }
+    fn send_to_connection(&mut self, message: SocketData) {
+        let hdr = message.header;
+        if let Some(active_connection) = self.active_connections.get_mut(&hdr.connection_id) {
+            // always write to the buffer even if payload is empty (because it could have been the keep-alive message)
+            active_connection.write_to_buf(hdr.seq, message.data, hdr.local_socket_closed);
 
             if let Some(payload) = active_connection.read_from_buf() {
                 if let Some(closed_at_index) = active_connection.closed_at_index {
-                    if payload.last_index > closed_at_index {
+                    if payload.last_sequence > closed_at_index {
                         active_connection.is_closed = true;
                     }
                 }
+
+                // however, don't send empty payload to the actual connection if it's not a close message
+                // TODO: or should we?
+                if payload.data.is_empty() && !active_connection.is_closed {
+                    return;
+                }
+
                 if let Err(err) = active_connection
                     .connection_sender
                     .as_mut()
@@ -207,34 +191,26 @@ impl Controller {
                         socket_closed: active_connection.is_closed,
                     })
                 {
-                    error!("WTF IS THIS: {err}");
+                    error!("failed to send on the active connection channel: {err}");
                 }
-
-                // TODO: ABOVE UNWRAP CAUSED A CRASH IN A NORMAL USE!!!!
-                // TODO:
-                // TODO: surprisingly it only happened on socks client, never on nSP
-                // TODO:
-                // TODO:
-                // TODO:
-                // TODO:
             }
-        } else if !self.recently_closed.contains(&conn_id) {
+        } else if !self.recently_closed.contains(&hdr.connection_id) {
             debug!("Received a 'Send' before 'Connect' - going to buffer the data");
             let pending = self
                 .pending_messages
-                .entry(conn_id)
+                .entry(hdr.connection_id)
                 .or_insert_with(Vec::new);
-            pending.push((payload, is_closed));
-        } else if !is_closed {
+            pending.push(message);
+        } else if !hdr.local_socket_closed {
             error!(
                 "Tried to write to closed connection {} ({} bytes were 'lost)",
-                conn_id,
-                payload.len()
+                hdr.connection_id,
+                message.data.len()
             );
         } else {
             debug!(
                 "Tried to write to closed connection {}, but remote is already closed",
-                conn_id
+                hdr.connection_id
             )
         }
     }
@@ -243,8 +219,8 @@ impl Controller {
         loop {
             tokio::select! {
                 command = self.receiver.next() => match command {
-                    Some(ControllerCommand::Send{connection_id, data, is_closed}) => {
-                        self.send_to_connection(connection_id, data, is_closed)
+                    Some(ControllerCommand::Send{data}) => {
+                        self.send_to_connection(data)
                     }
                     Some(ControllerCommand::Insert{connection_id, connection_sender}) => {
                         self.insert_connection(connection_id, connection_sender)

common/socks5/proxy-helpers/src/lib.rs

@@ -3,4 +3,5 @@
 
 pub mod available_reader;
 pub mod connection_controller;
+pub mod ordered_sender;
 pub mod proxy_runner;

common/socks5/proxy-helpers/src/ordered_sender.rs

@@ -0,0 +1,116 @@
+// Copyright 2023 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::proxy_runner::MixProxySender;
+use bytes::Bytes;
+use log::{debug, error};
+use nym_socks5_requests::{ConnectionId, SocketData};
+use std::io;
+
+pub(crate) struct OrderedMessageSender<F, S> {
+    connection_id: ConnectionId,
+    // addresses are provided for better logging
+    local_destination_address: String,
+    remote_source_address: String,
+    mixnet_sender: MixProxySender<S>,
+
+    next_message_seq: u64,
+    mix_message_adapter: F,
+}
+
+impl<F, S> OrderedMessageSender<F, S>
+where
+    F: Fn(SocketData) -> S,
+{
+    pub(crate) fn new(
+        local_destination_address: String,
+        remote_source_address: String,
+        connection_id: ConnectionId,
+        mixnet_sender: MixProxySender<S>,
+        mix_message_adapter: F,
+    ) -> Self {
+        OrderedMessageSender {
+            local_destination_address,
+            remote_source_address,
+            connection_id,
+            mixnet_sender,
+            next_message_seq: 0,
+            mix_message_adapter,
+        }
+    }
+
+    fn sequence(&mut self) -> u64 {
+        let next = self.next_message_seq;
+        self.next_message_seq += 1;
+        next
+    }
+
+    fn construct_message(&mut self, data: Vec<u8>, local_socket_closed: bool) -> S {
+        let data = SocketData::new(
+            self.sequence(),
+            self.connection_id,
+            local_socket_closed,
+            data,
+        );
+        (self.mix_message_adapter)(data)
+    }
+
+    async fn send_message(&self, message: S) {
+        if self.mixnet_sender.send(message).await.is_err() {
+            panic!("BatchRealMessageReceiver has stopped receiving!")
+        }
+    }
+
+    pub(crate) async fn send_empty_close(&mut self) {
+        let message = self.construct_message(Vec::new(), true);
+        self.send_message(message).await
+    }
+
+    pub(crate) async fn send_empty_keepalive(&mut self) {
+        log::trace!("Sending keepalive for connection: {}", self.connection_id);
+        let message = self.construct_message(Vec::new(), false);
+        self.send_message(message).await
+    }
+
+    pub(crate) fn process_data(&self, read_data: Option<io::Result<Bytes>>) -> ProcessedData {
+        let (read_data, is_finished) = match read_data {
+            Some(data) => match data {
+                Ok(data) => (data, false),
+                Err(err) => {
+                    error!(target: &*format!("({}) socks5 inbound", self.connection_id), "failed to read request from the socket - {err}");
+                    (Default::default(), true)
+                }
+            },
+            None => (Default::default(), true),
+        };
+
+        ProcessedData {
+            data: read_data,
+            is_done: is_finished,
+        }
+    }
+
+    fn log_sent_message(&self, data: &ProcessedData) {
+        debug!(
+            target: &*format!("({}) socks5 inbound", self.connection_id),
+            "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}",
+            data.data.len(),
+            self.local_destination_address,
+            self.remote_source_address,
+            data.is_done
+        );
+    }
+
+    /// Send data read from local socket into the mixnet
+    pub(crate) async fn send_data(&mut self, data: ProcessedData) {
+        self.log_sent_message(&data);
+        let message = self.construct_message(data.data.into(), data.is_done);
+        self.send_message(message).await;
+    }
+}
+
+// helper wrapper to keep track of field meanings
+pub(crate) struct ProcessedData {
+    data: Bytes,
+    pub(crate) is_done: bool,
+}

common/socks5/proxy-helpers/src/proxy_runner/inbound.rs

@@ -1,106 +1,22 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
+// Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use super::MixProxySender;
 use super::SHUTDOWN_TIMEOUT;
 use crate::available_reader::AvailableReader;
+use crate::ordered_sender::OrderedMessageSender;
 use crate::proxy_runner::KEEPALIVE_INTERVAL;
-use bytes::Bytes;
 use futures::FutureExt;
 use futures::StreamExt;
 use log::*;
-use nym_ordered_buffer::OrderedMessageSender;
-use nym_socks5_requests::ConnectionId;
+use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::LaneQueueLengths;
 use nym_task::connections::TransmissionLane;
 use nym_task::TaskClient;
-use std::fmt::Debug;
+use std::sync::Arc;
 use std::time::Duration;
-use std::{io, sync::Arc};
 use tokio::select;
 use tokio::{net::tcp::OwnedReadHalf, sync::Notify, time::sleep};
 
-async fn send_empty_close<F, S>(
-    connection_id: ConnectionId,
-    message_sender: &mut OrderedMessageSender,
-    mix_sender: &MixProxySender<S>,
-    adapter_fn: F,
-) where
-    F: Fn(ConnectionId, Vec<u8>, bool) -> S,
-    S: Debug,
-{
-    let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes();
-    mix_sender
-        .send(adapter_fn(connection_id, ordered_msg, true))
-        .await
-        .expect("BatchRealMessageReceiver has stopped receiving!");
-}
-
-async fn send_empty_keepalive<F, S>(
-    connection_id: ConnectionId,
-    message_sender: &mut OrderedMessageSender,
-    mix_sender: &MixProxySender<S>,
-    adapter_fn: F,
-) where
-    F: Fn(ConnectionId, Vec<u8>, bool) -> S,
-    S: Debug,
-{
-    log::trace!("Sending keepalive for connection: {connection_id}");
-    let ordered_msg = message_sender.wrap_message(Vec::new()).into_bytes();
-    mix_sender
-        .send(adapter_fn(connection_id, ordered_msg, false))
-        .await
-        .expect("BatchRealMessageReceiver has stopped receiving!");
-}
-
-async fn deal_with_data<F, S>(
-    read_data: Option<io::Result<Bytes>>,
-    local_destination_address: &str,
-    remote_source_address: &str,
-    connection_id: ConnectionId,
-    message_sender: &mut OrderedMessageSender,
-    mix_sender: &MixProxySender<S>,
-    adapter_fn: F,
-) -> bool
-where
-    F: Fn(ConnectionId, Vec<u8>, bool) -> S,
-    S: Debug,
-{
-    let (read_data, is_finished) = match read_data {
-        Some(data) => match data {
-            Ok(data) => (data, false),
-            Err(err) => {
-                error!(target: &*format!("({connection_id}) socks5 inbound"), "failed to read request from the socket - {err}");
-                (Default::default(), true)
-            }
-        },
-        None => (Default::default(), true),
-    };
-
-    debug!(
-        target: &*format!("({connection_id}) socks5 inbound"),
-        "[{} bytes]\t{} → local → mixnet → remote → {}. Local closed: {}",
-        read_data.len(),
-        local_destination_address,
-        remote_source_address,
-        is_finished
-    );
-
-    // if we're sending through the mixnet increase the sequence number...
-    let ordered_msg = message_sender.wrap_message(read_data.to_vec()).into_bytes();
-    log::trace!(
-        "pushing data down the input sender: size: {}",
-        ordered_msg.len()
-    );
-
-    mix_sender
-        .send(adapter_fn(connection_id, ordered_msg, is_finished))
-        .await
-        .expect("InputMessageReceiver has stopped receiving!");
-
-    is_finished
-}
-
 async fn wait_until_lane_empty(lane_queue_lengths: &Option<LaneQueueLengths>, connection_id: u64) {
     if let Some(lane_queue_lengths) = lane_queue_lengths {
         if tokio::time::timeout(
@@ -158,27 +74,21 @@ async fn wait_for_lane(
     }
 }
 
-#[allow(clippy::too_many_arguments)]
 pub(super) async fn run_inbound<F, S>(
     mut reader: OwnedReadHalf,
-    local_destination_address: String, // addresses are provided for better logging
-    remote_source_address: String,
+    mut message_sender: OrderedMessageSender<F, S>,
     connection_id: ConnectionId,
-    mix_sender: MixProxySender<S>,
     available_plaintext_per_mix_packet: usize,
-    adapter_fn: F,
     shutdown_notify: Arc<Notify>,
     lane_queue_lengths: Option<LaneQueueLengths>,
     mut shutdown_listener: TaskClient,
 ) -> OwnedReadHalf
 where
-    F: Fn(ConnectionId, Vec<u8>, bool) -> S + Send + 'static,
-    S: Debug,
+    F: Fn(SocketData) -> S + Send + 'static,
 {
     // TODO: this multiplication by 4 is completely arbitrary here
     let mut available_reader =
         AvailableReader::new(&mut reader, Some(available_plaintext_per_mix_packet * 4));
-    let mut message_sender = OrderedMessageSender::new();
 
     // Shutdown if outbound signled to shutdown
     let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT));
@@ -217,7 +127,7 @@ where
                 );
                 // inform remote just in case it was closed because of lack of heartbeat.
                 // worst case the remote will just have couple of false negatives
-                send_empty_close(connection_id, &mut message_sender, &mix_sender, &adapter_fn).await;
+                message_sender.send_empty_close().await;
                 break;
             }
             _ = shutdown_listener.recv() => {
@@ -233,7 +143,7 @@ where
                 break;
             }
             _ = keepalive_timer.tick() => {
-                send_empty_keepalive(connection_id, &mut message_sender, &mix_sender, &adapter_fn).await;
+                message_sender.send_empty_keepalive().await;
             }
             // Read the next data when there is space in the lane.
             // The purpose of chaining the wait here is that it makes sure we can cancel the
@@ -241,15 +151,12 @@ where
             read_data = wait_until_lane_almost_empty(&lane_queue_lengths, connection_id)
                 .then(|_| available_reader.next()), if !we_are_closed =>
             {
-                if deal_with_data(
-                    read_data,
-                    &local_destination_address,
-                    &remote_source_address,
-                    connection_id,
-                    &mut message_sender,
-                    &mix_sender,
-                    &adapter_fn,
-                ).await {
+                let processed = message_sender.process_data(read_data);
+                let is_done = processed.is_done;
+
+                message_sender.send_data(processed).await;
+
+                if is_done {
                     // After reading the last data, notify the closing_future to wait until the
                     // lane is clear before exiting.
                     // We don't wait here since we want to be able to cancel the wait on close or

common/socks5/proxy-helpers/src/proxy_runner/mod.rs

@@ -2,7 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::connection_controller::ConnectionReceiver;
-use nym_socks5_requests::ConnectionId;
+use crate::ordered_sender::OrderedMessageSender;
+use nym_socks5_requests::{ConnectionId, SocketData};
 use nym_task::connections::LaneQueueLengths;
 use nym_task::TaskClient;
 use std::fmt::Debug;
@@ -92,20 +93,24 @@ where
     // request/response as required by entity running particular side of the proxy.
     pub async fn run<F>(mut self, adapter_fn: F) -> Self
     where
-        F: Fn(ConnectionId, Vec<u8>, bool) -> S + Send + Sync + 'static,
+        F: Fn(SocketData) -> S + Send + Sync + 'static,
     {
         let (read_half, write_half) = self.socket.take().unwrap().into_split();
         let shutdown_notify = Arc::new(Notify::new());
 
         // should run until either inbound closes or is notified from outbound
-        let inbound_future = inbound::run_inbound(
-            read_half,
+        let ordered_sender = OrderedMessageSender::new(
             self.local_destination_address.clone(),
             self.remote_source_address.clone(),
             self.connection_id,
             self.mix_sender.clone(),
-            self.available_plaintext_per_mix_packet,
             adapter_fn,
+        );
+        let inbound_future = inbound::run_inbound(
+            read_half,
+            ordered_sender,
+            self.connection_id,
+            self.available_plaintext_per_mix_packet,
             Arc::clone(&shutdown_notify),
             self.lane_queue_lengths.clone(),
             self.shutdown_listener.clone(),

common/socks5/requests/src/lib.rs

@@ -3,6 +3,7 @@
 
 use nym_service_providers_common::interface;
 use nym_service_providers_common::interface::ServiceProviderMessagingError;
+use std::mem;
 use thiserror::Error;
 
 pub use request::*;
@@ -16,6 +17,185 @@ pub mod version;
 pub type Socks5ProviderRequest = interface::Request<Socks5Request>;
 pub type Socks5ProviderResponse = interface::Response<Socks5Request>;
 
+#[derive(Debug, Error, PartialEq, Eq)]
+#[error(
+    "didn't receive enough data to recover socket data. got {received}, but expected at least {expected}"
+)]
+pub struct InsufficientSocketDataError {
+    received: usize,
+    expected: usize,
+}
+
+#[derive(Debug, Clone, Copy, Eq, PartialEq)]
+pub struct SocketDataHeader {
+    pub seq: u64,
+    pub connection_id: ConnectionId,
+    pub local_socket_closed: bool,
+}
+
+impl SocketDataHeader {
+    const SERIALIZED_LEN: usize = mem::size_of::<ConnectionId>() + 1 + mem::size_of::<u64>();
+
+    // we need to have two serialization methods for backwards compatibility,
+    // since we serialized those fields differently depending on whether it was ingress vs egress...
+
+    pub fn try_from_request_bytes(
+        b: &[u8],
+    ) -> Result<SocketDataHeader, InsufficientSocketDataError> {
+        if b.len() != Self::SERIALIZED_LEN {
+            return Err(InsufficientSocketDataError {
+                received: b.len(),
+                expected: Self::SERIALIZED_LEN,
+            });
+        }
+
+        // the unwraps here are fine as we just ensured we have the exact amount of bytes we need
+        let connection_id = ConnectionId::from_be_bytes(b[0..8].try_into().unwrap());
+        let local_socket_closed = b[8] != 0;
+        let seq = u64::from_be_bytes(b[9..].try_into().unwrap());
+
+        Ok(SocketDataHeader {
+            seq,
+            connection_id,
+            local_socket_closed,
+        })
+    }
+
+    // the serialization of the header looks as follows:
+    // (it's vital it's not modified as we need this exact structure for backwards compatibility)
+    // CONNECTION_ID (8B) || SOCKET_CLOSED (1B) || SEQUENCE (8B)
+    pub fn into_request_bytes(self) -> Vec<u8> {
+        self.into_request_bytes_iter().collect()
+    }
+
+    pub fn into_request_bytes_iter(self) -> impl Iterator<Item = u8> {
+        self.connection_id
+            .to_be_bytes()
+            .into_iter()
+            .chain(std::iter::once(self.local_socket_closed as u8))
+            .chain(self.seq.to_be_bytes().into_iter())
+    }
+
+    pub fn try_from_response_bytes(
+        b: &[u8],
+    ) -> Result<SocketDataHeader, InsufficientSocketDataError> {
+        if b.len() != Self::SERIALIZED_LEN {
+            return Err(InsufficientSocketDataError {
+                received: b.len(),
+                expected: Self::SERIALIZED_LEN,
+            });
+        }
+
+        // the unwraps here are fine as we just ensured we have the exact amount of bytes we need
+        let local_socket_closed = b[0] != 0;
+        let connection_id = ConnectionId::from_be_bytes(b[1..9].try_into().unwrap());
+        let seq = u64::from_be_bytes(b[9..].try_into().unwrap());
+
+        Ok(SocketDataHeader {
+            seq,
+            connection_id,
+            local_socket_closed,
+        })
+    }
+
+    // the serialization of the header looks as follows:
+    // (it's vital it's not modified as we need this exact structure for backwards compatibility)
+    // SOCKET_CLOSED (1B) || CONNECTION_ID (8B) || SEQUENCE (8B)
+    pub fn into_response_bytes(self) -> Vec<u8> {
+        self.into_response_bytes_iter().collect()
+    }
+
+    pub fn into_response_bytes_iter(self) -> impl Iterator<Item = u8> {
+        std::iter::once(self.local_socket_closed as u8)
+            .chain(self.connection_id.to_be_bytes().into_iter())
+            .chain(self.seq.to_be_bytes().into_iter())
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SocketData {
+    pub header: SocketDataHeader,
+    pub data: Vec<u8>,
+}
+
+impl SocketData {
+    pub fn new(
+        seq: u64,
+        connection_id: ConnectionId,
+        local_socket_closed: bool,
+        data: Vec<u8>,
+    ) -> Self {
+        SocketData {
+            header: SocketDataHeader {
+                seq,
+                connection_id,
+                local_socket_closed,
+            },
+            data,
+        }
+    }
+
+    fn verify_deserialization_len(b: &[u8]) -> Result<(), InsufficientSocketDataError> {
+        if b.is_empty() {
+            return Err(InsufficientSocketDataError {
+                received: 0,
+                expected: SocketDataHeader::SERIALIZED_LEN,
+            });
+        }
+
+        if b.len() < SocketDataHeader::SERIALIZED_LEN {
+            return Err(InsufficientSocketDataError {
+                received: b.len(),
+                expected: SocketDataHeader::SERIALIZED_LEN,
+            });
+        }
+        Ok(())
+    }
+
+    // we need to have two serialization methods for backwards compatibility,
+    // since we serialized those fields differently depending on whether it was ingress vs egress...
+    pub fn try_from_request_bytes(b: &[u8]) -> Result<SocketData, InsufficientSocketDataError> {
+        Self::verify_deserialization_len(b)?;
+        let header =
+            SocketDataHeader::try_from_request_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?;
+        let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec();
+
+        Ok(SocketData { header, data })
+    }
+
+    // the serialization of the socket data looks as follows:
+    // HEADER || DATA
+    pub fn into_request_bytes(self) -> Vec<u8> {
+        self.into_request_bytes_iter().collect()
+    }
+
+    pub fn into_request_bytes_iter(self) -> impl Iterator<Item = u8> {
+        self.header
+            .into_request_bytes_iter()
+            .chain(self.data.into_iter())
+    }
+
+    pub fn try_from_response_bytes(b: &[u8]) -> Result<SocketData, InsufficientSocketDataError> {
+        Self::verify_deserialization_len(b)?;
+
+        let header =
+            SocketDataHeader::try_from_response_bytes(&b[..SocketDataHeader::SERIALIZED_LEN])?;
+        let data = b[SocketDataHeader::SERIALIZED_LEN..].to_vec();
+
+        Ok(SocketData { header, data })
+    }
+
+    pub fn into_response_bytes(self) -> Vec<u8> {
+        self.into_response_bytes_iter().collect()
+    }
+
+    pub fn into_response_bytes_iter(self) -> impl Iterator<Item = u8> {
+        self.header
+            .into_response_bytes_iter()
+            .chain(self.data.into_iter())
+    }
+}
+
 #[derive(Debug, Error)]
 pub enum Socks5RequestError {
     #[error("failed to deserialize received request: {source}")]
@@ -51,6 +231,103 @@ mod tests {
     use super::*;
     use nym_service_providers_common::interface::RequestContent;
 
+    #[cfg(test)]
+    mod socket_data_serialization {
+        use super::*;
+
+        #[test]
+        fn for_requests() {
+            assert_eq!(
+                InsufficientSocketDataError {
+                    received: 0,
+                    expected: SocketDataHeader::SERIALIZED_LEN
+                },
+                SocketData::try_from_request_bytes(&[]).unwrap_err()
+            );
+
+            assert_eq!(
+                InsufficientSocketDataError {
+                    received: 10,
+                    expected: SocketDataHeader::SERIALIZED_LEN
+                },
+                SocketData::try_from_request_bytes(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]).unwrap_err()
+            );
+
+            let good_data = SocketData::new(42, 12345, false, vec![2, 3]);
+            let serialized = good_data.clone().into_request_bytes();
+
+            assert_eq!(
+                good_data,
+                SocketData::try_from_request_bytes(&serialized).unwrap()
+            );
+            assert_ne!(
+                good_data,
+                SocketData::try_from_response_bytes(&serialized).unwrap()
+            );
+
+            let raw_bytes = [
+                6, 6, 6, 6, 6, 6, 6, 6, 0, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255,
+            ];
+            assert_eq!(
+                SocketData {
+                    header: SocketDataHeader {
+                        seq: u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
+                        connection_id: ConnectionId::from_be_bytes([6, 6, 6, 6, 6, 6, 6, 6]),
+                        local_socket_closed: false,
+                    },
+                    data: vec![255, 255, 255],
+                },
+                SocketData::try_from_request_bytes(&raw_bytes).unwrap()
+            )
+        }
+
+        #[test]
+        fn for_responses() {
+            assert_eq!(
+                InsufficientSocketDataError {
+                    received: 0,
+                    expected: SocketDataHeader::SERIALIZED_LEN
+                },
+                SocketData::try_from_response_bytes(&[]).unwrap_err()
+            );
+
+            assert_eq!(
+                InsufficientSocketDataError {
+                    received: 10,
+                    expected: SocketDataHeader::SERIALIZED_LEN
+                },
+                SocketData::try_from_response_bytes(&[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]).unwrap_err()
+            );
+
+            let good_data = SocketData::new(42, 12345, false, vec![2, 3]);
+            let serialized = good_data.clone().into_response_bytes();
+
+            assert_eq!(
+                good_data,
+                SocketData::try_from_response_bytes(&serialized).unwrap()
+            );
+            assert_ne!(
+                good_data,
+                SocketData::try_from_request_bytes(&serialized).unwrap()
+            );
+
+            let raw_bytes = [
+                0, 6, 6, 6, 6, 6, 6, 6, 6, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255,
+            ];
+            assert_eq!(
+                SocketData {
+                    header: SocketDataHeader {
+                        seq: u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
+                        connection_id: ConnectionId::from_be_bytes([6, 6, 6, 6, 6, 6, 6, 6]),
+                        local_socket_closed: false,
+                    },
+                    data: vec![255, 255, 255],
+                },
+                SocketData::try_from_response_bytes(&raw_bytes).unwrap()
+            )
+        }
+    }
+
     #[cfg(test)]
     mod interface_backwards_compatibility {
         use super::*;
@@ -99,9 +376,10 @@ mod tests {
             match new_deserialized.content {
                 RequestContent::ProviderData(req) => match req.content {
                     Socks5RequestContent::Send(send_req) => {
-                        assert_eq!(send_req.conn_id, 7810961472501196273);
-                        assert_eq!(send_req.data.len(), 111);
-                        assert!(!send_req.local_closed);
+                        assert_eq!(send_req.data.header.connection_id, 7810961472501196273);
+                        assert_eq!(send_req.data.header.seq, 0);
+                        assert_eq!(send_req.data.data.len(), 103);
+                        assert!(!send_req.data.header.local_socket_closed);
                     }
                     _ => panic!("unexpected request"),
                 },

common/socks5/requests/src/request.rs

@@ -1,7 +1,10 @@
 // Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::{make_bincode_serializer, Socks5ProtocolVersion, Socks5RequestError, Socks5Response};
+use crate::{
+    make_bincode_serializer, InsufficientSocketDataError, SocketData, Socks5ProtocolVersion,
+    Socks5RequestError, Socks5Response,
+};
 use nym_service_providers_common::interface::{Serializable, ServiceProviderRequest};
 use nym_sphinx_addressing::clients::{Recipient, RecipientFormattingError};
 use serde::{Deserialize, Serialize};
@@ -61,6 +64,9 @@ pub enum RequestDeserializationError {
         #[from]
         source: bincode::Error,
     },
+
+    #[error(transparent)]
+    InvalidSocketData(#[from] InsufficientSocketDataError),
 }
 
 impl RequestDeserializationError {
@@ -79,9 +85,7 @@ pub struct ConnectRequest {
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct SendRequest {
-    pub conn_id: ConnectionId,
-    pub data: Vec<u8>,
-    pub local_closed: bool,
+    pub data: SocketData,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -178,15 +182,10 @@ impl Socks5Request {
         }
     }
 
-    pub fn new_send(
-        protocol_version: Socks5ProtocolVersion,
-        conn_id: ConnectionId,
-        data: Vec<u8>,
-        local_closed: bool,
-    ) -> Socks5Request {
+    pub fn new_send(protocol_version: Socks5ProtocolVersion, data: SocketData) -> Socks5Request {
         Socks5Request {
             protocol_version,
-            content: Socks5RequestContent::new_send(conn_id, data, local_closed),
+            content: Socks5RequestContent::new_send(data),
         }
     }
 
@@ -231,16 +230,8 @@ impl Socks5RequestContent {
     }
 
     /// Construct a new Request::Send instance
-    pub fn new_send(
-        conn_id: ConnectionId,
-        data: Vec<u8>,
-        local_closed: bool,
-    ) -> Socks5RequestContent {
-        Socks5RequestContent::Send(SendRequest {
-            conn_id,
-            data,
-            local_closed,
-        })
+    pub fn new_send(data: SocketData) -> Socks5RequestContent {
+        Socks5RequestContent::Send(SendRequest { data })
     }
 
     /// Deserialize the request type, connection id, destination address and port,
@@ -257,6 +248,14 @@ impl Socks5RequestContent {
     /// The request_flag tells us whether this is a new connection request (`new_connect`),
     /// an already-established connection we should send up (`new_send`), or
     /// a request to close an established connection (`new_close`).
+
+    // connect:
+    // RequestFlag::Connect || CONN_ID || ADDR_LEN || ADDR || <RETURN_ADDR>
+    //
+    // send:
+    // RequestFlag::Send || CONN_ID || LOCAL_CLOSED || DATA
+    // where DATA: SEQ || TRUE_DATA
+
     pub fn try_from_bytes(b: &[u8]) -> Result<Socks5RequestContent, RequestDeserializationError> {
         // each request needs to at least contain flag and ConnectionId
         if b.is_empty() {
@@ -314,21 +313,9 @@ impl Socks5RequestContent {
                     return_address,
                 ))
             }
-            RequestFlag::Send => {
-                if b.len() < 9 {
-                    return Err(RequestDeserializationError::ConnectionIdTooShort);
-                }
-                let conn_id = u64::from_be_bytes([b[1], b[2], b[3], b[4], b[5], b[6], b[7], b[8]]);
-
-                let local_closed = b[9] != 0;
-                let data = b[10..].to_vec();
-
-                Ok(Socks5RequestContent::Send(SendRequest {
-                    conn_id,
-                    data,
-                    local_closed,
-                }))
-            }
+            RequestFlag::Send => Ok(Socks5RequestContent::Send(SendRequest {
+                data: SocketData::try_from_request_bytes(&b[1..])?,
+            })),
             RequestFlag::Query => {
                 use bincode::Options;
                 let query = make_bincode_serializer().deserialize(&b[1..])?;
@@ -359,9 +346,7 @@ impl Socks5RequestContent {
                 }
             }
             Socks5RequestContent::Send(req) => std::iter::once(RequestFlag::Send as u8)
-                .chain(req.conn_id.to_be_bytes().into_iter())
-                .chain(std::iter::once(req.local_closed as u8))
-                .chain(req.data.into_iter())
+                .chain(req.data.into_request_bytes_iter())
                 .collect(),
 
             Socks5RequestContent::Query(query) => {
@@ -618,26 +603,7 @@ mod request_deserialization_tests {
 
         #[test]
         fn works_when_request_is_sized_properly_even_without_data() {
-            // correct 8 bytes of connection_id, 1 byte of local_closed and 0 bytes request data
-            let request_bytes = [RequestFlag::Send as u8, 1, 2, 3, 4, 5, 6, 7, 8, 0].to_vec();
-            let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap();
-            match request {
-                Socks5RequestContent::Send(SendRequest {
-                    conn_id,
-                    data,
-                    local_closed,
-                }) => {
-                    assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id);
-                    assert_eq!(Vec::<u8>::new(), data);
-                    assert!(!local_closed)
-                }
-                _ => unreachable!(),
-            }
-        }
-
-        #[test]
-        fn works_when_request_is_sized_properly_and_has_data() {
-            // correct 8 bytes of connection_id, 1 byte of local_closed and 3 bytes request data (all 255)
+            // correct 8 bytes of connection_id, 1 byte of local_closed, 8 bytes of sequence and 0 bytes request data
             let request_bytes = [
                 RequestFlag::Send as u8,
                 1,
@@ -649,6 +615,53 @@ mod request_deserialization_tests {
                 7,
                 8,
                 0,
+                0,
+                0,
+                0,
+                0,
+                0,
+                0,
+                0,
+                1,
+            ]
+            .to_vec();
+            let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap();
+            match request {
+                Socks5RequestContent::Send(SendRequest { data }) => {
+                    assert_eq!(
+                        u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]),
+                        data.header.connection_id
+                    );
+                    assert!(!data.header.local_socket_closed);
+                    assert_eq!(1, data.header.seq);
+                    assert!(data.data.is_empty());
+                }
+                _ => unreachable!(),
+            }
+        }
+
+        #[test]
+        fn works_when_request_is_sized_properly_and_has_data() {
+            // correct 8 bytes of connection_id, 1 byte of local_closed, 8 bytes of sequence and 3 bytes request data (all 255)
+            let request_bytes = [
+                RequestFlag::Send as u8,
+                1,
+                2,
+                3,
+                4,
+                5,
+                6,
+                7,
+                8,
+                1,
+                0,
+                0,
+                0,
+                0,
+                0,
+                0,
+                0,
+                1,
                 255,
                 255,
                 255,
@@ -657,14 +670,14 @@ mod request_deserialization_tests {
 
             let request = Socks5RequestContent::try_from_bytes(&request_bytes).unwrap();
             match request {
-                Socks5RequestContent::Send(SendRequest {
-                    conn_id,
-                    data,
-                    local_closed,
-                }) => {
-                    assert_eq!(u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]), conn_id);
-                    assert_eq!(vec![255, 255, 255], data);
-                    assert!(!local_closed)
+                Socks5RequestContent::Send(SendRequest { data }) => {
+                    assert_eq!(
+                        u64::from_be_bytes([1, 2, 3, 4, 5, 6, 7, 8]),
+                        data.header.connection_id
+                    );
+                    assert!(data.header.local_socket_closed);
+                    assert_eq!(1, data.header.seq);
+                    assert_eq!(vec![255, 255, 255], data.data);
                 }
                 _ => unreachable!(),
             }

common/socks5/requests/src/response.rs

@@ -1,7 +1,10 @@
 // Copyright 2020-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::{make_bincode_serializer, ConnectionId, Socks5ProtocolVersion, Socks5RequestError};
+use crate::{
+    make_bincode_serializer, ConnectionId, InsufficientSocketDataError, SocketData,
+    Socks5ProtocolVersion, Socks5RequestError,
+};
 use nym_service_providers_common::interface::{Serializable, ServiceProviderResponse};
 use serde::{Deserialize, Serialize};
 use tap::TapFallible;
@@ -33,6 +36,12 @@ impl TryFrom<u8> for ResponseFlag {
 
 #[derive(Debug, Error)]
 pub enum ResponseDeserializationError {
+    #[error("the network data was malformed: {source}")]
+    MalformedNetworkData {
+        #[from]
+        source: InsufficientSocketDataError,
+    },
+
     #[error("not enough bytes to recover the connection id")]
     ConnectionIdTooShort,
 
@@ -115,23 +124,14 @@ impl Socks5Response {
 
     pub fn new_network_data(
         protocol_version: Socks5ProtocolVersion,
+        seq: u64,
         connection_id: ConnectionId,
         data: Vec<u8>,
         is_closed: bool,
     ) -> Socks5Response {
         Socks5Response {
             protocol_version,
-            content: Socks5ResponseContent::new_network_data(connection_id, data, is_closed),
-        }
-    }
-
-    pub fn new_closed_empty(
-        protocol_version: Socks5ProtocolVersion,
-        connection_id: ConnectionId,
-    ) -> Socks5Response {
-        Socks5Response {
-            protocol_version,
-            content: Socks5ResponseContent::new_closed_empty(connection_id),
+            content: Socks5ResponseContent::new_network_data(seq, connection_id, data, is_closed),
         }
     }
 
@@ -159,22 +159,21 @@ impl Socks5Response {
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub enum Socks5ResponseContent {
-    NetworkData(NetworkData),
+    NetworkData { content: SocketData },
     ConnectionError(ConnectionError),
     Query(QueryResponse),
 }
 
 impl Socks5ResponseContent {
     pub fn new_network_data(
+        seq: u64,
         connection_id: ConnectionId,
         data: Vec<u8>,
         is_closed: bool,
     ) -> Socks5ResponseContent {
-        Socks5ResponseContent::NetworkData(NetworkData::new(connection_id, data, is_closed))
-    }
-
-    pub fn new_closed_empty(connection_id: ConnectionId) -> Socks5ResponseContent {
-        Socks5ResponseContent::NetworkData(NetworkData::new_closed_empty(connection_id))
+        Socks5ResponseContent::NetworkData {
+            content: SocketData::new(seq, connection_id, is_closed, data),
+        }
     }
 
     pub fn new_connection_error(
@@ -186,9 +185,9 @@ impl Socks5ResponseContent {
 
     pub fn into_bytes(self) -> Vec<u8> {
         match self {
-            Socks5ResponseContent::NetworkData(res) => {
+            Socks5ResponseContent::NetworkData { content } => {
                 std::iter::once(ResponseFlag::NetworkData as u8)
-                    .chain(res.into_bytes().into_iter())
+                    .chain(content.into_response_bytes_iter())
                     .collect()
             }
             Socks5ResponseContent::ConnectionError(res) => {
@@ -220,9 +219,9 @@ impl Socks5ResponseContent {
 
         let response_flag = ResponseFlag::try_from(b[0])?;
         match response_flag {
-            ResponseFlag::NetworkData => Ok(Socks5ResponseContent::NetworkData(
-                NetworkData::try_from_bytes(&b[1..])?,
-            )),
+            ResponseFlag::NetworkData => Ok(Socks5ResponseContent::NetworkData {
+                content: SocketData::try_from_response_bytes(&b[1..])?,
+            }),
             ResponseFlag::ConnectionError => Ok(Socks5ResponseContent::ConnectionError(
                 ConnectionError::try_from_bytes(&b[1..])?,
             )),
@@ -235,73 +234,6 @@ impl Socks5ResponseContent {
     }
 }
 
-/// A remote network network data response retrieved by the Socks5 service provider. This
-/// can be serialized and sent back through the mixnet to the requesting
-/// application.
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct NetworkData {
-    pub data: Vec<u8>,
-    pub connection_id: ConnectionId,
-    pub is_closed: bool,
-}
-
-impl NetworkData {
-    /// Constructor for responses
-    pub fn new(connection_id: ConnectionId, data: Vec<u8>, is_closed: bool) -> Self {
-        NetworkData {
-            data,
-            connection_id,
-            is_closed,
-        }
-    }
-
-    pub fn new_closed_empty(connection_id: ConnectionId) -> Self {
-        NetworkData {
-            data: vec![],
-            connection_id,
-            is_closed: false,
-        }
-    }
-
-    pub fn try_from_bytes(b: &[u8]) -> Result<NetworkData, ResponseDeserializationError> {
-        if b.is_empty() {
-            return Err(ResponseDeserializationError::NoData);
-        }
-
-        let is_closed = b[0] != 0;
-
-        if b.len() < 9 {
-            return Err(ResponseDeserializationError::ConnectionIdTooShort);
-        }
-
-        let mut connection_id_bytes = b.to_vec();
-        let data = connection_id_bytes.split_off(9);
-
-        let connection_id = u64::from_be_bytes([
-            connection_id_bytes[1],
-            connection_id_bytes[2],
-            connection_id_bytes[3],
-            connection_id_bytes[4],
-            connection_id_bytes[5],
-            connection_id_bytes[6],
-            connection_id_bytes[7],
-            connection_id_bytes[8],
-        ]);
-
-        let response = NetworkData::new(connection_id, data, is_closed);
-        Ok(response)
-    }
-
-    /// Serializes the response into bytes so that it can be sent back through
-    /// the mixnet to the requesting application.
-    pub fn into_bytes(self) -> Vec<u8> {
-        std::iter::once(self.is_closed as u8)
-            .chain(self.connection_id.to_be_bytes().iter().cloned())
-            .chain(self.data.into_iter())
-            .collect()
-    }
-}
-
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct ConnectionError {
     pub connection_id: ConnectionId,
@@ -366,57 +298,6 @@ pub enum QueryResponse {
 mod tests {
     use super::*;
 
-    #[cfg(test)]
-    mod constructing_socks5_data_responses_from_bytes {
-        use super::*;
-
-        #[test]
-        fn fails_when_zero_bytes_are_supplied() {
-            let response_bytes = Vec::new();
-
-            let err = NetworkData::try_from_bytes(&response_bytes).unwrap_err();
-            assert!(matches!(err, ResponseDeserializationError::NoData));
-        }
-
-        #[test]
-        fn fails_when_connection_id_bytes_are_too_short() {
-            let response_bytes = vec![0, 1, 2, 3, 4, 5, 6];
-            let err = NetworkData::try_from_bytes(&response_bytes).unwrap_err();
-            assert!(matches!(
-                err,
-                ResponseDeserializationError::ConnectionIdTooShort,
-            ));
-        }
-
-        #[test]
-        fn works_when_there_is_no_data() {
-            let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7];
-            let expected = NetworkData::new(
-                u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
-                Vec::new(),
-                false,
-            );
-            let actual = NetworkData::try_from_bytes(&response_bytes).unwrap();
-            assert_eq!(expected.connection_id, actual.connection_id);
-            assert_eq!(expected.data, actual.data);
-            assert_eq!(expected.is_closed, actual.is_closed);
-        }
-
-        #[test]
-        fn works_when_there_is_data() {
-            let response_bytes = vec![0, 0, 1, 2, 3, 4, 5, 6, 7, 255, 255, 255];
-            let expected = NetworkData::new(
-                u64::from_be_bytes([0, 1, 2, 3, 4, 5, 6, 7]),
-                vec![255, 255, 255],
-                false,
-            );
-            let actual = NetworkData::try_from_bytes(&response_bytes).unwrap();
-            assert_eq!(expected.connection_id, actual.connection_id);
-            assert_eq!(expected.data, actual.data);
-            assert_eq!(expected.is_closed, actual.is_closed);
-        }
-    }
-
     #[cfg(test)]
     mod connection_error_response_serde_tests {
         use super::*;

service-providers/network-requester/Cargo.toml

@@ -43,7 +43,6 @@ nym-bin-common = { path = "../../common/bin-common", features = ["output_format"
 nym-network-defaults = { path = "../../common/network-defaults" }
 nym-sdk = { path = "../../sdk/rust/nym-sdk" }
 nym-sphinx = { path = "../../common/nymsphinx" }
-nym-ordered-buffer = {path = "../../common/socks5/ordered-buffer"}
 nym-socks5-proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
 nym-service-providers-common = { path = "../common" }
 nym-socks5-requests = { path = "../../common/socks5/requests" }

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

@@ -24,7 +24,7 @@ use nym_socks5_proxy_helpers::connection_controller::{
 };
 use nym_socks5_proxy_helpers::proxy_runner::{MixProxyReader, MixProxySender};
 use nym_socks5_requests::{
-    ConnectRequest, ConnectionId, NetworkData, QueryRequest, QueryResponse, SendRequest,
+    ConnectRequest, ConnectionId, QueryRequest, QueryResponse, SendRequest, SocketData,
     Socks5ProtocolVersion, Socks5ProviderRequest, Socks5Request, Socks5RequestContent,
     Socks5Response,
 };
@@ -137,13 +137,13 @@ impl ServiceProvider<Socks5Request> for NRServiceProvider {
                         .connected_services
                         .read()
                         .await
-                        .get(&req.conn_id)
+                        .get(&req.data.header.connection_id)
                     {
                         stats_collector
                             .request_stats_data
                             .write()
                             .await
-                            .processed(remote_addr, req.data.len() as u32);
+                            .processed(remote_addr, req.data.data.len() as u32);
                     }
                 }
                 self.handle_proxy_send(req)
@@ -359,7 +359,7 @@ impl NRServiceProvider {
                     return_address,
                     remote_version,
                     connection_id,
-                    NetworkData::new_closed_empty(connection_id),
+                    SocketData::new(0, connection_id, true, Vec::new()),
                 );
 
                 mix_input_sender
@@ -471,7 +471,9 @@ impl NRServiceProvider {
     }
 
     fn handle_proxy_send(&mut self, req: SendRequest) {
-        self.controller_sender.unbounded_send(req.into()).unwrap()
+        self.controller_sender
+            .unbounded_send(ControllerCommand::new_send(req.data))
+            .unwrap()
     }
 
     fn handle_query(

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

@@ -6,7 +6,7 @@ use nym_service_providers_common::interface::{
     ControlRequest, ControlResponse, ProviderInterfaceVersion, RequestVersion,
 };
 use nym_socks5_requests::{
-    ConnectionId, NetworkData, Socks5ProviderRequest, Socks5ProviderResponse, Socks5Request,
+    ConnectionId, SocketData, Socks5ProviderRequest, Socks5ProviderResponse, Socks5Request,
     Socks5RequestContent, Socks5Response, Socks5ResponseContent,
 };
 use nym_sphinx::addressing::clients::Recipient;
@@ -78,12 +78,12 @@ impl MixnetMessage {
         address: MixnetAddress,
         request_version: RequestVersion<Socks5Request>,
         connection_id: ConnectionId,
-        content: NetworkData,
+        content: SocketData,
     ) -> Self {
         // TODO: simplify by providing better constructor for `PlaceholderResponse`
         let res = Socks5Response::new(
             request_version.provider_protocol,
-            Socks5ResponseContent::NetworkData(content),
+            Socks5ResponseContent::NetworkData { content },
         );
         let msg =
             Socks5ProviderResponse::new_provider_data(request_version.provider_interface, res);
@@ -135,11 +135,15 @@ impl MixnetMessage {
     pub(crate) fn new_network_data_response_content(
         address: MixnetAddress,
         request_version: RequestVersion<Socks5Request>,
+        seq: u64,
         connection_id: ConnectionId,
         data: Vec<u8>,
         closed_socket: bool,
     ) -> Self {
-        let response_content = NetworkData::new(connection_id, data, closed_socket);
+        if seq == 0 && data.is_empty() {
+            println!("new empty response with 0 seq")
+        }
+        let response_content = SocketData::new(seq, connection_id, closed_socket, data);
         Self::new_network_data_response(address, request_version, connection_id, response_content)
     }
 

service-providers/network-requester/src/socks5/tcp.rs

@@ -66,13 +66,14 @@ impl Connection {
             Some(lane_queue_lengths),
             shutdown,
         )
-        .run(move |conn_id, read_data, socket_closed| {
+        .run(move |socket_data| {
             MixnetMessage::new_network_data_response_content(
                 return_address.clone(),
                 remote_version.clone(),
-                conn_id,
-                read_data,
-                socket_closed,
+                socket_data.header.seq,
+                socket_data.header.connection_id,
+                socket_data.data,
+                socket_data.header.local_socket_closed,
             )
         })
         .await

service-providers/network-requester/src/statistics/collector.rs

@@ -6,10 +6,11 @@ use crate::core::new_legacy_request_version;
 use crate::reply::MixnetMessage;
 use async_trait::async_trait;
 use log::*;
-use nym_ordered_buffer::OrderedMessageSender;
 use nym_service_providers_common::interface::RequestVersion;
 use nym_socks5_proxy_helpers::proxy_runner::MixProxySender;
-use nym_socks5_requests::{ConnectionId, RemoteAddress, Socks5Request, Socks5RequestContent};
+use nym_socks5_requests::{
+    ConnectionId, RemoteAddress, SocketData, Socks5Request, Socks5RequestContent,
+};
 use nym_sphinx::addressing::clients::Recipient;
 use nym_statistics_common::api::{
     build_statistics_request_bytes, DEFAULT_STATISTICS_SERVICE_ADDRESS,
@@ -190,9 +191,9 @@ impl StatisticsCollector for ServiceStatisticsCollector {
             .expect("MixProxyReader has stopped receiving!");
 
         trace!("Sending data to statistics service");
-        let mut message_sender = OrderedMessageSender::new();
-        let ordered_msg = message_sender.wrap_message(msg).into_bytes();
-        let send_req = Socks5RequestContent::new_send(conn_id, ordered_msg, true);
+
+        let message = SocketData::new(0, conn_id, true, msg);
+        let send_req = Socks5RequestContent::new_send(message);
 
         let mixnet_message = MixnetMessage::new_network_data_request(
             self.stats_provider_addr,