Merge pull request #3546 from twofaktor/patch-2

[UPDATE DOCS] socks5-client

.github/workflows/build-and-upload-binaries-ci.yml

@@ -63,7 +63,7 @@ jobs:
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: stable
+          toolchain: 1.69.0
 
       - name: Build all binaries
         uses: actions-rs/cargo@v1
@@ -74,7 +74,7 @@ jobs:
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: stable
+          toolchain: 1.69.0
           target: wasm32-unknown-unknown
           override: true
           components: rustfmt, clippy

CHANGELOG.md

@@ -4,6 +4,24 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ## [Unreleased]
 
+## [v1.1.21] (2023-06-13)
+
+- mixFetch: Change socks5 `SendRequest` to include OrderedMessage index as a field rather than making it serialized inside the `data` field
+ ([#3534])
+- Explorer - add more data columns to the Service Provider section: ([#3474])
+- network-requester: support report if they run an open proxy using `ControlRequest` API ([#3461])
+- Refactor client configs (London discussion) ([#3444])
+- Increase `DEFAULT_MAXIMUM_CONNECTION_BUFFER_SIZE` to 2000 to improve reliability ([#3433])
+- socks5: sender waits for lanes to clear even though the connection is closed ([#3366])
+- version bump for variables ([#3545])
+
+[#3534]: https://github.com/nymtech/nym/issues/3534
+[#3474]: https://github.com/nymtech/nym/issues/3474
+[#3461]: https://github.com/nymtech/nym/issues/3461
+[#3444]: https://github.com/nymtech/nym/issues/3444
+[#3433]: https://github.com/nymtech/nym/issues/3433
+[#3366]: https://github.com/nymtech/nym/issues/3366
+[#3545]: https://github.com/nymtech/nym/pull/3545
 
 ## [v1.1.20] (2023-06-06)
 

Cargo.lock

@@ -1655,7 +1655,7 @@ checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0"
 
 [[package]]
 name = "explorer-api"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "chrono",
  "clap 4.2.7",
@@ -3199,7 +3199,7 @@ dependencies = [
 
 [[package]]
 name = "nym-api"
-version = "1.1.21"
+version = "1.1.22"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3333,7 +3333,7 @@ dependencies = [
 
 [[package]]
 name = "nym-cli"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "anyhow",
  "base64 0.13.1",
@@ -3397,7 +3397,7 @@ dependencies = [
 
 [[package]]
 name = "nym-client"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "clap 4.2.7",
  "dirs",
@@ -3668,7 +3668,7 @@ dependencies = [
 
 [[package]]
 name = "nym-gateway"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3819,7 +3819,7 @@ dependencies = [
 
 [[package]]
 name = "nym-mixnode"
-version = "1.1.21"
+version = "1.1.22"
 dependencies = [
  "anyhow",
  "bs58",
@@ -3929,7 +3929,7 @@ dependencies = [
 
 [[package]]
 name = "nym-network-requester"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "anyhow",
  "async-file-watcher",
@@ -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",
@@ -3975,7 +3974,7 @@ dependencies = [
 
 [[package]]
 name = "nym-network-statistics"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "dirs",
  "log",
@@ -4118,7 +4117,7 @@ dependencies = [
 
 [[package]]
 name = "nym-socks5-client"
-version = "1.1.20"
+version = "1.1.21"
 dependencies = [
  "clap 4.2.7",
  "lazy_static",

clients/native/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-client"
-version = "1.1.20"
+version = "1.1.21"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 description = "Implementation of the Nym Client"
 edition = "2021"

clients/socks5/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-socks5-client"
-version = "1.1.20"
+version = "1.1.21"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 description = "A SOCKS5 localhost proxy that converts incoming messages to Sphinx and sends them to a Nym address"
 edition = "2021"

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::*;

documentation/dev-portal/book.toml

@@ -50,7 +50,7 @@ assets_version = "2.0.0" # do not edit: managed by `mdbook-admonish install`
 # code prerequisites versions
 minimum_rust_version = "1.66"
 # TODO remove this in place of develop in next release
-platform_release_version = "v1.1.20"
+platform_release_version = "v1.1.21"
 
 [preprocessor.last-changed]
 command = "mdbook-last-changed"

documentation/docs/book.toml

@@ -50,7 +50,7 @@ assets_version = "2.0.0" # do not edit: managed by `mdbook-admonish install`
 minimum_rust_version = "1.66"
 
 # old variables - this is still needed for links.. TODO change to develop?
-platform_release_version = "v1.1.20"
+platform_release_version = "v1.1.21"
 
 [preprocessor.last-changed]
 command = "mdbook-last-changed"

documentation/docs/src/clients/socks5-client.md

@@ -8,6 +8,7 @@
 ```
 
 ## What is this client for?
+
 Many existing applications are able to use either the SOCKS4, SOCKS4A, or SOCKS5 proxy protocols. If you want to send such an application's traffic through the mixnet, you can use the `nym-socks5-client` to bounce network traffic through the Nym network, like this:
 
 ```
@@ -65,6 +66,7 @@ The `nym-socks5-client` allows you to do the following from your local machine:
 The `nym-network-requester` then reassembles the original TCP stream using the packets' sequence numbers, and make the intended request. It will then chop up the response into Sphinx packets and send them back through the mixnet to your  `nym-socks5-client`. The application will then receive its data, without even noticing that it wasn't talking to a "normal" SOCKS5 proxy!
 
 ## Client setup
+
 ### Viewing command help
 
 You can check that your binaries are properly compiled with:
@@ -86,6 +88,7 @@ You can check the necessary parameters for the available commands by running:
 ```
 
 ### Initialising a new client instance
+
 Before you can use the client, you need to initalise a new instance of it, which can be done with the following command:
 
 ```
@@ -105,6 +108,7 @@ The `--provider` field needs to be filled with the Nym address of a Network Requ
 Since the nodes on this list are the infrastructure for [Nymconnect](https://nymtech.net/developers/quickstart/nymconnect-gui.html) they will support all apps on the [default whitelist](../nodes/network-requester-setup.md#network-requester-whitelist): Keybase, Telegram, Electrum, Blockstream Green, and Helios.
 
 #### Choosing a Gateway
+
 By default - as in the example above - your client will choose a random gateway to connect to.
 
 However, there are several options for choosing a gateway, if you do not want one that is randomly assigned to your client:
@@ -117,6 +121,7 @@ However, there are several options for choosing a gateway, if you do not want on
 > Note this doesn't mean that your client will pick the closest gateway to you, but it will be far more likely to connect to gateway with a 20ms ping rather than 200ms
 
 ### Configuring your client
+
 When you initalise a client instance, a configuration directory will be generated and stored in `$HOME_DIR/.nym/socks5-clients/<client-name>/`.
 
 ```
@@ -140,14 +145,13 @@ The `config.toml` file contains client configuration options, while the two `pem
 The generated files contain the client name, public/private keypairs, and gateway address. The name `<client_id>` in the example above is just a local identifier so that you can name your clients.
 
 #### Configuring your client for Docker
+
 By default, the native client listens to host `127.0.0.1`. However this can be an issue if you wish to run a client in a Dockerized environment, where it can be convenenient to listen on a different host such as `0.0.0.0`.
 
 You can set this via the `--host` flag during either the `init` or `run` commands.
 
 Alternatively, a custom host can be set in the `config.toml` file under the `socket` section. If you do this, remember to restart your client process.
 
-
-
 ### Running the socks5 client
 
 You can run the initalised client by doing this:
@@ -156,6 +160,38 @@ You can run the initalised client by doing this:
 ./nym-socks5-client run --id docs-example
 ```
 
+## Automating your socks5 client with systemd
+
+Stop the running process with `CTRL-C`, and create a service file for the socks5 client as we did with our client instance previously at `/etc/systemd/system/nym-socks5-client.service`:
+
+```ini
+[Unit]
+Description=Nym Socks5 Client ({{platform_release_version}})
+StartLimitInterval=350
+StartLimitBurst=10
+
+[Service]
+User=nym # replace this with whatever user you wish
+LimitNOFILE=65536
+ExecStart=/home/nym/nym-socks5-client run --id <your_id>
+KillSignal=SIGINT
+Restart=on-failure
+RestartSec=30
+
+[Install]
+WantedBy=multi-user.target
+```
+
+Now enable and start your socks5 client:
+
+```
+systemctl enable nym-socks5-client.service
+systemctl start nym-socks5-client.service
+
+# you can always check your socks5 client has succesfully started with:
+systemctl status nym-socks5-client.service
+```
+
 ## Using your Socks5 Client
 
 After completing the steps above, your local Socks5 Client will be listening on `localhost:1080` ready to proxy traffic to the Network Requester set as the `--provider` when initialising.

explorer-api/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "explorer-api"
-version = "1.1.20"
+version = "1.1.21"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

explorer-api/src/service_providers/http.rs

@@ -1,26 +1,93 @@
-use crate::service_providers::models::DirectoryService;
+use crate::service_providers::models::{
+    DirectoryService, DirectorySpDetailed, HarbourMasterService, PagedResult,
+};
 use okapi::openapi3::OpenApi;
-use reqwest::Error as ReqwestError;
-use rocket::{serde::json::Json, Route};
+use reqwest::{Client, Error as ReqwestError};
+use rocket::{http::Status, serde::json::Json, Route};
 use rocket_okapi::settings::OpenApiSettings;
 
-static SERVICE_PROVIDER_WELLKNOWN_URL: &str =
+const SERVICE_PROVIDER_WELLKNOWN_URL: &str =
     "https://nymtech.net/.wellknown/connect/service-providers.json";
 
+const HARBOUR_MASTER_URL: &str = "https://harbourmaster.nymtech.net/v1/services";
+const HM_SINCE_MIN: u32 = 120;
+const HM_SIZE: u8 = 100;
+
+#[derive(Debug)]
+pub enum GetSpError {
+    ReqwestError(ReqwestError),
+    Error(String),
+}
+
+impl From<ReqwestError> for GetSpError {
+    fn from(error: ReqwestError) -> Self {
+        GetSpError::ReqwestError(error)
+    }
+}
+
+impl From<&str> for GetSpError {
+    fn from(error: &str) -> Self {
+        GetSpError::Error(String::from(error))
+    }
+}
+
 pub fn service_providers_make_default_routes(settings: &OpenApiSettings) -> (Vec<Route>, OpenApi) {
     openapi_get_routes_spec![settings: get_service_providers]
 }
 
-pub async fn get_services() -> Result<Vec<DirectoryService>, ReqwestError> {
-    reqwest::get(SERVICE_PROVIDER_WELLKNOWN_URL)
+pub async fn get_services() -> Result<Vec<DirectorySpDetailed>, GetSpError> {
+    let reqw = Client::new();
+
+    let services_res = reqw
+        .get(SERVICE_PROVIDER_WELLKNOWN_URL)
+        .send()
         .await?
         .json::<Vec<DirectoryService>>()
-        .await
+        .await?;
+
+    let directory_sp = services_res
+        .iter()
+        .find(|item| item.id == "all")
+        .ok_or("NymConnect network requesters data not found in response")?;
+
+    let hm_services = reqw
+        .get(format!(
+            "{HARBOUR_MASTER_URL}?since_min={HM_SINCE_MIN}&size={HM_SIZE}"
+        ))
+        .send()
+        .await?
+        .json::<PagedResult<HarbourMasterService>>()
+        .await?;
+
+    let sp_list: Vec<_> = directory_sp
+        .items
+        .iter()
+        .map(|sp| {
+            let directory_sp = hm_services
+                .items
+                .iter()
+                .find(|item| item.service_provider_client_id == sp.address);
+            DirectorySpDetailed {
+                id: sp.id.clone(),
+                description: sp.description.clone(),
+                address: sp.address.clone(),
+                routing_score: directory_sp.map(|sp| sp.routing_score),
+                service_type: "Network requester".into(),
+            }
+        })
+        .collect();
+
+    Ok(sp_list)
 }
 
 #[openapi(tag = "service_providers")]
 #[get("/")]
-pub(crate) async fn get_service_providers() -> Json<Vec<DirectoryService>> {
-    let result = get_services().await.unwrap();
-    Json(result)
+pub(crate) async fn get_service_providers() -> Result<Json<Vec<DirectorySpDetailed>>, Status> {
+    match get_services().await {
+        Ok(res) => Ok(Json(res)),
+        Err(err) => {
+            log::error!("{:?}", err);
+            Err(Status::InternalServerError)
+        }
+    }
 }

explorer-api/src/service_providers/models.rs

@@ -8,9 +8,36 @@ pub struct DirectoryServiceProvider {
     pub gateway: String,
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
+pub struct DirectorySpDetailed {
+    pub id: String,
+    pub description: String,
+    pub address: String,
+    pub routing_score: Option<f32>,
+    pub service_type: String,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
 pub struct DirectoryService {
     pub id: String,
     pub description: String,
     pub items: Vec<DirectoryServiceProvider>,
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HarbourMasterService {
+    pub service_provider_client_id: String,
+    pub gateway_identity_key: String,
+    pub ip_address: String,
+    pub last_successful_ping_utc: String,
+    pub last_updated_utc: String,
+    pub routing_score: f32,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct PagedResult<T> {
+    pub page: u32,
+    pub size: u32,
+    pub total: i32,
+    pub items: Vec<T>,
+}

explorer/src/api/index.ts

@@ -31,7 +31,7 @@ import {
   Environment,
   GatewayBondAnnotated,
   GatewayBond,
-  DirectoryService,
+  DirectoryServiceProvider,
 } from '../typeDefs/explorer-api';
 
 function getFromCache(key: string) {
@@ -153,7 +153,7 @@ export class Api {
   static fetchUptimeStoryById = async (id: string): Promise<UptimeStoryResponse> =>
     (await fetch(`${UPTIME_STORY_API}/${id}/history`)).json();
 
-  static fetchServiceProviders = async (): Promise<DirectoryService[]> => {
+  static fetchServiceProviders = async (): Promise<DirectoryServiceProvider[]> => {
     const res = await fetch(SERVICE_PROVIDERS);
     const json = await res.json();
     return json;

explorer/src/components/Universal-DataGrid.tsx

@@ -4,6 +4,7 @@ import { DataGrid, GridColDef, useGridApiContext, useGridState } from '@mui/x-da
 import Pagination from '@mui/material/Pagination';
 import { SxProps } from '@mui/system';
 import { LinearProgress } from '@mui/material';
+import { GridInitialStateCommunity } from '@mui/x-data-grid/models/gridStateCommunity';
 
 const useStyles = makeStyles({
   root: {
@@ -49,46 +50,55 @@ type DataGridProps = {
   pageSize?: string | undefined;
   rows: any;
   loading?: boolean;
+  initialState?: GridInitialStateCommunity;
 };
-export const UniversalDataGrid: FCWithChildren<DataGridProps> = ({ rows, columns, loading, pagination, pageSize }) => {
+export const UniversalDataGrid: FCWithChildren<DataGridProps> = ({
+  rows,
+  columns,
+  loading,
+  pagination,
+  pageSize,
+  initialState,
+}) => {
   if (loading) return <LinearProgress />;
-  if (!loading)
-    return (
-      <DataGrid
-        pagination={pagination}
-        rows={rows}
-        components={{
-          Pagination: CustomPagination,
-        }}
-        columns={columns}
-        pageSize={Number(pageSize)}
-        disableSelectionOnClick
-        autoHeight
-        hideFooter={!pagination}
-        style={{
-          width: '100%',
-          border: 'none',
-        }}
-        sx={{
-          '*::-webkit-scrollbar': {
-            width: '1em',
-          },
-          '*::-webkit-scrollbar-track': {
-            background: (t) => t.palette.nym.networkExplorer.scroll.backgroud,
-            outline: (t) => `1px solid ${t.palette.nym.networkExplorer.scroll.border}`,
-            boxShadow: 'auto',
-            borderRadius: 'auto',
-          },
-          '*::-webkit-scrollbar-thumb': {
-            backgroundColor: (t) => t.palette.nym.networkExplorer.scroll.color,
-            borderRadius: '20px',
-            width: '.4em',
-            border: (t) => `3px solid ${t.palette.nym.networkExplorer.scroll.backgroud}`,
-            shadow: 'auto',
-          },
-        }}
-      />
-    );
+
+  return (
+    <DataGrid
+      pagination={pagination}
+      rows={rows}
+      components={{
+        Pagination: CustomPagination,
+      }}
+      columns={columns}
+      pageSize={Number(pageSize)}
+      disableSelectionOnClick
+      autoHeight
+      hideFooter={!pagination}
+      initialState={initialState}
+      style={{
+        width: '100%',
+        border: 'none',
+      }}
+      sx={{
+        '*::-webkit-scrollbar': {
+          width: '1em',
+        },
+        '*::-webkit-scrollbar-track': {
+          background: (t) => t.palette.nym.networkExplorer.scroll.backgroud,
+          outline: (t) => `1px solid ${t.palette.nym.networkExplorer.scroll.border}`,
+          boxShadow: 'auto',
+          borderRadius: 'auto',
+        },
+        '*::-webkit-scrollbar-thumb': {
+          backgroundColor: (t) => t.palette.nym.networkExplorer.scroll.color,
+          borderRadius: '20px',
+          width: '.4em',
+          border: (t) => `3px solid ${t.palette.nym.networkExplorer.scroll.backgroud}`,
+          shadow: 'auto',
+        },
+      }}
+    />
+  );
   return null;
 };
 

explorer/src/context/main.tsx

@@ -4,17 +4,18 @@ import {
   ApiState,
   BlockResponse,
   CountryDataResponse,
-  DirectoryService,
   GatewayResponse,
   MixNodeResponse,
   MixnodeStatus,
   SummaryOverviewResponse,
   ValidatorsResponse,
   Environment,
+  DirectoryServiceProvider,
 } from '../typeDefs/explorer-api';
 import { EnumFilterKey } from '../typeDefs/filters';
 import { Api, getEnvironment } from '../api';
 import { NavOptionType, originalNavOptions } from './nav';
+import { toPercentIntegerString } from '../utils';
 
 interface StateData {
   summaryOverview?: ApiState<SummaryOverviewResponse>;
@@ -27,7 +28,7 @@ interface StateData {
   navState: NavOptionType[];
   validators?: ApiState<ValidatorsResponse>;
   environment?: Environment;
-  serviceProviders?: ApiState<DirectoryService>;
+  serviceProviders?: ApiState<DirectoryServiceProvider[]>;
 }
 
 interface StateApi {
@@ -70,7 +71,7 @@ export const MainContextProvider: FCWithChildren = ({ children }) => {
   const [validators, setValidators] = React.useState<ApiState<ValidatorsResponse>>();
   const [block, setBlock] = React.useState<ApiState<BlockResponse>>();
   const [countryData, setCountryData] = React.useState<ApiState<CountryDataResponse>>();
-  const [serviceProviders, setServiceProviders] = React.useState<ApiState<DirectoryService>>();
+  const [serviceProviders, setServiceProviders] = React.useState<ApiState<DirectoryServiceProvider[]>>();
 
   const toggleMode = () => setMode((m) => (m !== 'light' ? 'light' : 'dark'));
 
@@ -168,8 +169,14 @@ export const MainContextProvider: FCWithChildren = ({ children }) => {
   const fetchServiceProviders = async () => {
     setServiceProviders({ data: undefined, isLoading: true });
     try {
-      const [res] = await Api.fetchServiceProviders();
-      setServiceProviders({ data: res, isLoading: false });
+      const res = await Api.fetchServiceProviders();
+      const resWithRoutingScorePercentage = res.map((item) => ({
+        ...item,
+        routing_score: item.routing_score
+          ? `${toPercentIntegerString(item.routing_score.toString())}%`
+          : item.routing_score,
+      }));
+      setServiceProviders({ data: resWithRoutingScorePercentage, isLoading: false });
     } catch (error) {
       setServiceProviders({
         error: error instanceof Error ? error : new Error('Service provider api fail'),

explorer/src/pages/Overview/index.tsx

@@ -78,7 +78,7 @@ export const PageOverview: FCWithChildren = () => {
                   onClick={() => navigate('/network-components/service-providers')}
                   title="Service providers"
                   icon={<PeopleAlt />}
-                  count={serviceProviders.data?.items.length}
+                  count={serviceProviders.data?.length}
                   errorMsg={summaryOverview?.error}
                 />
               </Grid>

explorer/src/pages/ServiceProviders/index.tsx

@@ -1,17 +1,50 @@
 import React from 'react';
 import { Button, Card, FormControl, Grid, ListItem, Menu, SelectChangeEvent, Typography } from '@mui/material';
+import { GridColDef, GridRenderCellParams } from '@mui/x-data-grid';
 import { TableToolbar } from '../../components/TableToolbar';
 import { Title } from '../../components/Title';
 import { UniversalDataGrid } from '../../components/Universal-DataGrid';
 import { useMainContext } from '../../context/main';
+import { CustomColumnHeading } from '../../components/CustomColumnHeading';
 
-const columns = [
+const columns: GridColDef[] = [
   {
     headerName: 'Client ID',
     field: 'address',
     disableColumnMenu: true,
+    flex: 3,
+  },
+  {
+    headerName: 'Type',
+    field: 'service_type',
+    disableColumnMenu: true,
     flex: 1,
   },
+  {
+    headerName: 'Routing score',
+    field: 'routing_score',
+    disableColumnMenu: true,
+    flex: 2,
+    sortingOrder: ['asc', 'desc'],
+    sortComparator: (a?: string, b?: string) => {
+      if (!a) return -1; // Place undefined values at the end
+      if (!b) return 1; // Place undefined values at the end
+
+      const aToNum = parseInt(a, 10);
+      const bToNum = parseInt(b, 10);
+
+      if (aToNum > bToNum) return 1;
+
+      return -1; // Sort numbers in ascending order
+    },
+    renderCell: (params: GridRenderCellParams) => (!params.value ? '-' : params.value),
+    renderHeader: () => (
+      <CustomColumnHeading
+        headingTitle="Routing score"
+        tooltipInfo="Routing score is only displayed for the service providers that had a successful ping within the last two hours"
+      />
+    ),
+  },
 ];
 
 const SupportedApps = () => {
@@ -37,7 +70,7 @@ const SupportedApps = () => {
       >
         Supported Apps
       </Button>
-      <Menu id="basic-menu" anchorEl={anchorEl} open={open} onClose={handleClose}>
+      <Menu anchorEl={anchorEl} open={open} onClose={handleClose}>
         <ListItem>Keybase</ListItem>
         <ListItem>Telegram</ListItem>
         <ListItem>Electrum</ListItem>
@@ -74,9 +107,19 @@ export const ServiceProviders = () => {
                 />
                 <UniversalDataGrid
                   pagination
-                  rows={serviceProviders?.data?.items}
+                  rows={serviceProviders.data}
                   columns={columns}
                   pageSize={pageSize}
+                  initialState={{
+                    sorting: {
+                      sortModel: [
+                        {
+                          field: 'routing_score',
+                          sort: 'desc',
+                        },
+                      ],
+                    },
+                  }}
                 />
               </>
             ) : (

explorer/src/typeDefs/explorer-api.ts

@@ -241,11 +241,15 @@ export type MixNodeEconomicDynamicsStatsResponse = {
 
 export type Environment = 'mainnet' | 'sandbox' | 'qa';
 
+export type ServiceProviderType = 'Network Requester';
+
 export type DirectoryServiceProvider = {
   id: string;
   description: string;
   address: string;
   gateway: string;
+  routing_score: string | null;
+  service_type: ServiceProviderType;
 };
 
 export type DirectoryService = {

gateway/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-gateway"
-version = "1.1.20"
+version = "1.1.21"
 authors = [
     "Dave Hrycyszyn <futurechimp@users.noreply.github.com>",
     "Jędrzej Stuczyński <andrew@nymtech.net>",

mixnode/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-mixnode"
-version = "1.1.21"
+version = "1.1.22"
 authors = [
     "Dave Hrycyszyn <futurechimp@users.noreply.github.com>",
     "Jędrzej Stuczyński <andrew@nymtech.net>",

nym-api/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-api"
-version = "1.1.21"
+version = "1.1.22"
 authors = [
     "Dave Hrycyszyn <futurechimp@users.noreply.github.com>",
     "Jędrzej Stuczyński <andrew@nymtech.net>",

service-providers/network-requester/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-network-requester"
-version = "1.1.20"
+version = "1.1.21"
 authors.workspace = true
 edition.workspace = true
 rust-version = "1.65"
@@ -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,

service-providers/network-statistics/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-network-statistics"
-version = "1.1.20"
+version = "1.1.21"
 edition = "2021"
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

tools/nym-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-cli"
-version = "1.1.20"
+version = "1.1.21"
 authors.workspace = true
 edition = "2021"
 

yarn.lock

@@ -4210,101 +4210,51 @@
   resolved "https://registry.yarnpkg.com/@tauri-apps/api/-/api-1.3.0.tgz#d0c853ab2cc7506bd826c5f7f260c67c7c15def5"
   integrity sha512-AH+3FonkKZNtfRtGrObY38PrzEj4d+1emCbwNGu0V2ENbXjlLHMZQlUh+Bhu/CRmjaIwZMGJ3yFvWaZZgTHoog==
 
-"@tauri-apps/api@^2.0.0-alpha.0":
-  version "2.0.0-alpha.3"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/api/-/api-2.0.0-alpha.3.tgz#7f467c188305944d67b5b6e82be0041b8b4dbbc9"
-  integrity sha512-F6seMDlcaxeCPy4gS0zJdp6Tet+0rd1qJi/fbKrOrhLM6Y5UtkiG1aSDnMPi+1udThSfadjhUwrLHINvfMCjzQ==
-
 "@tauri-apps/cli-darwin-arm64@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-darwin-arm64/-/cli-darwin-arm64-1.3.0.tgz#6c738af14f494546e0fbe64f6befa0eb08959150"
   integrity sha512-uuhx3/LaqFyHkoGOnOltBLKWGOzC6WzdXu+/Qv3NmNnyQWkY7O34z5V0oP6ibfuiOBZufKjOuBR+8YAIR8Qh9Q==
 
-"@tauri-apps/cli-darwin-arm64@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-darwin-arm64/-/cli-darwin-arm64-2.0.0-alpha.8.tgz#399bec52290ea85f74470bc329376142e17e9878"
-  integrity sha512-ZF9nkkYCDiAEKZFwjEbuqTcFVp+DBgem3edKjsZDYPQpWg0VcZOSYr0o3/RPC81T1/FAy1lq478mkcMe0efvEw==
-
 "@tauri-apps/cli-darwin-x64@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-darwin-x64/-/cli-darwin-x64-1.3.0.tgz#d3fda368b555663723337a70abe4b470592d9bda"
   integrity sha512-fj0VXHMDvb/H1CjaS/JoYd7xcourxndJn1IyM4afYbpXibT/fpmM6uZflDI6rRa220NfnBtQvy+asgwC9wuyLA==
 
-"@tauri-apps/cli-darwin-x64@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-darwin-x64/-/cli-darwin-x64-2.0.0-alpha.8.tgz#4935823acf880423f6e2c70097ab6584079321b3"
-  integrity sha512-N5V+tbP3qeAoXrrTZXvaLIeEWKCq11tqXoNFTkIZNGNC5yQdNpZSX0LqFqzmxVR1SHiOymebvcUlx+ADIpuXGw==
-
 "@tauri-apps/cli-linux-arm-gnueabihf@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm-gnueabihf/-/cli-linux-arm-gnueabihf-1.3.0.tgz#165ed5a299a944a8c0fc1ed5fc0d609db78a142b"
   integrity sha512-f80DmFPnH5ZskG61KIlAyMVk9YkrTq0XM2uiQjOo5gToIdJidSwhPQVeBLv+7UxhqaRBx082Dg2fOkWlO3LiOQ==
 
-"@tauri-apps/cli-linux-arm-gnueabihf@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm-gnueabihf/-/cli-linux-arm-gnueabihf-2.0.0-alpha.8.tgz#5b46f8f38be9b0c6fe8c343e7e589c958a225775"
-  integrity sha512-7LHfZA99ncMDUO/wCGMtrqmDHt1uEKZiNmuzPljWLUwVvbAn0pNWNygnvPNVLEOyav5NnZSGPqT1Zmn+L6Fpyg==
-
 "@tauri-apps/cli-linux-arm64-gnu@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm64-gnu/-/cli-linux-arm64-gnu-1.3.0.tgz#a68f2efc78e03d1ae59804e61ad9ee6bd78b08c0"
   integrity sha512-s6/OByuGoppoUSnOXv/b6Oe6cVFk2w/KHs19aJJpo9ov/dUAA1w9wXlXu2l6sOFGsu/plaVomF2cw3iAQmaUCQ==
 
-"@tauri-apps/cli-linux-arm64-gnu@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm64-gnu/-/cli-linux-arm64-gnu-2.0.0-alpha.8.tgz#56bfa1ec613bfff0aaecd8d38aaab470ae117d58"
-  integrity sha512-imq2MdhWdREvL2sqbU26mzH9sgSvfNWP0uvCPvPxUhK157xqdtGw+Gqm00hwnhTuT5bOFlsUNfnG2U19k1qMpA==
-
 "@tauri-apps/cli-linux-arm64-musl@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm64-musl/-/cli-linux-arm64-musl-1.3.0.tgz#eff2221408d1040d6f4cbf98a851c16e9eeb11e5"
   integrity sha512-yZfZAW4BG92cynL/D4wdrwBAl2oekRwiZnU5CM8k5yncalVEL0tyzuxQjjqbqrtDcw0rdkoBWrhFd+EB89vQaQ==
 
-"@tauri-apps/cli-linux-arm64-musl@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-arm64-musl/-/cli-linux-arm64-musl-2.0.0-alpha.8.tgz#6a0e2a3f1a494481ce391d379284f688c59ba2aa"
-  integrity sha512-7hXEyvCosBHIN6ahkbFOI5JoyWZAulc0sYd3hWh9V/MBfU+LlPiapsJi6fdde0zew5nnzwcCtfEKkoR737tAig==
-
 "@tauri-apps/cli-linux-x64-gnu@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-x64-gnu/-/cli-linux-x64-gnu-1.3.0.tgz#2faaea4272278837df09edae90a9907c4dfe99c4"
   integrity sha512-K3KRWSGKh7DTBr/ZKgWzeNX1Vdgx1ZBlUJXsm72R0Hb+93fDEp3TWgiwVkxqecB4aNWJhJsDcvRHuxw1G8xPlA==
 
-"@tauri-apps/cli-linux-x64-gnu@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-x64-gnu/-/cli-linux-x64-gnu-2.0.0-alpha.8.tgz#0ae1f0c19ab8fb9878161db4d40c3cbd581e37f3"
-  integrity sha512-S9T/trKpXcLc7hVIv7xFrRBlaivHD/HLUz0nYAkI2izNGFmbP3QpkqQDdwpWN/fxneodNgI2/mDFC3NULClj+A==
-
 "@tauri-apps/cli-linux-x64-musl@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-x64-musl/-/cli-linux-x64-musl-1.3.0.tgz#3f678ab7dc441da5582909230934204e23f85f6b"
   integrity sha512-99bVHqL1EtF7oESrlmEb5BWJsMNQ2ha70gesZhaVO2qI9Vg089XvrFZWC+aGiUsXNFrOw270+D9DKn03xO5+Zg==
 
-"@tauri-apps/cli-linux-x64-musl@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-linux-x64-musl/-/cli-linux-x64-musl-2.0.0-alpha.8.tgz#dbb668569ee71f041b38b05470697cc0ba761ff3"
-  integrity sha512-90mx6vSFNBbctaPaKhkH+um51gOoRJwLFOadOkklHS5Q6S6GjtSa1lmBEFyKUTAfFPtmiacuNYtoKx7nqm+z1Q==
-
 "@tauri-apps/cli-win32-ia32-msvc@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-win32-ia32-msvc/-/cli-win32-ia32-msvc-1.3.0.tgz#eeaaa3e59cb1f26f5e7d7ce5aa101d5821a5355b"
   integrity sha512-ckBUTqXXdnCiYyf2xvxiuqiKZurg7ET++f6yzfvYa+gofd5dagQJkGLlkIg2pJ2c8mhEG1Cfk1vxWPqqGfN2GQ==
 
-"@tauri-apps/cli-win32-ia32-msvc@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-win32-ia32-msvc/-/cli-win32-ia32-msvc-2.0.0-alpha.8.tgz#f87db96860ca883efa7a7d96095583cb69a4dc5f"
-  integrity sha512-VAMsLJYfp6iVI7oJ+uIkfe8DKPRMtWDiSEkfFqvDyFX0WMTQl23B0AzYyapVwZc+WkTkLuoMLpIWMQCgAoQWfQ==
-
 "@tauri-apps/cli-win32-x64-msvc@1.3.0":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli-win32-x64-msvc/-/cli-win32-x64-msvc-1.3.0.tgz#8d67ab0e9bd3ae9e3a9b6e54845e0ca34ab1526c"
   integrity sha512-vQ4wqRE0aziyRxgHIOLudGuxx4wETvFnmMvDBaNJRRrZQPlkOKnRxrvj1rNnI1845BdzSbDF4p7JDcFzToAfXA==
 
-"@tauri-apps/cli-win32-x64-msvc@2.0.0-alpha.8":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli-win32-x64-msvc/-/cli-win32-x64-msvc-2.0.0-alpha.8.tgz#d438b0231aaac35223d666d207b50c2794fc4ea6"
-  integrity sha512-zvWd13hRfRM0AEJJZ4t4CeB/cyru8hvbB6c+sxYDS9GPRWfHSH5dIeKoHhnMwP5fEOPZLN7VaeEP6tC88trD6g==
-
 "@tauri-apps/cli@^1.0.5", "@tauri-apps/cli@^1.2.2":
   version "1.3.0"
   resolved "https://registry.yarnpkg.com/@tauri-apps/cli/-/cli-1.3.0.tgz#587b7d6ed34dab04f56f29925626ba1a248e298c"
@@ -4320,21 +4270,6 @@
     "@tauri-apps/cli-win32-ia32-msvc" "1.3.0"
     "@tauri-apps/cli-win32-x64-msvc" "1.3.0"
 
-"@tauri-apps/cli@^2.0.0-alpha.2":
-  version "2.0.0-alpha.8"
-  resolved "https://registry.yarnpkg.com/@tauri-apps/cli/-/cli-2.0.0-alpha.8.tgz#84772a16889b38dcd6239fe159bb5a5f2b1e1cc9"
-  integrity sha512-cXt6pxh7oiV8Htz7eTPor7if4aN9f9emn10+5h2Y82YzST7I7wKXsrjuk0HIyzUiqiQjUgl3iT9gh791zgtI3w==
-  optionalDependencies:
-    "@tauri-apps/cli-darwin-arm64" "2.0.0-alpha.8"
-    "@tauri-apps/cli-darwin-x64" "2.0.0-alpha.8"
-    "@tauri-apps/cli-linux-arm-gnueabihf" "2.0.0-alpha.8"
-    "@tauri-apps/cli-linux-arm64-gnu" "2.0.0-alpha.8"
-    "@tauri-apps/cli-linux-arm64-musl" "2.0.0-alpha.8"
-    "@tauri-apps/cli-linux-x64-gnu" "2.0.0-alpha.8"
-    "@tauri-apps/cli-linux-x64-musl" "2.0.0-alpha.8"
-    "@tauri-apps/cli-win32-ia32-msvc" "2.0.0-alpha.8"
-    "@tauri-apps/cli-win32-x64-msvc" "2.0.0-alpha.8"
-
 "@tauri-apps/tauri-forage@^1.0.0-beta.2":
   version "1.0.0-beta.2"
   resolved "https://registry.yarnpkg.com/@tauri-apps/tauri-forage/-/tauri-forage-1.0.0-beta.2.tgz#c11b438ba32731dc7b4c8689d9fda4130baadc07"