Max/sdk streamer

Cargo.toml

@@ -115,7 +115,6 @@ members = [
     "common/wireguard-private-metadata/tests",
     "common/wireguard-types",
     "common/zulip-client",
-    "documentation/autodoc",
     "gateway",
     "nym-api",
     "nym-api/nym-api-requests",
@@ -124,7 +123,6 @@ members = [
     "nym-credential-proxy/nym-credential-proxy",
     "nym-credential-proxy/nym-credential-proxy-requests",
     "nym-credential-proxy/vpn-api-lib-wasm",
-    "nym-ip-packet-client",
     "nym-network-monitor",
     "nym-node",
     "nym-node-status-api/nym-node-status-agent",
@@ -142,6 +140,8 @@ members = [
     "sdk/ffi/go",
     "sdk/ffi/shared",
     "sdk/rust/nym-sdk",
+    "mixtcp",
+    # "service-providers/authenticator",
     "service-providers/common",
     "service-providers/ip-packet-router",
     "service-providers/network-requester",
@@ -154,6 +154,7 @@ members = [
     "tools/internal/ssl-inject",
     "tools/internal/testnet-manager",
     "tools/internal/testnet-manager/dkg-bypass-contract",
+    "tools/internal/testnet-manager/dkg-bypass-contract",
     "tools/internal/validator-status-check",
     "tools/nym-cli",
     "tools/nym-id-cli",

common/client-core/Cargo.toml

@@ -10,7 +10,8 @@ license.workspace = true
 
 [dependencies]
 async-trait = { workspace = true }
-base64 = { workspace = true }
+bincode = { workspace = true }
+base64 = "0.21.2"
 bs58 = { workspace = true }
 clap = { workspace = true, optional = true }
 cfg-if = { workspace = true }
@@ -25,6 +26,7 @@ sha2 = { workspace = true }
 si-scale = { workspace = true }
 thiserror = { workspace = true }
 url = { workspace = true, features = ["serde"] }
+tokio-util = { workspace = true, features = ["codec"] }
 time = { workspace = true }
 tokio = { workspace = true, features = ["sync", "macros"] }
 tracing = { workspace = true }

common/client-core/src/client/base_client/mod.rs

@@ -36,6 +36,7 @@ use crate::init::{
     types::{GatewaySetup, InitialisationResult},
 };
 use futures::channel::mpsc;
+use futures::SinkExt;
 use nym_bandwidth_controller::BandwidthController;
 use nym_client_core_config_types::{ForgetMe, RememberMe};
 use nym_client_core_gateways_storage::{GatewayDetails, GatewaysDetailsStore};
@@ -66,7 +67,7 @@ use std::os::raw::c_int as RawFd;
 use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tokio::sync::mpsc::Sender;
+use tokio_util::sync::{PollSendError, PollSender};
 use url::Url;
 
 #[cfg(target_arch = "wasm32")]
@@ -108,10 +109,7 @@ pub struct ClientInput {
 }
 
 impl ClientInput {
-    pub async fn send(
-        &self,
-        message: InputMessage,
-    ) -> Result<(), tokio::sync::mpsc::error::SendError<InputMessage>> {
+    pub async fn send(&mut self, message: InputMessage) -> Result<(), PollSendError<InputMessage>> {
         self.input_sender.send(message).await
     }
 }
@@ -756,7 +754,7 @@ where
         config: &Config,
         user_agent: Option<UserAgent>,
         client_stats_id: String,
-        input_sender: Sender<InputMessage>,
+        input_sender: PollSender<InputMessage>,
         shutdown_tracker: &ShutdownTracker,
     ) -> ClientStatsSender {
         tracing::info!("Starting statistics control...");
@@ -975,7 +973,7 @@ where
             &self.config,
             self.user_agent.clone(),
             generate_client_stats_id(*self_address.identity()),
-            input_sender.clone(),
+            tokio_util::sync::PollSender::new(input_sender.clone()),
             &shutdown_tracker.child_tracker(),
         );
 
@@ -1102,7 +1100,7 @@ where
             client_input: ClientInputStatus::AwaitingProducer {
                 client_input: ClientInput {
                     connection_command_sender: client_connection_tx,
-                    input_sender,
+                    input_sender: PollSender::new(input_sender),
                     client_request_sender,
                 },
             },

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

@@ -1,16 +1,26 @@
 // Copyright 2020-2023 - Nym Technologies SA <[email protected]>
 // SPDX-License-Identifier: Apache-2.0
-
 use nym_sphinx::addressing::clients::Recipient;
 use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
 use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
+use serde::{Deserialize, Serialize};
+use std::convert::TryInto;
+use tokio_util::{
+    bytes::Buf,
+    bytes::BytesMut,
+    codec::{Decoder, Encoder},
+};
+
+use crate::error::ClientCoreError;
 
-pub type InputMessageSender = tokio::sync::mpsc::Sender<InputMessage>;
+pub type InputMessageSender = tokio_util::sync::PollSender<InputMessage>;
 pub type InputMessageReceiver = tokio::sync::mpsc::Receiver<InputMessage>;
 
-#[derive(Debug)]
+const LENGHT_ENCODING_PREFIX_SIZE: usize = 4;
+
+#[derive(Serialize, Deserialize, Debug)]
 pub enum InputMessage {
     /// Fire an already prepared mix packets into the network.
     /// No guarantees are made about it. For example no retransmssion
@@ -65,6 +75,10 @@ pub enum InputMessage {
 }
 
 impl InputMessage {
+    pub fn simple(data: &[u8], recipient: Recipient) -> Self {
+        InputMessage::new_regular(recipient, data.to_vec(), TransmissionLane::General, None)
+    }
+
     pub fn new_premade(
         msgs: Vec<MixPacket>,
         lane: TransmissionLane,
@@ -185,4 +199,71 @@ impl InputMessage {
         self.set_max_retransmissions(max_retransmissions);
         self
     }
+    #[allow(clippy::expect_used)]
+    pub fn serialized_size(&self) -> u64 {
+        bincode::serialized_size(self).expect("failed to get serialized InputMessage size")
+            + LENGHT_ENCODING_PREFIX_SIZE as u64
+    }
+}
+
+// TODO: Tests
+pub struct AdressedInputMessageCodec(pub Recipient);
+
+impl Encoder<&[u8]> for AdressedInputMessageCodec {
+    type Error = ClientCoreError;
+
+    fn encode(&mut self, item: &[u8], buf: &mut BytesMut) -> Result<(), Self::Error> {
+        let mut codec = InputMessageCodec;
+        let input_message = InputMessage::simple(item, self.0);
+        codec.encode(input_message, buf)?;
+        Ok(())
+    }
+}
+
+pub struct InputMessageCodec;
+
+impl Encoder<InputMessage> for InputMessageCodec {
+    type Error = ClientCoreError;
+
+    fn encode(&mut self, item: InputMessage, buf: &mut BytesMut) -> Result<(), Self::Error> {
+        #[allow(clippy::expect_used)]
+        let encoded = bincode::serialize(&item).expect("failed to serialize InputMessage");
+        let encoded_len = encoded.len() as u32;
+        let mut encoded_with_len = encoded_len.to_le_bytes().to_vec();
+        encoded_with_len.extend(encoded);
+        buf.reserve(encoded_with_len.len());
+        buf.extend_from_slice(&encoded_with_len);
+        Ok(())
+    }
+}
+
+impl Decoder for InputMessageCodec {
+    type Item = InputMessage;
+    type Error = ClientCoreError;
+
+    fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
+        if buf.len() < LENGHT_ENCODING_PREFIX_SIZE {
+            return Ok(None);
+        }
+        #[allow(clippy::expect_used)]
+        let len = u32::from_le_bytes(
+            buf[0..LENGHT_ENCODING_PREFIX_SIZE]
+                .try_into()
+                .expect("Could not coarce to array"),
+        ) as usize;
+        if buf.len() < len + LENGHT_ENCODING_PREFIX_SIZE {
+            return Ok(None);
+        }
+
+        let decoded = match bincode::deserialize(
+            &buf[LENGHT_ENCODING_PREFIX_SIZE..len + LENGHT_ENCODING_PREFIX_SIZE],
+        ) {
+            Ok(decoded) => decoded,
+            Err(_) => return Ok(None),
+        };
+
+        buf.advance(len + LENGHT_ENCODING_PREFIX_SIZE);
+
+        Ok(Some(decoded))
+    }
 }

common/client-core/src/client/mix_traffic/transceiver.rs

@@ -13,7 +13,7 @@ use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
 use std::fmt::Debug;
 use std::os::raw::c_int as RawFd;
 use thiserror::Error;
-use tracing::{debug, error};
+use tracing::debug;
 
 #[cfg(not(target_arch = "wasm32"))]
 use futures::channel::oneshot;

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

@@ -8,7 +8,6 @@ use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
 use nym_sphinx::anonymous_replies::ReplySurbWithKeyRotation;
 use nym_task::connections::{ConnectionId, TransmissionLane};
 use std::sync::Weak;
-use tracing::error;
 
 pub(crate) fn new_control_channels() -> (ReplyControllerSender, ReplyControllerReceiver) {
     let (tx, rx) = mpsc::unbounded();

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

@@ -17,7 +17,7 @@
 #![warn(clippy::dbg_macro)]
 
 use crate::client::inbound_messages::{InputMessage, InputMessageSender};
-use futures::StreamExt;
+use futures::{SinkExt, StreamExt};
 use nym_client_core_config_types::StatsReporting;
 use nym_sphinx::addressing::Recipient;
 use nym_statistics_common::clients::{

common/client-core/src/init/helpers.rs

@@ -441,7 +441,7 @@ mod tests {
 
     #[test]
     fn test_multiple_urls_prepared_for_retries() {
-        let urls = vec![
+        let urls = [
             Url::parse("https://api1.nym.com").unwrap(),
             Url::parse("https://api2.nym.com").unwrap(),
             Url::parse("https://api3.nym.com").unwrap(),

common/client-core/src/lib.rs

@@ -1,3 +1,4 @@
+#![allow(deprecated)]
 use std::future::Future;
 
 #[cfg(all(

common/client-core/surb-storage/src/lib.rs

@@ -1,6 +1,7 @@
 // Copyright 2022 - Nym Technologies SA <[email protected]>
 // SPDX-License-Identifier: Apache-2.0
 
+#![allow(deprecated)]
 pub use backend::*;
 pub use combined::CombinedReplyStorage;
 pub use key_storage::SentReplyKeys;

common/client-libs/validator-client/src/nym_api/mod.rs

@@ -273,6 +273,49 @@ pub trait NymApiClientExt: ApiClient {
         Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
     }
 
+    async fn get_all_basic_exit_assigned_nodes_with_metadata(
+        &self,
+    ) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
+        // Get all nodes that can act as exit gateways
+        let mut page = 0;
+        let res = self
+            .get_basic_exit_assigned_nodes_v2(false, Some(page), None, false)
+            .await?;
+
+        let metadata = res.metadata;
+        let mut nodes = res.nodes.data;
+
+        if res.nodes.pagination.total == nodes.len() {
+            return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
+        }
+
+        page += 1;
+
+        loop {
+            let res = self
+                .get_basic_exit_assigned_nodes_v2(false, Some(page), None, false)
+                .await?;
+
+            if !metadata.consistency_check(&res.metadata) {
+                return Err(NymAPIError::InternalResponseInconsistency {
+                    url: self.api_url().clone(),
+                    details: "Inconsistent paged metadata".to_string(),
+                });
+            }
+
+            nodes.append(&mut res.nodes.data.clone());
+
+            // Check if we've got all nodes
+            if nodes.len() >= res.nodes.pagination.total {
+                break;
+            } else {
+                page += 1;
+            }
+        }
+
+        Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
+    }
+
     async fn get_all_described_nodes(&self) -> Result<Vec<NymNodeDescription>, NymAPIError> {
         // TODO: deal with paging in macro or some helper function or something, because it's the same pattern everywhere
         let mut page = 0;
@@ -467,6 +510,47 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
+    /// retrieve basic information for nodes are capable of operating as an exit gateway
+    /// this includes legacy gateways and nym-nodes
+    #[instrument(level = "debug", skip(self))]
+    async fn get_basic_exit_assigned_nodes_v2(
+        &self,
+        no_legacy: bool,
+        page: Option<u32>,
+        per_page: Option<u32>,
+        use_bincode: bool,
+    ) -> Result<PaginatedCachedNodesResponseV2<SkimmedNode>, NymAPIError> {
+        let mut params = Vec::new();
+
+        if no_legacy {
+            params.push(("no_legacy", "true".to_string()))
+        }
+
+        if let Some(page) = page {
+            params.push(("page", page.to_string()))
+        }
+
+        if let Some(per_page) = per_page {
+            params.push(("per_page", per_page.to_string()))
+        }
+
+        if use_bincode {
+            params.push(("output", "bincode".to_string()))
+        }
+
+        self.get_response(
+            &[
+                routes::V2_API_VERSION,
+                "unstable",
+                routes::NYM_NODES_ROUTES,
+                "skimmed",
+                "exit-gateways",
+            ],
+            &params,
+        )
+        .await
+    }
+
     /// retrieve basic information for nodes that got assigned 'mixing' node in this epoch
     /// this includes legacy mixnodes and nym-nodes
     #[deprecated(note = "use get_basic_active_mixing_assigned_nodes_v2")]

common/cosmwasm-smart-contracts/coconut-dkg/src/types.rs

@@ -1,6 +1,8 @@
 // Copyright 2022-2023 - Nym Technologies SA <[email protected]>
 // SPDX-License-Identifier: Apache-2.0
 
+#![allow(clippy::derivable_impls)]
+// MAX: surpressing warning for the moment, will be dealt with in a different PR (TODO)
 use cosmwasm_schema::cw_serde;
 use std::fmt::{Display, Formatter};
 use std::str::FromStr;

common/credential-storage/src/ephemeral_storage.rs

@@ -246,7 +246,7 @@ mod tests {
         let _exp_date_sigs = generate_expiration_date_signatures(
             sig_req.expiration_date.ecash_unix_timestamp(),
             &[signing_keys.secret_key()],
-            &vec![signing_keys.verification_key()],
+            &[signing_keys.verification_key()],
             &signing_keys.verification_key(),
             &[1],
         )?;
@@ -263,7 +263,7 @@ mod tests {
 
         let wallet = issuance.aggregate_signature_shares(
             &signing_keys.verification_key(),
-            &vec![partial_wallet],
+            &[partial_wallet],
             sig_req,
         )?;