wip

* New transactions for increasing amount of pledged tokens

* unit tests

* Added an option to pledge extra tokens through the vesting contract

* Introduced wallet endpoints for new operations

* Using updated pledge cap in the vesting contract

* Changelog update

CHANGELOG.md

@@ -6,6 +6,13 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ### Added
 
+- socks5-client/network-requester: add support for socks4a protocol
+
+
+## [v1.1.1](https://github.com/nymtech/nym/tree/v1.1.1) (2022-11-29)
+
+### Added
+
 - binaries: add `-c` shortform for `--config-env-file`
 - websocket-requests: add server response signalling current packet queue length in the client
 - contracts: DKG contract that handles coconut key generation ([#1678][#1708][#1747])

Cargo.lock

@@ -586,7 +586,7 @@ dependencies = [
 
 [[package]]
 name = "client-core"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "client-connections",
  "config",
@@ -609,6 +609,7 @@ dependencies = [
  "tempfile",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "topology",
  "url",
  "validator-client",
@@ -1633,7 +1634,7 @@ dependencies = [
 
 [[package]]
 name = "explorer-api"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "chrono",
  "clap 3.2.8",
@@ -3122,7 +3123,7 @@ dependencies = [
 
 [[package]]
 name = "nym-cli"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "base64",
@@ -3176,7 +3177,7 @@ dependencies = [
 
 [[package]]
 name = "nym-client"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "clap 3.2.8",
  "client-connections",
@@ -3216,7 +3217,7 @@ dependencies = [
 
 [[package]]
 name = "nym-gateway"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3263,7 +3264,7 @@ dependencies = [
 
 [[package]]
 name = "nym-mixnode"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "bs58",
@@ -3305,7 +3306,7 @@ dependencies = [
 
 [[package]]
 name = "nym-network-requester"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "async-trait",
  "clap 3.2.8",
@@ -3351,9 +3352,18 @@ dependencies = [
  "tokio",
 ]
 
+[[package]]
+name = "nym-sdk"
+version = "0.1.0"
+dependencies = [
+ "client-core",
+ "rand 0.7.3",
+ "tokio",
+]
+
 [[package]]
 name = "nym-socks5-client"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "clap 3.2.8",
  "client-connections",
@@ -3420,7 +3430,7 @@ dependencies = [
 
 [[package]]
 name = "nym-validator-api"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "anyhow",
  "async-trait",

Cargo.toml

@@ -72,6 +72,7 @@ members = [
     "gateway/gateway-requests",
     "integrations/bity",
     "mixnode",
+    "sdk/rust/nym-sdk",
     "service-providers/network-requester",
     "service-providers/network-statistics",
     "validator-api",

clients/client-core/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "client-core"
-version = "1.1.0"
+version = "1.1.1"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
 
@@ -17,6 +17,7 @@ sled = { version = "0.34", optional = true }
 tap = "1.0.1"
 thiserror = "1.0.34"
 url = { version ="2.2", features = ["serde"] }
+tokio = { version = "1.21.2", features = ["time", "macros"]}
 
 # internal
 config = { path = "../../common/config" }
@@ -30,8 +31,11 @@ nymsphinx = { path = "../../common/nymsphinx" }
 pemstore = { path = "../../common/pemstore" }
 topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client", default-features = false }
+task = { path = "../../common/task" }
 
-tokio = { version = "1.21.2", features = ["time", "macros"]}
+[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-stream]
+version = "0.1.9"
+features = ["time"]
 
 [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen-futures]
 version = "0.4"
@@ -47,8 +51,8 @@ rev = "b9d1a54ad514c2f230a026afe0dde341e98cd7b6"
 version = "0.2.4"
 features = ["futures"]
 
-[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task]
+#[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task]
-path = "../../common/task"
+#path = "../../common/task"
 
 [dev-dependencies]
 tempfile = "3.1.0"

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

@@ -0,0 +1,441 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::client::cover_traffic_stream::LoopCoverTrafficStream;
+use crate::client::inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender};
+use crate::client::key_manager::KeyManager;
+use crate::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
+use crate::client::real_messages_control;
+use crate::client::real_messages_control::RealMessagesController;
+use crate::client::received_buffer::{
+    ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController,
+};
+use crate::client::topology_control::{
+    TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
+};
+use crate::config::{Config, DebugConfig, GatewayEndpointConfig};
+use crate::error::ClientCoreError;
+use client_connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
+use crypto::asymmetric::identity;
+use futures::channel::mpsc;
+use gateway_client::bandwidth::BandwidthController;
+use gateway_client::{
+    AcknowledgementReceiver, AcknowledgementSender, GatewayClient, MixnetMessageReceiver,
+    MixnetMessageSender,
+};
+use log::info;
+use nymsphinx::addressing::clients::Recipient;
+use nymsphinx::addressing::nodes::NodeIdentity;
+#[cfg(feature = "reply-surb")]
+use std::path::PathBuf;
+#[cfg(feature = "reply-surb")]
+use tap::TapFallible;
+use task::{ShutdownListener, ShutdownNotifier};
+use url::Url;
+
+// it's fine to do this disgusting compilation flag business here as this problem
+// is going to go away in 1.2.0
+#[cfg(feature = "reply-surb")]
+use crate::client::reply_key_storage::ReplyKeyStorage;
+
+pub struct ClientInput {
+    pub shared_lane_queue_lengths: LaneQueueLengths,
+    pub connection_command_sender: ConnectionCommandSender,
+    pub input_sender: InputMessageSender,
+}
+
+pub struct ClientOutput {
+    pub received_buffer_request_sender: ReceivedBufferRequestSender,
+}
+
+pub enum ClientInputStatus {
+    AwaitingProducer { client_input: ClientInput },
+    Connected,
+}
+
+impl ClientInputStatus {
+    pub fn register_producer(&mut self) -> ClientInput {
+        match std::mem::replace(self, ClientInputStatus::Connected) {
+            ClientInputStatus::AwaitingProducer { client_input } => client_input,
+            ClientInputStatus::Connected => panic!("producer was already registered before"),
+        }
+    }
+}
+
+pub enum ClientOutputStatus {
+    AwaitingConsumer { client_output: ClientOutput },
+    Connected,
+}
+
+impl ClientOutputStatus {
+    pub fn register_consumer(&mut self) -> ClientOutput {
+        match std::mem::replace(self, ClientOutputStatus::Connected) {
+            ClientOutputStatus::AwaitingConsumer { client_output } => client_output,
+            ClientOutputStatus::Connected => panic!("consumer was already registered before"),
+        }
+    }
+}
+
+pub struct BaseClientBuilder<'a> {
+    // due to wasm limitations I had to split it like this : (
+    gateway_config: &'a GatewayEndpointConfig,
+    debug_config: &'a DebugConfig,
+    disabled_credentials: bool,
+    validator_api_endpoints: Vec<Url>,
+
+    #[cfg(feature = "reply-surb")]
+    reply_surb_keys_store_path: PathBuf,
+
+    bandwidth_controller: Option<BandwidthController>,
+    key_manager: KeyManager,
+}
+
+impl<'a> BaseClientBuilder<'a> {
+    pub fn new_from_base_config<T>(
+        base_config: &'a Config<T>,
+        key_manager: KeyManager,
+        bandwidth_controller: Option<BandwidthController>,
+    ) -> BaseClientBuilder<'a> {
+        BaseClientBuilder {
+            gateway_config: base_config.get_gateway_endpoint_config(),
+            debug_config: base_config.get_debug_config(),
+            disabled_credentials: base_config.get_disabled_credentials_mode(),
+            validator_api_endpoints: base_config.get_validator_api_endpoints(),
+            bandwidth_controller,
+            key_manager,
+            #[cfg(feature = "reply-surb")]
+            reply_surb_keys_store_path: base_config.get_reply_encryption_key_store_path(),
+        }
+    }
+
+    pub fn new(
+        gateway_config: &'a GatewayEndpointConfig,
+        debug_config: &'a DebugConfig,
+        key_manager: KeyManager,
+        bandwidth_controller: Option<BandwidthController>,
+        disabled_credentials: bool,
+        validator_api_endpoints: Vec<Url>,
+        #[cfg(feature = "reply-surb")] reply_surb_keys_store_path: PathBuf,
+    ) -> BaseClientBuilder<'a> {
+        BaseClientBuilder {
+            gateway_config,
+            debug_config,
+            disabled_credentials,
+            validator_api_endpoints,
+            bandwidth_controller,
+            key_manager,
+            #[cfg(feature = "reply-surb")]
+            reply_surb_keys_store_path,
+        }
+    }
+
+    pub fn as_mix_recipient(&self) -> Recipient {
+        Recipient::new(
+            *self.key_manager.identity_keypair().public_key(),
+            *self.key_manager.encryption_keypair().public_key(),
+            // TODO: below only works under assumption that gateway address == gateway id
+            // (which currently is true)
+            NodeIdentity::from_base58_string(&self.gateway_config.gateway_id).unwrap(),
+        )
+    }
+
+    // future constantly pumping loop cover traffic at some specified average rate
+    // the pumped traffic goes to the MixTrafficController
+    fn start_cover_traffic_stream(
+        &self,
+        topology_accessor: TopologyAccessor,
+        mix_tx: BatchMixMessageSender,
+        shutdown: ShutdownListener,
+    ) {
+        info!("Starting loop cover traffic stream...");
+
+        let mut stream = LoopCoverTrafficStream::new(
+            self.key_manager.ack_key(),
+            self.debug_config.average_ack_delay,
+            self.debug_config.average_packet_delay,
+            self.debug_config.loop_cover_traffic_average_delay,
+            mix_tx,
+            self.as_mix_recipient(),
+            topology_accessor,
+        );
+
+        if let Some(size) = self.debug_config.use_extended_packet_size {
+            log::debug!("Setting extended packet size: {:?}", size);
+            stream.set_custom_packet_size(size.into());
+        }
+
+        stream.start_with_shutdown(shutdown);
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn start_real_traffic_controller(
+        &self,
+        topology_accessor: TopologyAccessor,
+        ack_receiver: AcknowledgementReceiver,
+        input_receiver: InputMessageReceiver,
+        mix_sender: BatchMixMessageSender,
+        lane_queue_lengths: LaneQueueLengths,
+        client_connection_rx: ConnectionCommandReceiver,
+        shutdown: ShutdownListener,
+        #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
+    ) {
+        let mut controller_config = real_messages_control::Config::new(
+            self.key_manager.ack_key(),
+            self.debug_config.ack_wait_multiplier,
+            self.debug_config.ack_wait_addition,
+            self.debug_config.average_ack_delay,
+            self.debug_config.message_sending_average_delay,
+            self.debug_config.average_packet_delay,
+            self.debug_config.disable_main_poisson_packet_distribution,
+            self.as_mix_recipient(),
+        );
+
+        if let Some(size) = self.debug_config.use_extended_packet_size {
+            log::debug!("Setting extended packet size: {:?}", size);
+            controller_config.set_custom_packet_size(size.into());
+        }
+
+        info!("Starting real traffic stream...");
+
+        RealMessagesController::new(
+            controller_config,
+            ack_receiver,
+            input_receiver,
+            mix_sender,
+            topology_accessor,
+            lane_queue_lengths,
+            client_connection_rx,
+            #[cfg(feature = "reply-surb")]
+            reply_key_storage,
+        )
+        .start_with_shutdown(shutdown);
+    }
+
+    // buffer controlling all messages fetched from provider
+    // required so that other components would be able to use them (say the websocket)
+    fn start_received_messages_buffer_controller(
+        &self,
+        query_receiver: ReceivedBufferRequestReceiver,
+        mixnet_receiver: MixnetMessageReceiver,
+        shutdown: ShutdownListener,
+        #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
+    ) {
+        info!("Starting received messages buffer controller...");
+        ReceivedMessagesBufferController::new(
+            self.key_manager.encryption_keypair(),
+            query_receiver,
+            mixnet_receiver,
+            #[cfg(feature = "reply-surb")]
+            reply_key_storage,
+        )
+        .start_with_shutdown(shutdown)
+    }
+
+    async fn start_gateway_client(
+        &mut self,
+        mixnet_message_sender: MixnetMessageSender,
+        ack_sender: AcknowledgementSender,
+        shutdown: ShutdownListener,
+    ) -> GatewayClient {
+        let gateway_id = self.gateway_config.gateway_id.clone();
+        if gateway_id.is_empty() {
+            panic!("The identity of the gateway is unknown - did you run `nym-client` init?")
+        }
+        let gateway_owner = self.gateway_config.gateway_owner.clone();
+        if gateway_owner.is_empty() {
+            panic!("The owner of the gateway is unknown - did you run `nym-client` init?")
+        }
+        let gateway_address = self.gateway_config.gateway_listener.clone();
+        if gateway_address.is_empty() {
+            panic!("The address of the gateway is unknown - did you run `nym-client` init?")
+        }
+
+        let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
+            .expect("provided gateway id is invalid!");
+
+        // disgusting wasm workaround since there's no key persistence there (nor `client init`)
+        let shared_key = if self.key_manager.gateway_key_set() {
+            Some(self.key_manager.gateway_shared_key())
+        } else {
+            None
+        };
+
+        let mut gateway_client = GatewayClient::new(
+            gateway_address,
+            self.key_manager.identity_keypair(),
+            gateway_identity,
+            gateway_owner,
+            shared_key,
+            mixnet_message_sender,
+            ack_sender,
+            self.debug_config.gateway_response_timeout,
+            self.bandwidth_controller.take(),
+            Some(shutdown),
+        );
+
+        gateway_client.set_disabled_credentials_mode(self.disabled_credentials);
+
+        gateway_client
+            .authenticate_and_start()
+            .await
+            .expect("could not authenticate and start up the gateway connection");
+
+        gateway_client
+    }
+
+    // future responsible for periodically polling directory server and updating
+    // the current global view of topology
+    async fn start_topology_refresher(
+        &mut self,
+        topology_accessor: TopologyAccessor,
+        shutdown: ShutdownListener,
+    ) -> Result<(), ClientCoreError> {
+        let topology_refresher_config = TopologyRefresherConfig::new(
+            self.validator_api_endpoints.clone(),
+            self.debug_config.topology_refresh_rate,
+            env!("CARGO_PKG_VERSION").to_string(),
+        );
+        let mut topology_refresher =
+            TopologyRefresher::new(topology_refresher_config, topology_accessor);
+        // before returning, block entire runtime to refresh the current network view so that any
+        // components depending on topology would see a non-empty view
+        info!("Obtaining initial network topology");
+        topology_refresher.refresh().await;
+
+        // TODO: a slightly more graceful termination here
+        if !topology_refresher.is_topology_routable().await {
+            log::error!(
+                "The current network topology seem to be insufficient to route any packets through \
+                - check if enough nodes and a gateway are online"
+            );
+            return Err(ClientCoreError::InsufficientNetworkTopology);
+        }
+
+        info!("Starting topology refresher...");
+        topology_refresher.start_with_shutdown(shutdown);
+        Ok(())
+    }
+
+    // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
+    // TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
+    // over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
+    // requests?
+    fn start_mix_traffic_controller(
+        gateway_client: GatewayClient,
+        shutdown: ShutdownListener,
+    ) -> BatchMixMessageSender {
+        info!("Starting mix traffic controller...");
+        let (mix_traffic_controller, mix_tx) = MixTrafficController::new(gateway_client);
+        mix_traffic_controller.start_with_shutdown(shutdown);
+        mix_tx
+    }
+
+    pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError> {
+        info!("Starting nym client");
+        // channels for inter-component communication
+        // TODO: make the channels be internally created by the relevant components
+        // rather than creating them here, so say for example the buffer controller would create the request channels
+        // and would allow anyone to clone the sender channel
+
+        // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway
+        // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer
+        let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
+
+        // used for announcing connection or disconnection of a channel for pushing re-assembled messages to
+        let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
+
+        // channels responsible for controlling real messages
+        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
+
+        // channels responsible for controlling ack messages
+        let (ack_sender, ack_receiver) = mpsc::unbounded();
+        let shared_topology_accessor = TopologyAccessor::new();
+
+        #[cfg(feature = "reply-surb")]
+        let reply_key_storage =
+            ReplyKeyStorage::load(&self.reply_surb_keys_store_path).tap_err(|err| {
+                log::error!("Failed to load reply key storage - is it perhaps already in use?");
+                log::error!("{:?}", err);
+            })?;
+
+        // Shutdown notifier for signalling tasks to stop
+        let shutdown = ShutdownNotifier::default();
+
+        // the components are started in very specific order. Unless you know what you are doing,
+        // do not change that.
+        self.start_topology_refresher(shared_topology_accessor.clone(), shutdown.subscribe())
+            .await?;
+        self.start_received_messages_buffer_controller(
+            received_buffer_request_receiver,
+            mixnet_messages_receiver,
+            shutdown.subscribe(),
+            #[cfg(feature = "reply-surb")]
+            reply_key_storage.clone(),
+        );
+
+        let gateway_client = self
+            .start_gateway_client(mixnet_messages_sender, ack_sender, shutdown.subscribe())
+            .await;
+
+        // The sphinx_message_sender is the transmitter for any component generating sphinx packets
+        // that are to be sent to the mixnet. They are used by cover traffic stream and real
+        // traffic stream.
+        // The MixTrafficController then sends the actual traffic
+        let sphinx_message_sender =
+            Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe());
+
+        // Channels that the websocket listener can use to signal downstream to the real traffic
+        // controller that connections are closed.
+        let (client_connection_tx, client_connection_rx) = mpsc::unbounded();
+
+        // Shared queue length data. Published by the `OutQueueController` in the client, and used
+        // primarily to throttle incoming connections (e.g socks5 for attached network-requesters)
+        let shared_lane_queue_lengths = LaneQueueLengths::new();
+
+        self.start_real_traffic_controller(
+            shared_topology_accessor.clone(),
+            ack_receiver,
+            input_receiver,
+            sphinx_message_sender.clone(),
+            shared_lane_queue_lengths.clone(),
+            client_connection_rx,
+            shutdown.subscribe(),
+            #[cfg(feature = "reply-surb")]
+            reply_key_storage,
+        );
+
+        if !self.debug_config.disable_loop_cover_traffic_stream {
+            self.start_cover_traffic_stream(
+                shared_topology_accessor,
+                sphinx_message_sender,
+                shutdown.subscribe(),
+            );
+        }
+
+        info!("Client startup finished!");
+        info!("The address of this client is: {}", self.as_mix_recipient());
+
+        Ok(BaseClient {
+            client_input: ClientInputStatus::AwaitingProducer {
+                client_input: ClientInput {
+                    shared_lane_queue_lengths,
+                    connection_command_sender: client_connection_tx,
+                    input_sender,
+                },
+            },
+            client_output: ClientOutputStatus::AwaitingConsumer {
+                client_output: ClientOutput {
+                    received_buffer_request_sender,
+                },
+            },
+            shutdown_notifier: shutdown,
+        })
+    }
+}
+
+pub struct BaseClient {
+    pub client_input: ClientInputStatus,
+    pub client_output: ClientOutputStatus,
+
+    pub shutdown_notifier: ShutdownNotifier,
+}

clients/client-core/src/client/cover_traffic_stream.rs

@@ -143,6 +143,16 @@ impl LoopCoverTrafficStream<OsRng> {
         self.packet_size = packet_size;
     }
 
+    fn set_next_delay(&mut self, amount: Duration) {
+        #[cfg(not(target_arch = "wasm32"))]
+        let next_delay = Box::pin(time::sleep(amount));
+
+        #[cfg(target_arch = "wasm32")]
+        let next_delay = Box::pin(wasm_timer::Delay::new(amount));
+
+        self.next_delay = next_delay;
+    }
+
     async fn on_new_message(&mut self) {
         trace!("next cover message!");
 
@@ -202,12 +212,11 @@ impl LoopCoverTrafficStream<OsRng> {
         tokio::task::yield_now().await;
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) {
         // we should set initial delay only when we actually start the stream
         let sampled =
             sample_poisson_duration(&mut self.rng, self.average_cover_message_sending_delay);
-        self.next_delay = Box::pin(time::sleep(sampled));
+        self.set_next_delay(sampled);
 
         spawn_future(async move {
             debug!("Started LoopCoverTrafficStream with graceful shutdown support");
@@ -228,19 +237,16 @@ impl LoopCoverTrafficStream<OsRng> {
                     }
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            shutdown.recv_timeout().await;
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("LoopCoverTrafficStream: Exiting");
         })
     }
 
-    #[cfg(target_arch = "wasm32")]
     pub fn start(mut self) {
         // we should set initial delay only when we actually start the stream
         let sampled =
             sample_poisson_duration(&mut self.rng, self.average_cover_message_sending_delay);
-        self.next_delay = Box::pin(wasm_timer::Delay::new(sampled));
+        self.set_next_delay(sampled);
 
         spawn_future(async move {
             debug!("Started LoopCoverTrafficStream without graceful shutdown support");

clients/client-core/src/client/key_manager.rs

@@ -149,6 +149,10 @@ impl KeyManager {
         )
     }
 
+    pub fn gateway_key_set(&self) -> bool {
+        self.gateway_shared_key.is_some()
+    }
+
     /// Gets an atomically reference counted pointer to [`AckKey`].
     pub fn ack_key(&self) -> Arc<AckKey> {
         Arc::clone(&self.ack_key)

clients/client-core/src/client/mix_traffic.rs

@@ -67,10 +67,7 @@ impl MixTrafficController {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         spawn_future(async move {
             debug!("Started MixTrafficController with graceful shutdown support");
 
@@ -90,14 +87,11 @@ impl MixTrafficController {
                     }
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            shutdown.recv_timeout().await;
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("MixTrafficController: Exiting");
         })
     }
 
-    #[cfg(target_arch = "wasm32")]
     pub fn start(mut self) {
         spawn_future(async move {
             debug!("Started MixTrafficController without graceful shutdown support");

clients/client-core/src/client/mod.rs

@@ -1,3 +1,7 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+pub mod base_client;
 pub mod cover_traffic_stream;
 pub mod inbound_messages;
 pub mod key_manager;

clients/client-core/src/client/real_messages_control/acknowledgement_control/acknowledgement_listener.rs

@@ -70,10 +70,7 @@ impl AcknowledgementListener {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started AcknowledgementListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
@@ -90,13 +87,12 @@ impl AcknowledgementListener {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        shutdown.recv_timeout().await;
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("AcknowledgementListener: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started AcknowledgementListener without graceful shutdown support");
 

clients/client-core/src/client/real_messages_control/acknowledgement_control/action_controller.rs

@@ -245,7 +245,6 @@ impl ActionController {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
         debug!("Started ActionController with graceful shutdown support");
 
@@ -272,13 +271,15 @@ impl ActionController {
                 }
             }
         }
+        #[cfg(not(target_arch = "wasm32"))]
         tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
             .await
             .expect("Task stopped without shutdown called");
         log::debug!("ActionController: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started ActionController without graceful shutdown support");
 

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

@@ -194,10 +194,7 @@ where
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started InputMessageListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
@@ -216,13 +213,12 @@ where
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        shutdown.recv_timeout().await;
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("InputMessageListener: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started InputMessageListener without graceful shutdown support");
         while let Some(input_msg) = self.input_receiver.recv().await {

clients/client-core/src/client/real_messages_control/acknowledgement_control/mod.rs

@@ -234,7 +234,6 @@ where
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) fn start_with_shutdown(self, shutdown: task::ShutdownListener) {
         let mut acknowledgement_listener = self.acknowledgement_listener;
         let mut input_message_listener = self.input_message_listener;
@@ -280,7 +279,8 @@ where
         });
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) fn start(self) {
         let mut acknowledgement_listener = self.acknowledgement_listener;
         let mut input_message_listener = self.input_message_listener;

clients/client-core/src/client/real_messages_control/acknowledgement_control/retransmission_request_listener.rs

@@ -125,10 +125,7 @@ where
             .expect("BatchRealMessageReceiver has stopped receiving!");
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started RetransmissionRequestListener with graceful shutdown support");
 
         while !shutdown.is_shutdown() {
@@ -145,13 +142,12 @@ where
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        shutdown.recv_timeout().await;
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("RetransmissionRequestListener: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started RetransmissionRequestListener without graceful shutdown support");
 

clients/client-core/src/client/real_messages_control/acknowledgement_control/sent_notification_listener.rs

@@ -42,7 +42,6 @@ impl SentNotificationListener {
             .unwrap();
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
         debug!("Started SentNotificationListener with graceful shutdown support");
 
@@ -66,7 +65,8 @@ impl SentNotificationListener {
         log::debug!("SentNotificationListener: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started SentNotificationListener without graceful shutdown support");
 

clients/client-core/src/client/real_messages_control/mod.rs

@@ -113,9 +113,9 @@ impl RealMessagesController<OsRng> {
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
         topology_access: TopologyAccessor,
-        #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
+        #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage,
     ) -> Self {
         let rng = OsRng;
 
@@ -175,7 +175,6 @@ impl RealMessagesController<OsRng> {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(self, shutdown: task::ShutdownListener) {
         let mut out_queue_control = self.out_queue_control;
         let ack_control = self.ack_control;

clients/client-core/src/client/real_messages_control/real_traffic_stream.rs

@@ -495,40 +495,61 @@ where
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub(super) async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
         debug!("Started OutQueueControl with graceful shutdown support");
 
-        let mut status_timer = tokio::time::interval(Duration::from_secs(5));
+        #[cfg(not(target_arch = "wasm32"))]
-        let mut infrequent_status_timer = tokio::time::interval(Duration::from_secs(60));
+        {
+            let mut status_timer = tokio::time::interval(Duration::from_secs(5));
+            let mut infrequent_status_timer = tokio::time::interval(Duration::from_secs(60));
 
-        while !shutdown.is_shutdown() {
+            while !shutdown.is_shutdown() {
-            tokio::select! {
+                tokio::select! {
-                biased;
+                    biased;
-                _ = shutdown.recv() => {
+                    _ = shutdown.recv() => {
-                    log::trace!("OutQueueControl: Received shutdown");
+                        log::trace!("OutQueueControl: Received shutdown");
+                    }
+                    _ = status_timer.tick() => {
+                        self.log_status();
+                    }
+                    _ = infrequent_status_timer.tick() => {
+                        self.log_status_infrequent();
+                    }
+                    next_message = self.next() => if let Some(next_message) = next_message {
+                        self.on_message(next_message).await;
+                    } else {
+                        log::trace!("OutQueueControl: Stopping since channel closed");
+                        break;
+                    }
                 }
-                _ = status_timer.tick() => {
+            }
-                    self.log_status();
+            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
-                }
+                .await
-                _ = infrequent_status_timer.tick() => {
+                .expect("Task stopped without shutdown called");
-                    self.log_status_infrequent();
+        }
-                }
+
-                next_message = self.next() => if let Some(next_message) = next_message {
+        #[cfg(target_arch = "wasm32")]
-                    self.on_message(next_message).await;
+        {
-                } else {
+            while !shutdown.is_shutdown() {
-                    log::trace!("OutQueueControl: Stopping since channel closed");
+                tokio::select! {
-                    break;
+                    biased;
+                    _ = shutdown.recv() => {
+                        log::trace!("OutQueueControl: Received shutdown");
+                    }
+                    next_message = self.next() => if let Some(next_message) = next_message {
+                        self.on_message(next_message).await;
+                    } else {
+                        log::trace!("OutQueueControl: Stopping since channel closed");
+                        break;
+                    }
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("OutQueueControl: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     pub(super) async fn run(&mut self) {
         debug!("Started OutQueueControl without graceful shutdown support");
 

clients/client-core/src/client/received_buffer.rs

@@ -320,10 +320,7 @@ impl RequestReceiver {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started RequestReceiver with graceful shutdown support");
         while !shutdown.is_shutdown() {
             tokio::select! {
@@ -342,13 +339,12 @@ impl RequestReceiver {
                 },
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        shutdown.recv_timeout().await;
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("RequestReceiver: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     async fn run(&mut self) {
         debug!("Started RequestReceiver without graceful shutdown support");
 
@@ -374,10 +370,7 @@ impl FragmentedMessageReceiver {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     async fn run_with_shutdown(&mut self, mut shutdown: task::ShutdownListener) {
-        use std::time::Duration;
-
         debug!("Started FragmentedMessageReceiver with graceful shutdown support");
         while !shutdown.is_shutdown() {
             tokio::select! {
@@ -395,13 +388,12 @@ impl FragmentedMessageReceiver {
                 }
             }
         }
-        tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+        shutdown.recv_timeout().await;
-            .await
-            .expect("Task stopped without shutdown called");
         log::debug!("FragmentedMessageReceiver: Exiting");
     }
 
-    #[cfg(target_arch = "wasm32")]
+    // todo: think whether this is still required
+    #[allow(dead_code)]
     async fn run(&mut self) {
         debug!("Started FragmentedMessageReceiver without graceful shutdown support");
 
@@ -438,7 +430,6 @@ impl ReceivedMessagesBufferController {
         }
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(self, shutdown: task::ShutdownListener) {
         let mut fragmented_message_receiver = self.fragmented_message_receiver;
         let mut request_receiver = self.request_receiver;

clients/client-core/src/client/topology_control.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::spawn_future;
+use futures::StreamExt;
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::params::DEFAULT_NUM_MIX_HOPS;
@@ -296,14 +297,22 @@ impl TopologyRefresher {
         self.topology_accessor.is_routable().await
     }
 
-    #[cfg(not(target_arch = "wasm32"))]
     pub fn start_with_shutdown(mut self, mut shutdown: task::ShutdownListener) {
         spawn_future(async move {
             debug!("Started TopologyRefresher with graceful shutdown support");
 
+            #[cfg(not(target_arch = "wasm32"))]
+            let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(
+                self.refresh_rate,
+            ));
+
+            #[cfg(target_arch = "wasm32")]
+            let mut interval =
+                gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
+
             while !shutdown.is_shutdown() {
                 tokio::select! {
-                    _ = tokio::time::sleep(self.refresh_rate) => {
+                    _ = interval.next() => {
                         self.refresh().await;
                     },
                     _ = shutdown.recv() => {
@@ -311,21 +320,23 @@ impl TopologyRefresher {
                     },
                 }
             }
-            tokio::time::timeout(Duration::from_secs(5), shutdown.recv())
+            shutdown.recv_timeout().await;
-                .await
-                .expect("Task stopped without shutdown called");
             log::debug!("TopologyRefresher: Exiting");
         })
     }
 
-    #[cfg(target_arch = "wasm32")]
     pub fn start(mut self) {
-        use futures::StreamExt;
-
         spawn_future(async move {
+            #[cfg(not(target_arch = "wasm32"))]
+            let mut interval = tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(
+                self.refresh_rate,
+            ));
+
+            #[cfg(target_arch = "wasm32")]
             let mut interval =
                 gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
-            while let Some(_) = interval.next().await {
+
+            while (interval.next().await).is_some() {
                 self.refresh().await;
             }
         })

clients/client-core/src/config/mod.rs

@@ -34,7 +34,7 @@ pub fn missing_string_value() -> String {
     MISSING_VALUE.to_string()
 }
 
-#[derive(Debug, Deserialize, PartialEq, Serialize)]
+#[derive(Debug, Clone, Deserialize, PartialEq, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct Config<T> {
     client: Client<T>,
@@ -42,17 +42,23 @@ pub struct Config<T> {
     #[serde(default)]
     logging: Logging,
     #[serde(default)]
-    debug: Debug,
+    debug: DebugConfig,
 }
 
-impl<T: NymConfig> Config<T> {
+impl<T> Config<T> {
-    pub fn new<S: Into<String>>(id: S) -> Self {
+    pub fn new<S: Into<String>>(id: S) -> Self
+    where
+        T: NymConfig,
+    {
         let mut cfg = Config::default();
         cfg.with_id(id);
         cfg
     }
 
-    pub fn with_id<S: Into<String>>(&mut self, id: S) {
+    pub fn with_id<S: Into<String>>(&mut self, id: S)
+    where
+        T: NymConfig,
+    {
         let id = id.into();
 
         // identity key setting
@@ -117,7 +123,7 @@ impl<T: NymConfig> Config<T> {
         self.client.disabled_credentials_mode = disabled_credentials_mode;
     }
 
-    pub fn with_gateway_endpoint(&mut self, gateway_endpoint: GatewayEndpoint) {
+    pub fn with_gateway_endpoint(&mut self, gateway_endpoint: GatewayEndpointConfig) {
         self.client.gateway_endpoint = gateway_endpoint;
     }
 
@@ -203,7 +209,11 @@ impl<T: NymConfig> Config<T> {
         self.client.gateway_endpoint.gateway_listener.clone()
     }
 
-    pub fn get_gateway_endpoint(&self) -> &GatewayEndpoint {
+    pub fn get_gateway_endpoint_config(&self) -> &GatewayEndpointConfig {
+        &self.client.gateway_endpoint
+    }
+
+    pub fn get_gateway_endpoint(&self) -> &GatewayEndpointConfig {
         &self.client.gateway_endpoint
     }
 
@@ -212,6 +222,10 @@ impl<T: NymConfig> Config<T> {
     }
 
     // Debug getters
+    pub fn get_debug_config(&self) -> &DebugConfig {
+        &self.debug
+    }
+
     pub fn get_average_packet_delay(&self) -> Duration {
         self.debug.average_packet_delay
     }
@@ -257,7 +271,7 @@ impl<T: NymConfig> Config<T> {
     }
 
     pub fn get_use_extended_packet_size(&self) -> Option<ExtendedPacketSize> {
-        self.debug.use_extended_packet_size.clone()
+        self.debug.use_extended_packet_size
     }
 
     pub fn get_version(&self) -> &str {
@@ -277,7 +291,7 @@ impl<T: NymConfig> Default for Config<T> {
 
 #[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
 #[cfg_attr(target_arch = "wasm32", wasm_bindgen(getter_with_clone))]
-pub struct GatewayEndpoint {
+pub struct GatewayEndpointConfig {
     /// gateway_id specifies ID of the gateway to which the client should send messages.
     /// If initially omitted, a random gateway will be chosen from the available topology.
     pub gateway_id: String,
@@ -289,10 +303,10 @@ pub struct GatewayEndpoint {
     pub gateway_listener: String,
 }
 
-impl From<topology::gateway::Node> for GatewayEndpoint {
+impl From<topology::gateway::Node> for GatewayEndpointConfig {
-    fn from(node: topology::gateway::Node) -> GatewayEndpoint {
+    fn from(node: topology::gateway::Node) -> GatewayEndpointConfig {
         let gateway_listener = node.clients_address();
-        GatewayEndpoint {
+        GatewayEndpointConfig {
             gateway_id: node.identity_key.to_base58_string(),
             gateway_owner: node.owner,
             gateway_listener,
@@ -300,7 +314,7 @@ impl From<topology::gateway::Node> for GatewayEndpoint {
     }
 }
 
-#[derive(Debug, Deserialize, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, Deserialize, PartialEq, Eq, Serialize)]
 pub struct Client<T> {
     /// Version of the client for which this configuration was created.
     #[serde(default = "missing_string_value")]
@@ -346,7 +360,7 @@ pub struct Client<T> {
     reply_encryption_key_store_path: PathBuf,
 
     /// Information regarding how the client should send data to gateway.
-    gateway_endpoint: GatewayEndpoint,
+    gateway_endpoint: GatewayEndpointConfig,
 
     /// Path to the database containing bandwidth credentials of this client.
     database_path: PathBuf,
@@ -416,13 +430,13 @@ impl<T: NymConfig> Client<T> {
     }
 }
 
-#[derive(Debug, Default, Deserialize, PartialEq, Eq, Serialize)]
+#[derive(Debug, Clone, Default, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct Logging {}
 
-#[derive(Debug, Deserialize, PartialEq, Serialize)]
+#[derive(Debug, Clone, Deserialize, PartialEq, Serialize)]
 #[serde(default, deny_unknown_fields)]
-pub struct Debug {
+pub struct DebugConfig {
     /// The parameter of Poisson distribution determining how long, on average,
     /// sent packet is going to be delayed at any given mix node.
     /// So for a packet going through three mix nodes, on average, it will take three times this value
@@ -488,7 +502,7 @@ pub struct Debug {
     pub use_extended_packet_size: Option<ExtendedPacketSize>,
 }
 
-#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
 pub enum ExtendedPacketSize {
     Extended8,
@@ -496,9 +510,9 @@ pub enum ExtendedPacketSize {
     Extended32,
 }
 
-impl Default for Debug {
+impl Default for DebugConfig {
     fn default() -> Self {
-        Debug {
+        DebugConfig {
             average_packet_delay: DEFAULT_AVERAGE_PACKET_DELAY,
             average_ack_delay: DEFAULT_AVERAGE_PACKET_DELAY,
             ack_wait_multiplier: DEFAULT_ACK_WAIT_MULTIPLIER,

clients/client-core/src/error.rs

@@ -1,6 +1,8 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+#[cfg(feature = "reply-surb")]
+use crate::client::reply_key_storage::ReplyKeyStorageError;
 use crypto::asymmetric::identity::Ed25519RecoveryError;
 use gateway_client::error::GatewayClientError;
 use validator_client::ValidatorClientError;
@@ -16,6 +18,10 @@ pub enum ClientCoreError {
     #[error("Validator client error: {0}")]
     ValidatorClientError(#[from] ValidatorClientError),
 
+    #[cfg(feature = "reply-surb")]
+    #[error("Reply key storage error: {0}")]
+    ReplyKeyStorageError(#[from] ReplyKeyStorageError),
+
     #[error("No gateway with id: {0}")]
     NoGatewayWithId(String),
     #[error("No gateways on network")]

clients/client-core/src/init.rs

@@ -90,7 +90,6 @@ async fn register_with_gateway(
         gateway.owner.clone(),
         our_identity.clone(),
         timeout,
-        #[cfg(not(target_arch = "wasm32"))]
         None,
     );
     gateway_client

clients/native/Cargo.toml

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

clients/native/src/client/config/mod.rs

@@ -27,6 +27,10 @@ impl SocketType {
             _ => SocketType::None,
         }
     }
+
+    pub fn is_websocket(&self) -> bool {
+        matches!(self, SocketType::WebSocket)
+    }
 }
 
 #[derive(Debug, Default, Deserialize, PartialEq, Serialize)]

clients/native/src/client/mod.rs

@@ -1,205 +1,46 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use client_connections::{
+use crate::client::config::Config;
-    ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths, TransmissionLane,
-};
-use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
-use client_core::client::inbound_messages::{
-    InputMessage, InputMessageReceiver, InputMessageSender,
-};
-use client_core::client::key_manager::KeyManager;
-use client_core::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
-use client_core::client::real_messages_control;
-use client_core::client::real_messages_control::RealMessagesController;
-use client_core::client::received_buffer::{
-    ReceivedBufferMessage, ReceivedBufferRequestReceiver, ReceivedBufferRequestSender,
-    ReceivedMessagesBufferController, ReconstructedMessagesReceiver,
-};
-use client_core::client::reply_key_storage::ReplyKeyStorage;
-use client_core::client::topology_control::{
-    TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
-};
-use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
-use client_core::error::ClientCoreError;
-use crypto::asymmetric::identity;
-use futures::channel::mpsc;
-use gateway_client::bandwidth::BandwidthController;
-use gateway_client::{
-    AcknowledgementReceiver, AcknowledgementSender, GatewayClient, MixnetMessageReceiver,
-    MixnetMessageSender,
-};
-use log::*;
-use nymsphinx::addressing::clients::Recipient;
-use nymsphinx::addressing::nodes::NodeIdentity;
-use nymsphinx::anonymous_replies::ReplySurb;
-use nymsphinx::receiver::ReconstructedMessage;
-use tap::TapFallible;
-use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
-
-use crate::client::config::{Config, SocketType};
 use crate::error::ClientError;
 use crate::websocket;
+use client_connections::TransmissionLane;
+use client_core::client::base_client::{BaseClientBuilder, ClientInput, ClientOutput};
+use client_core::client::inbound_messages::InputMessage;
+use client_core::client::key_manager::KeyManager;
+use client_core::client::received_buffer::{ReceivedBufferMessage, ReconstructedMessagesReceiver};
+use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
+use futures::channel::mpsc;
+use gateway_client::bandwidth::BandwidthController;
+use log::*;
+use nymsphinx::addressing::clients::Recipient;
+use nymsphinx::anonymous_replies::ReplySurb;
+use nymsphinx::receiver::ReconstructedMessage;
+use task::{wait_for_signal, ShutdownNotifier};
 
 pub(crate) mod config;
 
-pub struct NymClient {
+pub struct SocketClient {
     /// Client configuration options, including, among other things, packet sending rates,
     /// key filepaths, etc.
     config: Config,
 
     /// KeyManager object containing smart pointers to all relevant keys used by the client.
     key_manager: KeyManager,
-
-    /// Channel used for transforming 'raw' messages into sphinx packets and sending them
-    /// through the mix network.
-    /// It is only available if the client started with the websocket listener disabled.
-    input_tx: Option<InputMessageSender>,
-
-    /// Channel used for obtaining reconstructed messages received from the mix network.
-    /// It is only available if the client started with the websocket listener disabled.
-    receive_tx: Option<ReconstructedMessagesReceiver>,
 }
 
-impl NymClient {
+impl SocketClient {
     pub fn new(config: Config) -> Self {
         let pathfinder = ClientKeyPathfinder::new_from_config(config.get_base());
         let key_manager = KeyManager::load_keys(&pathfinder).expect("failed to load stored keys");
 
-        NymClient {
+        SocketClient {
             config,
             key_manager,
-            input_tx: None,
-            receive_tx: None,
         }
     }
 
-    pub fn as_mix_recipient(&self) -> Recipient {
+    async fn create_bandwidth_controller(config: &Config) -> BandwidthController {
-        Recipient::new(
-            *self.key_manager.identity_keypair().public_key(),
-            *self.key_manager.encryption_keypair().public_key(),
-            // TODO: below only works under assumption that gateway address == gateway id
-            // (which currently is true)
-            NodeIdentity::from_base58_string(self.config.get_base().get_gateway_id()).unwrap(),
-        )
-    }
-
-    // future constantly pumping loop cover traffic at some specified average rate
-    // the pumped traffic goes to the MixTrafficController
-    fn start_cover_traffic_stream(
-        &self,
-        topology_accessor: TopologyAccessor,
-        mix_tx: BatchMixMessageSender,
-        shutdown: ShutdownListener,
-    ) {
-        info!("Starting loop cover traffic stream...");
-
-        let mut stream = LoopCoverTrafficStream::new(
-            self.key_manager.ack_key(),
-            self.config.get_base().get_average_ack_delay(),
-            self.config.get_base().get_average_packet_delay(),
-            self.config
-                .get_base()
-                .get_loop_cover_traffic_average_delay(),
-            mix_tx,
-            self.as_mix_recipient(),
-            topology_accessor,
-        );
-
-        if let Some(size) = self.config.get_base().get_use_extended_packet_size() {
-            log::debug!("Setting extended packet size: {:?}", size);
-            stream.set_custom_packet_size(size.into());
-        }
-
-        stream.start_with_shutdown(shutdown);
-    }
-
-    #[allow(clippy::too_many_arguments)]
-    fn start_real_traffic_controller(
-        &self,
-        topology_accessor: TopologyAccessor,
-        reply_key_storage: ReplyKeyStorage,
-        ack_receiver: AcknowledgementReceiver,
-        input_receiver: InputMessageReceiver,
-        mix_sender: BatchMixMessageSender,
-        lane_queue_lengths: LaneQueueLengths,
-        client_connection_rx: ConnectionCommandReceiver,
-        shutdown: ShutdownListener,
-    ) {
-        let mut controller_config = real_messages_control::Config::new(
-            self.key_manager.ack_key(),
-            self.config.get_base().get_ack_wait_multiplier(),
-            self.config.get_base().get_ack_wait_addition(),
-            self.config.get_base().get_average_ack_delay(),
-            self.config.get_base().get_message_sending_average_delay(),
-            self.config.get_base().get_average_packet_delay(),
-            self.config
-                .get_base()
-                .get_disabled_main_poisson_packet_distribution(),
-            self.as_mix_recipient(),
-        );
-
-        if let Some(size) = self.config.get_base().get_use_extended_packet_size() {
-            log::debug!("Setting extended packet size: {:?}", size);
-            controller_config.set_custom_packet_size(size.into());
-        }
-
-        info!("Starting real traffic stream...");
-
-        RealMessagesController::new(
-            controller_config,
-            ack_receiver,
-            input_receiver,
-            mix_sender,
-            topology_accessor,
-            reply_key_storage,
-            lane_queue_lengths,
-            client_connection_rx,
-        )
-        .start_with_shutdown(shutdown);
-    }
-
-    // buffer controlling all messages fetched from provider
-    // required so that other components would be able to use them (say the websocket)
-    fn start_received_messages_buffer_controller(
-        &self,
-        query_receiver: ReceivedBufferRequestReceiver,
-        mixnet_receiver: MixnetMessageReceiver,
-        reply_key_storage: ReplyKeyStorage,
-        shutdown: ShutdownListener,
-    ) {
-        info!("Starting received messages buffer controller...");
-        ReceivedMessagesBufferController::new(
-            self.key_manager.encryption_keypair(),
-            query_receiver,
-            mixnet_receiver,
-            reply_key_storage,
-        )
-        .start_with_shutdown(shutdown)
-    }
-
-    async fn start_gateway_client(
-        &mut self,
-        mixnet_message_sender: MixnetMessageSender,
-        ack_sender: AcknowledgementSender,
-        shutdown: ShutdownListener,
-    ) -> GatewayClient {
-        let gateway_id = self.config.get_base().get_gateway_id();
-        if gateway_id.is_empty() {
-            panic!("The identity of the gateway is unknown - did you run `nym-client` init?")
-        }
-        let gateway_owner = self.config.get_base().get_gateway_owner();
-        if gateway_owner.is_empty() {
-            panic!("The owner of the gateway is unknown - did you run `nym-client` init?")
-        }
-        let gateway_address = self.config.get_base().get_gateway_listener();
-        if gateway_address.is_empty() {
-            panic!("The address of the gateway is unknown - did you run `nym-client` init?")
-        }
-
-        let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
-            .expect("provided gateway id is invalid!");
-
         #[cfg(feature = "coconut")]
         let bandwidth_controller = {
             let details = network_defaults::NymNetworkDetails::new_from_env();
@@ -212,163 +53,48 @@ impl NymClient {
                     .await
                     .expect("Could not query api clients");
             BandwidthController::new(
-                credential_storage::initialise_storage(self.config.get_base().get_database_path())
+                credential_storage::initialise_storage(config.get_base().get_database_path()).await,
-                    .await,
                 coconut_api_clients,
             )
         };
         #[cfg(not(feature = "coconut"))]
         let bandwidth_controller = BandwidthController::new(
-            credential_storage::initialise_storage(self.config.get_base().get_database_path())
+            credential_storage::initialise_storage(config.get_base().get_database_path()).await,
-                .await,
         )
         .expect("Could not create bandwidth controller");
-
+        bandwidth_controller
-        let mut gateway_client = GatewayClient::new(
-            gateway_address,
-            self.key_manager.identity_keypair(),
-            gateway_identity,
-            gateway_owner,
-            Some(self.key_manager.gateway_shared_key()),
-            mixnet_message_sender,
-            ack_sender,
-            self.config.get_base().get_gateway_response_timeout(),
-            Some(bandwidth_controller),
-            Some(shutdown),
-        );
-
-        gateway_client
-            .set_disabled_credentials_mode(self.config.get_base().get_disabled_credentials_mode());
-
-        gateway_client
-            .authenticate_and_start()
-            .await
-            .expect("could not authenticate and start up the gateway connection");
-
-        gateway_client
-    }
-
-    // future responsible for periodically polling directory server and updating
-    // the current global view of topology
-    async fn start_topology_refresher(
-        &mut self,
-        topology_accessor: TopologyAccessor,
-        shutdown: ShutdownListener,
-    ) -> Result<(), ClientError> {
-        let topology_refresher_config = TopologyRefresherConfig::new(
-            self.config.get_base().get_validator_api_endpoints(),
-            self.config.get_base().get_topology_refresh_rate(),
-            env!("CARGO_PKG_VERSION").to_string(),
-        );
-        let mut topology_refresher =
-            TopologyRefresher::new(topology_refresher_config, topology_accessor);
-        // before returning, block entire runtime to refresh the current network view so that any
-        // components depending on topology would see a non-empty view
-        info!("Obtaining initial network topology");
-        topology_refresher.refresh().await;
-
-        // TODO: a slightly more graceful termination here
-        if !topology_refresher.is_topology_routable().await {
-            log::error!(
-                "The current network topology seem to be insufficient to route any packets through \
-                - check if enough nodes and a gateway are online"
-            );
-            return Err(ClientCoreError::InsufficientNetworkTopology.into());
-        }
-
-        info!("Starting topology refresher...");
-        topology_refresher.start_with_shutdown(shutdown);
-        Ok(())
-    }
-
-    // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
-    // TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
-    // over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
-    // requests?
-    fn start_mix_traffic_controller(
-        gateway_client: GatewayClient,
-        shutdown: ShutdownListener,
-    ) -> BatchMixMessageSender {
-        info!("Starting mix traffic controller...");
-        let (mix_traffic_controller, mix_tx) = MixTrafficController::new(gateway_client);
-        mix_traffic_controller.start_with_shutdown(shutdown);
-        mix_tx
     }
 
     fn start_websocket_listener(
-        &self,
+        config: &Config,
-        buffer_requester: ReceivedBufferRequestSender,
+        client_input: ClientInput,
-        msg_input: InputMessageSender,
+        client_output: ClientOutput,
-        shared_lane_queue_lengths: LaneQueueLengths,
+        self_address: Recipient,
-        client_connection_tx: ConnectionCommandSender,
     ) {
         info!("Starting websocket listener...");
 
+        let ClientInput {
+            shared_lane_queue_lengths,
+            connection_command_sender,
+            input_sender,
+        } = client_input;
+
+        let received_buffer_request_sender = client_output.received_buffer_request_sender;
+
         let websocket_handler = websocket::Handler::new(
-            msg_input,
+            input_sender,
-            client_connection_tx,
+            connection_command_sender,
-            buffer_requester,
+            received_buffer_request_sender,
-            &self.as_mix_recipient(),
+            self_address,
             shared_lane_queue_lengths,
         );
 
-        websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler);
+        websocket::Listener::new(config.get_listening_port()).start(websocket_handler);
     }
 
-    /// EXPERIMENTAL DIRECT RUST API
+    /// blocking version of `start_socket` method. Will run forever (or until SIGINT is sent)
-    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
+    pub async fn run_socket_forever(self) -> Result<(), ClientError> {
-    /// well enough in local tests)
+        let shutdown = self.start_socket().await?;
-    pub async fn send_message(
-        &mut self,
-        recipient: Recipient,
-        message: Vec<u8>,
-        with_reply_surb: bool,
-    ) {
-        let lane = TransmissionLane::General;
-        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane);
-
-        self.input_tx
-            .as_ref()
-            .expect("start method was not called before!")
-            .send(input_msg)
-            .await
-            .expect("InputMessageReceiver has stopped receiving!");
-    }
-
-    /// EXPERIMENTAL DIRECT RUST API
-    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
-    /// well enough in local tests)
-    pub async fn send_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) {
-        let input_msg = InputMessage::new_reply(reply_surb, message);
-
-        self.input_tx
-            .as_ref()
-            .expect("start method was not called before!")
-            .send(input_msg)
-            .await
-            .expect("InputMessageReceiver has stopped receiving!");
-    }
-
-    /// EXPERIMENTAL DIRECT RUST API
-    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
-    /// well enough in local tests)
-    /// Note: it waits for the first occurrence of messages being sent to ourselves. If you expect multiple
-    /// messages, you might have to call this function repeatedly.
-    // TODO: I guess this should really return something that `impl Stream<Item=ReconstructedMessage>`
-    pub async fn wait_for_messages(&mut self) -> Vec<ReconstructedMessage> {
-        use futures::StreamExt;
-
-        self.receive_tx
-            .as_mut()
-            .expect("start method was not called before!")
-            .next()
-            .await
-            .expect("buffer controller seems to have somehow died!")
-    }
-
-    /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
-    pub async fn run_forever(&mut self) -> Result<(), ClientError> {
-        let shutdown = self.start().await?;
         wait_for_signal().await;
 
         println!(
@@ -388,117 +114,117 @@ impl NymClient {
         Ok(())
     }
 
-    pub async fn start(&mut self) -> Result<ShutdownNotifier, ClientError> {
+    pub async fn start_socket(self) -> Result<ShutdownNotifier, ClientError> {
-        info!("Starting nym client");
+        if !self.config.get_socket_type().is_websocket() {
-        // channels for inter-component communication
+            return Err(ClientError::InvalidSocketMode);
-        // TODO: make the channels be internally created by the relevant components
-        // rather than creating them here, so say for example the buffer controller would create the request channels
-        // and would allow anyone to clone the sender channel
-
-        // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway
-        // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer
-        let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
-
-        // used for announcing connection or disconnection of a channel for pushing re-assembled messages to
-        let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
-
-        // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
-
-        // channels responsible for controlling ack messages
-        let (ack_sender, ack_receiver) = mpsc::unbounded();
-        let shared_topology_accessor = TopologyAccessor::new();
-
-        let reply_key_storage =
-            ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .tap_err(|err| {
-                    log::error!("Failed to load reply key storage - is it perhaps already in use?");
-                    log::error!("{}", err);
-                })?;
-
-        // Shutdown notifier for signalling tasks to stop
-        let shutdown = ShutdownNotifier::default();
-
-        // the components are started in very specific order. Unless you know what you are doing,
-        // do not change that.
-        self.start_topology_refresher(shared_topology_accessor.clone(), shutdown.subscribe())
-            .await?;
-        self.start_received_messages_buffer_controller(
-            received_buffer_request_receiver,
-            mixnet_messages_receiver,
-            reply_key_storage.clone(),
-            shutdown.subscribe(),
-        );
-
-        let gateway_client = self
-            .start_gateway_client(mixnet_messages_sender, ack_sender, shutdown.subscribe())
-            .await;
-
-        // The sphinx_message_sender is the transmitter for any component generating sphinx packets
-        // that are to be sent to the mixnet. They are used by cover traffic stream and real
-        // traffic stream.
-        // The MixTrafficController then sends the actual traffic
-        let sphinx_message_sender =
-            Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe());
-
-        // Channels that the websocket listener can use to signal downstream to the real traffic
-        // controller that connections are closed.
-        let (client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections (e.g socks5 for attached network-requesters)
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
-        self.start_real_traffic_controller(
-            shared_topology_accessor.clone(),
-            reply_key_storage,
-            ack_receiver,
-            input_receiver,
-            sphinx_message_sender.clone(),
-            shared_lane_queue_lengths.clone(),
-            client_connection_rx,
-            shutdown.subscribe(),
-        );
-
-        if !self
-            .config
-            .get_base()
-            .get_disabled_loop_cover_traffic_stream()
-        {
-            self.start_cover_traffic_stream(
-                shared_topology_accessor,
-                sphinx_message_sender,
-                shutdown.subscribe(),
-            );
         }
 
-        match self.config.get_socket_type() {
+        let base_builder = BaseClientBuilder::new_from_base_config(
-            SocketType::WebSocket => self.start_websocket_listener(
+            self.config.get_base(),
-                received_buffer_request_sender,
+            self.key_manager,
-                input_sender,
+            Some(Self::create_bandwidth_controller(&self.config).await),
-                shared_lane_queue_lengths,
+        );
-                client_connection_tx,
-            ),
-            SocketType::None => {
-                // if we did not start the socket, it means we're running (supposedly) in the native mode
-                // and hence we should announce 'ourselves' to the buffer
-                let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded();
 
-                // tell the buffer to start sending stuff to us
+        let self_address = base_builder.as_mix_recipient();
-                received_buffer_request_sender
+        let mut started_client = base_builder.start_base().await?;
-                    .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
+        let client_input = started_client.client_input.register_producer();
-                        reconstructed_sender,
+        let client_output = started_client.client_output.register_consumer();
-                    ))
-                    .expect("the buffer request failed!");
 
-                self.receive_tx = Some(reconstructed_receiver);
+        Self::start_websocket_listener(&self.config, client_input, client_output, self_address);
-                self.input_tx = Some(input_sender);
-            }
-        }
 
         info!("Client startup finished!");
-        info!("The address of this client is: {}", self.as_mix_recipient());
+        info!("The address of this client is: {}", self_address);
 
-        Ok(shutdown)
+        Ok(started_client.shutdown_notifier)
+    }
+
+    pub async fn start_direct(self) -> Result<DirectClient, ClientError> {
+        if self.config.get_socket_type().is_websocket() {
+            return Err(ClientError::InvalidSocketMode);
+        }
+
+        let base_client = BaseClientBuilder::new_from_base_config(
+            self.config.get_base(),
+            self.key_manager,
+            Some(Self::create_bandwidth_controller(&self.config).await),
+        );
+
+        let mut started_client = base_client.start_base().await?;
+        let client_input = started_client.client_input.register_producer();
+        let client_output = started_client.client_output.register_consumer();
+
+        // register our receiver
+        let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded();
+
+        // tell the buffer to start sending stuff to us
+        client_output
+            .received_buffer_request_sender
+            .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
+                reconstructed_sender,
+            ))
+            .expect("the buffer request failed!");
+
+        Ok(DirectClient {
+            client_input,
+            reconstructed_receiver,
+            _shutdown_notifier: started_client.shutdown_notifier,
+        })
+    }
+}
+
+pub struct DirectClient {
+    client_input: ClientInput,
+    reconstructed_receiver: ReconstructedMessagesReceiver,
+
+    // we need to keep reference to this guy otherwise things will start dropping
+    _shutdown_notifier: ShutdownNotifier,
+}
+
+impl DirectClient {
+    /// EXPERIMENTAL DIRECT RUST API
+    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
+    /// well enough in local tests)
+    pub async fn send_message(
+        &mut self,
+        recipient: Recipient,
+        message: Vec<u8>,
+        with_reply_surb: bool,
+    ) {
+        let lane = TransmissionLane::General;
+        let input_msg = InputMessage::new_fresh(recipient, message, with_reply_surb, lane);
+
+        self.client_input
+            .input_sender
+            .send(input_msg)
+            .await
+            .expect("InputMessageReceiver has stopped receiving!");
+    }
+
+    /// EXPERIMENTAL DIRECT RUST API
+    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
+    /// well enough in local tests)
+    pub async fn send_reply(&mut self, reply_surb: ReplySurb, message: Vec<u8>) {
+        let input_msg = InputMessage::new_reply(reply_surb, message);
+
+        self.client_input
+            .input_sender
+            .send(input_msg)
+            .await
+            .expect("InputMessageReceiver has stopped receiving!");
+    }
+
+    /// EXPERIMENTAL DIRECT RUST API
+    /// It's untested and there are absolutely no guarantees about it (but seems to have worked
+    /// well enough in local tests)
+    /// Note: it waits for the first occurrence of messages being sent to ourselves. If you expect multiple
+    /// messages, you might have to call this function repeatedly.
+    // TODO: I guess this should really return something that `impl Stream<Item=ReconstructedMessage>`
+    pub async fn wait_for_messages(&mut self) -> Vec<ReconstructedMessage> {
+        use futures::StreamExt;
+
+        self.reconstructed_receiver
+            .next()
+            .await
+            .expect("buffer controller seems to have somehow died!")
     }
 }

clients/native/src/commands/init.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::Args;
-use client_core::{config::GatewayEndpoint, error::ClientCoreError};
+use client_core::{config::GatewayEndpointConfig, error::ClientCoreError};
 use config::NymConfig;
 
 use crate::{
@@ -132,7 +132,7 @@ async fn setup_gateway(
     register: bool,
     user_chosen_gateway_id: Option<&str>,
     config: &Config,
-) -> Result<GatewayEndpoint, ClientCoreError> {
+) -> Result<GatewayEndpointConfig, ClientCoreError> {
     if register {
         // Get the gateway details by querying the validator-api. Either pick one at random or use
         // the chosen one if it's among the available ones.

clients/native/src/commands/run.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::{
-    client::{config::Config, NymClient},
+    client::{config::Config, SocketClient},
     commands::{override_config, OverrideConfig},
     error::ClientError,
 };
@@ -98,5 +98,5 @@ pub(crate) async fn execute(args: &Run) -> Result<(), ClientError> {
         return Err(ClientError::FailedLocalVersionCheck);
     }
 
-    NymClient::new(config).run_forever().await
+    SocketClient::new(config).run_socket_forever().await
 }

clients/native/src/error.rs

@@ -1,25 +1,15 @@
-use client_core::{client::reply_key_storage::ReplyKeyStorageError, error::ClientCoreError};
+use client_core::error::ClientCoreError;
-use crypto::asymmetric::identity::Ed25519RecoveryError;
-use gateway_client::error::GatewayClientError;
-use validator_client::ValidatorClientError;
 
 #[derive(thiserror::Error, Debug)]
 pub enum ClientError {
-    #[error("I/O error: {0}")]
-    IoError(#[from] std::io::Error),
-    #[error("Gateway client error: {0}")]
-    GatewayClientError(#[from] GatewayClientError),
-    #[error("Ed25519 error: {0}")]
-    Ed25519RecoveryError(#[from] Ed25519RecoveryError),
-    #[error("Validator client error: {0}")]
-    ValidatorClientError(#[from] ValidatorClientError),
     #[error("client-core error: {0}")]
     ClientCoreError(#[from] ClientCoreError),
-    #[error("Reply key storage error: {0}")]
-    ReplyKeyStorageError(#[from] ReplyKeyStorageError),
 
     #[error("Failed to load config for: {0}")]
     FailedToLoadConfig(String),
     #[error("Failed local version check, client and config mismatch")]
     FailedLocalVersionCheck,
+
+    #[error("Attempted to start the client in invalid socket mode")]
+    InvalidSocketMode,
 }

clients/native/src/websocket/handler.rs

@@ -73,14 +73,14 @@ impl Handler {
         msg_input: InputMessageSender,
         client_connection_tx: ConnectionCommandSender,
         buffer_requester: ReceivedBufferRequestSender,
-        self_full_address: &Recipient,
+        self_full_address: Recipient,
         lane_queue_lengths: LaneQueueLengths,
     ) -> Self {
         Handler {
             msg_input,
             client_connection_tx,
             buffer_requester,
-            self_full_address: *self_full_address,
+            self_full_address,
             socket: None,
             received_response_type: Default::default(),
             lane_queue_lengths,

clients/socks5/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-socks5-client"
-version = "1.1.0"
+version = "1.1.1"
 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"

clients/socks5/src/client/mod.rs

@@ -1,43 +1,21 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use std::error::Error;
-
 use crate::client::config::Config;
 use crate::error::Socks5ClientError;
 use crate::socks::{
     authentication::{AuthenticationMethods, Authenticator, User},
     server::SphinxSocksServer,
 };
-use client_connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
+use client_core::client::base_client::{BaseClientBuilder, ClientInput, ClientOutput};
-use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
-use client_core::client::inbound_messages::{
-    InputMessage, InputMessageReceiver, InputMessageSender,
-};
 use client_core::client::key_manager::KeyManager;
-use client_core::client::mix_traffic::{BatchMixMessageSender, MixTrafficController};
-use client_core::client::real_messages_control::RealMessagesController;
-use client_core::client::received_buffer::{
-    ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController,
-};
-use client_core::client::reply_key_storage::ReplyKeyStorage;
-use client_core::client::topology_control::{
-    TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
-};
 use client_core::config::persistence::key_pathfinder::ClientKeyPathfinder;
-use client_core::error::ClientCoreError;
-use crypto::asymmetric::identity;
 use futures::channel::mpsc;
 use futures::StreamExt;
 use gateway_client::bandwidth::BandwidthController;
-use gateway_client::{
-    AcknowledgementReceiver, AcknowledgementSender, GatewayClient, MixnetMessageReceiver,
-    MixnetMessageSender,
-};
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
-use nymsphinx::addressing::nodes::NodeIdentity;
+use std::error::Error;
-use tap::TapFallible;
 use task::{wait_for_signal_and_error, ShutdownListener, ShutdownNotifier};
 
 pub mod config;
@@ -72,132 +50,7 @@ impl NymClient {
         }
     }
 
-    pub fn as_mix_recipient(&self) -> Recipient {
+    async fn create_bandwidth_controller(config: &Config) -> BandwidthController {
-        Recipient::new(
-            *self.key_manager.identity_keypair().public_key(),
-            *self.key_manager.encryption_keypair().public_key(),
-            // TODO: below only works under assumption that gateway address == gateway id
-            // (which currently is true)
-            NodeIdentity::from_base58_string(self.config.get_base().get_gateway_id()).unwrap(),
-        )
-    }
-
-    // future constantly pumping loop cover traffic at some specified average rate
-    // the pumped traffic goes to the MixTrafficController
-    fn start_cover_traffic_stream(
-        &self,
-        topology_accessor: TopologyAccessor,
-        mix_tx: BatchMixMessageSender,
-        shutdown: ShutdownListener,
-    ) {
-        info!("Starting loop cover traffic stream...");
-
-        let mut stream = LoopCoverTrafficStream::new(
-            self.key_manager.ack_key(),
-            self.config.get_base().get_average_ack_delay(),
-            self.config.get_base().get_average_packet_delay(),
-            self.config
-                .get_base()
-                .get_loop_cover_traffic_average_delay(),
-            mix_tx,
-            self.as_mix_recipient(),
-            topology_accessor,
-        );
-
-        if let Some(size) = self.config.get_base().get_use_extended_packet_size() {
-            log::debug!("Setting extended packet size: {:?}", size);
-            stream.set_custom_packet_size(size.into());
-        }
-
-        stream.start_with_shutdown(shutdown);
-    }
-
-    #[allow(clippy::too_many_arguments)]
-    fn start_real_traffic_controller(
-        &self,
-        topology_accessor: TopologyAccessor,
-        reply_key_storage: ReplyKeyStorage,
-        ack_receiver: AcknowledgementReceiver,
-        input_receiver: InputMessageReceiver,
-        mix_sender: BatchMixMessageSender,
-        client_connection_rx: ConnectionCommandReceiver,
-        lane_queue_lengths: LaneQueueLengths,
-        shutdown: ShutdownListener,
-    ) {
-        let mut controller_config = client_core::client::real_messages_control::Config::new(
-            self.key_manager.ack_key(),
-            self.config.get_base().get_ack_wait_multiplier(),
-            self.config.get_base().get_ack_wait_addition(),
-            self.config.get_base().get_average_ack_delay(),
-            self.config.get_base().get_message_sending_average_delay(),
-            self.config.get_base().get_average_packet_delay(),
-            self.config
-                .get_base()
-                .get_disabled_main_poisson_packet_distribution(),
-            self.as_mix_recipient(),
-        );
-
-        if let Some(size) = self.config.get_base().get_use_extended_packet_size() {
-            log::debug!("Setting extended packet size: {:?}", size);
-            controller_config.set_custom_packet_size(size.into());
-        }
-
-        info!("Starting real traffic stream...");
-
-        RealMessagesController::new(
-            controller_config,
-            ack_receiver,
-            input_receiver,
-            mix_sender,
-            topology_accessor,
-            reply_key_storage,
-            lane_queue_lengths,
-            client_connection_rx,
-        )
-        .start_with_shutdown(shutdown);
-    }
-
-    // buffer controlling all messages fetched from provider
-    // required so that other components would be able to use them (say the websocket)
-    fn start_received_messages_buffer_controller(
-        &self,
-        query_receiver: ReceivedBufferRequestReceiver,
-        mixnet_receiver: MixnetMessageReceiver,
-        reply_key_storage: ReplyKeyStorage,
-        shutdown: ShutdownListener,
-    ) {
-        info!("Starting received messages buffer controller...");
-        ReceivedMessagesBufferController::new(
-            self.key_manager.encryption_keypair(),
-            query_receiver,
-            mixnet_receiver,
-            reply_key_storage,
-        )
-        .start_with_shutdown(shutdown);
-    }
-
-    async fn start_gateway_client(
-        &mut self,
-        mixnet_message_sender: MixnetMessageSender,
-        ack_sender: AcknowledgementSender,
-        shutdown: ShutdownListener,
-    ) -> GatewayClient {
-        let gateway_id = self.config.get_base().get_gateway_id();
-        if gateway_id.is_empty() {
-            panic!("The identity of the gateway is unknown - did you run `nym-client` init?")
-        }
-        let gateway_owner = self.config.get_base().get_gateway_owner();
-        if gateway_owner.is_empty() {
-            panic!("The owner of the gateway is unknown - did you run `nym-client` init?")
-        }
-        let gateway_address = self.config.get_base().get_gateway_listener();
-        if gateway_address.is_empty() {
-            panic!("The address of the gateway is unknown - did you run `nym-client` init?")
-        }
-
-        let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
-            .expect("provided gateway id is invalid!");
-
         #[cfg(feature = "coconut")]
         let bandwidth_controller = {
             let details = network_defaults::NymNetworkDetails::new_from_env();
@@ -210,108 +63,44 @@ impl NymClient {
                     .await
                     .expect("Could not query api clients");
             BandwidthController::new(
-                credential_storage::initialise_storage(self.config.get_base().get_database_path())
+                credential_storage::initialise_storage(config.get_base().get_database_path()).await,
-                    .await,
                 coconut_api_clients,
             )
         };
         #[cfg(not(feature = "coconut"))]
         let bandwidth_controller = BandwidthController::new(
-            credential_storage::initialise_storage(self.config.get_base().get_database_path())
+            credential_storage::initialise_storage(config.get_base().get_database_path()).await,
-                .await,
         )
         .expect("Could not create bandwidth controller");
-
+        bandwidth_controller
-        let mut gateway_client = GatewayClient::new(
-            gateway_address,
-            self.key_manager.identity_keypair(),
-            gateway_identity,
-            gateway_owner,
-            Some(self.key_manager.gateway_shared_key()),
-            mixnet_message_sender,
-            ack_sender,
-            self.config.get_base().get_gateway_response_timeout(),
-            Some(bandwidth_controller),
-            Some(shutdown),
-        );
-
-        gateway_client
-            .set_disabled_credentials_mode(self.config.get_base().get_disabled_credentials_mode());
-
-        gateway_client
-            .authenticate_and_start()
-            .await
-            .expect("could not authenticate and start up the gateway connection");
-
-        gateway_client
-    }
-
-    // future responsible for periodically polling directory server and updating
-    // the current global view of topology
-    async fn start_topology_refresher(
-        &mut self,
-        topology_accessor: TopologyAccessor,
-        shutdown: ShutdownListener,
-    ) -> Result<(), Socks5ClientError> {
-        let topology_refresher_config = TopologyRefresherConfig::new(
-            self.config.get_base().get_validator_api_endpoints(),
-            self.config.get_base().get_topology_refresh_rate(),
-            env!("CARGO_PKG_VERSION").to_string(),
-        );
-        let mut topology_refresher =
-            TopologyRefresher::new(topology_refresher_config, topology_accessor);
-        // before returning, block entire runtime to refresh the current network view so that any
-        // components depending on topology would see a non-empty view
-        info!("Obtaining initial network topology");
-        topology_refresher.refresh().await;
-
-        // TODO: a slightly more graceful termination here
-        if !topology_refresher.is_topology_routable().await {
-            log::error!(
-                "The current network topology seem to be insufficient to route any packets through \
-                - check if enough nodes and a gateway are online"
-            );
-            return Err(ClientCoreError::InsufficientNetworkTopology.into());
-        }
-
-        info!("Starting topology refresher...");
-        topology_refresher.start_with_shutdown(shutdown);
-        Ok(())
-    }
-
-    // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
-    // TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
-    // over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
-    // requests?
-    fn start_mix_traffic_controller(
-        gateway_client: GatewayClient,
-        shutdown: ShutdownListener,
-    ) -> BatchMixMessageSender {
-        info!("Starting mix traffic controller...");
-        let (mix_traffic_controller, mix_tx) = MixTrafficController::new(gateway_client);
-        mix_traffic_controller.start_with_shutdown(shutdown);
-        mix_tx
     }
 
     fn start_socks5_listener(
-        &self,
+        config: &Config,
-        buffer_requester: ReceivedBufferRequestSender,
+        client_input: ClientInput,
-        msg_input: InputMessageSender,
+        client_output: ClientOutput,
-        client_connection_tx: ConnectionCommandSender,
+        self_address: Recipient,
-        lane_queue_lengths: LaneQueueLengths,
         mut shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
         let auth_methods = vec![AuthenticationMethods::NoAuth as u8];
         let allowed_users: Vec<User> = Vec::new();
 
+        let ClientInput {
+            shared_lane_queue_lengths,
+            connection_command_sender,
+            input_sender,
+        } = client_input;
+
+        let received_buffer_request_sender = client_output.received_buffer_request_sender;
+
         let authenticator = Authenticator::new(auth_methods, allowed_users);
         let mut sphinx_socks = SphinxSocksServer::new(
-            self.config.get_listening_port(),
+            config.get_listening_port(),
             authenticator,
-            self.config.get_provider_mix_address(),
+            config.get_provider_mix_address(),
-            self.as_mix_recipient(),
+            self_address,
-            lane_queue_lengths,
+            shared_lane_queue_lengths,
             shutdown.clone(),
         );
         tokio::spawn(async move {
@@ -323,7 +112,11 @@ impl NymClient {
             // TODO: replace this by a generic solution, such as a task manager that stores all
             // JoinHandles of all spawned tasks.
             if let Err(res) = sphinx_socks
-                .serve(msg_input, buffer_requester, client_connection_tx)
+                .serve(
+                    input_sender,
+                    received_buffer_request_sender,
+                    connection_command_sender,
+                )
                 .await
             {
                 shutdown.send_we_stopped(Box::new(res));
@@ -332,7 +125,7 @@ impl NymClient {
     }
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
-    pub async fn run_forever(&mut self) -> Result<(), Box<dyn Error + Send>> {
+    pub async fn run_forever(self) -> Result<(), Box<dyn Error + Send>> {
         let mut shutdown = self
             .start()
             .await
@@ -352,7 +145,7 @@ impl NymClient {
 
     // Variant of `run_forever` that listends for remote control messages
     pub async fn run_and_listen(
-        &mut self,
+        self,
         mut receiver: Socks5ControlMessageReceiver,
     ) -> Result<(), Box<dyn Error + Send>> {
         // Start the main task
@@ -395,101 +188,29 @@ impl NymClient {
         res
     }
 
-    pub async fn start(&mut self) -> Result<ShutdownNotifier, Socks5ClientError> {
+    pub async fn start(self) -> Result<ShutdownNotifier, Socks5ClientError> {
-        info!("Starting nym client");
+        let base_builder = BaseClientBuilder::new_from_base_config(
-        // channels for inter-component communication
+            self.config.get_base(),
-        // TODO: make the channels be internally created by the relevant components
+            self.key_manager,
-        // rather than creating them here, so say for example the buffer controller would create the request channels
+            Some(Self::create_bandwidth_controller(&self.config).await),
-        // and would allow anyone to clone the sender channel
-
-        // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway
-        // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer
-        let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
-
-        // used for announcing connection or disconnection of a channel for pushing re-assembled messages to
-        let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
-
-        // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
-
-        // channels responsible for controlling ack messages
-        let (ack_sender, ack_receiver) = mpsc::unbounded();
-        let shared_topology_accessor = TopologyAccessor::new();
-
-        let reply_key_storage =
-            ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
-                .tap_err(|err| {
-                    log::error!("Failed to load reply key storage - is it perhaps already in use?");
-                    log::error!("{}", err);
-                })?;
-
-        // Shutdown notifier for signalling tasks to stop
-        let shutdown = ShutdownNotifier::default();
-
-        // the components are started in very specific order. Unless you know what you are doing,
-        // do not change that.
-        self.start_topology_refresher(shared_topology_accessor.clone(), shutdown.subscribe())
-            .await?;
-        self.start_received_messages_buffer_controller(
-            received_buffer_request_receiver,
-            mixnet_messages_receiver,
-            reply_key_storage.clone(),
-            shutdown.subscribe(),
         );
 
-        let gateway_client = self
+        let self_address = base_builder.as_mix_recipient();
-            .start_gateway_client(mixnet_messages_sender, ack_sender, shutdown.subscribe())
+        let mut started_client = base_builder.start_base().await?;
-            .await;
+        let client_input = started_client.client_input.register_producer();
+        let client_output = started_client.client_output.register_consumer();
 
-        // The sphinx_message_sender is the transmitter for any component generating sphinx packets
+        Self::start_socks5_listener(
-        // that are to be sent to the mixnet. They are used by cover traffic stream and real
+            &self.config,
-        // traffic stream.
+            client_input,
-        // The MixTrafficController then sends the actual traffic
+            client_output,
-        let sphinx_message_sender =
+            self_address,
-            Self::start_mix_traffic_controller(gateway_client, shutdown.subscribe());
+            started_client.shutdown_notifier.subscribe(),
-
-        // Channel for announcing closed (socks5) connections by the controller.
-        // This will be forwarded to `OutQueueControl`
-        let (client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
-        self.start_real_traffic_controller(
-            shared_topology_accessor.clone(),
-            reply_key_storage,
-            ack_receiver,
-            input_receiver,
-            sphinx_message_sender.clone(),
-            client_connection_rx,
-            shared_lane_queue_lengths.clone(),
-            shutdown.subscribe(),
-        );
-
-        if !self
-            .config
-            .get_base()
-            .get_disabled_loop_cover_traffic_stream()
-        {
-            self.start_cover_traffic_stream(
-                shared_topology_accessor,
-                sphinx_message_sender,
-                shutdown.subscribe(),
-            );
-        }
-
-        self.start_socks5_listener(
-            received_buffer_request_sender,
-            input_sender,
-            client_connection_tx,
-            shared_lane_queue_lengths,
-            shutdown.subscribe(),
         );
 
         info!("Client startup finished!");
-        info!("The address of this client is: {}", self.as_mix_recipient());
+        info!("The address of this client is: {}", self_address);
 
-        Ok(shutdown)
+        Ok(started_client.shutdown_notifier)
     }
 }

clients/socks5/src/commands/init.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::Args;
-use client_core::{config::GatewayEndpoint, error::ClientCoreError};
+use client_core::{config::GatewayEndpointConfig, error::ClientCoreError};
 use config::NymConfig;
 
 use crate::{
@@ -131,7 +131,7 @@ async fn setup_gateway(
     register: bool,
     user_chosen_gateway_id: Option<&str>,
     config: &Config,
-) -> Result<GatewayEndpoint, ClientCoreError> {
+) -> Result<GatewayEndpointConfig, ClientCoreError> {
     if register {
         // Get the gateway details by querying the validator-api. Either pick one at random or use
         // the chosen one if it's among the available ones.

clients/socks5/src/error.rs

@@ -1,7 +1,4 @@
-use client_core::{client::reply_key_storage::ReplyKeyStorageError, error::ClientCoreError};
+use client_core::error::ClientCoreError;
-use crypto::asymmetric::identity::Ed25519RecoveryError;
-use gateway_client::error::GatewayClientError;
-use validator_client::ValidatorClientError;
 
 use crate::socks::types::SocksProxyError;
 
@@ -9,16 +6,8 @@ use crate::socks::types::SocksProxyError;
 pub enum Socks5ClientError {
     #[error("I/O error: {0}")]
     IoError(#[from] std::io::Error),
-    #[error("Gateway client error: {0}")]
-    GatewayClientError(#[from] GatewayClientError),
-    #[error("Ed25519 error: {0}")]
-    Ed25519RecoveryError(#[from] Ed25519RecoveryError),
-    #[error("Validator client error: {0}")]
-    ValidatorClientError(#[from] ValidatorClientError),
     #[error("client-core error: {0}")]
     ClientCoreError(#[from] ClientCoreError),
-    #[error("Reply key storage error: {0}")]
-    ReplyKeyStorageError(#[from] ReplyKeyStorageError),
 
     #[error("SOCKS proxy error")]
     SocksProxyError(SocksProxyError),

clients/socks5/src/socks/client.rs

@@ -2,8 +2,8 @@
 
 use super::authentication::{AuthenticationMethods, Authenticator, User};
 use super::request::{SocksCommand, SocksRequest};
-use super::types::{ResponseCode, SocksProxyError};
+use super::types::{ResponseCodeV4, ResponseCodeV5, SocksProxyError};
-use super::{RESERVED, SOCKS_VERSION};
+use super::{SocksVersion, RESERVED, SOCKS4_VERSION, SOCKS5_VERSION};
 use client_connections::{LaneQueueLengths, TransmissionLane};
 use client_core::client::inbound_messages::{InputMessage, InputMessageSender};
 use futures::channel::mpsc;
@@ -129,13 +129,13 @@ impl AsyncWrite for StreamState {
 /// A client connecting to the Socks proxy server, because
 /// it wants to make a Nym-protected outbound request. Typically, this is
 /// something like e.g. a wallet app running on your laptop connecting to
-/// SphinxSocksServer.
+/// `SphinxSocksServer`.
 pub(crate) struct SocksClient {
     controller_sender: ControllerSender,
     stream: StreamState,
     auth_nmethods: u8,
     authenticator: Authenticator,
-    socks_version: u8,
+    socks_version: Option<SocksVersion>,
     input_sender: InputMessageSender,
     connection_id: ConnectionId,
     service_provider: Recipient,
@@ -158,29 +158,32 @@ impl Drop for SocksClient {
 }
 
 impl SocksClient {
-    /// Create a new SOCKClient
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         stream: TcpStream,
         authenticator: Authenticator,
         input_sender: InputMessageSender,
-        service_provider: Recipient,
+        service_provider: &Recipient,
         controller_sender: ControllerSender,
-        self_address: Recipient,
+        self_address: &Recipient,
         lane_queue_lengths: LaneQueueLengths,
-        shutdown_listener: ShutdownListener,
+        mut shutdown_listener: ShutdownListener,
     ) -> Self {
+        // If this task fails and exits, we don't want to send shutdown signal
+        shutdown_listener.mark_as_success();
+
         let connection_id = Self::generate_random();
+
         SocksClient {
             controller_sender,
             connection_id,
             stream: StreamState::Available(stream),
             auth_nmethods: 0,
-            socks_version: 0,
+            socks_version: None,
             authenticator,
             input_sender,
-            service_provider,
+            service_provider: *service_provider,
-            self_address,
+            self_address: *self_address,
             started_proxy: false,
             lane_queue_lengths,
             shutdown_listener,
@@ -192,13 +195,45 @@ impl SocksClient {
         rng.next_u64()
     }
 
+    pub async fn send_error(&mut self, err: SocksProxyError) -> Result<(), SocksProxyError> {
+        let error_text = format!("{}", err);
+        let Some(ref version) = self.socks_version else {
+            log::error!("Trying to send error without knowing the version");
+            return Ok(());
+        };
+
+        match version {
+            SocksVersion::V4 => {
+                let response = ResponseCodeV4::RequestRejected;
+                self.send_error_v4(response).await
+            }
+            SocksVersion::V5 => {
+                let response = if error_text.contains("Host") {
+                    ResponseCodeV5::HostUnreachable
+                } else if error_text.contains("Network") {
+                    ResponseCodeV5::NetworkUnreachable
+                } else if error_text.contains("ttl") {
+                    ResponseCodeV5::TtlExpired
+                } else {
+                    ResponseCodeV5::Failure
+                };
+                self.send_error_v5(response).await
+            }
+        }
+    }
+
     // Send an error back to the client
-    pub async fn error(&mut self, r: ResponseCode) -> Result<(), SocksProxyError> {
+    pub async fn send_error_v4(&mut self, r: ResponseCodeV4) -> Result<(), SocksProxyError> {
-        self.stream.write_all(&[5, r as u8]).await?;
+        self.stream.write_all(&[SOCKS4_VERSION, r as u8]).await?;
         Ok(())
     }
 
-    /// Shutdown the TcpStream to the client and end the session
+    pub async fn send_error_v5(&mut self, r: ResponseCodeV5) -> Result<(), SocksProxyError> {
+        self.stream.write_all(&[SOCKS5_VERSION, r as u8]).await?;
+        Ok(())
+    }
+
+    /// Shutdown the `TcpStream` to the client and end the session
     pub async fn shutdown(&mut self) -> Result<(), SocksProxyError> {
         info!("client is shutting down its TCP stream");
         self.stream.shutdown().await?;
@@ -210,25 +245,27 @@ impl SocksClient {
     /// is in use and that the client is authenticated, then runs the request.
     pub async fn run(&mut self) -> Result<(), SocksProxyError> {
         debug!("New connection from: {}", self.stream.peer_addr()?.ip());
-        let mut header = [0u8; 2];
+
         // Read a byte from the stream and determine the version being requested
+        let mut header = [0u8];
         self.stream.read_exact(&mut header).await?;
 
-        self.socks_version = header[0];
+        self.socks_version = match SocksVersion::try_from(header[0]) {
-        self.auth_nmethods = header[1];
+            Ok(version) => Some(version),
+            Err(_err) => {
+                warn!("Init: Unsupported version: SOCKS{}", header[0]);
+                return self.shutdown().await;
+            }
+        };
 
-        // Handle SOCKS4 requests
+        if self.socks_version == Some(SocksVersion::V5) {
-        if header[0] != SOCKS_VERSION {
+            let mut auth = [0u8];
-            warn!("Init: Unsupported version: SOCKS{}", self.socks_version);
+            self.stream.read_exact(&mut auth).await?;
-            self.shutdown().await
+            self.auth_nmethods = auth[0];
-        }
+            self.authenticate_socks5().await?;
-        // Valid SOCKS5
-        else {
-            // Authenticate w/ client
-            self.authenticate().await?;
-            // Handle requests
-            self.handle_request().await
         }
+
+        self.handle_request().await
     }
 
     async fn send_connect_to_mixnet(&mut self, remote_address: RemoteAddress) {
@@ -252,10 +289,15 @@ impl SocksClient {
             .await;
 
         let stream = self.stream.run_proxy();
-        let local_stream_remote = stream
+        let peer_addr = match stream.peer_addr() {
-            .peer_addr()
+            Ok(peer_addr) => peer_addr,
-            .expect("failed to extract peer address")
+            Err(err) => {
-            .to_string();
+                log::error!("Unable to extract the remote peer address: {err}");
+                return;
+            }
+        };
+        let local_stream_remote = peer_addr.to_string();
+
         let connection_id = self.connection_id;
         let input_sender = self.input_sender.clone();
 
@@ -286,8 +328,17 @@ impl SocksClient {
     async fn handle_request(&mut self) -> Result<(), SocksProxyError> {
         debug!("Handling CONNECT Command");
 
-        let request = SocksRequest::from_stream(&mut self.stream).await?;
+        let version = self
-        let remote_address = request.to_string();
+            .socks_version
+            .as_ref()
+            .expect("Must read version before parsing request");
+
+        let request = match version {
+            SocksVersion::V4 => SocksRequest::from_stream_socks4(&mut self.stream).await?,
+            SocksVersion::V5 => SocksRequest::from_stream_socks5(&mut self.stream).await?,
+        };
+
+        let remote_address = request.address_string();
 
         // setup for receiving from the mixnet
         let (mix_sender, mix_receiver) = mpsc::unbounded();
@@ -296,7 +347,10 @@ impl SocksClient {
             // Use the Proxy to connect to the specified addr/port
             SocksCommand::Connect => {
                 trace!("Connecting to: {:?}", remote_address.clone());
-                self.acknowledge_socks5().await;
+                match version {
+                    SocksVersion::V4 => self.acknowledge_socks4().await,
+                    SocksVersion::V5 => self.acknowledge_socks5().await,
+                }
 
                 self.started_proxy = true;
                 self.controller_sender
@@ -319,7 +373,6 @@ impl SocksClient {
             SocksCommand::UdpAssociate => unimplemented!(), // not handled
         };
 
-        self.shutdown_listener.mark_as_success();
         Ok(())
     }
 
@@ -328,8 +381,8 @@ impl SocksClient {
     async fn acknowledge_socks5(&mut self) {
         self.stream
             .write_all(&[
-                SOCKS_VERSION,
+                SOCKS5_VERSION,
-                ResponseCode::Success as u8,
+                ResponseCodeV5::Success as u8,
                 RESERVED,
                 1,
                 127,
@@ -343,13 +396,30 @@ impl SocksClient {
             .unwrap();
     }
 
+    /// Writes a Socks4 header back to the requesting client's TCP stream,
+    async fn acknowledge_socks4(&mut self) {
+        self.stream
+            .write_all(&[
+                0, //SOCKS4_VERSION,
+                ResponseCodeV4::Granted as u8,
+                0,
+                0,
+                127,
+                0,
+                0,
+                1,
+            ])
+            .await
+            .unwrap();
+    }
+
     /// Authenticate the incoming request. Each request is checked for its
     /// authentication method. A user/password request will extract the
     /// username and password from the stream, then check with the Authenticator
     /// to see if the resulting user is allowed.
     ///
     /// A lot of this could probably be put into the `SocksRequest::from_stream()`
-    /// constructor, and/or cleaned up with tokio::codec. It's mostly just
+    /// constructor, and/or cleaned up with `tokio::codec`. It's mostly just
     /// read-a-byte-or-two. The bytes being extracted look like this:
     ///
     /// +----+------+----------+------+------------+
@@ -361,7 +431,7 @@ impl SocksClient {
     /// Pulling out the stream code into its own home, and moving the if/else logic
     /// into the Authenticator (where it'll be more easily testable)
     /// would be a good next step.
-    async fn authenticate(&mut self) -> Result<(), SocksProxyError> {
+    async fn authenticate_socks5(&mut self) -> Result<(), SocksProxyError> {
         debug!("Authenticating w/ {}", self.stream.peer_addr()?.ip());
         // Get valid auth methods
         let methods = self.get_available_methods().await?;
@@ -370,7 +440,7 @@ impl SocksClient {
         let mut response = [0u8; 2];
 
         // Set the version in the response
-        response[0] = SOCKS_VERSION;
+        response[0] = SOCKS5_VERSION;
         if methods.contains(&(AuthenticationMethods::UserPass as u8)) {
             // Set the default auth method (NO AUTH)
             response[1] = AuthenticationMethods::UserPass as u8;
@@ -406,11 +476,11 @@ impl SocksClient {
             // Authenticate passwords
             if self.authenticator.is_allowed(&user) {
                 debug!("Access Granted. User: {}", user.username);
-                let response = [1, ResponseCode::Success as u8];
+                let response = [1, ResponseCodeV5::Success as u8];
                 self.stream.write_all(&response).await?;
             } else {
                 debug!("Access Denied. User: {}", user.username);
-                let response = [1, ResponseCode::Failure as u8];
+                let response = [1, ResponseCodeV5::Failure as u8];
                 self.stream.write_all(&response).await?;
 
                 // Shutdown
@@ -429,7 +499,7 @@ impl SocksClient {
             response[1] = AuthenticationMethods::NoMethods as u8;
             self.stream.write_all(&response).await?;
             self.shutdown().await?;
-            Err(ResponseCode::Failure.into())
+            Err(ResponseCodeV5::Failure.into())
         }
     }
 

clients/socks5/src/socks/mixnet_responses.rs

@@ -105,6 +105,7 @@ impl MixnetResponseListener {
                 }
             }
         }
+        #[cfg(not(target_arch = "wasm32"))]
         tokio::time::timeout(Duration::from_secs(5), self.shutdown.recv())
             .await
             .expect("Task stopped without shutdown called");

clients/socks5/src/socks/mod.rs

@@ -1,5 +1,9 @@
 #![forbid(unsafe_code)]
 
+use std::convert::TryFrom;
+
+use self::types::SocksProxyError;
+
 pub mod authentication;
 mod client;
 pub(crate) mod mixnet_responses;
@@ -9,6 +13,27 @@ pub mod types;
 pub mod utils;
 
 /// Version of socks
-const SOCKS_VERSION: u8 = 0x05;
+const SOCKS4_VERSION: u8 = 0x04;
+const SOCKS5_VERSION: u8 = 0x05;
 
 const RESERVED: u8 = 0x00;
+
+#[derive(Clone, PartialEq, Eq)]
+pub enum SocksVersion {
+    V4 = 0x04,
+    V5 = 0x05,
+}
+
+pub struct InvalidSocksVersion;
+
+impl TryFrom<u8> for SocksVersion {
+    type Error = SocksProxyError;
+
+    fn try_from(version: u8) -> Result<Self, Self::Error> {
+        match version {
+            SOCKS4_VERSION => Ok(Self::V4),
+            SOCKS5_VERSION => Ok(Self::V5),
+            _ => Err(SocksProxyError::UnsupportedProxyVersion(version)),
+        }
+    }
+}

clients/socks5/src/socks/request.rs

@@ -1,5 +1,7 @@
-use super::types::{AddrType, ResponseCode, SocksProxyError};
+use crate::socks::SOCKS4_VERSION;
-use super::{utils as socks_utils, SOCKS_VERSION};
+
+use super::types::{AddrType, ResponseCodeV5, SocksProxyError};
+use super::{utils as socks_utils, SOCKS5_VERSION};
 use log::*;
 use std::fmt::{self, Display};
 use tokio::io::{AsyncRead, AsyncReadExt};
@@ -15,80 +17,114 @@ pub(crate) struct SocksRequest {
 }
 
 impl SocksRequest {
-    /// Parse a SOCKS5 request from a TcpStream
+    /// Parse a SOCKS4 request from a `TcpStream`
-    pub async fn from_stream<R>(stream: &mut R) -> Result<Self, SocksProxyError>
+    /// From documents at:
+    ///  - SOCKS4: https://www.openssh.com/txt/socks4.protocol
+    ///  - SOCKS4a: https://www.openssh.com/txt/socks4a.protocol
+    pub async fn from_stream_socks4<R>(stream: &mut R) -> Result<Self, SocksProxyError>
     where
         R: AsyncRead + Unpin,
     {
+        log::trace!("read from stream socks4");
+
+        let mut packet = [0u8; 3];
+        stream.read_exact(&mut packet).await?;
+
+        // CD (command)
+        let Some(command) = SocksCommand::from(packet[0] as usize) else {
+            log::warn!("Invalid Command");
+            return Err(ResponseCodeV5::CommandNotSupported.into());
+        };
+
+        // DSTPORT
+        let mut port = [0u8; 2];
+        port.copy_from_slice(&packet[1..]);
+        let port = merge_u8_into_u16(port[0], port[1]);
+
+        // DSTIP
+        let mut ip = [0u8; 4];
+        stream.read_exact(&mut ip).await?;
+
+        // USERID
+        let _userid = read_until_zero(stream).await;
+
+        // SOCKS4a extension
+        // https://www.openssh.com/txt/socks4a.protocol
+        // If the IP is 0.0.0.x with x nonzero, read the domain name
+        let (addr, addr_type) = if ip[..3] == [0, 0, 0] && ip[3] != 0 {
+            (read_until_zero(stream).await?, AddrType::Domain)
+        } else {
+            (ip.to_vec(), AddrType::V4)
+        };
+
+        // Return parsed request
+        Ok(SocksRequest {
+            version: SOCKS4_VERSION,
+            command,
+            addr_type,
+            addr,
+            port,
+        })
+    }
+    /// Parse a SOCKS5 request from a `TcpStream`
+    /// From: https://www.rfc-editor.org/rfc/rfc1928
+    pub async fn from_stream_socks5<R>(stream: &mut R) -> Result<Self, SocksProxyError>
+    where
+        R: AsyncRead + Unpin,
+    {
+        log::info!("read from stream socks5");
+
         let mut packet = [0u8; 4];
         // Read a byte from the stream and determine the version being requested
         stream.read_exact(&mut packet).await?;
 
-        if packet[0] != SOCKS_VERSION {
+        // VER
-            warn!("from_stream Unsupported version: SOCKS{}", packet[0]);
+        if packet[0] != SOCKS5_VERSION {
+            warn!("Unsupported version: SOCKS{}", packet[0]);
             return Err(SocksProxyError::UnsupportedProxyVersion(packet[0]));
         }
 
-        // Get command
+        // CMD
-        let mut command: SocksCommand = SocksCommand::Connect;
+        let Some(command) = SocksCommand::from(packet[1] as usize) else {
-        match SocksCommand::from(packet[1] as usize) {
+            warn!("Invalid Command");
-            Some(com) => {
+            return Err(ResponseCodeV5::CommandNotSupported.into());
-                command = com;
+        };
-                Ok(())
-            }
-            None => {
-                warn!("Invalid Command");
-                Err(ResponseCode::CommandNotSupported)
-            }
-        }?;
 
-        // DST.address
+        // RSV
+        // packet[2] is reserved
 
-        let mut addr_type: AddrType = AddrType::V6;
+        // ATYP
-        match AddrType::from(packet[3] as usize) {
+        let Some(addr_type) = AddrType::from(packet[3] as usize) else {
-            Some(addr) => {
+            error!("No Addr");
-                addr_type = addr;
+            return Err(ResponseCodeV5::AddrTypeNotSupported.into())
-                Ok(())
+        };
-            }
-            None => {
-                error!("No Addr");
-                Err(ResponseCode::AddrTypeNotSupported)
-            }
-        }?;
 
-        trace!("Getting Addr");
+        // DST.ADDR
-        // Get Addr from addr_type and stream
+        let addr = match addr_type {
-        let addr: Result<Vec<u8>, SocksProxyError> = match addr_type {
             AddrType::Domain => {
-                let mut domain_length = [0u8; 1];
+                let mut domain_length = [0u8];
                 stream.read_exact(&mut domain_length).await?;
-
                 let mut domain = vec![0u8; domain_length[0] as usize];
                 stream.read_exact(&mut domain).await?;
-
+                domain
-                Ok(domain)
             }
             AddrType::V4 => {
                 let mut addr = [0u8; 4];
                 stream.read_exact(&mut addr).await?;
-                Ok(addr.to_vec())
+                addr.to_vec()
             }
             AddrType::V6 => {
                 let mut addr = [0u8; 16];
                 stream.read_exact(&mut addr).await?;
-                Ok(addr.to_vec())
+                addr.to_vec()
             }
         };
 
-        let addr = addr?;
+        // DST.PORT
-
-        // read DST.port
         let mut port = [0u8; 2];
         stream.read_exact(&mut port).await?;
-        // Merge two u8s into u16
+        let port = merge_u8_into_u16(port[0], port[1]);
-        let port = (u16::from(port[0]) << 8) | u16::from(port[1]);
 
-        // Return parsed request
         Ok(SocksRequest {
             version: packet[0],
             command,
@@ -97,14 +133,18 @@ impl SocksRequest {
             port,
         })
     }
+
+    /// Print out the address and port to a String.
+    /// This might return domain:port, ipv6:port, or ipv4:port.
+    pub fn address_string(&self) -> String {
+        let address = socks_utils::pretty_print_addr(&self.addr_type, &self.addr);
+        format!("{}:{}", address, self.port)
+    }
 }
 
 impl Display for SocksRequest {
-    /// Print out the address and port to a String.
-    /// This might return domain:port, ipv6:port, or ipv4:port.
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let address = socks_utils::pretty_print_addr(&self.addr_type, &self.addr);
+        write!(f, "{}", self.address_string())
-        write!(f, "{}:{}", address, self.port)
     }
 }
 
@@ -127,3 +167,23 @@ impl SocksCommand {
         }
     }
 }
+
+fn merge_u8_into_u16(a: u8, b: u8) -> u16 {
+    (u16::from(a) << 8) | u16::from(b)
+}
+
+async fn read_until_zero<R>(stream: &mut R) -> Result<Vec<u8>, SocksProxyError>
+where
+    R: AsyncRead + Unpin,
+{
+    let mut result = Vec::new();
+    let mut char = [0u8];
+    loop {
+        stream.read_exact(&mut char).await?;
+        if char[0] == 0 {
+            break;
+        }
+        result.push(char[0]);
+    }
+    Ok(result)
+}

clients/socks5/src/socks/server.rs

@@ -1,8 +1,8 @@
 use crate::error::Socks5ClientError;
 
-use super::authentication::Authenticator;
+use super::{
-use super::client::SocksClient;
+    authentication::Authenticator, client::SocksClient, mixnet_responses::MixnetResponseListener,
-use super::{mixnet_responses::MixnetResponseListener, types::ResponseCode};
+};
 use client_connections::{ConnectionCommandSender, LaneQueueLengths};
 use client_core::client::{
     inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender,
@@ -85,47 +85,26 @@ impl SphinxSocksServer {
         loop {
             tokio::select! {
                 Ok((stream, _remote)) = listener.accept() => {
-                    // TODO Optimize this
                     let mut client = SocksClient::new(
                         stream,
                         self.authenticator.clone(),
                         input_sender.clone(),
-                        self.service_provider,
+                        &self.service_provider,
                         controller_sender.clone(),
-                        self.self_address,
+                        &self.self_address,
                         self.lane_queue_lengths.clone(),
                         self.shutdown.clone(),
                     );
 
                     tokio::spawn(async move {
-                        {
+                        if let Err(err) = client.run().await {
-                            match client.run().await {
+                            error!("Error! {}", err);
-                                Ok(_) => {}
+                            if client.send_error(err).await.is_err() {
-                                Err(error) => {
+                                warn!("Failed to send error code");
-                                    error!("Error! {}", error);
+                            };
-                                    let error_text = format!("{}", error);
+                            if client.shutdown().await.is_err() {
-
+                                warn!("Failed to shutdown TcpStream");
-                                    let response: ResponseCode;
-
-                                    if error_text.contains("Host") {
-                                        response = ResponseCode::HostUnreachable;
-                                    } else if error_text.contains("Network") {
-                                        response = ResponseCode::NetworkUnreachable;
-                                    } else if error_text.contains("ttl") {
-                                        response = ResponseCode::TtlExpired
-                                    } else {
-                                        response = ResponseCode::Failure
-                                    }
-
-                                    if client.error(response).await.is_err() {
-                                        warn!("Failed to send error code");
-                                    };
-                                    if client.shutdown().await.is_err() {
-                                        warn!("Failed to shutdown TcpStream");
-                                    };
-                                }
                             };
-                            // client gets dropped here
                         }
                     });
                 },

clients/socks5/src/socks/types.rs

@@ -1,7 +1,17 @@
 use snafu::Snafu;
-#[derive(Debug, Snafu)]
+
+/// SOCKS4 Response codes
+#[allow(dead_code)]
+pub(crate) enum ResponseCodeV4 {
+    Granted = 0x5a,
+    RequestRejected = 0x5b,
+    CannotConnectToIdent = 0x5c,
+    DifferentUserId = 0x5d,
+}
+
 /// Possible SOCKS5 Response Codes
-pub(crate) enum ResponseCode {
+#[derive(Debug, Snafu)]
+pub(crate) enum ResponseCodeV5 {
     Success = 0x00,
     #[snafu(display("SOCKS5 Server Failure"))]
     Failure = 0x01,
@@ -48,7 +58,7 @@ where
 }
 
 /// DST.addr variant types
-#[derive(PartialEq)]
+#[derive(Debug, PartialEq)]
 pub(crate) enum AddrType {
     V4 = 0x01,
     Domain = 0x03,

clients/webassembly/Cargo.toml

@@ -39,7 +39,7 @@ topology = { path = "../../common/topology" }
 gateway-client = { path = "../../common/client-libs/gateway-client", default-features = false, features = ["wasm", "coconut"] }
 validator-client = { path = "../../common/client-libs/validator-client", default-features = false }
 wasm-utils = { path = "../../common/wasm-utils" }
-
+task = { path = "../../common/task" }
 # The `console_error_panic_hook` crate provides better debugging of panics by
 # logging them with `console.error`. This is great for development, but requires
 # all the `std::fmt` and `std::panicking` infrastructure, so isn't great for

clients/webassembly/js-example/package-lock.json

@@ -24,7 +24,7 @@
     },
     "../pkg": {
       "name": "@nymproject/nym-client-wasm",
-      "version": "1.0.0",
+      "version": "1.1.0",
       "license": "Apache-2.0"
     },
     "node_modules/@discoveryjs/json-ext": {

clients/webassembly/js-example/worker.js

@@ -61,16 +61,9 @@ async function main() {
   // sets up better stack traces in case of in-rust panics
   set_panic_hook();
 
-  console.error("the current mainnet is not compatible with v2! - either use the pre-merge branch or explicitly set the client to use one of V2 QA networks")
-  return
-
   // validator server we will use to get topology from
-  // MAINNET (V1):
   const validator = 'https://validator.nymtech.net/api'; //"http://localhost:8081";
   const preferredGateway = 'E3mvZTHQCdBvhfr178Swx9g4QG3kkRUun7YnToLMcMbM';
-  // QA (V2):
-  // const validator = 'https://qa-validator-api.nymtech.net/api'; //"http://localhost:8081";
-  // const preferredGateway = 'CgQrYP8etksSBf4nALNqp93SHPpgFwEUyTsjBNNLj5WM';
 
   const gatewayEndpoint = await get_gateway(validator, preferredGateway);
   gatewayEndpoint.gateway_listener = "wss://gateway1.nymtech.net:443"; // this is needed if we want it to work on the web. However this gateway is a v1 gateway, we will need to change for v2 once we get there

clients/webassembly/src/client/config.rs

@@ -4,12 +4,15 @@
 // due to expansion of #[wasm_bindgen] macro on `Debug` Config struct
 #![allow(clippy::drop_non_drop)]
 
-use client_core::config::{Debug as ConfigDebug, ExtendedPacketSize, GatewayEndpoint};
+use client_core::config::{DebugConfig as ConfigDebug, ExtendedPacketSize, GatewayEndpointConfig};
+use serde::{Deserialize, Serialize};
 use std::time::Duration;
 use url::Url;
 use wasm_bindgen::prelude::*;
 
 #[wasm_bindgen]
+#[derive(Debug, Deserialize, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct Config {
     /// ID specifies the human readable ID of this particular client.
     pub(crate) id: String,
@@ -19,7 +22,7 @@ pub struct Config {
     pub(crate) disabled_credentials_mode: bool,
 
     /// Information regarding how the client should send data to gateway.
-    pub(crate) gateway_endpoint: GatewayEndpoint,
+    pub(crate) gateway_endpoint: GatewayEndpointConfig,
 
     pub(crate) debug: ConfigDebug,
 }
@@ -30,7 +33,7 @@ impl Config {
     pub fn new(
         id: String,
         validator_server: String,
-        gateway_endpoint: GatewayEndpoint,
+        gateway_endpoint: GatewayEndpointConfig,
         debug: Option<Debug>,
     ) -> Self {
         Config {

clients/webassembly/src/client/mod.rs

@@ -2,51 +2,44 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use self::config::Config;
-use client_connections::{ConnectionCommandReceiver, LaneQueueLengths, TransmissionLane};
+use crate::client::response_pusher::ResponsePusher;
-use client_core::client::{
+use client_connections::TransmissionLane;
-    cover_traffic_stream::LoopCoverTrafficStream,
+use client_core::client::base_client::{BaseClientBuilder, ClientInput, ClientOutput};
-    inbound_messages::{InputMessage, InputMessageReceiver, InputMessageSender},
+use client_core::client::{inbound_messages::InputMessage, key_manager::KeyManager};
-    key_manager::KeyManager,
-    mix_traffic::{BatchMixMessageSender, MixTrafficController},
-    real_messages_control::{self, RealMessagesController},
-    received_buffer::{
-        ReceivedBufferMessage, ReceivedBufferRequestReceiver, ReceivedBufferRequestSender,
-        ReceivedMessagesBufferController,
-    },
-    topology_control::{TopologyAccessor, TopologyRefresher, TopologyRefresherConfig},
-};
 use crypto::asymmetric::identity;
-use futures::channel::mpsc;
-use futures::StreamExt;
-use gateway_client::{
-    AcknowledgementReceiver, AcknowledgementSender, GatewayClient, MixnetMessageReceiver,
-    MixnetMessageSender,
-};
 use nymsphinx::addressing::clients::Recipient;
 use rand::rngs::OsRng;
+use task::ShutdownNotifier;
 use wasm_bindgen::prelude::*;
-use wasm_bindgen_futures::spawn_local;
+use wasm_utils::{console_error, console_log};
-use wasm_utils::console_log;
 
 pub mod config;
+mod response_pusher;
 
 #[wasm_bindgen]
 pub struct NymClient {
     config: Config,
 
     /// KeyManager object containing smart pointers to all relevant keys used by the client.
-    key_manager: KeyManager,
+    // due to disgusting workaround I had to wrap the key_manager in an Option
+    // so that the interface wouldn't change (i.e. both `start` and `new` would still return a `NymClient`)
+    key_manager: Option<KeyManager>,
+    self_address: Option<String>,
 
     // TODO: this should be stored somewhere persistently
     // received_keys: HashSet<SURBEncryptionKey>,
     /// Channel used for transforming 'raw' messages into sphinx packets and sending them
     /// through the mix network.
-    input_tx: Option<InputMessageSender>,
+    client_input: Option<ClientInput>,
 
     // callbacks
     on_message: Option<js_sys::Function>,
     on_binary_message: Option<js_sys::Function>,
     on_gateway_connect: Option<js_sys::Function>,
+
+    // even though we don't use graceful shutdowns, other components rely on existence of this struct
+    // and if it's dropped, everything will start going offline
+    _shutdown: Option<ShutdownNotifier>,
 }
 
 #[wasm_bindgen]
@@ -55,14 +48,19 @@ impl NymClient {
     pub fn new(config: Config) -> Self {
         Self {
             config,
-            key_manager: Self::setup_key_manager(),
+            key_manager: Some(Self::setup_key_manager()),
             on_message: None,
             on_binary_message: None,
             on_gateway_connect: None,
-            input_tx: None,
+            client_input: None,
+            self_address: None,
+            _shutdown: None,
         }
     }
 
+    // TODO: once we make keys persistent, we'll require some kind of `init` method to generate
+    // a prior shared keypair between the client and the gateway
+
     // perhaps this should be public?
     fn setup_key_manager() -> KeyManager {
         let mut rng = OsRng;
@@ -84,298 +82,26 @@ impl NymClient {
     }
 
     fn as_mix_recipient(&self) -> Recipient {
+        // another disgusting (and hopefully temporary) workaround
+        let key_manager_ref = self
+            .key_manager
+            .as_ref()
+            .expect("attempting to call 'as_mix_recipient' after 'start'");
+
         Recipient::new(
-            *self.key_manager.identity_keypair().public_key(),
+            *key_manager_ref.identity_keypair().public_key(),
-            *self.key_manager.encryption_keypair().public_key(),
+            *key_manager_ref.encryption_keypair().public_key(),
             identity::PublicKey::from_base58_string(&self.config.gateway_endpoint.gateway_id)
                 .expect("no gateway has been selected"),
         )
     }
 
     pub fn self_address(&self) -> String {
-        self.as_mix_recipient().to_string()
+        if let Some(address) = &self.self_address {
-    }
+            address.clone()
-
+        } else {
-    // future constantly pumping loop cover traffic at some specified average rate
+            self.as_mix_recipient().to_string()
-    // the pumped traffic goes to the MixTrafficController
-    fn start_cover_traffic_stream(
-        &self,
-        topology_accessor: TopologyAccessor,
-        mix_tx: BatchMixMessageSender,
-    ) {
-        console_log!("Starting loop cover traffic stream...");
-
-        let mut stream = LoopCoverTrafficStream::new(
-            self.key_manager.ack_key(),
-            self.config.debug.average_ack_delay,
-            self.config.debug.average_packet_delay,
-            self.config.debug.loop_cover_traffic_average_delay,
-            mix_tx,
-            self.as_mix_recipient(),
-            topology_accessor,
-        );
-
-        if let Some(size) = &self.config.debug.use_extended_packet_size {
-            stream.set_custom_packet_size(size.clone().into());
         }
-
-        stream.start();
-    }
-
-    fn start_real_traffic_controller(
-        &self,
-        topology_accessor: TopologyAccessor,
-        ack_receiver: AcknowledgementReceiver,
-        input_receiver: InputMessageReceiver,
-        mix_sender: BatchMixMessageSender,
-        client_connection_rx: ConnectionCommandReceiver,
-        lane_queue_lengths: LaneQueueLengths,
-    ) {
-        let mut controller_config = real_messages_control::Config::new(
-            self.key_manager.ack_key(),
-            self.config.debug.ack_wait_multiplier,
-            self.config.debug.ack_wait_addition,
-            self.config.debug.average_ack_delay,
-            self.config.debug.message_sending_average_delay,
-            self.config.debug.average_packet_delay,
-            self.config.debug.disable_main_poisson_packet_distribution,
-            self.as_mix_recipient(),
-        );
-
-        if let Some(size) = &self.config.debug.use_extended_packet_size {
-            controller_config.set_custom_packet_size(size.clone().into());
-        }
-
-        console_log!("Starting real traffic stream...");
-
-        RealMessagesController::new(
-            controller_config,
-            ack_receiver,
-            input_receiver,
-            mix_sender,
-            topology_accessor,
-            lane_queue_lengths,
-            client_connection_rx,
-        )
-        .start();
-    }
-
-    // buffer controlling all messages fetched from provider
-    // required so that other components would be able to use them (say the websocket)
-    fn start_received_messages_buffer_controller(
-        &self,
-        query_receiver: ReceivedBufferRequestReceiver,
-        mixnet_receiver: MixnetMessageReceiver,
-    ) {
-        console_log!("Starting received messages buffer controller...");
-        ReceivedMessagesBufferController::new(
-            self.key_manager.encryption_keypair(),
-            query_receiver,
-            mixnet_receiver,
-        )
-        .start()
-    }
-
-    async fn start_gateway_client(
-        &mut self,
-        mixnet_message_sender: MixnetMessageSender,
-        ack_sender: AcknowledgementSender,
-    ) -> GatewayClient {
-        let gateway_id = self.config.gateway_endpoint.gateway_id.clone();
-        if gateway_id.is_empty() {
-            panic!("The identity of the gateway is unknown - did you run `get_gateway()`?")
-        }
-        let gateway_owner = self.config.gateway_endpoint.gateway_owner.clone();
-        if gateway_owner.is_empty() {
-            panic!("The owner of the gateway is unknown - did you run `get_gateway()`?")
-        }
-        let gateway_address = self.config.gateway_endpoint.gateway_listener.clone();
-        if gateway_address.is_empty() {
-            panic!("The address of the gateway is unknown - did you run `get_gateway()`?")
-        }
-
-        let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
-            .expect("provided gateway id is invalid!");
-
-        let mut gateway_client = GatewayClient::new(
-            gateway_address,
-            self.key_manager.identity_keypair(),
-            gateway_identity,
-            gateway_owner,
-            None,
-            mixnet_message_sender,
-            ack_sender,
-            self.config.debug.gateway_response_timeout,
-            None,
-        );
-
-        gateway_client.set_disabled_credentials_mode(self.config.disabled_credentials_mode);
-
-        let shared_keys = gateway_client
-            .authenticate_and_start()
-            .await
-            .expect("could not authenticate and start up the gateway connection");
-        self.key_manager.insert_gateway_shared_key(shared_keys);
-
-        match self.on_gateway_connect.as_ref() {
-            Some(callback) => {
-                callback
-                    .call0(&JsValue::null())
-                    .expect("on connect callback failed!");
-            }
-            None => console_log!("Gateway connection established - no callback specified"),
-        };
-
-        gateway_client
-    }
-
-    // future responsible for periodically polling directory server and updating
-    // the current global view of topology
-    async fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
-        let topology_refresher_config = TopologyRefresherConfig::new(
-            vec![self.config.validator_api_url.clone()],
-            self.config.debug.topology_refresh_rate,
-            env!("CARGO_PKG_VERSION").to_string(),
-        );
-        let mut topology_refresher =
-            TopologyRefresher::new(topology_refresher_config, topology_accessor);
-        // before returning, block entire runtime to refresh the current network view so that any
-        // components depending on topology would see a non-empty view
-        console_log!("Obtaining initial network topology");
-        topology_refresher.refresh().await;
-
-        // TODO: a slightly more graceful termination here
-        if !topology_refresher.is_topology_routable().await {
-            panic!(
-                "The current network topology seem to be insufficient to route any packets through\
-                - check if enough nodes and a gateway are online"
-            );
-        }
-
-        console_log!("Starting topology refresher...");
-
-        // TODO: re-enable
-        topology_refresher.start();
-    }
-
-    // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
-    // TODO: if we want to send control messages to gateway_client, this CAN'T take the ownership
-    // over it. Perhaps GatewayClient needs to be thread-shareable or have some channel for
-    // requests?
-    fn start_mix_traffic_controller(gateway_client: GatewayClient) -> BatchMixMessageSender {
-        console_log!("Starting mix traffic controller...");
-        let (mix_traffic_controller, mix_tx) = MixTrafficController::new(gateway_client);
-        mix_traffic_controller.start();
-        mix_tx
-    }
-
-    // TODO: this procedure is extremely overcomplicated, because it's based off native client's behaviour
-    // which doesn't fully apply in this case
-    fn start_reconstructed_pusher(
-        &mut self,
-        received_buffer_request_sender: ReceivedBufferRequestSender,
-    ) {
-        let on_message = self.on_message.take();
-        let on_binary_message = self.on_binary_message.take();
-
-        spawn_local(async move {
-            let (reconstructed_sender, mut reconstructed_receiver) = mpsc::unbounded();
-
-            // tell the buffer to start sending stuff to us
-            received_buffer_request_sender
-                .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
-                    reconstructed_sender,
-                ))
-                .expect("the buffer request failed!");
-
-            let this = JsValue::null();
-
-            while let Some(reconstructed) = reconstructed_receiver.next().await {
-                for msg in reconstructed {
-                    if let Some(ref callback_binary) = on_binary_message {
-                        let arg1 = serde_wasm_bindgen::to_value(&msg.message).unwrap();
-                        callback_binary
-                            .call1(&this, &arg1)
-                            .expect("on binary message failed!");
-                    }
-                    if let Some(ref callback) = on_message {
-                        if msg.reply_surb.is_some() {
-                            console_log!("the received message contained a reply-surb that we do not know how to handle (yet)")
-                        }
-                        let stringified = String::from_utf8_lossy(&msg.message).into_owned();
-                        let arg1 = serde_wasm_bindgen::to_value(&stringified).unwrap();
-                        callback.call1(&this, &arg1).expect("on message failed!");
-                    }
-                }
-            }
-        });
-    }
-
-    pub async fn start(mut self) -> NymClient {
-        console_log!("Starting wasm client '{}'", self.config.id);
-        // channels for inter-component communication
-        // TODO: make the channels be internally created by the relevant components
-        // rather than creating them here, so say for example the buffer controller would create the request channels
-        // and would allow anyone to clone the sender channel
-
-        // unwrapped_sphinx_sender is the transmitter of mixnet messages received from the gateway
-        // unwrapped_sphinx_receiver is the receiver for said messages - used by ReceivedMessagesBuffer
-        let (mixnet_messages_sender, mixnet_messages_receiver) = mpsc::unbounded();
-
-        // used for announcing connection or disconnection of a channel for pushing re-assembled messages to
-        let (received_buffer_request_sender, received_buffer_request_receiver) = mpsc::unbounded();
-
-        // channels responsible for controlling real messages
-        let (input_sender, input_receiver) = tokio::sync::mpsc::channel::<InputMessage>(1);
-
-        // channels responsible for controlling ack messages
-        let (ack_sender, ack_receiver) = mpsc::unbounded();
-        let shared_topology_accessor = TopologyAccessor::new();
-
-        // Channel that the real traffix controller can listed to for closing connections.
-        // Currently unused in the wasm client.
-        let (_client_connection_tx, client_connection_rx) = mpsc::unbounded();
-
-        // the components are started in very specific order. Unless you know what you are doing,
-        // do not change that.
-        self.start_topology_refresher(shared_topology_accessor.clone())
-            .await;
-        self.start_received_messages_buffer_controller(
-            received_buffer_request_receiver,
-            mixnet_messages_receiver,
-        );
-
-        let gateway_client = self
-            .start_gateway_client(mixnet_messages_sender, ack_sender)
-            .await;
-
-        // The sphinx_message_sender is the transmitter for any component generating sphinx packets
-        // that are to be sent to the mixnet. They are used by cover traffic stream and real
-        // traffic stream.
-        // The MixTrafficController then sends the actual traffic
-        let sphinx_message_sender = Self::start_mix_traffic_controller(gateway_client);
-
-        // Shared queue length data. Published by the `OutQueueController` in the client, and used
-        // primarily to throttle incoming connections
-        let shared_lane_queue_lengths = LaneQueueLengths::new();
-
-        self.start_real_traffic_controller(
-            shared_topology_accessor.clone(),
-            ack_receiver,
-            input_receiver,
-            sphinx_message_sender.clone(),
-            client_connection_rx,
-            shared_lane_queue_lengths,
-        );
-
-        if !self.config.debug.disable_loop_cover_traffic_stream {
-            self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
-        }
-
-        self.start_reconstructed_pusher(received_buffer_request_sender);
-        self.input_tx = Some(input_sender);
-
-        self
     }
 
     // Right now it's impossible to have async exported functions to take `&mut self` rather than mut self
@@ -395,13 +121,65 @@ impl NymClient {
 
         let input_msg = InputMessage::new_fresh(recipient, message, false, lane);
 
-        self.input_tx
+        self.client_input
             .as_ref()
             .expect("start method was not called before!")
+            .input_sender
             .send(input_msg)
             .await
             .expect("InputMessageReceiver has stopped receiving!");
 
         self
     }
+
+    fn start_reconstructed_pusher(
+        client_output: ClientOutput,
+        on_message: Option<js_sys::Function>,
+        on_binary_message: Option<js_sys::Function>,
+    ) {
+        ResponsePusher::new(client_output, on_message, on_binary_message).start()
+    }
+
+    pub async fn start(mut self) -> NymClient {
+        console_log!("Starting the wasm client");
+
+        let base_builder = BaseClientBuilder::new(
+            &self.config.gateway_endpoint,
+            &self.config.debug,
+            self.key_manager.take().unwrap(),
+            None,
+            true,
+            vec![self.config.validator_api_url.clone()],
+        );
+
+        self.self_address = Some(base_builder.as_mix_recipient().to_string());
+        let mut started_client = match base_builder.start_base().await {
+            Ok(base_client) => base_client,
+            Err(err) => {
+                console_error!("failed to start base client components - {}", err);
+                // proper error handling is left here as an exercise for the reader (hi Mark : ))
+                panic!("failed to start base client components - {err}")
+            }
+        };
+        match self.on_gateway_connect.as_ref() {
+            Some(callback) => {
+                callback
+                    .call0(&JsValue::null())
+                    .expect("on connect callback failed!");
+            }
+            None => console_log!("Gateway connection established - no callback specified"),
+        };
+
+        // those should be moved to a completely different struct, but I don't want to break compatibility for now
+        let client_input = started_client.client_input.register_producer();
+        let client_output = started_client.client_output.register_consumer();
+
+        let on_message = self.on_message.take();
+        let on_binary_message = self.on_binary_message.take();
+        Self::start_reconstructed_pusher(client_output, on_message, on_binary_message);
+        self.client_input = Some(client_input);
+        self._shutdown = Some(started_client.shutdown_notifier);
+
+        self
+    }
 }

clients/webassembly/src/client/response_pusher.rs

@@ -0,0 +1,71 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use client_core::client::base_client::ClientOutput;
+use client_core::client::received_buffer::{ReceivedBufferMessage, ReconstructedMessagesReceiver};
+use futures::channel::mpsc;
+use futures::StreamExt;
+use wasm_bindgen::JsValue;
+use wasm_bindgen_futures::spawn_local;
+use wasm_utils::console_log;
+
+pub(crate) struct ResponsePusher {
+    reconstructed_receiver: ReconstructedMessagesReceiver,
+    on_message: Option<js_sys::Function>,
+    on_binary_message: Option<js_sys::Function>,
+}
+
+impl ResponsePusher {
+    pub(crate) fn new(
+        client_output: ClientOutput,
+        on_message: Option<js_sys::Function>,
+        on_binary_message: Option<js_sys::Function>,
+    ) -> Self {
+        if on_message.is_none() && on_binary_message.is_none() {
+            // exercise for the reader : )
+            panic!("neither 'on_message' nor 'on_binary_message' was set!")
+        }
+
+        // register our output
+        let (reconstructed_sender, reconstructed_receiver) = mpsc::unbounded();
+
+        // tell the buffer to start sending stuff to us
+        client_output
+            .received_buffer_request_sender
+            .unbounded_send(ReceivedBufferMessage::ReceiverAnnounce(
+                reconstructed_sender,
+            ))
+            .expect("the buffer request failed!");
+
+        ResponsePusher {
+            reconstructed_receiver,
+            on_message,
+            on_binary_message,
+        }
+    }
+
+    pub(crate) fn start(mut self) {
+        spawn_local(async move {
+            let this = JsValue::null();
+
+            while let Some(reconstructed) = self.reconstructed_receiver.next().await {
+                for msg in reconstructed {
+                    if let Some(ref callback_binary) = self.on_binary_message {
+                        let arg1 = serde_wasm_bindgen::to_value(&msg.message).unwrap();
+                        callback_binary
+                            .call1(&this, &arg1)
+                            .expect("on binary message failed!");
+                    }
+                    if let Some(ref callback) = self.on_message {
+                        if msg.reply_surb.is_some() {
+                            console_log!("the received message contained a reply-surb that we do not know how to handle (yet)")
+                        }
+                        let stringified = String::from_utf8_lossy(&msg.message).into_owned();
+                        let arg1 = serde_wasm_bindgen::to_value(&stringified).unwrap();
+                        callback.call1(&this, &arg1).expect("on message failed!");
+                    }
+                }
+            }
+        })
+    }
+}

clients/webassembly/src/gateway_selector.rs

@@ -1,15 +1,15 @@
 // Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use client_core::config::GatewayEndpoint;
+use client_core::config::GatewayEndpointConfig;
 use wasm_bindgen::prelude::*;
 
 #[wasm_bindgen]
-pub async fn get_gateway(api_server: String, preferred: Option<String>) -> GatewayEndpoint {
+pub async fn get_gateway(api_server: String, preferred: Option<String>) -> GatewayEndpointConfig {
     let validator_client = validator_client::client::ApiClient::new(api_server.parse().unwrap());
 
     let gateways = match validator_client.get_cached_gateways().await {
-        Err(err) => panic!("failed to obtain list of all gateways - {}", err),
+        Err(err) => panic!("failed to obtain list of all gateways - {err}"),
         Ok(gateways) => gateways,
     };
 
@@ -18,7 +18,7 @@ pub async fn get_gateway(api_server: String, preferred: Option<String>) -> Gatew
             .iter()
             .find(|g| g.gateway.identity_key == preferred)
         {
-            return GatewayEndpoint {
+            return GatewayEndpointConfig {
                 gateway_id: details.gateway.identity_key.clone(),
                 gateway_owner: details.owner.to_string(),
                 gateway_listener: format!(
@@ -33,7 +33,7 @@ pub async fn get_gateway(api_server: String, preferred: Option<String>) -> Gatew
         .first()
         .expect("current topology holds no gateways");
 
-    GatewayEndpoint {
+    GatewayEndpointConfig {
         gateway_id: details.gateway.identity_key.clone(),
         gateway_owner: details.owner.to_string(),
         gateway_listener: format!(

common/client-libs/gateway-client/Cargo.toml

@@ -26,6 +26,7 @@ network-defaults = { path = "../../network-defaults" }
 nymsphinx = { path = "../../nymsphinx" }
 pemstore = { path = "../../pemstore" }
 validator-client = { path = "../validator-client", optional = true }
+task = { path = "../../task" }
 
 
 [dependencies.tungstenite]
@@ -47,9 +48,6 @@ version = "0.14"
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.credential-storage]
 path = "../../credential-storage"
 
-[target."cfg(not(target_arch = \"wasm32\"))".dependencies.task]
-path = "../../task"
-
 # wasm-only dependencies
 [target."cfg(target_arch = \"wasm32\")".dependencies.wasm-bindgen]
 version = "0.2"

common/client-libs/gateway-client/src/bandwidth.rs

@@ -26,8 +26,15 @@ use {
     },
 };
 
+// TODO: make it nicer for wasm (I don't want to touch it for this experiment)
+#[cfg(target_arch = "wasm32")]
+use crate::wasm_storage::PersistentStorage;
+
+#[cfg(not(target_arch = "wasm32"))]
+use credential_storage::PersistentStorage;
+
 #[derive(Clone)]
-pub struct BandwidthController<St: Storage> {
+pub struct BandwidthController<St: Storage = PersistentStorage> {
     #[allow(dead_code)]
     storage: St,
     #[cfg(feature = "coconut")]

common/client-libs/gateway-client/src/client.rs

@@ -22,6 +22,7 @@ use rand::rngs::OsRng;
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::Duration;
+use task::ShutdownListener;
 use tungstenite::protocol::Message;
 
 #[cfg(feature = "coconut")]
@@ -30,8 +31,6 @@ use coconut_interface::Credential;
 #[cfg(not(target_arch = "wasm32"))]
 use credential_storage::PersistentStorage;
 #[cfg(not(target_arch = "wasm32"))]
-use task::ShutdownListener;
-#[cfg(not(target_arch = "wasm32"))]
 use tokio_tungstenite::connect_async;
 
 #[cfg(target_arch = "wasm32")]
@@ -67,8 +66,9 @@ pub struct GatewayClient {
     /// Delay between each subsequent reconnection attempt.
     reconnection_backoff: Duration,
 
-    #[cfg(not(target_arch = "wasm32"))]
     /// Listen to shutdown messages.
+    // TODO: fix this
+    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
     shutdown: Option<ShutdownListener>,
 }
 
@@ -85,7 +85,7 @@ impl GatewayClient {
         ack_sender: AcknowledgementSender,
         response_timeout_duration: Duration,
         bandwidth_controller: Option<BandwidthController<PersistentStorage>>,
-        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
+        shutdown: Option<ShutdownListener>,
     ) -> Self {
         GatewayClient {
             authenticated: false,
@@ -97,18 +97,12 @@ impl GatewayClient {
             local_identity,
             shared_key,
             connection: SocketState::NotConnected,
-            packet_router: PacketRouter::new(
+            packet_router: PacketRouter::new(ack_sender, mixnet_message_sender, shutdown.clone()),
-                ack_sender,
-                mixnet_message_sender,
-                #[cfg(not(target_arch = "wasm32"))]
-                shutdown.clone(),
-            ),
             response_timeout_duration,
             bandwidth_controller,
             should_reconnect_on_failure: true,
             reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS,
             reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF,
-            #[cfg(not(target_arch = "wasm32"))]
             shutdown,
         }
     }
@@ -136,7 +130,7 @@ impl GatewayClient {
         gateway_owner: String,
         local_identity: Arc<identity::KeyPair>,
         response_timeout_duration: Duration,
-        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
+        shutdown: Option<ShutdownListener>,
     ) -> Self {
         use futures::channel::mpsc;
 
@@ -144,12 +138,7 @@ impl GatewayClient {
         // perfectly fine here, because it's not meant to be used
         let (ack_tx, _) = mpsc::unbounded();
         let (mix_tx, _) = mpsc::unbounded();
-        let packet_router = PacketRouter::new(
+        let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown.clone());
-            ack_tx,
-            mix_tx,
-            #[cfg(not(target_arch = "wasm32"))]
-            shutdown.clone(),
-        );
 
         GatewayClient {
             authenticated: false,
@@ -167,7 +156,6 @@ impl GatewayClient {
             should_reconnect_on_failure: false,
             reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS,
             reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF,
-            #[cfg(not(target_arch = "wasm32"))]
             shutdown,
         }
     }
@@ -403,13 +391,10 @@ impl GatewayClient {
                     .batch_send_without_response(messages)
                     .await
                 {
-                    error!("failed to batch send messages - {}...", err);
+                    error!("failed to batch send messages - {err}...");
                     // we must ensure we do not leave the task still active
                     if let Err(err) = self.recover_socket_connection().await {
-                        error!(
+                        error!("... and the delegated stream has also errored out - {err}")
-                            "... and the delegated stream has also errored out - {}",
-                            err
-                        )
                     }
                     Err(err)
                 } else {
@@ -429,13 +414,10 @@ impl GatewayClient {
             SocketState::Available(ref mut conn) => Ok(conn.send(msg).await?),
             SocketState::PartiallyDelegated(ref mut partially_delegated) => {
                 if let Err(err) = partially_delegated.send_without_response(msg).await {
-                    error!("failed to send message without response - {}...", err);
+                    error!("failed to send message without response - {err}...");
                     // we must ensure we do not leave the task still active
                     if let Err(err) = self.recover_socket_connection().await {
-                        error!(
+                        error!("... and the delegated stream has also errored out - {err}")
-                            "... and the delegated stream has also errored out - {}",
-                            err
-                        )
                     }
                     Err(err)
                 } else {

common/client-libs/gateway-client/src/packet_router.rs

@@ -4,15 +4,13 @@
 // JS: I personally don't like this name very much, but could not think of anything better.
 // I will gladly take any suggestions on how to rename this.
 
+use crate::error::GatewayClientError;
 use futures::channel::mpsc;
 use log::*;
 use nymsphinx::addressing::nodes::MAX_NODE_ADDRESS_UNPADDED_LEN;
 use nymsphinx::params::packet_sizes::PacketSize;
-#[cfg(not(target_arch = "wasm32"))]
 use task::ShutdownListener;
 
-use crate::error::GatewayClientError;
-
 pub type MixnetMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
 pub type MixnetMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
 
@@ -23,7 +21,6 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
 pub struct PacketRouter {
     ack_sender: AcknowledgementSender,
     mixnet_message_sender: MixnetMessageSender,
-    #[cfg(not(target_arch = "wasm32"))]
     shutdown: Option<ShutdownListener>,
 }
 
@@ -31,12 +28,11 @@ impl PacketRouter {
     pub fn new(
         ack_sender: AcknowledgementSender,
         mixnet_message_sender: MixnetMessageSender,
-        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
+        shutdown: Option<ShutdownListener>,
     ) -> Self {
         PacketRouter {
             ack_sender,
             mixnet_message_sender,
-            #[cfg(not(target_arch = "wasm32"))]
             shutdown,
         }
     }
@@ -86,7 +82,6 @@ impl PacketRouter {
         if !received_messages.is_empty() {
             trace!("routing 'real'");
             if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) {
-                #[cfg(not(target_arch = "wasm32"))]
                 if let Some(shutdown) = &mut self.shutdown {
                     if shutdown.is_shutdown_poll() {
                         // This should ideally not happen, but it's ok

common/client-libs/validator-client/src/nymd/traits/mixnet_signing_client.rs

@@ -180,6 +180,35 @@ pub trait MixnetSigningClient {
         .await
     }
 
+    async fn pledge_more(
+        &self,
+        additional_pledge: Coin,
+        fee: Option<Fee>,
+    ) -> Result<ExecuteResult, NymdError> {
+        self.execute_mixnet_contract(
+            fee,
+            MixnetExecuteMsg::PledgeMore {},
+            vec![additional_pledge],
+        )
+        .await
+    }
+
+    async fn pledge_more_on_behalf(
+        &self,
+        owner: AccountId,
+        additional_pledge: Coin,
+        fee: Option<Fee>,
+    ) -> Result<ExecuteResult, NymdError> {
+        self.execute_mixnet_contract(
+            fee,
+            MixnetExecuteMsg::PledgeMoreOnBehalf {
+                owner: owner.to_string(),
+            },
+            vec![additional_pledge],
+        )
+        .await
+    }
+
     async fn unbond_mixnode(&self, fee: Option<Fee>) -> Result<ExecuteResult, NymdError> {
         self.execute_mixnet_contract(fee, MixnetExecuteMsg::UnbondMixnode {}, vec![])
             .await

common/client-libs/validator-client/src/nymd/traits/vesting_signing_client.rs

@@ -64,6 +64,21 @@ pub trait VestingSigningClient {
         fee: Option<Fee>,
     ) -> Result<ExecuteResult, NymdError>;
 
+    async fn vesting_pledge_more(
+        &self,
+        additional_pledge: Coin,
+        fee: Option<Fee>,
+    ) -> Result<ExecuteResult, NymdError> {
+        self.execute_vesting_contract(
+            fee,
+            VestingExecuteMsg::PledgeMore {
+                amount: additional_pledge.into(),
+            },
+            vec![],
+        )
+        .await
+    }
+
     async fn vesting_unbond_mixnode(&self, fee: Option<Fee>) -> Result<ExecuteResult, NymdError>;
 
     async fn vesting_track_unbond_mixnode(

common/cosmwasm-smart-contracts/mixnet-contract/src/events.rs

@@ -12,6 +12,8 @@ pub const EVENT_VERSION_PREFIX: &str = "v2_";
 
 pub enum MixnetEventType {
     MixnodeBonding,
+    PendingPledgeIncrease,
+    PledgeIncrease,
     GatewayBonding,
     GatewayUnbonding,
     PendingMixnodeUnbonding,
@@ -51,6 +53,8 @@ impl ToString for MixnetEventType {
     fn to_string(&self) -> String {
         let event_name = match self {
             MixnetEventType::MixnodeBonding => "mixnode_bonding",
+            MixnetEventType::PendingPledgeIncrease => "pending_pledge_increase",
+            MixnetEventType::PledgeIncrease => "pledge_increase",
             MixnetEventType::GatewayBonding => "gateway_bonding",
             MixnetEventType::GatewayUnbonding => "gateway_unbonding",
             MixnetEventType::PendingMixnodeUnbonding => "pending_mixnode_unbonding",
@@ -330,6 +334,19 @@ pub fn new_mixnode_bonding_event(
         .add_attribute(AMOUNT_KEY, amount.to_string())
 }
 
+pub fn new_pending_pledge_increase_event(mix_id: MixId, amount: &Coin) -> Event {
+    Event::new(MixnetEventType::PendingPledgeIncrease)
+        .add_attribute(MIX_ID_KEY, mix_id.to_string())
+        .add_attribute(AMOUNT_KEY, amount.to_string())
+}
+
+pub fn new_pledge_increase_event(created_at: BlockHeight, mix_id: MixId, amount: &Coin) -> Event {
+    Event::new(MixnetEventType::PledgeIncrease)
+        .add_attribute(EVENT_CREATION_HEIGHT_KEY, created_at.to_string())
+        .add_attribute(MIX_ID_KEY, mix_id.to_string())
+        .add_attribute(AMOUNT_KEY, amount.to_string())
+}
+
 pub fn new_mixnode_unbonding_event(created_at: BlockHeight, mix_id: MixId) -> Event {
     Event::new(MixnetEventType::MixnodeUnbonding)
         .add_attribute(EVENT_CREATION_HEIGHT_KEY, created_at.to_string())

common/cosmwasm-smart-contracts/mixnet-contract/src/mixnode.rs

@@ -325,6 +325,14 @@ impl MixNodeRewarding {
         Ok(())
     }
 
+    pub fn increase_operator_uint128(
+        &mut self,
+        amount: Uint128,
+    ) -> Result<(), MixnetContractError> {
+        self.operator += amount.into_base_decimal()?;
+        Ok(())
+    }
+
     pub fn increase_delegates_uint128(
         &mut self,
         amount: Uint128,

common/cosmwasm-smart-contracts/mixnet-contract/src/msg.rs

@@ -113,6 +113,10 @@ pub enum ExecuteMsg {
         owner_signature: String,
         owner: String,
     },
+    PledgeMore {},
+    PledgeMoreOnBehalf {
+        owner: String,
+    },
     UnbondMixnode {},
     UnbondMixnodeOnBehalf {
         owner: String,
@@ -223,6 +227,8 @@ impl ExecuteMsg {
             ExecuteMsg::BondMixnodeOnBehalf { mix_node, .. } => {
                 format!("bonding mixnode {} on behalf", mix_node.identity_key)
             }
+            ExecuteMsg::PledgeMore {} => "pledging additional tokens".into(),
+            ExecuteMsg::PledgeMoreOnBehalf { .. } => "pledging additional tokens on behalf".into(),
             ExecuteMsg::UnbondMixnode { .. } => "unbonding mixnode".into(),
             ExecuteMsg::UnbondMixnodeOnBehalf { .. } => "unbonding mixnode on behalf".into(),
             ExecuteMsg::UpdateMixnodeCostParams { .. } => "updating mixnode cost parameters".into(),

common/cosmwasm-smart-contracts/mixnet-contract/src/pending_events.rs

@@ -34,6 +34,10 @@ pub enum PendingEpochEventKind {
         mix_id: MixId,
         proxy: Option<Addr>,
     },
+    PledgeMore {
+        mix_id: MixId,
+        amount: Coin,
+    },
     UnbondMixnode {
         mix_id: MixId,
     },
@@ -78,7 +82,6 @@ pub enum PendingIntervalEventKind {
         mix_id: MixId,
         new_costs: MixNodeCostParams,
     },
-
     UpdateRewardingParams {
         update: IntervalRewardingParamsUpdate,
     },

common/cosmwasm-smart-contracts/vesting-contract/src/events.rs

@@ -14,6 +14,7 @@ pub const VESTING_UNDELEGATION_EVENT_TYPE: &str = "vesting_undelegation";
 pub const VESTING_GATEWAY_BONDING_EVENT_TYPE: &str = "vesting_gateway_bonding";
 pub const VESTING_GATEWAY_UNBONDING_EVENT_TYPE: &str = "vesting_gateway_unbonding";
 pub const VESTING_MIXNODE_BONDING_EVENT_TYPE: &str = "vesting_mixnode_bonding";
+pub const VESTING_PLEDGE_MORE_EVENT_TYPE: &str = "vesting_pledge_more";
 pub const VESTING_MIXNODE_UNBONDING_EVENT_TYPE: &str = "vesting_mixnode_unbonding";
 pub const VESTING_UPDATE_MIXNODE_CONFIG_EVENT_TYPE: &str = "vesting_update_mixnode_config";
 pub const VESTING_UPDATE_MIXNODE_COST_PARAMS_EVENT_TYPE: &str =
@@ -112,6 +113,10 @@ pub fn new_vesting_mixnode_bonding_event() -> Event {
     Event::new(VESTING_MIXNODE_BONDING_EVENT_TYPE)
 }
 
+pub fn new_vesting_pledge_more_event() -> Event {
+    Event::new(VESTING_PLEDGE_MORE_EVENT_TYPE)
+}
+
 pub fn new_vesting_update_mixnode_config_event() -> Event {
     Event::new(VESTING_UPDATE_MIXNODE_CONFIG_EVENT_TYPE)
 }

common/cosmwasm-smart-contracts/vesting-contract/src/messages.rs

@@ -101,6 +101,9 @@ pub enum ExecuteMsg {
         owner_signature: String,
         amount: Coin,
     },
+    PledgeMore {
+        amount: Coin,
+    },
     UnbondMixnode {},
     TrackUnbondMixnode {
         owner: String,
@@ -145,6 +148,7 @@ impl ExecuteMsg {
             ExecuteMsg::WithdrawVestedCoins { .. } => "VestingExecuteMsg::WithdrawVestedCoins",
             ExecuteMsg::TrackUndelegation { .. } => "VestingExecuteMsg::TrackUndelegation",
             ExecuteMsg::BondMixnode { .. } => "VestingExecuteMsg::BondMixnode",
+            ExecuteMsg::PledgeMore { .. } => "VestingExecuteMsg::PledgeMore",
             ExecuteMsg::UnbondMixnode { .. } => "VestingExecuteMsg::UnbondMixnode",
             ExecuteMsg::TrackUnbondMixnode { .. } => "VestingExecuteMsg::TrackUnbondMixnode",
             ExecuteMsg::BondGateway { .. } => "VestingExecuteMsg::BondGateway",

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

@@ -254,6 +254,7 @@ impl Controller {
                 },
             }
         }
+        #[cfg(not(target_arch = "wasm32"))]
         tokio::time::timeout(Duration::from_secs(5), self.shutdown.recv())
             .await
             .expect("Task stopped without shutdown called");

common/task/Cargo.toml

@@ -9,7 +9,11 @@ edition = "2021"
 futures = "0.3"
 log = "0.4"
 thiserror = "1.0.37"
-tokio = { version = "1.21.2", features = ["macros", "signal", "time", "sync"] }
+tokio = { version = "1.21.2", features = ["macros", "sync"] }
+
+[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio]
+version = "1.21.2"
+features = ["signal", "time"]
 
 [dev-dependencies]
 tokio = { version = "1.21.2", features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }

common/task/src/lib.rs

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod shutdown;
+#[cfg(not(target_arch = "wasm32"))]
 pub mod signal;
 
 pub use shutdown::{ShutdownListener, ShutdownNotifier};
+#[cfg(not(target_arch = "wasm32"))]
 pub use signal::{wait_for_signal, wait_for_signal_and_error};

common/task/src/shutdown.rs

@@ -31,6 +31,7 @@ pub struct ShutdownNotifier {
     // track of which tasks we are still waiting for.
     notify_tx: watch::Sender<()>,
     notify_rx: Option<watch::Receiver<()>>,
+    #[cfg_attr(target_arch = "wasm32", allow(dead_code))]
     shutdown_timer_secs: u64,
 
     // If any task failed, it needs to report separately
@@ -113,6 +114,11 @@ impl ShutdownNotifier {
             drop(notify_rx);
         }
 
+        // in wasm we'll never get our shutdown anyway...
+        #[cfg(target_arch = "wasm32")]
+        futures::future::pending::<()>().await;
+
+        #[cfg(not(target_arch = "wasm32"))]
         tokio::select! {
             _ = self.notify_tx.closed() => {
                 log::info!("All registered tasks succesfully shutdown");
@@ -149,6 +155,9 @@ pub struct ShutdownListener {
 }
 
 impl ShutdownListener {
+    #[cfg(not(target_arch = "wasm32"))]
+    const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+
     fn new(
         notify: watch::Receiver<()>,
         return_error: ErrorSender,
@@ -175,6 +184,13 @@ impl ShutdownListener {
         self.shutdown = true;
     }
 
+    pub async fn recv_timeout(&mut self) {
+        #[cfg(not(target_arch = "wasm32"))]
+        tokio::time::timeout(Self::SHUTDOWN_TIMEOUT, self.recv())
+            .await
+            .expect("Task stopped without shutdown called");
+    }
+
     pub fn is_shutdown_poll(&mut self) -> bool {
         if self.shutdown {
             return true;

common/types/src/pending_events.rs

@@ -57,6 +57,10 @@ pub enum PendingEpochEventData {
         mix_id: MixId,
         proxy: Option<String>,
     },
+    PledgeMore {
+        mix_id: MixId,
+        amount: DecCoin,
+    },
     UnbondMixnode {
         mix_id: MixId,
     },
@@ -91,6 +95,12 @@ impl PendingEpochEventData {
                 mix_id,
                 proxy: proxy.map(|p| p.into_string()),
             }),
+            MixnetContractPendingEpochEventKind::PledgeMore { mix_id, amount } => {
+                Ok(PendingEpochEventData::PledgeMore {
+                    mix_id,
+                    amount: reg.attempt_convert_to_display_dec_coin(amount.into())?,
+                })
+            }
             MixnetContractPendingEpochEventKind::UnbondMixnode { mix_id } => {
                 Ok(PendingEpochEventData::UnbondMixnode { mix_id })
             }

contracts/CHANGELOG.md

@@ -3,7 +3,9 @@
 ## Added
 
 - Added migration code to the mixnet contract to allow updating stored vesting contract address to make it easier to deploy any future environments ([#1759],[#1769])
+- Added an option to pledge additional tokens without the need to rebond minxode ([#1679])
 
+[#1679]: https://github.com/nymtech/nym/pull/1679
 [#1759]: https://github.com/nymtech/nym/pull/1759
 [#1769]: https://github.com/nymtech/nym/pull/1769
 

contracts/mixnet/src/contract.rs

@@ -168,6 +168,12 @@ pub fn execute(
             owner,
             owner_signature,
         ),
+        ExecuteMsg::PledgeMore {} => {
+            crate::mixnodes::transactions::try_increase_pledge(deps, env, info)
+        }
+        ExecuteMsg::PledgeMoreOnBehalf { owner } => {
+            crate::mixnodes::transactions::try_increase_pledge_on_behalf(deps, env, info, owner)
+        }
         ExecuteMsg::UnbondMixnode {} => {
             crate::mixnodes::transactions::try_remove_mixnode(deps, env, info)
         }

contracts/mixnet/src/interval/pending_events.rs

@@ -7,13 +7,14 @@ use crate::interval::helpers::change_interval_config;
 use crate::interval::storage;
 use crate::mixnet_contract_settings::storage as mixnet_params_storage;
 use crate::mixnodes::helpers::{cleanup_post_unbond_mixnode_storage, get_mixnode_details_by_id};
+use crate::mixnodes::storage as mixnodes_storage;
 use crate::rewards::storage as rewards_storage;
 use crate::support::helpers::send_to_proxy_or_owner;
 use cosmwasm_std::{wasm_execute, Addr, Coin, DepsMut, Env, Response};
 use mixnet_contract_common::error::MixnetContractError;
 use mixnet_contract_common::events::{
     new_active_set_update_event, new_delegation_event, new_delegation_on_unbonded_node_event,
-    new_mixnode_cost_params_update_event, new_mixnode_unbonding_event,
+    new_mixnode_cost_params_update_event, new_mixnode_unbonding_event, new_pledge_increase_event,
     new_rewarding_params_update_event, new_undelegation_event,
 };
 use mixnet_contract_common::mixnode::MixNodeCostParams;
@@ -258,6 +259,42 @@ pub(crate) fn update_active_set_size(
     Ok(Response::new().add_event(new_active_set_update_event(created_at, active_set_size)))
 }
 
+pub(crate) fn increase_pledge(
+    deps: DepsMut<'_>,
+    created_at: BlockHeight,
+    mix_id: MixId,
+    increase: Coin,
+) -> Result<Response, MixnetContractError> {
+    // note: we have already validated the amount to know it has the correct denomination
+
+    // the target node MUST exist - we have checked it at the time of putting this event onto the queue
+    // we have also verified there were no preceding unbond events
+    let mix_details = get_mixnode_details_by_id(deps.storage, mix_id)?.ok_or(
+        MixnetContractError::InconsistentState {
+            comment:
+                "mixnode getting processed to increase its pledge doesn't exist in the storage"
+                    .into(),
+        },
+    )?;
+
+    let mut updated_bond = mix_details.bond_information.clone();
+    let mut updated_rewarding = mix_details.rewarding_details;
+
+    updated_bond.original_pledge.amount += increase.amount;
+    updated_rewarding.increase_operator_uint128(increase.amount)?;
+
+    // update both, bond information and rewarding details
+    mixnodes_storage::mixnode_bonds().replace(
+        deps.storage,
+        mix_id,
+        Some(&updated_bond),
+        Some(&mix_details.bond_information),
+    )?;
+    rewards_storage::MIXNODE_REWARDING.save(deps.storage, mix_id, &updated_rewarding)?;
+
+    Ok(Response::new().add_event(new_pledge_increase_event(created_at, mix_id, &increase)))
+}
+
 impl ContractExecutableEvent for PendingEpochEventData {
     fn execute(self, deps: DepsMut<'_>, env: &Env) -> Result<Response, MixnetContractError> {
         // note that the basic validation on all those events was already performed before
@@ -274,6 +311,9 @@ impl ContractExecutableEvent for PendingEpochEventData {
                 mix_id,
                 proxy,
             } => undelegate(deps, self.created_at, owner, mix_id, proxy),
+            PendingEpochEventKind::PledgeMore { mix_id, amount } => {
+                increase_pledge(deps, self.created_at, mix_id, amount)
+            }
             PendingEpochEventKind::UnbondMixnode { mix_id } => {
                 unbond_mixnode(deps, env, self.created_at, mix_id)
             }
@@ -1135,6 +1175,228 @@ mod tests {
         }
     }
 
+    #[cfg(test)]
+    mod increasing_pledge {
+        use super::*;
+        use cosmwasm_std::Uint128;
+        use mixnet_contract_common::rewarding::helpers::truncate_reward_amount;
+
+        #[test]
+        fn returns_hard_error_if_mixnode_doesnt_exist() {
+            // this should have never happened so hard error MUST be thrown here
+            let mut test = TestSetup::new();
+
+            let amount = test.coin(123);
+            let res = increase_pledge(test.deps_mut(), 123, 1, amount);
+            assert!(matches!(
+                res,
+                Err(MixnetContractError::InconsistentState { .. })
+            ));
+        }
+
+        #[test]
+        fn updates_stored_bond_information_and_rewarding_details() {
+            let mut test = TestSetup::new();
+            let mix_id = test.add_dummy_mixnode("mix-owner", None);
+
+            let old_details = get_mixnode_details_by_id(test.deps().storage, mix_id)
+                .unwrap()
+                .unwrap();
+
+            let amount = test.coin(12345);
+            increase_pledge(test.deps_mut(), 123, mix_id, amount.clone()).unwrap();
+
+            let updated_details = get_mixnode_details_by_id(test.deps().storage, mix_id)
+                .unwrap()
+                .unwrap();
+
+            assert_eq!(
+                updated_details.bond_information.original_pledge.amount,
+                old_details.bond_information.original_pledge.amount + amount.amount
+            );
+
+            assert_eq!(
+                updated_details.rewarding_details.operator,
+                old_details.rewarding_details.operator
+                    + Decimal::from_atomics(amount.amount, 0).unwrap()
+            );
+        }
+
+        #[test]
+        fn without_any_events_in_between_is_equivalent_to_pledging_the_same_amount_immediately() {
+            let mut test = TestSetup::new();
+            let pledge1 = Uint128::new(150_000_000);
+            let pledge2 = Uint128::new(50_000_000);
+            let pledge3 = Uint128::new(200_000_000);
+
+            let mix_id_repledge = test.add_dummy_mixnode("mix-owner1", Some(pledge1));
+            let increase = test.coin(pledge2.u128());
+            increase_pledge(test.deps_mut(), 123, mix_id_repledge, increase).unwrap();
+
+            let mix_id_full_pledge = test.add_dummy_mixnode("mix-owner2", Some(pledge3));
+
+            test.add_immediate_delegation("alice", 123_456_789u128, mix_id_repledge);
+            test.add_immediate_delegation("bob", 500_000_000u128, mix_id_repledge);
+            test.add_immediate_delegation("carol", 111_111_111u128, mix_id_repledge);
+
+            test.add_immediate_delegation("alice", 123_456_789u128, mix_id_full_pledge);
+            test.add_immediate_delegation("bob", 500_000_000u128, mix_id_full_pledge);
+            test.add_immediate_delegation("carol", 111_111_111u128, mix_id_full_pledge);
+
+            test.skip_to_next_epoch_end();
+            test.update_rewarded_set(vec![mix_id_repledge, mix_id_full_pledge]);
+
+            let dist1 =
+                test.reward_with_distribution(mix_id_repledge, test_helpers::performance(100.0));
+            let dist2 =
+                test.reward_with_distribution(mix_id_full_pledge, test_helpers::performance(100.0));
+
+            assert_eq!(dist1, dist2)
+        }
+
+        #[test]
+        fn correctly_increases_future_rewards() {
+            let mut test = TestSetup::new();
+            let pledge1 = Uint128::new(150_000_000_000);
+            let pledge2 = Uint128::new(50_000_000_000);
+
+            let mix_id_repledge = test.add_dummy_mixnode("mix-owner1", Some(pledge1));
+
+            test.add_immediate_delegation("alice", 123_456_789_000u128, mix_id_repledge);
+            test.add_immediate_delegation("bob", 500_000_000_000u128, mix_id_repledge);
+            test.add_immediate_delegation("carol", 111_111_111_000u128, mix_id_repledge);
+
+            test.skip_to_next_epoch_end();
+            test.update_rewarded_set(vec![mix_id_repledge]);
+
+            let dist =
+                test.reward_with_distribution(mix_id_repledge, test_helpers::performance(100.0));
+
+            let increase = test.coin(pledge2.u128());
+            increase_pledge(test.deps_mut(), 123, mix_id_repledge, increase).unwrap();
+
+            let pledge3 = Uint128::new(200_000_000_000) + truncate_reward_amount(dist.operator);
+            let mix_id_full_pledge = test.add_dummy_mixnode("mix-owner2", Some(pledge3));
+
+            test.add_immediate_delegation("alice", 123_456_789_000u128, mix_id_full_pledge);
+            test.add_immediate_delegation("bob", 500_000_000_000u128, mix_id_full_pledge);
+            test.add_immediate_delegation("carol", 111_111_111_000u128, mix_id_full_pledge);
+
+            let lost_operator = dist.operator
+                - Decimal::from_atomics(truncate_reward_amount(dist.operator), 0).unwrap();
+            let lost_delegates = dist.delegates
+                - Decimal::from_atomics(truncate_reward_amount(dist.delegates), 0).unwrap();
+
+            // add the tiny bit of lost precision manually
+            let mut mix_rewarding_full = test.mix_rewarding(mix_id_full_pledge);
+            mix_rewarding_full.delegates += lost_delegates;
+            mix_rewarding_full.operator += lost_operator;
+            rewards_storage::MIXNODE_REWARDING
+                .save(
+                    test.deps_mut().storage,
+                    mix_id_full_pledge,
+                    &mix_rewarding_full,
+                )
+                .unwrap();
+
+            test.add_immediate_delegation(
+                "dave",
+                truncate_reward_amount(dist.delegates).u128(),
+                mix_id_full_pledge,
+            );
+
+            test.skip_to_next_epoch_end();
+            test.update_rewarded_set(vec![mix_id_repledge, mix_id_full_pledge]);
+
+            // go through few epochs of rewarding
+            for _ in 0..500 {
+                test.skip_to_next_epoch_end();
+                let dist1 = test
+                    .reward_with_distribution(mix_id_repledge, test_helpers::performance(100.0));
+                let dist2 = test
+                    .reward_with_distribution(mix_id_full_pledge, test_helpers::performance(100.0));
+
+                assert_eq!(dist1, dist2)
+            }
+        }
+
+        #[test]
+        fn correctly_increases_future_rewards_with_more_passed_epochs() {
+            let mut test = TestSetup::new();
+            let pledge1 = Uint128::new(150_000_000_000);
+            let pledge2 = Uint128::new(50_000_000_000);
+
+            let mix_id_repledge = test.add_dummy_mixnode("mix-owner1", Some(pledge1));
+
+            test.add_immediate_delegation("alice", 123_456_789_000u128, mix_id_repledge);
+            test.add_immediate_delegation("bob", 500_000_000_000u128, mix_id_repledge);
+            test.add_immediate_delegation("carol", 111_111_111_000u128, mix_id_repledge);
+
+            test.skip_to_next_epoch_end();
+            test.update_rewarded_set(vec![mix_id_repledge]);
+
+            let mut cumulative_op_reward = Decimal::zero();
+            let mut cumulative_del_reward = Decimal::zero();
+
+            // go few epochs of rewarding before adding more pledge
+            for _ in 0..500 {
+                test.skip_to_next_epoch_end();
+                let dist = test
+                    .reward_with_distribution(mix_id_repledge, test_helpers::performance(100.0));
+                cumulative_op_reward += dist.operator;
+                cumulative_del_reward += dist.delegates;
+            }
+
+            let increase = test.coin(pledge2.u128());
+            increase_pledge(test.deps_mut(), 123, mix_id_repledge, increase).unwrap();
+
+            let pledge3 =
+                Uint128::new(200_000_000_000) + truncate_reward_amount(cumulative_op_reward);
+            let mix_id_full_pledge = test.add_dummy_mixnode("mix-owner2", Some(pledge3));
+
+            test.add_immediate_delegation("alice", 123_456_789_000u128, mix_id_full_pledge);
+            test.add_immediate_delegation("bob", 500_000_000_000u128, mix_id_full_pledge);
+            test.add_immediate_delegation("carol", 111_111_111_000u128, mix_id_full_pledge);
+
+            let lost_operator = cumulative_op_reward
+                - Decimal::from_atomics(truncate_reward_amount(cumulative_op_reward), 0).unwrap();
+            let lost_delegates = cumulative_del_reward
+                - Decimal::from_atomics(truncate_reward_amount(cumulative_del_reward), 0).unwrap();
+
+            // add the tiny bit of lost precision manually
+            let mut mix_rewarding_full = test.mix_rewarding(mix_id_full_pledge);
+            mix_rewarding_full.delegates += lost_delegates;
+            mix_rewarding_full.operator += lost_operator;
+            rewards_storage::MIXNODE_REWARDING
+                .save(
+                    test.deps_mut().storage,
+                    mix_id_full_pledge,
+                    &mix_rewarding_full,
+                )
+                .unwrap();
+
+            test.add_immediate_delegation(
+                "dave",
+                truncate_reward_amount(cumulative_del_reward).u128(),
+                mix_id_full_pledge,
+            );
+
+            test.skip_to_next_epoch_end();
+            test.update_rewarded_set(vec![mix_id_repledge, mix_id_full_pledge]);
+
+            // go through few more epochs of rewarding
+            for _ in 0..500 {
+                test.skip_to_next_epoch_end();
+                let dist1 = test
+                    .reward_with_distribution(mix_id_repledge, test_helpers::performance(100.0));
+                let dist2 = test
+                    .reward_with_distribution(mix_id_full_pledge, test_helpers::performance(100.0));
+
+                assert_eq!(dist1, dist2)
+            }
+        }
+    }
+
     #[test]
     fn updating_active_set_updates_rewarding_params() {
         let mut test = TestSetup::new();

contracts/mixnet/src/mixnet_contract_settings/storage.rs

@@ -37,7 +37,6 @@ pub(crate) fn minimum_delegation_stake(
         .map(|state| state.params.minimum_mixnode_delegation)?)
 }
 
-#[allow(unused)]
 pub(crate) fn rewarding_denom(storage: &dyn Storage) -> Result<String, MixnetContractError> {
     Ok(CONTRACT_STATE
         .load(storage)

contracts/mixnet/src/mixnodes/helpers.rs

@@ -147,7 +147,7 @@ pub(crate) fn cleanup_post_unbond_mixnode_storage(
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
     use crate::support::tests::fixtures::{
         mix_node_cost_params_fixture, mix_node_fixture, TEST_COIN_DENOM,
@@ -155,13 +155,13 @@ mod tests {
     use crate::support::tests::test_helpers::TestSetup;
     use cosmwasm_std::coin;
 
-    const OWNER_EXISTS: &str = "mix-owner-existing";
+    pub(crate) const OWNER_EXISTS: &str = "mix-owner-existing";
-    const OWNER_UNBONDING: &str = "mix-owner-unbonding";
+    pub(crate) const OWNER_UNBONDING: &str = "mix-owner-unbonding";
-    const OWNER_UNBONDED: &str = "mix-owner-unbonded";
+    pub(crate) const OWNER_UNBONDED: &str = "mix-owner-unbonded";
-    const OWNER_UNBONDED_LEFTOVER: &str = "mix-owner-unbonded-leftover";
+    pub(crate) const OWNER_UNBONDED_LEFTOVER: &str = "mix-owner-unbonded-leftover";
 
     // create a mixnode that is bonded, unbonded, in the process of unbonding and unbonded with leftover mix rewarding details
-    fn setup_mix_combinations(test: &mut TestSetup) -> Vec<MixId> {
+    pub(crate) fn setup_mix_combinations(test: &mut TestSetup) -> Vec<MixId> {
         let mix_id_exists = test.add_dummy_mixnode(OWNER_EXISTS, None);
         let mix_id_unbonding = test.add_dummy_mixnode(OWNER_UNBONDING, None);
         let mix_id_unbonded = test.add_dummy_mixnode(OWNER_UNBONDED, None);

contracts/mixnet/src/mixnodes/transactions.rs

@@ -5,16 +5,20 @@ use super::storage;
 use crate::interval::storage as interval_storage;
 use crate::interval::storage::push_new_interval_event;
 use crate::mixnet_contract_settings::storage as mixnet_params_storage;
-use crate::mixnodes::helpers::{must_get_mixnode_bond_by_owner, save_new_mixnode};
+use crate::mixnet_contract_settings::storage::rewarding_denom;
+use crate::mixnodes::helpers::{
+    get_mixnode_details_by_owner, must_get_mixnode_bond_by_owner, save_new_mixnode,
+};
 use crate::support::helpers::{
     ensure_bonded, ensure_no_existing_bond, ensure_proxy_match, validate_node_identity_signature,
     validate_pledge,
 };
-use cosmwasm_std::{Addr, Coin, DepsMut, Env, MessageInfo, Response};
+use cosmwasm_std::{coin, Addr, Coin, DepsMut, Env, MessageInfo, Response};
 use mixnet_contract_common::error::MixnetContractError;
 use mixnet_contract_common::events::{
     new_mixnode_bonding_event, new_mixnode_config_update_event,
     new_mixnode_pending_cost_params_update_event, new_pending_mixnode_unbonding_event,
+    new_pending_pledge_increase_event,
 };
 use mixnet_contract_common::mixnode::{MixNodeConfigUpdate, MixNodeCostParams};
 use mixnet_contract_common::pending_events::{PendingEpochEventKind, PendingIntervalEventKind};
@@ -117,6 +121,54 @@ fn _try_add_mixnode(
     )))
 }
 
+pub fn try_increase_pledge(
+    deps: DepsMut<'_>,
+    env: Env,
+    info: MessageInfo,
+) -> Result<Response, MixnetContractError> {
+    _try_increase_pledge(deps, env, info.funds, info.sender, None)
+}
+
+pub fn try_increase_pledge_on_behalf(
+    deps: DepsMut<'_>,
+    env: Env,
+    info: MessageInfo,
+    owner: String,
+) -> Result<Response, MixnetContractError> {
+    let proxy = info.sender;
+    let owner = deps.api.addr_validate(&owner)?;
+    _try_increase_pledge(deps, env, info.funds, owner, Some(proxy))
+}
+
+pub fn _try_increase_pledge(
+    deps: DepsMut<'_>,
+    env: Env,
+    increase: Vec<Coin>,
+    owner: Addr,
+    proxy: Option<Addr>,
+) -> Result<Response, MixnetContractError> {
+    let mix_details = get_mixnode_details_by_owner(deps.storage, owner.clone())?
+        .ok_or(MixnetContractError::NoAssociatedMixNodeBond { owner })?;
+    let mix_id = mix_details.mix_id();
+
+    ensure_proxy_match(&proxy, &mix_details.bond_information.proxy)?;
+    ensure_bonded(&mix_details.bond_information)?;
+
+    let rewarding_denom = rewarding_denom(deps.storage)?;
+    let pledge_increase = validate_pledge(increase, coin(1, rewarding_denom))?;
+
+    let cosmos_event = new_pending_pledge_increase_event(mix_id, &pledge_increase);
+
+    // push the event to execute it at the end of the epoch
+    let epoch_event = PendingEpochEventKind::PledgeMore {
+        mix_id,
+        amount: pledge_increase,
+    };
+    interval_storage::push_new_epoch_event(deps.storage, &env, epoch_event)?;
+
+    Ok(Response::new().add_event(cosmos_event))
+}
+
 pub fn try_remove_mixnode_on_behalf(
     deps: DepsMut<'_>,
     env: Env,
@@ -680,4 +732,187 @@ pub mod tests {
         )
         .is_err());
     }
+
+    #[cfg(test)]
+    mod increasing_mixnode_pledge {
+        use super::*;
+        use crate::mixnodes::helpers::tests::{
+            setup_mix_combinations, OWNER_UNBONDED, OWNER_UNBONDED_LEFTOVER, OWNER_UNBONDING,
+        };
+        use crate::support::tests::test_helpers::TestSetup;
+
+        #[test]
+        fn is_not_allowed_if_account_doesnt_own_mixnode() {
+            let mut test = TestSetup::new();
+            let env = test.env();
+            let sender = mock_info("not-mix-owner", &[]);
+
+            let res = try_increase_pledge(test.deps_mut(), env, sender);
+            assert_eq!(
+                res,
+                Err(MixnetContractError::NoAssociatedMixNodeBond {
+                    owner: Addr::unchecked("not-mix-owner")
+                })
+            )
+        }
+
+        #[test]
+        fn is_not_allowed_if_theres_proxy_mismatch() {
+            let mut test = TestSetup::new();
+            let env = test.env();
+
+            let owner_without_proxy = Addr::unchecked("no-proxy");
+            let owner_with_proxy = Addr::unchecked("with-proxy");
+            let proxy = Addr::unchecked("proxy");
+            let wrong_proxy = Addr::unchecked("unrelated-proxy");
+
+            test.add_dummy_mixnode(owner_without_proxy.as_str(), None);
+            test.add_dummy_mixnode_with_proxy(owner_with_proxy.as_str(), None, proxy.clone());
+
+            let res = _try_increase_pledge(
+                test.deps_mut(),
+                env.clone(),
+                Vec::new(),
+                owner_without_proxy.clone(),
+                Some(proxy),
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::ProxyMismatch {
+                    existing: "None".to_string(),
+                    incoming: "proxy".to_string()
+                })
+            );
+
+            let res = _try_increase_pledge(
+                test.deps_mut(),
+                env.clone(),
+                Vec::new(),
+                owner_with_proxy.clone(),
+                None,
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::ProxyMismatch {
+                    existing: "proxy".to_string(),
+                    incoming: "None".to_string()
+                })
+            );
+
+            let res = _try_increase_pledge(
+                test.deps_mut(),
+                env,
+                Vec::new(),
+                owner_with_proxy.clone(),
+                Some(wrong_proxy),
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::ProxyMismatch {
+                    existing: "proxy".to_string(),
+                    incoming: "unrelated-proxy".to_string()
+                })
+            )
+        }
+
+        #[test]
+        fn is_not_allowed_if_mixnode_has_unbonded_or_is_unbonding() {
+            let mut test = TestSetup::new();
+            let env = test.env();
+
+            // TODO: I dislike this cross-test access, but it provides us with exactly what we need
+            // perhaps it should be refactored a bit?
+            let owner_unbonding = Addr::unchecked(OWNER_UNBONDING);
+            let owner_unbonded = Addr::unchecked(OWNER_UNBONDED);
+            let owner_unbonded_leftover = Addr::unchecked(OWNER_UNBONDED_LEFTOVER);
+
+            let ids = setup_mix_combinations(&mut test);
+            let mix_id_unbonding = ids[1];
+
+            let res = try_increase_pledge(
+                test.deps_mut(),
+                env.clone(),
+                mock_info(owner_unbonding.as_str(), &[]),
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::MixnodeIsUnbonding {
+                    mix_id: mix_id_unbonding
+                })
+            );
+
+            // if the nodes are gone we treat them as tey never existed in the first place
+            // (regardless of if there's some leftover data)
+            let res = try_increase_pledge(
+                test.deps_mut(),
+                env.clone(),
+                mock_info(owner_unbonded_leftover.as_str(), &[]),
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::NoAssociatedMixNodeBond {
+                    owner: owner_unbonded_leftover
+                })
+            );
+
+            let res = try_increase_pledge(
+                test.deps_mut(),
+                env,
+                mock_info(owner_unbonded.as_str(), &[]),
+            );
+            assert_eq!(
+                res,
+                Err(MixnetContractError::NoAssociatedMixNodeBond {
+                    owner: owner_unbonded
+                })
+            )
+        }
+
+        #[test]
+        fn is_not_allowed_if_no_tokens_were_sent() {
+            let mut test = TestSetup::new();
+            let env = test.env();
+            let owner = "mix-owner";
+
+            test.add_dummy_mixnode(owner, None);
+
+            let sender_empty = mock_info(owner, &[]);
+            let res = try_increase_pledge(test.deps_mut(), env.clone(), sender_empty);
+            assert_eq!(res, Err(MixnetContractError::NoBondFound));
+
+            let sender_zero = mock_info(owner, &[test.coin(0)]);
+            let res = try_increase_pledge(test.deps_mut(), env, sender_zero);
+            assert_eq!(
+                res,
+                Err(MixnetContractError::InsufficientPledge {
+                    received: test.coin(0),
+                    minimum: test.coin(1)
+                })
+            )
+        }
+
+        #[test]
+        fn with_valid_information_creates_pending_event() {
+            let mut test = TestSetup::new();
+            let env = test.env();
+            let owner = "mix-owner";
+            let mix_id = test.add_dummy_mixnode(owner, None);
+
+            let events = test.pending_epoch_events();
+            assert!(events.is_empty());
+
+            let sender = mock_info(owner, &[test.coin(1000)]);
+            try_increase_pledge(test.deps_mut(), env, sender).unwrap();
+
+            let events = test.pending_epoch_events();
+
+            assert_eq!(
+                events[0].kind,
+                PendingEpochEventKind::PledgeMore {
+                    mix_id,
+                    amount: test.coin(1000)
+                }
+            );
+        }
+    }
 }

contracts/mixnet/src/support/tests/mod.rs

@@ -28,6 +28,9 @@ pub mod test_helpers {
     use crate::mixnodes::transactions::{
         try_add_mixnode, try_add_mixnode_on_behalf, try_remove_mixnode,
     };
+    use crate::rewards::queries::{
+        query_pending_delegator_reward, query_pending_mixnode_operator_reward,
+    };
     use crate::rewards::storage as rewards_storage;
     use crate::rewards::transactions::try_reward_mixnode;
     use crate::support::tests;
@@ -37,7 +40,7 @@ pub mod test_helpers {
     use cosmwasm_std::testing::mock_info;
     use cosmwasm_std::testing::MockApi;
     use cosmwasm_std::testing::MockQuerier;
-    use cosmwasm_std::{Addr, BankMsg, CosmosMsg, Storage};
+    use cosmwasm_std::{coin, Addr, BankMsg, CosmosMsg, Storage};
     use cosmwasm_std::{Coin, Order};
     use cosmwasm_std::{Decimal, Empty, MemoryStorage};
     use cosmwasm_std::{Deps, OwnedDeps};
@@ -138,6 +141,10 @@ pub mod test_helpers {
                 .vesting_contract_address
         }
 
+        pub fn coin(&self, amount: u128) -> Coin {
+            coin(amount, rewarding_denom(self.deps().storage).unwrap())
+        }
+
         pub fn current_interval(&self) -> Interval {
             interval_storage::current_interval(self.deps().storage).unwrap()
         }
@@ -308,6 +315,22 @@ pub mod test_helpers {
             .unwrap();
         }
 
+        #[allow(unused)]
+        pub fn pending_operator_reward(&mut self, mix: MixId) -> Decimal {
+            query_pending_mixnode_operator_reward(self.deps(), mix)
+                .unwrap()
+                .amount_earned_detailed
+                .expect("no reward!")
+        }
+
+        #[allow(unused)]
+        pub fn pending_delegator_reward(&mut self, delegator: &str, target: MixId) -> Decimal {
+            query_pending_delegator_reward(self.deps(), delegator.into(), target, None)
+                .unwrap()
+                .amount_earned_detailed
+                .expect("no reward!")
+        }
+
         pub fn skip_to_next_epoch_end(&mut self) {
             self.skip_to_next_epoch();
             self.skip_to_current_epoch_end();
@@ -337,8 +360,19 @@ pub mod test_helpers {
             self.env.block.time = Timestamp::from_seconds(epoch_end as u64 + 1);
             let advanced = interval.advance_epoch();
 
-            if interval.current_epoch_id() != interval.epochs_in_interval() {
+            assert_eq!(
+                interval.current_epoch_absolute_id() + 1,
+                advanced.current_epoch_absolute_id()
+            );
+
+            if interval.current_epoch_id() != interval.epochs_in_interval() - 1 {
                 assert_eq!(interval.current_epoch_id() + 1, advanced.current_epoch_id())
+            } else {
+                assert_eq!(advanced.current_epoch_id(), 0);
+                assert_eq!(
+                    interval.current_interval_id() + 1,
+                    advanced.current_interval_id()
+                )
             }
 
             interval_storage::save_interval(self.deps_mut().storage, &advanced).unwrap()

contracts/vesting/src/contract.rs

@@ -121,6 +121,7 @@ pub fn execute(
             env,
             deps,
         ),
+        ExecuteMsg::PledgeMore { amount } => try_pledge_more(deps, env, info, amount),
         ExecuteMsg::UnbondMixnode {} => try_unbond_mixnode(info, deps),
         ExecuteMsg::TrackUnbondMixnode { owner, amount } => {
             try_track_unbond_mixnode(&owner, amount, info, deps)
@@ -331,6 +332,19 @@ pub fn try_bond_mixnode(
     )
 }
 
+pub fn try_pledge_more(
+    deps: DepsMut<'_>,
+    env: Env,
+    info: MessageInfo,
+    amount: Coin,
+) -> Result<Response, ContractError> {
+    let mix_denom = MIX_DENOM.load(deps.storage)?;
+    let additional_pledge = validate_funds(&[amount], mix_denom)?;
+
+    let account = account_from_address(info.sender.as_str(), deps.storage, deps.api)?;
+    account.try_pledge_additional_tokens(additional_pledge, &env, deps.storage)
+}
+
 /// Unbond a mixnode, sends [mixnet_contract_common::ExecuteMsg::UnbondMixnodeOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_unbond_mixnode(info: MessageInfo, deps: DepsMut<'_>) -> Result<Response, ContractError> {
     let account = account_from_address(info.sender.as_str(), deps.storage, deps.api)?;

contracts/vesting/src/traits/bonding_account.rs

@@ -18,6 +18,13 @@ pub trait MixnodeBondingAccount {
         storage: &mut dyn Storage,
     ) -> Result<Response, ContractError>;
 
+    fn try_pledge_additional_tokens(
+        &self,
+        additional_pledge: Coin,
+        env: &Env,
+        storage: &mut dyn Storage,
+    ) -> Result<Response, ContractError>;
+
     fn try_unbond_mixnode(&self, storage: &dyn Storage) -> Result<Response, ContractError>;
 
     fn try_track_unbond_mixnode(

contracts/vesting/src/vesting/account/mixnode_bonding_account.rs

@@ -12,7 +12,8 @@ use mixnet_contract_common::mixnode::MixNodeCostParams;
 use mixnet_contract_common::{ExecuteMsg as MixnetExecuteMsg, MixNode};
 use vesting_contract_common::events::{
     new_vesting_mixnode_bonding_event, new_vesting_mixnode_unbonding_event,
-    new_vesting_update_mixnode_config_event, new_vesting_update_mixnode_cost_params_event,
+    new_vesting_pledge_more_event, new_vesting_update_mixnode_config_event,
+    new_vesting_update_mixnode_cost_params_event,
 };
 use vesting_contract_common::PledgeData;
 
@@ -84,6 +85,64 @@ impl MixnodeBondingAccount for Account {
             .add_event(new_vesting_mixnode_bonding_event()))
     }
 
+    fn try_pledge_additional_tokens(
+        &self,
+        additional_pledge: Coin,
+        env: &Env,
+        storage: &mut dyn Storage,
+    ) -> Result<Response, ContractError> {
+        let current_balance = self.load_balance(storage)?;
+        let total_pledged_after =
+            self.total_pledged_locked(storage, env)? + additional_pledge.amount;
+        let locked_pledge_cap = self.absolute_pledge_cap()?;
+
+        if locked_pledge_cap < total_pledged_after {
+            return Err(ContractError::LockedPledgeCapReached {
+                current: total_pledged_after,
+                cap: locked_pledge_cap,
+            });
+        }
+
+        if current_balance < additional_pledge.amount {
+            return Err(ContractError::InsufficientBalance(
+                self.owner_address().as_str().to_string(),
+                current_balance.u128(),
+            ));
+        }
+
+        let mut pledge_data = if let Some(pledge_data) = self.load_mixnode_pledge(storage)? {
+            pledge_data
+        } else {
+            return Err(ContractError::NoBondFound(
+                self.owner_address().as_str().to_string(),
+            ));
+        };
+
+        // need a second pair of eyes here to make sure updating existing timestamp on pledge data
+        // is not going to have some unexpected consequences
+        pledge_data.amount.amount += additional_pledge.amount;
+        pledge_data.block_time = env.block.time;
+
+        let msg = MixnetExecuteMsg::PledgeMoreOnBehalf {
+            owner: self.owner_address().into_string(),
+        };
+
+        let new_balance = Uint128::new(current_balance.u128() - additional_pledge.amount.u128());
+
+        let pledge_more_mag = wasm_execute(
+            MIXNET_CONTRACT_ADDRESS.load(storage)?,
+            &msg,
+            vec![additional_pledge],
+        )?;
+
+        self.save_balance(new_balance, storage)?;
+        self.save_mixnode_pledge(pledge_data, storage)?;
+
+        Ok(Response::new()
+            .add_message(pledge_more_mag)
+            .add_event(new_vesting_pledge_more_event()))
+    }
+
     fn try_unbond_mixnode(&self, storage: &dyn Storage) -> Result<Response, ContractError> {
         let msg = MixnetExecuteMsg::UnbondMixnodeOnBehalf {
             owner: self.owner_address().into_string(),

explorer-api/Cargo.toml

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

gateway/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-gateway"
-version = "1.1.0"
+version = "1.1.1"
 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.0"
+version = "1.1.1"
 authors = [
     "Dave Hrycyszyn <futurechimp@users.noreply.github.com>",
     "Jędrzej Stuczyński <andrew@nymtech.net>",

nym-connect/CHANGELOG.md

@@ -1,3 +1,23 @@
+## [nym-connect-v1.1.1](https://github.com/nymtech/nym/tree/nym-connect-v1.1.1) (2022-11-29)
+
+- socks5-client: fix multiplex concurrent connections ([#1720], [#1777])
+- socks5-client: fix wait closing inbound connection until data is sent, and throttle incoming data in general ([#1772], [#1783],[#1789])
+- socks5-client: fix shutting down all background workers if anyone of them panics or errors out. This fixes an issue where the nym-connect UI was showing connected even though the socks5 tunnel was non-functional. ([#1805])
+- gateway-libs: fix decryping messages stored on the gateway between reconnects ([#1786])
+
+- nymconnect: updated UI
+- nymconnect: new help area
+- nymconnect: listen for service errors and display on frontend
+
+[#1720]: https://github.com/nymtech/nym/pull/1720
+[#1772]: https://github.com/nymtech/nym/pull/1772
+[#1777]: https://github.com/nymtech/nym/pull/1777
+[#1783]: https://github.com/nymtech/nym/pull/1783
+[#1786]: https://github.com/nymtech/nym/pull/1786
+[#1789]: https://github.com/nymtech/nym/pull/1789
+[#1805]: https://github.com/nymtech/nym/pull/1805
+
+
 ## [nym-connect-v1.1.0](https://github.com/nymtech/nym/tree/nym-connect-v1.1.0) (2022-11-09)
 
 - nym-connect: rework of rewarding changes the directory data structures that describe the mixnet topology ([#1472])
@@ -39,4 +59,4 @@
 
 - nym-connect: reuse config id instead of creating a new id on each connection
 
-[#1427]: https://github.com/nymtech/nym/pull/1427
+[#1427]: https://github.com/nymtech/nym/pull/1427

nym-connect/Cargo.lock

@@ -630,7 +630,7 @@ dependencies = [
 
 [[package]]
 name = "client-core"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "client-connections",
  "config",
@@ -652,6 +652,7 @@ dependencies = [
  "task",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "topology",
  "url",
  "validator-client",
@@ -3351,7 +3352,7 @@ dependencies = [
 
 [[package]]
 name = "nym-socks5-client"
-version = "1.1.0"
+version = "1.1.1"
 dependencies = [
  "clap",
  "client-connections",

nym-connect/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nym/nym-connect",
-  "version": "1.0.0",
+  "version": "1.1.0",
   "main": "index.js",
   "license": "MIT",
   "scripts": {
@@ -107,4 +107,4 @@
     "webpack-favicons": "^1.3.8",
     "webpack-merge": "^5.8.0"
   }
-}
+}

nym-connect/src-tauri/src/config/mod.rs

@@ -1,6 +1,6 @@
 use std::path::PathBuf;
 
-use client_core::config::GatewayEndpoint;
+use client_core::config::GatewayEndpointConfig;
 use std::sync::Arc;
 use tap::TapFallible;
 use tokio::sync::RwLock;
@@ -169,7 +169,7 @@ async fn setup_gateway(
     register: bool,
     user_chosen_gateway_id: Option<&str>,
     config: &Socks5Config,
-) -> Result<GatewayEndpoint> {
+) -> Result<GatewayEndpointConfig> {
     if register {
         // Get the gateway details by querying the validator-api. Either pick one at random or use
         // the chosen one if it's among the available ones.

nym-connect/src-tauri/src/tasks.rs

@@ -1,4 +1,4 @@
-use client_core::config::GatewayEndpoint;
+use client_core::config::GatewayEndpointConfig;
 use futures::channel::mpsc;
 use std::sync::Arc;
 use tap::TapFallible;
@@ -25,13 +25,17 @@ pub enum Socks5StatusMessage {
 /// The main SOCKS5 client task. It loads the configuration from file determined by the `id`.
 pub fn start_nym_socks5_client(
     id: &str,
-) -> Result<(Socks5ControlMessageSender, StatusReceiver, GatewayEndpoint)> {
+) -> Result<(
+    Socks5ControlMessageSender,
+    StatusReceiver,
+    GatewayEndpointConfig,
+)> {
     log::info!("Loading config from file: {id}");
     let config = Socks5Config::load_from_file(Some(id))
         .tap_err(|_| log::warn!("Failed to load configuration file"))?;
     let used_gateway = config.get_base().get_gateway_endpoint().clone();
 
-    let mut socks5_client = Socks5NymClient::new(config);
+    let socks5_client = Socks5NymClient::new(config);
     log::info!("Starting socks5 client");
 
     // Channel to send control messages to the socks5 client

nym-wallet/src-tauri/src/main.rs

@@ -55,6 +55,7 @@ fn main() {
             mixnet::admin::update_contract_settings,
             mixnet::bond::bond_gateway,
             mixnet::bond::bond_mixnode,
+            mixnet::bond::pledge_more,
             mixnet::bond::gateway_bond_details,
             mixnet::bond::get_pending_operator_rewards,
             mixnet::bond::mixnode_bond_details,
@@ -105,6 +106,7 @@ fn main() {
             vesting::rewards::vesting_claim_operator_reward,
             vesting::bond::vesting_bond_gateway,
             vesting::bond::vesting_bond_mixnode,
+            vesting::bond::vesting_pledge_more,
             vesting::bond::vesting_unbond_gateway,
             vesting::bond::vesting_unbond_mixnode,
             vesting::bond::vesting_update_mixnode_cost_params,
@@ -130,6 +132,7 @@ fn main() {
             simulate::mixnet::simulate_bond_gateway,
             simulate::mixnet::simulate_unbond_gateway,
             simulate::mixnet::simulate_bond_mixnode,
+            simulate::mixnet::simulate_pledge_more,
             simulate::mixnet::simulate_unbond_mixnode,
             simulate::mixnet::simulate_update_mixnode_config,
             simulate::mixnet::simulate_update_mixnode_cost_params,
@@ -140,6 +143,7 @@ fn main() {
             simulate::vesting::simulate_vesting_bond_gateway,
             simulate::vesting::simulate_vesting_unbond_gateway,
             simulate::vesting::simulate_vesting_bond_mixnode,
+            simulate::vesting::simulate_vesting_pledge_more,
             simulate::vesting::simulate_vesting_unbond_mixnode,
             simulate::vesting::simulate_vesting_update_mixnode_config,
             simulate::vesting::simulate_vesting_update_mixnode_cost_params,

nym-wallet/src-tauri/src/operations/mixnet/bond.rs

@@ -102,6 +102,33 @@ pub async fn bond_mixnode(
     )?)
 }
 
+#[tauri::command]
+pub async fn pledge_more(
+    fee: Option<Fee>,
+    additional_pledge: DecCoin,
+    state: tauri::State<'_, WalletState>,
+) -> Result<TransactionExecuteResult, BackendError> {
+    let guard = state.read().await;
+    let additional_pledge_base = guard.attempt_convert_to_base_coin(additional_pledge.clone())?;
+    let fee_amount = guard.convert_tx_fee(fee.as_ref());
+    log::info!(
+        ">>> Pledge more, additional_pledge_display = {}, additional_pledge_base = {}, fee = {:?}",
+        additional_pledge,
+        additional_pledge_base,
+        fee,
+    );
+    let res = guard
+        .current_client()?
+        .nymd
+        .pledge_more(additional_pledge_base, fee)
+        .await?;
+    log::info!("<<< tx hash = {}", res.transaction_hash);
+    log::trace!("<<< {:?}", res);
+    Ok(TransactionExecuteResult::from_execute_result(
+        res, fee_amount,
+    )?)
+}
+
 #[tauri::command]
 pub async fn unbond_mixnode(
     fee: Option<Fee>,

nym-wallet/src-tauri/src/operations/simulate/mixnet.rs

@@ -84,6 +84,14 @@ pub async fn simulate_bond_mixnode(
     .await
 }
 
+#[tauri::command]
+pub async fn simulate_pledge_more(
+    additional_pledge: DecCoin,
+    state: tauri::State<'_, WalletState>,
+) -> Result<FeeDetails, BackendError> {
+    simulate_mixnet_operation(ExecuteMsg::PledgeMore {}, Some(additional_pledge), &state).await
+}
+
 #[tauri::command]
 pub async fn simulate_unbond_mixnode(
     state: tauri::State<'_, WalletState>,

nym-wallet/src-tauri/src/operations/simulate/vesting.rs

@@ -91,6 +91,19 @@ pub async fn simulate_vesting_bond_mixnode(
     .await
 }
 
+#[tauri::command]
+pub async fn simulate_vesting_pledge_more(
+    additional_pledge: DecCoin,
+    state: tauri::State<'_, WalletState>,
+) -> Result<FeeDetails, BackendError> {
+    let guard = state.read().await;
+    let amount = guard
+        .attempt_convert_to_base_coin(additional_pledge)?
+        .into();
+
+    simulate_vesting_operation(ExecuteMsg::PledgeMore { amount }, None, &state).await
+}
+
 #[tauri::command]
 pub async fn simulate_vesting_unbond_mixnode(
     state: tauri::State<'_, WalletState>,

nym-wallet/src-tauri/src/operations/vesting/bond.rs

@@ -93,6 +93,33 @@ pub async fn vesting_bond_mixnode(
     )?)
 }
 
+#[tauri::command]
+pub async fn vesting_pledge_more(
+    fee: Option<Fee>,
+    additional_pledge: DecCoin,
+    state: tauri::State<'_, WalletState>,
+) -> Result<TransactionExecuteResult, BackendError> {
+    let guard = state.read().await;
+    let additional_pledge_base = guard.attempt_convert_to_base_coin(additional_pledge.clone())?;
+    let fee_amount = guard.convert_tx_fee(fee.as_ref());
+    log::info!(
+        ">>> Pledge more with locked tokens, additional_pledge_display = {}, additional_pledge_base = {}, fee = {:?}",
+        additional_pledge,
+        additional_pledge_base,
+        fee,
+    );
+    let res = guard
+        .current_client()?
+        .nymd
+        .vesting_pledge_more(additional_pledge_base, fee)
+        .await?;
+    log::info!("<<< tx hash = {}", res.transaction_hash);
+    log::trace!("<<< {:?}", res);
+    Ok(TransactionExecuteResult::from_execute_result(
+        res, fee_amount,
+    )?)
+}
+
 #[tauri::command]
 pub async fn vesting_unbond_mixnode(
     fee: Option<Fee>,

sdk/rust/nym-sdk/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "nym-sdk"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+client-core = { path = "../../../clients/client-core" }
+rand = { version = "0.7.3" }
+
+[dev-dependencies]
+tokio = { version = "1", features = ["full"] }

sdk/rust/nym-sdk/examples/complex_config.rs

@@ -0,0 +1,24 @@
+use std::path::PathBuf;
+
+use nym_sdk::mixnet;
+
+#[tokio::main]
+async fn main() {
+    // specify some config options
+    let keys = Some(PathBuf::from("~/.nym/clients/superfoomp"));
+    let config = mixnet::Config { keys };
+
+    let client = mixnet::Client::new(Some(config)); // passing a config allows the user to set values
+
+    // let show_receive = move || println!("got a message from the mixnet: {}", message); // might need to bury this in a struct as a `FnOnce`, see https://stackoverflow.com/questions/41081240/idiomatic-callbacks-in-rust
+    // client.on_receive(show_receive); // have some way to pipe any received info to a function for processing
+
+    // connect to the mixnet, now we're listening for incoming
+    client.connect_to_mixnet();
+
+    // be able to get our client address
+    println!("Our client address is {}", client.nym_address);
+
+    // send important info up the pipe to a buddy
+    client.send_str("foo.bar@blah", "flappappa");
+}

sdk/rust/nym-sdk/examples/simple.rs

@@ -0,0 +1,22 @@
+use nym_sdk::mixnet;
+
+#[tokio::main]
+async fn main() {
+    // here's what I'd actually like to write
+    let client = mixnet::Client::new(None); // passing no config makes the client fire up an ephemeral session and figure shit out on its own
+
+    // let show_receive = move || println!("got a message from the mixnet: {}", message); // might need to bury this in a struct as a `FnMut`, see https://stackoverflow.com/questions/41081240/idiomatic-callbacks-in-rust
+    // connect to the mixnet, now we're listening for incoming
+    client.connect_to_mixnet();
+
+    // be able to get our client address
+    println!("Our client address is {}", client.nym_address);
+
+    // send important string info up the pipe to a buddy
+    client.send_str("foo.bar@blah", "flappappa");
+
+    // send some bytes to a buddy
+    client.send_bytes("foo.bar@blah", "flappappa".as_bytes().to_vec());
+
+    
+}

sdk/rust/nym-sdk/src/lib.rs

@@ -0,0 +1 @@
+pub mod mixnet;

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

@@ -0,0 +1,62 @@
+use std::path::PathBuf;
+
+pub struct Client {
+    pub nym_address: String,
+    config: Config,
+}
+
+impl Client {
+    /// Create a new mixnet client. If no config options are supplied, creates a new client with ephemeral keys
+    /// stored in RAM, which will be discarded at application close.
+    ///
+    /// Callers have the option of supplying futher parameters to store persistent identities at a location on-disk,
+    /// if desired.
+    pub fn new(config_option: Option<Config>) -> Client {
+        if config_option.is_none() {
+            let config = Self::some_minimal_config();
+            let nym_address = Self::new_nym_address();
+            Client {
+                config,
+                nym_address,
+            }
+        } else {
+            let self_address = Self::new_nym_address();
+            let nym_address = Self::new_nym_address();
+
+            Client {
+                config: config_option.unwrap(),
+                nym_address,
+            }
+        }
+    }
+
+    /// Connects to the mixnet via the gateway in the client config
+    pub fn connect_to_mixnet(&self) {}
+
+    /// Sets the callback function which is run when the client receives a message
+    /// from the mixnet
+    pub fn on_receive(&mut self, message: &str) {
+        println!("Message received: {}", message);
+    }
+
+    /// Sends stringy data to the supplied Nym address
+    pub fn send_str(&self, address: &str, message: &str) {}
+
+    /// Sends bytes to the supplied Nym address
+    pub fn send_bytes(&self, address: &str, message: Vec<u8>) {}
+
+    fn new_nym_address() -> String {
+        "the.nym@address".to_string()
+    }
+
+    fn some_minimal_config() -> Config {
+        let mut keys_path = PathBuf::new();
+        Config {
+            keys: Some(keys_path),
+        }
+    }
+}
+
+pub struct Config {
+    pub keys: Option<PathBuf>,
+}

service-providers/network-requester/Cargo.toml

@@ -3,7 +3,7 @@
 
 [package]
 name = "nym-network-requester"
-version = "1.1.0"
+version = "1.1.1"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jędrzej Stuczyński <andrew@nymtech.net>"]
 edition = "2021"
 rust-version = "1.65"

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

@@ -157,9 +157,13 @@ impl ServiceProvider {
         lane_queue_lengths: LaneQueueLengths,
     ) -> Option<ReconstructedMessage> {
         while let Some(msg) = websocket_reader.next().await {
-            let data = msg
+            let data = match msg {
-                .expect("we failed to read from the websocket!")
+                Ok(msg) => msg.into_data(),
-                .into_data();
+                Err(err) => {
+                    log::error!("Failed to read from the websocket: {err}");
+                    continue;
+                }
+            };
 
             // try to recover the actual message from the mix network...
             let deserialized_message = match ServerResponse::deserialize(&data) {

tools/nym-cli/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym-cli"
-version = "1.1.0"
+version = "1.1.1"
 authors = ["Nym Technologies SA"]
 edition = "2021"
 

validator-api/Cargo.toml

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