fix build and refactor from CR

* Rename to nym-wallet-recovery-cli

* Rename crate name too

* Rename workspace member too

* Cargo.lock

.github/workflows/nightly_build.yml

@@ -44,6 +44,12 @@ jobs:
           command: build
           args: --workspace
 
+      - name: Reclaim some disk space (because Windows is being annoying)
+        uses: actions-rs/cargo@v1
+        if: ${{ matrix.os == 'windows-latest' }}
+        with:
+          command: clean
+
       - name: Run all tests
         uses: actions-rs/cargo@v1
         with:
@@ -101,6 +107,12 @@ jobs:
           command: build
           args: --workspace --features=coconut
 
+      - name: Reclaim some disk space (because Windows is being annoying)
+        uses: actions-rs/cargo@v1
+        if: ${{ matrix.os == 'windows-latest' }}
+        with:
+          command: clean
+
       - name: Run all tests with coconut enabled
         uses: actions-rs/cargo@v1
         with:

.github/workflows/nightly_build_on_dispatch.yml

@@ -167,7 +167,7 @@ jobs:
           KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
           KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
           KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMTECH_TEAM }}"
-          KEYBASE_NYM_CHANNEL: "test"
+          KEYBASE_NYM_CHANNEL: "${{ secrets.KEYBASE_CHANNEL_DEV_CORE_ID }}"
           IS_SUCCESS: "${{ env.WORKFLOW_CONCLUSION == 'success' }}"
         uses: docker://keybaseio/client:stable-node
         with:

.github/workflows/support-files/notifications/send_message.js

@@ -89,7 +89,7 @@ async function sendKeybaseMessage(messageBody) {
     });
 
     const channel = {
-      name: 'nymtech_bot',
+      name: context.env.KEYBASE_NYMBOT_TEAM || 'nymtech_bot',
       membersType: 'team',
       topicName: context.keybase.channel,
       topic_type: 'CHAT',

CHANGELOG.md

@@ -8,14 +8,19 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 - nym-cli: added CLI tool for interacting with the Nyx blockchain and Nym mixnet smart contracts ([#1577])
 - validator-client: added `query_contract_smart` and `query_contract_raw` on `NymdClient` ([#1558])
+- network-requester: added additional Blockstream Green wallet endpoint to `example.allowed.list` ([#1611](https://github.com/nymtech/nym/pull/1611))
 
 ### Changed
 
 - validator-client: made `fee` argument optional for `execute` and `execute_multiple` ([#1541])
+- socks5 client: graceful shutdown should fix error on disconnect in nym-connect ([#1591])
+- wasm-client: fixed build errors on MacOS and changed example JS code to use mainnet ([#1585])
 
 [#1541]: https://github.com/nymtech/nym/pull/1541
 [#1558]: https://github.com/nymtech/nym/pull/1558
 [#1577]: https://github.com/nymtech/nym/pull/1577
+[#1585]: https://github.com/nymtech/nym/pull/1585
+[#1591]: https://github.com/nymtech/nym/pull/1591
 
 
 ## [nym-binaries-1.0.2](https://github.com/nymtech/nym/tree/nym-binaries-1.0.2)

Cargo.toml

@@ -85,4 +85,4 @@ default-members = [
     "explorer-api",
 ]
 
-exclude = ["explorer", "contracts", "tokenomics-py", "clients/webassembly", "nym-wallet", "nym-connect"]
+exclude = ["explorer", "contracts", "clients/webassembly", "nym-wallet", "nym-connect"]

Makefile

@@ -2,10 +2,10 @@ test: clippy-all cargo-test wasm fmt
 test-all: test cargo-test-expensive
 no-clippy: build cargo-test wasm fmt
 happy: fmt clippy-happy test
-clippy-all: clippy-all-main clippy-all-contracts clippy-all-wallet clippy-all-connect
+clippy-all: clippy-main clippy-coconut clippy-all-contracts clippy-all-wallet clippy-all-connect
 clippy-happy: clippy-happy-main clippy-happy-contracts clippy-happy-wallet clippy-happy-connect
-cargo-test: test-main test-contracts test-wallet test-connect
-cargo-test-expensive: test-main-expensive test-contracts-expensive test-wallet-expensive test-connect-expensive
+cargo-test: test-main test-contracts test-wallet test-connect test-coconut
+cargo-test-expensive: test-main-expensive test-contracts-expensive test-wallet-expensive test-connect-expensive test-coconut-expensive
 build: build-contracts build-wallet build-main build-connect
 fmt: fmt-main fmt-contracts fmt-wallet fmt-connect
 
@@ -21,8 +21,15 @@ clippy-happy-wallet:
 clippy-happy-connect:
 	cargo clippy --manifest-path nym-connect/Cargo.toml
 
-clippy-all-main:
-	cargo clippy --workspace --all-features -- -D warnings
+clippy-main:
+	cargo clippy --workspace -- -D warnings
+
+clippy-coconut:
+	cargo clippy --workspace --features coconut -- -D warnings
+
+clippy-wasm:
+	cargo clippy --workspace --features wasm -- -D warnings
+
 
 clippy-all-contracts:
 	cargo clippy --workspace --manifest-path contracts/Cargo.toml --all-features --target wasm32-unknown-unknown -- -D warnings
@@ -34,10 +41,20 @@ clippy-all-connect:
 	cargo clippy --workspace --manifest-path nym-connect/Cargo.toml --all-features -- -D warnings
 
 test-main:
-	cargo test --all-features --workspace
+	cargo test --workspace
+
+test-coconut:
+	cargo test --workspace --features coconut
+
+test-wasm:
+	cargo test --workspace --features wasm
+
 
 test-main-expensive:
-	cargo test --all-features --workspace -- --ignored
+	cargo test --workspace -- --ignored
+
+test-coconut-expensive:
+	cargo test --workspace --features coconut -- --ignored
 
 test-contracts:
 	cargo test --manifest-path contracts/Cargo.toml --all-features

clients/client-core/Cargo.toml

@@ -14,6 +14,7 @@ log = "0.4"
 rand = { version = "0.7.3", features = ["wasm-bindgen"] }
 serde = { version = "1.0", features = ["derive"] }
 sled = "0.34"
+thiserror = "1.0.34"
 tokio = { version = "1.19.1", features = ["macros"] }
 url = { version ="2.2", features = ["serde"] }
 
@@ -25,8 +26,10 @@ gateway-requests = { path = "../../gateway/gateway-requests" }
 nonexhaustive-delayqueue = { path = "../../common/nonexhaustive-delayqueue" }
 nymsphinx = { path = "../../common/nymsphinx" }
 pemstore = { path = "../../common/pemstore" }
+task = { path = "../../common/task" }
 topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client" }
+tap = "1.0.1"
 
 [dev-dependencies]
 tempfile = "3.1.0"

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

@@ -13,6 +13,7 @@ use nymsphinx::utils::sample_poisson_duration;
 use rand::{rngs::OsRng, CryptoRng, Rng};
 use std::pin::Pin;
 use std::sync::Arc;
+use task::ShutdownListener;
 use tokio::task::JoinHandle;
 use tokio::time;
 
@@ -48,6 +49,9 @@ where
 
     /// Accessor to the common instance of network topology.
     topology_access: TopologyAccessor,
+
+    /// Listen to shutdown signals.
+    shutdown: ShutdownListener,
 }
 
 impl<R> Stream for LoopCoverTrafficStream<R>
@@ -84,6 +88,7 @@ where
 // obviously when we finally make shared rng that is on 'higher' level, this should become
 // generic `R`
 impl LoopCoverTrafficStream<OsRng> {
+    #[allow(clippy::too_many_arguments)]
     pub fn new(
         ack_key: Arc<AckKey>,
         average_ack_delay: time::Duration,
@@ -92,6 +97,7 @@ impl LoopCoverTrafficStream<OsRng> {
         mix_tx: BatchMixMessageSender,
         our_full_destination: Recipient,
         topology_access: TopologyAccessor,
+        shutdown: ShutdownListener,
     ) -> Self {
         let rng = OsRng;
 
@@ -105,6 +111,7 @@ impl LoopCoverTrafficStream<OsRng> {
             our_full_destination,
             rng,
             topology_access,
+            shutdown,
         }
     }
 
@@ -159,9 +166,25 @@ impl LoopCoverTrafficStream<OsRng> {
             self.average_cover_message_sending_delay,
         )));
 
-        while self.next().await.is_some() {
-            self.on_new_message().await;
+        let mut shutdown = self.shutdown.clone();
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("LoopCoverTrafficStream: Received shutdown");
+                }
+                next = self.next() => {
+                    if next.is_some() {
+                        self.on_new_message().await;
+                    } else {
+                        log::trace!("LoopCoverTrafficStream: Stopping since channel closed");
+                        break;
+                    }
+                }
+            }
         }
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("LoopCoverTrafficStream: Exiting");
     }
 
     pub fn start(mut self) -> JoinHandle<()> {

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

@@ -6,6 +6,7 @@ use futures::StreamExt;
 use gateway_client::GatewayClient;
 use log::*;
 use nymsphinx::forwarding::packet::MixPacket;
+use task::ShutdownListener;
 use tokio::task::JoinHandle;
 
 pub type BatchMixMessageSender = mpsc::UnboundedSender<Vec<MixPacket>>;
@@ -18,6 +19,7 @@ pub struct MixTrafficController {
     // later on gateway_client will need to be accessible by other entities
     gateway_client: GatewayClient,
     mix_rx: BatchMixMessageReceiver,
+    shutdown: ShutdownListener,
 
     // TODO: this is temporary work-around.
     // in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
@@ -28,10 +30,12 @@ impl MixTrafficController {
     pub fn new(
         mix_rx: BatchMixMessageReceiver,
         gateway_client: GatewayClient,
+        shutdown: ShutdownListener,
     ) -> MixTrafficController {
         MixTrafficController {
             gateway_client,
             mix_rx,
+            shutdown,
             consecutive_gateway_failure_count: 0,
         }
     }
@@ -66,9 +70,24 @@ impl MixTrafficController {
     }
 
     pub async fn run(&mut self) {
-        while let Some(mix_packets) = self.mix_rx.next().await {
-            self.on_messages(mix_packets).await;
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                mix_packets = self.mix_rx.next() => match mix_packets {
+                    Some(mix_packets) => {
+                        self.on_messages(mix_packets).await;
+                    },
+                    None => {
+                        log::trace!("MixTrafficController: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("MixTrafficController: Received shutdown");
+                }
+            }
         }
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("MixTrafficController: Exiting");
     }
 
     pub fn start(mut self) -> JoinHandle<()> {

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

@@ -1,3 +1,5 @@
+use std::sync::atomic::AtomicBool;
+
 pub mod cover_traffic_stream;
 pub mod inbound_messages;
 pub mod key_manager;
@@ -6,3 +8,10 @@ pub mod real_messages_control;
 pub mod received_buffer;
 pub mod reply_key_storage;
 pub mod topology_control;
+
+// This is *NOT* used to signal shutdown.
+// It's critical that we don't have any tasks finishing early, this is an additional safety check
+// that tasks exiting are doing so because shutdown has been signalled, and no other reason.
+// In particular for tasks that rely on their associated channel being closed to signal shutdown,
+// and don't have access to a shutdown listener channel.
+pub static SHUTDOWN_HAS_BEEN_SIGNALLED: AtomicBool = AtomicBool::new(false);

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

@@ -10,6 +10,7 @@ use nymsphinx::{
     chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
 };
 use std::sync::Arc;
+use task::ShutdownListener;
 
 /// Module responsible for listening for any data resembling acknowledgements from the network
 /// and firing actions to remove them from the 'Pending' state.
@@ -17,6 +18,7 @@ pub(super) struct AcknowledgementListener {
     ack_key: Arc<AckKey>,
     ack_receiver: AcknowledgementReceiver,
     action_sender: ActionSender,
+    shutdown: ShutdownListener,
 }
 
 impl AcknowledgementListener {
@@ -24,11 +26,13 @@ impl AcknowledgementListener {
         ack_key: Arc<AckKey>,
         ack_receiver: AcknowledgementReceiver,
         action_sender: ActionSender,
+        shutdown: ShutdownListener,
     ) -> Self {
         AcknowledgementListener {
             ack_key,
             ack_receiver,
             action_sender,
+            shutdown,
         }
     }
 
@@ -65,12 +69,26 @@ impl AcknowledgementListener {
 
     pub(super) async fn run(&mut self) {
         debug!("Started AcknowledgementListener");
-        while let Some(acks) = self.ack_receiver.next().await {
-            // realistically we would only be getting one ack at the time
-            for ack in acks {
-                self.on_ack(ack).await;
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                acks = self.ack_receiver.next() => match acks {
+                    Some(acks) => {
+                        // realistically we would only be getting one ack at the time
+                        for ack in acks {
+                            self.on_ack(ack).await;
+                        }
+                    },
+                    None => {
+                        log::trace!("AcknowledgementListener: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("AcknowledgementListener: Received shutdown");
+                }
             }
         }
-        error!("TODO: error msg. Or maybe panic?")
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("AcknowledgementListener: Exiting");
     }
 }

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

@@ -12,6 +12,7 @@ use nymsphinx::Delay as SphinxDelay;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
+use task::ShutdownListener;
 
 pub(crate) type ActionSender = UnboundedSender<Action>;
 
@@ -99,12 +100,16 @@ pub(super) struct ActionController {
 
     /// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
     retransmission_sender: RetransmissionRequestSender,
+
+    /// Listen for shutdown notifications
+    shutdown: ShutdownListener,
 }
 
 impl ActionController {
     pub(super) fn new(
         config: Config,
         retransmission_sender: RetransmissionRequestSender,
+        shutdown: ShutdownListener,
     ) -> (Self, ActionSender) {
         let (sender, receiver) = mpsc::unbounded();
         (
@@ -114,6 +119,7 @@ impl ActionController {
                 pending_acks_timers: NonExhaustiveDelayQueue::new(),
                 incoming_actions: receiver,
                 retransmission_sender,
+                shutdown,
             },
             sender,
         )
@@ -246,14 +252,30 @@ impl ActionController {
     }
 
     pub(super) async fn run(&mut self) {
-        loop {
-            // at some point there will be a global shutdown signal here as the third option
+        while !self.shutdown.is_shutdown() {
             tokio::select! {
-                // we NEVER expect for ANY sender to get dropped so unwrap here is fine
-                action = self.incoming_actions.next() => self.process_action(action.unwrap()),
-                // pending ack queue Stream CANNOT return a `None` so unwrap here is fine
-                expired_ack = self.pending_acks_timers.next() => self.handle_expired_ack_timer(expired_ack.unwrap())
+                action = self.incoming_actions.next() => match action {
+                    Some(action) => self.process_action(action),
+                    None => {
+                        log::trace!(
+                            "ActionController: Stopping since incoming actions channel closed"
+                        );
+                        break;
+                    }
+                },
+                expired_ack = self.pending_acks_timers.next() => match expired_ack {
+                    Some(expired_ack) => self.handle_expired_ack_timer(expired_ack),
+                    None => {
+                        log::trace!("ActionController: Stopping since ack channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("ActionController: Received shutdown");
+                }
             }
         }
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("ActionController: Exiting");
     }
 }

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

@@ -16,6 +16,7 @@ use nymsphinx::preparer::MessagePreparer;
 use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
 use rand::{CryptoRng, Rng};
 use std::sync::Arc;
+use task::ShutdownListener;
 
 /// Module responsible for dealing with the received messages: splitting them, creating acknowledgements,
 /// putting everything into sphinx packets, etc.
@@ -32,6 +33,7 @@ where
     real_message_sender: BatchRealMessageSender,
     topology_access: TopologyAccessor,
     reply_key_storage: ReplyKeyStorage,
+    shutdown: ShutdownListener,
 }
 
 impl<R> InputMessageListener<R>
@@ -50,6 +52,7 @@ where
         real_message_sender: BatchRealMessageSender,
         topology_access: TopologyAccessor,
         reply_key_storage: ReplyKeyStorage,
+        shutdown: ShutdownListener,
     ) -> Self {
         InputMessageListener {
             ack_key,
@@ -60,6 +63,7 @@ where
             real_message_sender,
             topology_access,
             reply_key_storage,
+            shutdown,
         }
     }
 
@@ -182,9 +186,23 @@ where
 
     pub(super) async fn run(&mut self) {
         debug!("Started InputMessageListener");
-        while let Some(input_msg) = self.input_receiver.next().await {
-            self.on_input_message(input_msg).await;
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                input_msg = self.input_receiver.next() => match input_msg {
+                    Some(input_msg) => {
+                        self.on_input_message(input_msg).await;
+                    },
+                    None => {
+                        log::trace!("InputMessageListener: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("InputMessageListener: Received shutdown");
+                }
+            }
         }
-        error!("TODO: error msg. Or maybe panic?")
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("InputMessageListener: Exiting");
     }
 }

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

@@ -25,6 +25,7 @@ use std::{
     sync::{Arc, Weak},
     time::Duration,
 };
+use task::ShutdownListener;
 use tokio::task::JoinHandle;
 
 mod acknowledgement_listener;
@@ -152,6 +153,7 @@ impl<R> AcknowledgementController<R>
 where
     R: 'static + CryptoRng + Rng + Clone + Send,
 {
+    #[allow(clippy::too_many_arguments)]
     pub(super) fn new(
         config: Config,
         rng: R,
@@ -160,13 +162,14 @@ where
         ack_recipient: Recipient,
         reply_key_storage: ReplyKeyStorage,
         connectors: AcknowledgementControllerConnectors,
+        shutdown: ShutdownListener,
     ) -> Self {
         let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
 
         let action_config =
             action_controller::Config::new(config.ack_wait_addition, config.ack_wait_multiplier);
         let (action_controller, action_sender) =
-            ActionController::new(action_config, retransmission_tx);
+            ActionController::new(action_config, retransmission_tx, shutdown.clone());
 
         let message_preparer = MessagePreparer::new(
             rng,
@@ -180,6 +183,7 @@ where
             Arc::clone(&ack_key),
             connectors.ack_receiver,
             action_sender.clone(),
+            shutdown.clone(),
         );
 
         // will listen for any new messages from the client
@@ -192,6 +196,7 @@ where
             connectors.real_message_sender.clone(),
             topology_access.clone(),
             reply_key_storage,
+            shutdown.clone(),
         );
 
         // will listen for any ack timeouts and trigger retransmission
@@ -203,12 +208,13 @@ where
             connectors.real_message_sender,
             retransmission_rx,
             topology_access,
+            shutdown.clone(),
         );
 
         // will listen for events indicating the packet was sent through the network so that
         // the retransmission timer should be started.
         let sent_notification_listener =
-            SentNotificationListener::new(connectors.sent_notifier, action_sender);
+            SentNotificationListener::new(connectors.sent_notifier, action_sender, shutdown);
 
         AcknowledgementController {
             acknowledgement_listener: Some(acknowledgement_listener),
@@ -232,27 +238,27 @@ where
         // graceful shutdowns.
         let ack_listener_fut = tokio::spawn(async move {
             acknowledgement_listener.run().await;
-            error!("The acknowledgement listener has finished execution!");
+            debug!("The acknowledgement listener has finished execution!");
             acknowledgement_listener
         });
         let input_listener_fut = tokio::spawn(async move {
             input_message_listener.run().await;
-            error!("The input listener has finished execution!");
+            debug!("The input listener has finished execution!");
             input_message_listener
         });
         let retransmission_req_fut = tokio::spawn(async move {
             retransmission_request_listener.run().await;
-            error!("The retransmission request listener has finished execution!");
+            debug!("The retransmission request listener has finished execution!");
             retransmission_request_listener
         });
         let sent_notification_fut = tokio::spawn(async move {
             sent_notification_listener.run().await;
-            error!("The sent notification listener has finished execution!");
+            debug!("The sent notification listener has finished execution!");
             sent_notification_listener
         });
         let action_controller_fut = tokio::spawn(async move {
             action_controller.run().await;
-            error!("The controller has finished execution!");
+            debug!("The controller has finished execution!");
             action_controller
         });
 

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

@@ -14,6 +14,7 @@ use nymsphinx::preparer::MessagePreparer;
 use nymsphinx::{acknowledgements::AckKey, addressing::clients::Recipient};
 use rand::{CryptoRng, Rng};
 use std::sync::{Arc, Weak};
+use task::ShutdownListener;
 
 // responsible for packet retransmission upon fired timer
 pub(super) struct RetransmissionRequestListener<R>
@@ -27,12 +28,14 @@ where
     real_message_sender: BatchRealMessageSender,
     request_receiver: RetransmissionRequestReceiver,
     topology_access: TopologyAccessor,
+    shutdown: ShutdownListener,
 }
 
 impl<R> RetransmissionRequestListener<R>
 where
     R: CryptoRng + Rng,
 {
+    #[allow(clippy::too_many_arguments)]
     pub(super) fn new(
         ack_key: Arc<AckKey>,
         ack_recipient: Recipient,
@@ -41,6 +44,7 @@ where
         real_message_sender: BatchRealMessageSender,
         request_receiver: RetransmissionRequestReceiver,
         topology_access: TopologyAccessor,
+        shutdown: ShutdownListener,
     ) -> Self {
         RetransmissionRequestListener {
             ack_key,
@@ -50,6 +54,7 @@ where
             real_message_sender,
             request_receiver,
             topology_access,
+            shutdown,
         }
     }
 
@@ -121,9 +126,22 @@ where
 
     pub(super) async fn run(&mut self) {
         debug!("Started RetransmissionRequestListener");
-        while let Some(timed_out_ack) = self.request_receiver.next().await {
-            self.on_retransmission_request(timed_out_ack).await;
+
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                timed_out_ack = self.request_receiver.next() => match timed_out_ack {
+                    Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
+                    None => {
+                        log::trace!("RetransmissionRequestListener: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("RetransmissionRequestListener: Received shutdown");
+                }
+            }
         }
-        error!("TODO: error msg. Or maybe panic?")
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("RetransmissionRequestListener: Exiting");
     }
 }

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

@@ -6,6 +6,7 @@ use super::SentPacketNotificationReceiver;
 use futures::StreamExt;
 use log::*;
 use nymsphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
+use task::ShutdownListener;
 
 /// Module responsible for starting up retransmission timers.
 /// It is required because when we send our packet to the `real traffic stream` controlled
@@ -14,16 +15,19 @@ use nymsphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
 pub(super) struct SentNotificationListener {
     sent_notifier: SentPacketNotificationReceiver,
     action_sender: ActionSender,
+    shutdown: ShutdownListener,
 }
 
 impl SentNotificationListener {
     pub(super) fn new(
         sent_notifier: SentPacketNotificationReceiver,
         action_sender: ActionSender,
+        shutdown: ShutdownListener,
     ) -> Self {
         SentNotificationListener {
             sent_notifier,
             action_sender,
+            shutdown,
         }
     }
 
@@ -44,9 +48,23 @@ impl SentNotificationListener {
 
     pub(super) async fn run(&mut self) {
         debug!("Started SentNotificationListener");
-        while let Some(frag_id) = self.sent_notifier.next().await {
-            self.on_sent_message(frag_id).await;
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                frag_id = self.sent_notifier.next() => match frag_id {
+                    Some(frag_id) => {
+                        self.on_sent_message(frag_id).await;
+                    }
+                    None => {
+                        log::trace!("SentNotificationListener: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("SentNotificationListener: Received shutdown");
+                }
+            }
         }
-        error!("TODO: error msg. Or maybe panic?")
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("SentNotificationListener: Exiting");
     }
 }

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

@@ -22,6 +22,7 @@ use nymsphinx::addressing::clients::Recipient;
 use rand::{rngs::OsRng, CryptoRng, Rng};
 use std::sync::Arc;
 use std::time::Duration;
+use task::ShutdownListener;
 use tokio::task::JoinHandle;
 
 mod acknowledgement_control;
@@ -91,6 +92,7 @@ impl RealMessagesController<OsRng> {
         mix_sender: BatchMixMessageSender,
         topology_access: TopologyAccessor,
         reply_key_storage: ReplyKeyStorage,
+        shutdown: ShutdownListener,
     ) -> Self {
         let rng = OsRng;
 
@@ -119,6 +121,7 @@ impl RealMessagesController<OsRng> {
             config.self_recipient,
             reply_key_storage,
             ack_controller_connectors,
+            shutdown.clone(),
         );
 
         let out_queue_config = real_traffic_stream::Config::new(
@@ -136,6 +139,7 @@ impl RealMessagesController<OsRng> {
             rng,
             config.self_recipient,
             topology_access,
+            shutdown,
         );
 
         RealMessagesController {
@@ -153,12 +157,12 @@ impl RealMessagesController<OsRng> {
         // graceful shutdowns.
         let out_queue_control_fut = tokio::spawn(async move {
             out_queue_control.run_out_queue_control().await;
-            error!("The out queue controller has finished execution!");
+            debug!("The out queue controller has finished execution!");
             out_queue_control
         });
         let ack_control_fut = tokio::spawn(async move {
             ack_control.run().await;
-            error!("The acknowledgement controller has finished execution!");
+            debug!("The acknowledgement controller has finished execution!");
             ack_control
         });
 

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

@@ -19,6 +19,7 @@ use std::collections::VecDeque;
 use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
+use task::ShutdownListener;
 use tokio::time;
 
 /// Configurable parameters of the `OutQueueControl`
@@ -83,6 +84,9 @@ where
 
     /// Buffer containing all real messages received. It is first exhausted before more are pulled.
     received_buffer: VecDeque<RealMessage>,
+
+    /// Listens for shutdown signals
+    shutdown: ShutdownListener,
 }
 
 pub(crate) struct RealMessage {
@@ -174,6 +178,7 @@ where
         rng: R,
         our_full_destination: Recipient,
         topology_access: TopologyAccessor,
+        shutdown: ShutdownListener,
     ) -> Self {
         OutQueueControl {
             config,
@@ -186,6 +191,7 @@ where
             rng,
             topology_access,
             received_buffer: VecDeque::with_capacity(0), // we won't be putting any data into this guy directly
+            shutdown,
         }
     }
 
@@ -239,7 +245,15 @@ where
         // - we run out of memory
         // - the receiver channel is closed
         // in either case there's no recovery and we can only panic
-        self.mix_tx.unbounded_send(vec![next_message]).unwrap();
+        if let Err(err) = self.mix_tx.unbounded_send(vec![next_message]) {
+            if self.shutdown.is_shutdown_poll() {
+                log::info!("Failed to send (shutdown detected)");
+            } else {
+                // We don't try to limp along, panic to avoid continuing in a potentially
+                // inconsistent state
+                panic!("{err}");
+            }
+        }
 
         // JS: Not entirely sure why or how it fixes stuff, but without the yield call,
         // the UnboundedReceiver [of mix_rx] will not get a chance to read anything
@@ -257,9 +271,26 @@ where
             self.config.average_message_sending_delay,
         )));
 
-        while let Some(next_message) = self.next().await {
-            self.on_message(next_message).await;
+        let mut shutdown = self.shutdown.clone();
+        while !shutdown.is_shutdown() {
+            tokio::select! {
+                biased;
+                _ = shutdown.recv() => {
+                    log::trace!("OutQueueControl: Received shutdown");
+                }
+                next_message = self.next() => match next_message {
+                    Some(next_message) => {
+                        self.on_message(next_message).await;
+                    },
+                    None => {
+                        log::trace!("OutQueueControl: Stopping since channel closed");
+                        break;
+                    }
+                }
+            }
         }
+        assert!(shutdown.is_shutdown_poll());
+        log::debug!("OutQueueControl: Exiting");
     }
 
     pub(crate) async fn run_out_queue_control(&mut self) {

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

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::client::reply_key_storage::ReplyKeyStorage;
+use crate::client::SHUTDOWN_HAS_BEEN_SIGNALLED;
 use crypto::asymmetric::encryption;
 use crypto::symmetric::stream_cipher;
 use crypto::Digest;
@@ -14,7 +15,9 @@ use nymsphinx::anonymous_replies::{encryption_key::EncryptionKeyDigest, SurbEncr
 use nymsphinx::params::{ReplySurbEncryptionAlgorithm, ReplySurbKeyDigestAlgorithm};
 use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage};
 use std::collections::HashSet;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use task::ShutdownListener;
 use tokio::task::JoinHandle;
 
 // Buffer Requests to say "hey, send any reconstructed messages to this channel"
@@ -292,16 +295,27 @@ impl RequestReceiver {
 
     fn start(mut self) -> JoinHandle<()> {
         tokio::spawn(async move {
-            while let Some(request) = self.query_receiver.next().await {
-                match request {
-                    ReceivedBufferMessage::ReceiverAnnounce(sender) => {
-                        self.received_buffer.connect_sender(sender).await;
-                    }
-                    ReceivedBufferMessage::ReceiverDisconnect => {
-                        self.received_buffer.disconnect_sender().await
-                    }
-                }
+            loop {
+                tokio::select! {
+                    request = self.query_receiver.next() => {
+                        match request {
+                            Some(ReceivedBufferMessage::ReceiverAnnounce(sender)) => {
+                                self.received_buffer.connect_sender(sender).await;
+                            }
+                            Some(ReceivedBufferMessage::ReceiverDisconnect) => {
+                                self.received_buffer.disconnect_sender().await
+                            }
+                            None => {
+                                log::trace!("RequestReceiver: Stopping since channel closed");
+                                break;
+                            },
+                        }
+                    },
+                };
             }
+
+            assert!(SHUTDOWN_HAS_BEEN_SIGNALLED.load(Ordering::Relaxed));
+            log::debug!("RequestReceiver: Exiting");
         })
     }
 }
@@ -309,23 +323,41 @@ impl RequestReceiver {
 struct FragmentedMessageReceiver {
     received_buffer: ReceivedMessagesBuffer,
     mixnet_packet_receiver: MixnetMessageReceiver,
+    shutdown: ShutdownListener,
 }
 
 impl FragmentedMessageReceiver {
     fn new(
         received_buffer: ReceivedMessagesBuffer,
         mixnet_packet_receiver: MixnetMessageReceiver,
+        shutdown: ShutdownListener,
     ) -> Self {
         FragmentedMessageReceiver {
             received_buffer,
             mixnet_packet_receiver,
+            shutdown,
         }
     }
     fn start(mut self) -> JoinHandle<()> {
         tokio::spawn(async move {
-            while let Some(new_messages) = self.mixnet_packet_receiver.next().await {
-                self.received_buffer.handle_new_received(new_messages).await;
+            while !self.shutdown.is_shutdown() {
+                tokio::select! {
+                    new_messages = self.mixnet_packet_receiver.next() => match new_messages {
+                        Some(new_messages) => {
+                            self.received_buffer.handle_new_received(new_messages).await;
+                        }
+                        None => {
+                            log::trace!("FragmentedMessageReceiver: Stopping since channel closed");
+                            break;
+                        }
+                    },
+                    _ = self.shutdown.recv() => {
+                        log::trace!("FragmentedMessageReceiver: Received shutdown");
+                    }
+                }
             }
+            assert!(self.shutdown.is_shutdown_poll());
+            log::debug!("FragmentedMessageReceiver: Exiting");
         })
     }
 }
@@ -341,6 +373,7 @@ impl ReceivedMessagesBufferController {
         query_receiver: ReceivedBufferRequestReceiver,
         mixnet_packet_receiver: MixnetMessageReceiver,
         reply_key_storage: ReplyKeyStorage,
+        shutdown: ShutdownListener,
     ) -> Self {
         let received_buffer =
             ReceivedMessagesBuffer::new(local_encryption_keypair, reply_key_storage);
@@ -349,6 +382,7 @@ impl ReceivedMessagesBufferController {
             fragmented_message_receiver: FragmentedMessageReceiver::new(
                 received_buffer.clone(),
                 mixnet_packet_receiver,
+                shutdown,
             ),
             request_receiver: RequestReceiver::new(received_buffer, query_receiver),
         }

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

@@ -10,6 +10,7 @@ use std::ops::Deref;
 use std::sync::Arc;
 use std::time;
 use std::time::Duration;
+use task::ShutdownListener;
 use tokio::sync::{RwLock, RwLockReadGuard};
 use tokio::task::JoinHandle;
 use topology::{nym_topology_from_bonds, NymTopology};
@@ -303,12 +304,20 @@ impl TopologyRefresher {
         self.topology_accessor.is_routable().await
     }
 
-    pub fn start(mut self) -> JoinHandle<()> {
+    pub fn start(mut self, mut shutdown: ShutdownListener) -> JoinHandle<()> {
         tokio::spawn(async move {
-            loop {
-                tokio::time::sleep(self.refresh_rate).await;
-                self.refresh().await;
+            while !shutdown.is_shutdown() {
+                tokio::select! {
+                    _ = tokio::time::sleep(self.refresh_rate) => {
+                        self.refresh().await;
+                    },
+                    _ = shutdown.recv() => {
+                        log::trace!("TopologyRefresher: Received shutdown");
+                    },
+                }
             }
+            assert!(shutdown.is_shutdown_poll());
+            log::debug!("TopologyRefresher: Exiting");
         })
     }
 }

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

@@ -121,16 +121,6 @@ impl<T: NymConfig> Config<T> {
         self.client.gateway_endpoint.gateway_id = id.into();
     }
 
-    #[cfg(not(feature = "coconut"))]
-    pub fn with_eth_private_key<S: Into<String>>(&mut self, eth_private_key: S) {
-        self.client.eth_private_key = eth_private_key.into();
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    pub fn with_eth_endpoint<S: Into<String>>(&mut self, eth_endpoint: S) {
-        self.client.eth_endpoint = eth_endpoint.into();
-    }
-
     pub fn set_custom_validator_apis(&mut self, validator_api_urls: Vec<Url>) {
         self.client.validator_api_urls = validator_api_urls;
     }
@@ -209,16 +199,6 @@ impl<T: NymConfig> Config<T> {
         self.client.database_path.clone()
     }
 
-    #[cfg(not(feature = "coconut"))]
-    pub fn get_eth_endpoint(&self) -> String {
-        self.client.eth_endpoint.clone()
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    pub fn get_eth_private_key(&self) -> String {
-        self.client.eth_private_key.clone()
-    }
-
     // Debug getters
     pub fn get_average_packet_delay(&self) -> Duration {
         self.debug.average_packet_delay
@@ -342,14 +322,6 @@ pub struct Client<T> {
     /// Path to the database containing bandwidth credentials of this client.
     database_path: PathBuf,
 
-    /// Ethereum private key.
-    #[cfg(not(feature = "coconut"))]
-    eth_private_key: String,
-
-    /// Address to an Ethereum full node.
-    #[cfg(not(feature = "coconut"))]
-    eth_endpoint: String,
-
     /// nym_home_directory specifies absolute path to the home nym Clients directory.
     /// It is expected to use default value and hence .toml file should not redefine this field.
     nym_root_directory: PathBuf,
@@ -375,10 +347,6 @@ impl<T: NymConfig> Default for Client<T> {
             reply_encryption_key_store_path: Default::default(),
             gateway_endpoint: Default::default(),
             database_path: Default::default(),
-            #[cfg(not(feature = "coconut"))]
-            eth_private_key: "".to_string(),
-            #[cfg(not(feature = "coconut"))]
-            eth_endpoint: "".to_string(),
             nym_root_directory: T::default_root_directory(),
             super_struct: Default::default(),
         }

clients/client-core/src/error.rs

@@ -0,0 +1,27 @@
+// Copyright 2022 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crypto::asymmetric::identity::Ed25519RecoveryError;
+use gateway_client::error::GatewayClientError;
+use validator_client::ValidatorClientError;
+
+#[derive(thiserror::Error, Debug)]
+pub enum ClientCoreError {
+    #[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("No gateway with id: {0}")]
+    NoGatewayWithId(String),
+    #[error("No gateways on network")]
+    NoGatewaysOnNetwork,
+    #[error("List of validator apis is empty")]
+    ListOfValidatorApisIsEmpty,
+    #[error("Could not load existing gateway configuration: {0}")]
+    CouldNotLoadExistingGatewayConfiguration(std::io::Error),
+}

clients/client-core/src/init.rs

@@ -14,25 +14,27 @@ use nymsphinx::addressing::nodes::NodeIdentity;
 use rand::rngs::OsRng;
 use rand::seq::SliceRandom;
 use rand::thread_rng;
+use tap::TapFallible;
 use topology::{filter::VersionFilterable, gateway};
 use url::Url;
 
 use crate::{
     client::key_manager::KeyManager,
     config::{persistence::key_pathfinder::ClientKeyPathfinder, Config},
+    error::ClientCoreError,
 };
 
 pub async fn query_gateway_details(
     validator_servers: Vec<Url>,
     chosen_gateway_id: Option<&str>,
-) -> gateway::Node {
+) -> Result<gateway::Node, ClientCoreError> {
     let validator_api = validator_servers
         .choose(&mut thread_rng())
-        .expect("The list of validator apis is empty");
+        .ok_or(ClientCoreError::ListOfValidatorApisIsEmpty)?;
     let validator_client = validator_client::ApiClient::new(validator_api.clone());
 
     log::trace!("Fetching list of gateways from: {}", validator_api);
-    let gateways = validator_client.get_cached_gateways().await.unwrap();
+    let gateways = validator_client.get_cached_gateways().await?;
     let valid_gateways = gateways
         .into_iter()
         .filter_map(|gateway| gateway.try_into().ok())
@@ -47,38 +49,40 @@ pub async fn query_gateway_details(
         filtered_gateways
             .iter()
             .find(|gateway| gateway.identity_key.to_base58_string() == gateway_id)
-            .expect(&*format!("no gateway with id {} exists!", gateway_id))
-            .clone()
+            .ok_or_else(|| ClientCoreError::NoGatewayWithId(gateway_id.to_string()))
+            .cloned()
     } else {
         filtered_gateways
             .choose(&mut rand::thread_rng())
-            .expect("there are no gateways on the network!")
-            .clone()
+            .ok_or(ClientCoreError::NoGatewaysOnNetwork)
+            .cloned()
     }
 }
 
 pub async fn register_with_gateway_and_store_keys<T>(
     gateway_details: gateway::Node,
     config: &Config<T>,
-) where
+) -> Result<(), ClientCoreError>
+where
     T: NymConfig,
 {
     let mut rng = OsRng;
     let mut key_manager = KeyManager::new(&mut rng);
 
-    let shared_keys = register_with_gateway(&gateway_details, key_manager.identity_keypair()).await;
+    let shared_keys =
+        register_with_gateway(&gateway_details, key_manager.identity_keypair()).await?;
     key_manager.insert_gateway_shared_key(shared_keys);
 
     let pathfinder = ClientKeyPathfinder::new_from_config(config);
-    key_manager
+    Ok(key_manager
         .store_keys(&pathfinder)
-        .expect("Failed to generated keys");
+        .tap_err(|err| log::error!("Failed to generate keys: {err}"))?)
 }
 
 async fn register_with_gateway(
     gateway: &gateway::Node,
     our_identity: Arc<identity::KeyPair>,
-) -> Arc<SharedKeys> {
+) -> Result<Arc<SharedKeys>, ClientCoreError> {
     let timeout = Duration::from_millis(1500);
     let mut gateway_client = GatewayClient::new_init(
         gateway.clients_address(),
@@ -86,52 +90,59 @@ async fn register_with_gateway(
         gateway.owner.clone(),
         our_identity.clone(),
         timeout,
+        None,
     );
     gateway_client
         .establish_connection()
         .await
-        .expect("failed to establish connection with the gateway!");
-    gateway_client
+        .tap_err(|_| log::warn!("Failed to establish connection with gateway!"))?;
+    let shared_keys = gateway_client
         .perform_initial_authentication()
         .await
-        .expect("failed to register with the gateway!")
+        .tap_err(|_| log::warn!("Failed to register with the gateway!"))?;
+    Ok(shared_keys)
 }
 
-pub fn show_address<T>(config: &Config<T>)
+pub fn show_address<T>(config: &Config<T>) -> Result<(), ClientCoreError>
 where
     T: config::NymConfig,
 {
-    fn load_identity_keys(pathfinder: &ClientKeyPathfinder) -> identity::KeyPair {
+    fn load_identity_keys(
+        pathfinder: &ClientKeyPathfinder,
+    ) -> Result<identity::KeyPair, ClientCoreError> {
         let identity_keypair: identity::KeyPair =
             pemstore::load_keypair(&pemstore::KeyPairPath::new(
                 pathfinder.private_identity_key().to_owned(),
                 pathfinder.public_identity_key().to_owned(),
             ))
-            .expect("Failed to read stored identity key files");
-        identity_keypair
+            .tap_err(|_| log::error!("Failed to read stored identity key files"))?;
+        Ok(identity_keypair)
     }
 
-    fn load_sphinx_keys(pathfinder: &ClientKeyPathfinder) -> encryption::KeyPair {
+    fn load_sphinx_keys(
+        pathfinder: &ClientKeyPathfinder,
+    ) -> Result<encryption::KeyPair, ClientCoreError> {
         let sphinx_keypair: encryption::KeyPair =
             pemstore::load_keypair(&pemstore::KeyPairPath::new(
                 pathfinder.private_encryption_key().to_owned(),
                 pathfinder.public_encryption_key().to_owned(),
             ))
-            .expect("Failed to read stored sphinx key files");
-        sphinx_keypair
+            .tap_err(|_| log::error!("Failed to read stored sphinx key files"))?;
+        Ok(sphinx_keypair)
     }
 
     let pathfinder = ClientKeyPathfinder::new_from_config(config);
-    let identity_keypair = load_identity_keys(&pathfinder);
-    let sphinx_keypair = load_sphinx_keys(&pathfinder);
+    let identity_keypair = load_identity_keys(&pathfinder)?;
+    let sphinx_keypair = load_sphinx_keys(&pathfinder)?;
 
     let client_recipient = Recipient::new(
         *identity_keypair.public_key(),
         *sphinx_keypair.public_key(),
         // TODO: below only works under assumption that gateway address == gateway id
         // (which currently is true)
-        NodeIdentity::from_base58_string(config.get_gateway_id()).unwrap(),
+        NodeIdentity::from_base58_string(config.get_gateway_id())?,
     );
 
     println!("\nThe address of this client is: {}", client_recipient);
+    Ok(())
 }

clients/client-core/src/lib.rs

@@ -1,3 +1,4 @@
 pub mod client;
 pub mod config;
+pub mod error;
 pub mod init;

clients/native/Cargo.toml

@@ -33,23 +33,23 @@ tokio-tungstenite = "0.14" # websocket
 ## internal
 client-core = { path = "../client-core" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
-credentials = { path = "../../common/credentials", optional = true }
-credential-storage = { path = "../../common/credential-storage" }
 config = { path = "../../common/config" }
+credential-storage = { path = "../../common/credential-storage" }
+credentials = { path = "../../common/credentials", optional = true }
 crypto = { path = "../../common/crypto" }
 gateway-client = { path = "../../common/client-libs/gateway-client" }
 gateway-requests = { path = "../../gateway/gateway-requests" }
+network-defaults = { path = "../../common/network-defaults" }
 nymsphinx = { path = "../../common/nymsphinx" }
 pemstore = { path = "../../common/pemstore" }
+task = { path = "../../common/task" }
 topology = { path = "../../common/topology" }
-websocket-requests = { path = "websocket-requests" }
 validator-client = { path = "../../common/client-libs/validator-client", features = ["nymd-client"] }
 version-checker = { path = "../../common/version-checker" }
-network-defaults = { path = "../../common/network-defaults" }
+websocket-requests = { path = "websocket-requests" }
 
 [features]
 coconut = ["coconut-interface", "credentials", "credentials/coconut", "gateway-requests/coconut", "gateway-client/coconut", "client-core/coconut"]
-eth = []
 
 [dev-dependencies]
 serde_json = "1.0" # for the "textsend" example

clients/native/examples/js-examples/websocket/package-lock.json

@@ -27,6 +27,58 @@
         "node": ">=10.0.0"
       }
     },
+    "node_modules/@jridgewell/gen-mapping": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.2.tgz",
+      "integrity": "sha512-mh65xKQAzI6iBcFzwv28KVWSmCkdRBWoOh+bYQGW3+6OZvbbN3TqMGo5hqYxQniRcH9F2VZIoJCm4pa3BPDK/A==",
+      "dependencies": {
+        "@jridgewell/set-array": "^1.0.1",
+        "@jridgewell/sourcemap-codec": "^1.4.10",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      },
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/resolve-uri": {
+      "version": "3.1.0",
+      "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.0.tgz",
+      "integrity": "sha512-F2msla3tad+Mfht5cJq7LSXcdudKTWCVYUgw6pLFOOHSTtZlj6SWNYAp+AhuqLmWdBO2X5hPrLcu8cVP8fy28w==",
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/set-array": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/set-array/-/set-array-1.1.2.tgz",
+      "integrity": "sha512-xnkseuNADM0gt2bs+BvhO0p78Mk762YnZdsuzFV018NoG1Sj1SCQvpSqa7XUaTam5vAGasABV9qXASMKnFMwMw==",
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/source-map": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/source-map/-/source-map-0.3.2.tgz",
+      "integrity": "sha512-m7O9o2uR8k2ObDysZYzdfhb08VuEml5oWGiosa1VdaPZ/A6QyPkAJuwN0Q1lhULOf6B7MtQmHENS743hWtCrgw==",
+      "dependencies": {
+        "@jridgewell/gen-mapping": "^0.3.0",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "node_modules/@jridgewell/sourcemap-codec": {
+      "version": "1.4.14",
+      "resolved": "https://registry.npmjs.org/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.4.14.tgz",
+      "integrity": "sha512-XPSJHWmi394fuUuzDnGz1wiKqWfo1yXecHQMRf2l6hztTO+nPru658AyDngaBe7isIxEkRsPR3FZh+s7iVa4Uw=="
+    },
+    "node_modules/@jridgewell/trace-mapping": {
+      "version": "0.3.15",
+      "resolved": "https://registry.npmjs.org/@jridgewell/trace-mapping/-/trace-mapping-0.3.15.tgz",
+      "integrity": "sha512-oWZNOULl+UbhsgB51uuZzglikfIKSUBO/M9W2OfEjn7cmqoAiCgmv9lyACTUacZwBz0ITnJ2NqjU8Tx0DHL88g==",
+      "dependencies": {
+        "@jridgewell/resolve-uri": "^3.0.3",
+        "@jridgewell/sourcemap-codec": "^1.4.10"
+      }
+    },
     "node_modules/@nodelib/fs.scandir": {
       "version": "2.1.5",
       "resolved": "https://registry.npmjs.org/@nodelib/fs.scandir/-/fs.scandir-2.1.5.tgz",
@@ -3529,13 +3581,13 @@
       }
     },
     "node_modules/terser": {
-      "version": "5.12.1",
-      "resolved": "https://registry.npmjs.org/terser/-/terser-5.12.1.tgz",
-      "integrity": "sha512-NXbs+7nisos5E+yXwAD+y7zrcTkMqb0dEJxIGtSKPdCBzopf7ni4odPul2aechpV7EXNvOudYOX2bb5tln1jbQ==",
+      "version": "5.15.0",
+      "resolved": "https://registry.npmjs.org/terser/-/terser-5.15.0.tgz",
+      "integrity": "sha512-L1BJiXVmheAQQy+as0oF3Pwtlo4s3Wi1X2zNZ2NxOB4wx9bdS9Vk67XQENLFdLYGCK/Z2di53mTj/hBafR+dTA==",
       "dependencies": {
+        "@jridgewell/source-map": "^0.3.2",
         "acorn": "^8.5.0",
         "commander": "^2.20.0",
-        "source-map": "~0.7.2",
         "source-map-support": "~0.5.20"
       },
       "bin": {
@@ -3583,14 +3635,6 @@
       "resolved": "https://registry.npmjs.org/commander/-/commander-2.20.3.tgz",
       "integrity": "sha512-GpVkmM8vF2vQUkj2LvZmD35JxeJOLCwJ9cUkugyk2nuhbv3+mJvpLYYt+0+USMxE+oj+ey/lJEnhZw75x/OMcQ=="
     },
-    "node_modules/terser/node_modules/source-map": {
-      "version": "0.7.3",
-      "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.7.3.tgz",
-      "integrity": "sha512-CkCj6giN3S+n9qrYiBTX5gystlENnRW5jZeNLHpe6aue+SrHcG5VYwujhW9s4dY31mEGsxBDrHR6oI69fTXsaQ==",
-      "engines": {
-        "node": ">= 8"
-      }
-    },
     "node_modules/thunky": {
       "version": "1.1.0",
       "resolved": "https://registry.npmjs.org/thunky/-/thunky-1.1.0.tgz",
@@ -4330,6 +4374,49 @@
       "integrity": "sha512-dBVuXR082gk3jsFp7Rd/JI4kytwGHecnCoTtXFb7DB6CNHp4rg5k1bhg0nWdLGLnOV71lmDzGQaLMy8iPLY0pw==",
       "dev": true
     },
+    "@jridgewell/gen-mapping": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.2.tgz",
+      "integrity": "sha512-mh65xKQAzI6iBcFzwv28KVWSmCkdRBWoOh+bYQGW3+6OZvbbN3TqMGo5hqYxQniRcH9F2VZIoJCm4pa3BPDK/A==",
+      "requires": {
+        "@jridgewell/set-array": "^1.0.1",
+        "@jridgewell/sourcemap-codec": "^1.4.10",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "@jridgewell/resolve-uri": {
+      "version": "3.1.0",
+      "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.0.tgz",
+      "integrity": "sha512-F2msla3tad+Mfht5cJq7LSXcdudKTWCVYUgw6pLFOOHSTtZlj6SWNYAp+AhuqLmWdBO2X5hPrLcu8cVP8fy28w=="
+    },
+    "@jridgewell/set-array": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/set-array/-/set-array-1.1.2.tgz",
+      "integrity": "sha512-xnkseuNADM0gt2bs+BvhO0p78Mk762YnZdsuzFV018NoG1Sj1SCQvpSqa7XUaTam5vAGasABV9qXASMKnFMwMw=="
+    },
+    "@jridgewell/source-map": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/source-map/-/source-map-0.3.2.tgz",
+      "integrity": "sha512-m7O9o2uR8k2ObDysZYzdfhb08VuEml5oWGiosa1VdaPZ/A6QyPkAJuwN0Q1lhULOf6B7MtQmHENS743hWtCrgw==",
+      "requires": {
+        "@jridgewell/gen-mapping": "^0.3.0",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "@jridgewell/sourcemap-codec": {
+      "version": "1.4.14",
+      "resolved": "https://registry.npmjs.org/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.4.14.tgz",
+      "integrity": "sha512-XPSJHWmi394fuUuzDnGz1wiKqWfo1yXecHQMRf2l6hztTO+nPru658AyDngaBe7isIxEkRsPR3FZh+s7iVa4Uw=="
+    },
+    "@jridgewell/trace-mapping": {
+      "version": "0.3.15",
+      "resolved": "https://registry.npmjs.org/@jridgewell/trace-mapping/-/trace-mapping-0.3.15.tgz",
+      "integrity": "sha512-oWZNOULl+UbhsgB51uuZzglikfIKSUBO/M9W2OfEjn7cmqoAiCgmv9lyACTUacZwBz0ITnJ2NqjU8Tx0DHL88g==",
+      "requires": {
+        "@jridgewell/resolve-uri": "^3.0.3",
+        "@jridgewell/sourcemap-codec": "^1.4.10"
+      }
+    },
     "@nodelib/fs.scandir": {
       "version": "2.1.5",
       "resolved": "https://registry.npmjs.org/@nodelib/fs.scandir/-/fs.scandir-2.1.5.tgz",
@@ -7058,13 +7145,13 @@
       "integrity": "sha512-GNzQvQTOIP6RyTfE2Qxb8ZVlNmw0n88vp1szwWRimP02mnTsx3Wtn5qRdqY9w2XduFNUgvOwhNnQsjwCp+kqaQ=="
     },
     "terser": {
-      "version": "5.12.1",
-      "resolved": "https://registry.npmjs.org/terser/-/terser-5.12.1.tgz",
-      "integrity": "sha512-NXbs+7nisos5E+yXwAD+y7zrcTkMqb0dEJxIGtSKPdCBzopf7ni4odPul2aechpV7EXNvOudYOX2bb5tln1jbQ==",
+      "version": "5.15.0",
+      "resolved": "https://registry.npmjs.org/terser/-/terser-5.15.0.tgz",
+      "integrity": "sha512-L1BJiXVmheAQQy+as0oF3Pwtlo4s3Wi1X2zNZ2NxOB4wx9bdS9Vk67XQENLFdLYGCK/Z2di53mTj/hBafR+dTA==",
       "requires": {
+        "@jridgewell/source-map": "^0.3.2",
         "acorn": "^8.5.0",
         "commander": "^2.20.0",
-        "source-map": "~0.7.2",
         "source-map-support": "~0.5.20"
       },
       "dependencies": {
@@ -7072,11 +7159,6 @@
           "version": "2.20.3",
           "resolved": "https://registry.npmjs.org/commander/-/commander-2.20.3.tgz",
           "integrity": "sha512-GpVkmM8vF2vQUkj2LvZmD35JxeJOLCwJ9cUkugyk2nuhbv3+mJvpLYYt+0+USMxE+oj+ey/lJEnhZw75x/OMcQ=="
-        },
-        "source-map": {
-          "version": "0.7.3",
-          "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.7.3.tgz",
-          "integrity": "sha512-CkCj6giN3S+n9qrYiBTX5gystlENnRW5jZeNLHpe6aue+SrHcG5VYwujhW9s4dY31mEGsxBDrHR6oI69fTXsaQ=="
         }
       }
     },

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

@@ -49,12 +49,6 @@ reply_encryption_key_store_path = '{{ client.reply_encryption_key_store_path }}'
 # Path to the database containing bandwidth credentials
 database_path = '{{ client.database_path }}'
 
-# Ethereum private key.
-eth_private_key = '{{ client.eth_private_key }}'
-
-# Addess to an Ethereum full node.
-eth_endpoint = '{{ client.eth_endpoint }}'
-
 ##### additional client config options #####
 
 # A gateway specific, optional, base58 stringified shared key used for

clients/native/src/client/mod.rs

@@ -32,6 +32,7 @@ use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::addressing::nodes::NodeIdentity;
 use nymsphinx::anonymous_replies::ReplySurb;
 use nymsphinx::receiver::ReconstructedMessage;
+use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
 
 use crate::client::config::{Config, SocketType};
 use crate::websocket;
@@ -85,6 +86,7 @@ impl NymClient {
         &self,
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting loop cover traffic stream...");
 
@@ -98,6 +100,7 @@ impl NymClient {
             mix_tx,
             self.as_mix_recipient(),
             topology_accessor,
+            shutdown,
         )
         .start();
     }
@@ -109,6 +112,7 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
+        shutdown: ShutdownListener,
     ) {
         let controller_config = real_messages_control::Config::new(
             self.key_manager.ack_key(),
@@ -129,6 +133,7 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            shutdown,
         )
         .start();
     }
@@ -140,6 +145,7 @@ impl NymClient {
         query_receiver: ReceivedBufferRequestReceiver,
         mixnet_receiver: MixnetMessageReceiver,
         reply_key_storage: ReplyKeyStorage,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting received messages buffer controller...");
         ReceivedMessagesBufferController::new(
@@ -147,6 +153,7 @@ impl NymClient {
             query_receiver,
             mixnet_receiver,
             reply_key_storage,
+            shutdown,
         )
         .start()
     }
@@ -155,6 +162,7 @@ impl NymClient {
         &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() {
@@ -182,8 +190,6 @@ impl NymClient {
         let bandwidth_controller = BandwidthController::new(
             credential_storage::initialise_storage(self.config.get_base().get_database_path())
                 .await,
-            self.config.get_base().get_eth_endpoint(),
-            self.config.get_base().get_eth_private_key(),
         )
         .expect("Could not create bandwidth controller");
 
@@ -197,6 +203,7 @@ impl NymClient {
             ack_sender,
             self.config.get_base().get_gateway_response_timeout(),
             Some(bandwidth_controller),
+            Some(shutdown),
         );
 
         gateway_client
@@ -212,7 +219,11 @@ impl NymClient {
 
     // 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) {
+    async fn start_topology_refresher(
+        &mut self,
+        topology_accessor: TopologyAccessor,
+        shutdown: ShutdownListener,
+    ) {
         let topology_refresher_config = TopologyRefresherConfig::new(
             self.config.get_base().get_validator_api_endpoints(),
             self.config.get_base().get_topology_refresh_rate(),
@@ -234,7 +245,7 @@ impl NymClient {
         }
 
         info!("Starting topology refresher...");
-        topology_refresher.start();
+        topology_refresher.start(shutdown);
     }
 
     // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -245,9 +256,10 @@ impl NymClient {
         &mut self,
         mix_rx: BatchMixMessageReceiver,
         gateway_client: GatewayClient,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting mix traffic controller...");
-        MixTrafficController::new(mix_rx, gateway_client).start();
+        MixTrafficController::new(mix_rx, gateway_client, shutdown).start();
     }
 
     fn start_websocket_listener(
@@ -308,20 +320,26 @@ impl NymClient {
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
     pub async fn run_forever(&mut self) {
-        self.start().await;
-        if let Err(e) = tokio::signal::ctrl_c().await {
-            error!(
-                "There was an error while capturing SIGINT - {:?}. We will terminate regardless",
-                e
-            );
-        }
+        let shutdown = self.start().await;
+        wait_for_signal().await;
 
         println!(
-            "Received SIGINT - the client will terminate now (threads are not yet nicely stopped, if you see stack traces that's alright)."
+            "Received signal - the client will terminate now (threads are not yet nicely stopped, if you see stack traces that's alright)."
         );
+
+        log::info!("Sending shutdown");
+        shutdown.signal_shutdown().ok();
+
+        // Some of these components have shutdown signalling implemented as part of socks5 work,
+        // but since it's not fully implemented (yet) for all the components of the native client,
+        // we don't try to wait and instead just stop immediately.
+        //log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
+        //shutdown.wait_for_shutdown().await;
+
+        log::info!("Stopping nym-client");
     }
 
-    pub async fn start(&mut self) {
+    pub async fn start(&mut self) -> ShutdownNotifier {
         info!("Starting nym client");
         // channels for inter-component communication
         // TODO: make the channels be internally created by the relevant components
@@ -351,30 +369,43 @@ impl NymClient {
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
                 .expect("Failed to load reply key storage!");
 
+        // 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())
+        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)
+            .start_gateway_client(mixnet_messages_sender, ack_sender, shutdown.subscribe())
             .await;
 
-        self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client);
+        self.start_mix_traffic_controller(
+            sphinx_message_receiver,
+            gateway_client,
+            shutdown.subscribe(),
+        );
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
+            shutdown.subscribe(),
         );
 
-        self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
+        self.start_cover_traffic_stream(
+            shared_topology_accessor,
+            sphinx_message_sender,
+            shutdown.subscribe(),
+        );
 
         match self.config.get_socket_type() {
             SocketType::WebSocket => {
@@ -399,5 +430,7 @@ impl NymClient {
 
         info!("Client startup finished!");
         info!("The address of this client is: {}", self.as_mix_recipient());
+
+        shutdown
     }
 }

clients/native/src/commands/init.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::Args;
-use client_core::config::GatewayEndpoint;
+use client_core::{config::GatewayEndpoint, error::ClientCoreError};
 use config::NymConfig;
 
 use crate::{
@@ -10,9 +10,6 @@ use crate::{
     commands::{override_config, OverrideConfig},
 };
 
-#[cfg(all(feature = "eth", not(feature = "coconut")))]
-use crate::commands::{DEFAULT_ETH_ENDPOINT, DEFAULT_ETH_PRIVATE_KEY};
-
 #[derive(Args, Clone)]
 pub(crate) struct Init {
     /// Id of the nym-mixnet-client we want to create config for.
@@ -47,27 +44,9 @@ pub(crate) struct Init {
 
     /// Set this client to work in a enabled credentials mode that would attempt to use gateway
     /// with bandwidth credential requirement.
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     #[clap(long)]
     enabled_credentials_mode: bool,
-
-    /// URL of an Ethereum full node that we want to use for getting bandwidth tokens from ERC20
-    /// tokens. If you don't want to set this value, use --enabled-credentials-mode instead
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(
-        long,
-        default_value_if("enabled-credentials-mode", None, Some(DEFAULT_ETH_ENDPOINT))
-    )]
-    eth_endpoint: String,
-
-    /// Ethereum private key used for obtaining bandwidth tokens from ERC20 tokens. If you don't
-    /// want to set this value, use --enabled-credentials-mode instead")
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(
-        long,
-        default_value_if("enabled-credentials-mode", None, Some(DEFAULT_ETH_PRIVATE_KEY))
-    )]
-    eth_private_key: String,
 }
 
 impl From<Init> for OverrideConfig {
@@ -78,14 +57,8 @@ impl From<Init> for OverrideConfig {
             port: init_config.port,
             fastmode: init_config.fastmode,
 
-            #[cfg(any(feature = "eth", feature = "coconut"))]
+            #[cfg(feature = "coconut")]
             enabled_credentials_mode: init_config.enabled_credentials_mode,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_private_key: Some(init_config.eth_private_key),
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_endpoint: Some(init_config.eth_endpoint),
         }
     }
 }
@@ -120,7 +93,12 @@ pub(crate) async fn execute(args: &Init) {
     let override_config_fields = OverrideConfig::from(args.clone());
     config = override_config(config, override_config_fields);
 
-    let gateway = setup_gateway(id, register_gateway, user_chosen_gateway_id, &config).await;
+    let gateway = setup_gateway(id, register_gateway, user_chosen_gateway_id, &config)
+        .await
+        .unwrap_or_else(|err| {
+            eprintln!("Failed to setup gateway\nError: {err}");
+            std::process::exit(1)
+        });
     config.get_base_mut().with_gateway_endpoint(gateway);
 
     let config_save_location = config.get_config_file_save_location();
@@ -138,7 +116,10 @@ pub(crate) async fn execute(args: &Init) {
     );
     println!("Client configuration completed.");
 
-    client_core::init::show_address(config.get_base());
+    client_core::init::show_address(config.get_base()).unwrap_or_else(|err| {
+        eprintln!("Failed to show address\nError: {err}");
+        std::process::exit(1)
+    });
 }
 
 async fn setup_gateway(
@@ -146,7 +127,7 @@ async fn setup_gateway(
     register: bool,
     user_chosen_gateway_id: Option<&str>,
     config: &Config,
-) -> GatewayEndpoint {
+) -> Result<GatewayEndpoint, 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.
@@ -155,16 +136,16 @@ async fn setup_gateway(
             config.get_base().get_validator_api_endpoints(),
             user_chosen_gateway_id,
         )
-        .await;
+        .await?;
         log::debug!("Querying gateway gives: {}", gateway);
 
         // Registering with gateway by setting up and writing shared keys to disk
         log::trace!("Registering gateway");
         client_core::init::register_with_gateway_and_store_keys(gateway.clone(), config.get_base())
-            .await;
+            .await?;
         println!("Saved all generated keys");
 
-        gateway.into()
+        Ok(gateway.into())
     } else if user_chosen_gateway_id.is_some() {
         // Just set the config, don't register or create any keys
         // This assumes that the user knows what they are doing, and that the existing keys are
@@ -174,22 +155,22 @@ async fn setup_gateway(
             config.get_base().get_validator_api_endpoints(),
             user_chosen_gateway_id,
         )
-        .await;
+        .await?;
         log::debug!("Querying gateway gives: {}", gateway);
-        gateway.into()
+        Ok(gateway.into())
     } else {
         println!("Not registering gateway, will reuse existing config and keys");
-        match Config::load_from_file(Some(id)) {
-            Ok(existing_config) => existing_config.get_base().get_gateway_endpoint().clone(),
-            Err(err) => {
-                panic!(
-                    "Unable to configure gateway: {err}. \n
-                    Seems like the client was already initialized but it was not possible to read \
-                    the existing configuration file. \n
-                    CAUTION: Consider backing up your gateway keys and try force gateway registration, or \
-                    removing the existing configuration and starting over."
-                )
-            }
-        }
+        let existing_config = Config::load_from_file(Some(id)).map_err(|err| {
+            log::error!(
+                "Unable to configure gateway: {err}. \n
+                Seems like the client was already initialized but it was not possible to read \
+                the existing configuration file. \n
+                CAUTION: Consider backing up your gateway keys and try force gateway registration, or \
+                removing the existing configuration and starting over."
+            );
+            ClientCoreError::CouldNotLoadExistingGatewayConfiguration(err)
+        })?;
+
+        Ok(existing_config.get_base().get_gateway_endpoint().clone())
     }
 }

clients/native/src/commands/mod.rs

@@ -4,13 +4,6 @@
 use crate::client::config::{Config, SocketType};
 use clap::{Parser, Subcommand};
 
-#[cfg(not(feature = "coconut"))]
-pub(crate) const DEFAULT_ETH_ENDPOINT: &str =
-    "https://rinkeby.infura.io/v3/00000000000000000000000000000000";
-#[cfg(not(feature = "coconut"))]
-pub(crate) const DEFAULT_ETH_PRIVATE_KEY: &str =
-    "0000000000000000000000000000000000000000000000000000000000000001";
-
 pub(crate) mod init;
 pub(crate) mod run;
 pub(crate) mod upgrade;
@@ -78,14 +71,8 @@ pub(crate) struct OverrideConfig {
     port: Option<u16>,
     fastmode: bool,
 
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     enabled_credentials_mode: bool,
-
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    eth_private_key: Option<String>,
-
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    eth_endpoint: Option<String>,
 }
 
 pub(crate) async fn execute(args: &Cli) {
@@ -117,32 +104,13 @@ pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Confi
         config = config.with_port(port);
     }
 
-    #[cfg(all(not(feature = "eth"), not(feature = "coconut")))]
-    {
-        config
-            .get_base_mut()
-            .with_eth_endpoint(DEFAULT_ETH_ENDPOINT.to_string());
-        config
-            .get_base_mut()
-            .with_eth_private_key(DEFAULT_ETH_PRIVATE_KEY.to_string());
-    }
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     {
         if args.enabled_credentials_mode {
             config.get_base_mut().with_disabled_credentials(false)
         }
     }
 
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    {
-        if let Some(eth_endpoint) = args.eth_endpoint {
-            config.get_base_mut().with_eth_endpoint(eth_endpoint);
-        }
-        if let Some(eth_private_key) = args.eth_private_key {
-            config.get_base_mut().with_eth_private_key(eth_private_key);
-        }
-    }
-
     if args.fastmode {
         config.get_base_mut().set_high_default_traffic_volume();
     }

clients/native/src/commands/run.rs

@@ -36,21 +36,9 @@ pub(crate) struct Run {
 
     /// Set this client to work in a enabled credentials mode that would attempt to use gateway
     /// with bandwidth credential requirement.
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     #[clap(long)]
     enabled_credentials_mode: bool,
-
-    /// URL of an Ethereum full node that we want to use for getting bandwidth tokens from ERC20
-    /// tokens. If you don't want to set this value, use --enabled-credentials-mode instead
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(long)]
-    eth_endpoint: Option<String>,
-
-    /// Ethereum private key used for obtaining bandwidth tokens from ERC20 tokens. If you don't
-    /// want to set this value, use --enabled-credentials-mode instead")
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(long)]
-    eth_private_key: Option<String>,
 }
 
 impl From<Run> for OverrideConfig {
@@ -60,15 +48,8 @@ impl From<Run> for OverrideConfig {
             disable_socket: run_config.disable_socket,
             port: run_config.port,
             fastmode: false,
-
-            #[cfg(any(feature = "eth", feature = "coconut"))]
+            #[cfg(feature = "coconut")]
             enabled_credentials_mode: run_config.enabled_credentials_mode,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_private_key: run_config.eth_private_key,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_endpoint: run_config.eth_endpoint,
         }
     }
 }

clients/socks5/Cargo.toml

@@ -26,21 +26,22 @@ url = "2.2"
 # internal
 client-core = { path = "../client-core" }
 coconut-interface = { path = "../../common/coconut-interface", optional = true }
-credentials = { path = "../../common/credentials", optional = true }
-credential-storage = { path = "../../common/credential-storage" }
 config = { path = "../../common/config" }
+credential-storage = { path = "../../common/credential-storage" }
+credentials = { path = "../../common/credentials", optional = true }
 crypto = { path = "../../common/crypto" }
 gateway-client = { path = "../../common/client-libs/gateway-client" }
 gateway-requests = { path = "../../gateway/gateway-requests" }
+network-defaults = { path = "../../common/network-defaults" }
 nymsphinx = { path = "../../common/nymsphinx" }
 ordered-buffer = { path = "../../common/socks5/ordered-buffer" }
-socks5-requests = { path = "../../common/socks5/requests" }
-topology = { path = "../../common/topology" }
 pemstore = { path = "../../common/pemstore" }
 proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
+socks5-requests = { path = "../../common/socks5/requests" }
+task = { path = "../../common/task" }
+topology = { path = "../../common/topology" }
 validator-client = { path = "../../common/client-libs/validator-client", features = ["nymd-client"] }
 version-checker = { path = "../../common/version-checker" }
-network-defaults = { path = "../../common/network-defaults" }
 
 [features]
 coconut = ["coconut-interface", "credentials", "gateway-requests/coconut", "gateway-client/coconut", "credentials/coconut", "client-core/coconut"]

clients/socks5/src/client/config/template.rs

@@ -49,12 +49,6 @@ reply_encryption_key_store_path = '{{ client.reply_encryption_key_store_path }}'
 # Path to the database containing bandwidth credentials
 database_path = '{{ client.database_path }}'
 
-# Ethereum private key.
-eth_private_key = '{{ client.eth_private_key }}'
-
-# Addess to an Ethereum full node.
-eth_endpoint = '{{ client.eth_endpoint }}'
-
 ##### additional client config options #####
 
 # A gateway specific, optional, base58 stringified shared key used for

clients/socks5/src/client/mod.rs

@@ -1,6 +1,8 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use std::sync::atomic::Ordering;
+
 use client_core::client::cover_traffic_stream::LoopCoverTrafficStream;
 use client_core::client::inbound_messages::{
     InputMessage, InputMessageReceiver, InputMessageSender,
@@ -29,6 +31,7 @@ use gateway_client::{
 use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use nymsphinx::addressing::nodes::NodeIdentity;
+use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
 
 use crate::client::config::Config;
 use crate::socks::{
@@ -84,6 +87,7 @@ impl NymClient {
         &self,
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting loop cover traffic stream...");
 
@@ -97,6 +101,7 @@ impl NymClient {
             mix_tx,
             self.as_mix_recipient(),
             topology_accessor,
+            shutdown,
         )
         .start();
     }
@@ -108,6 +113,7 @@ impl NymClient {
         ack_receiver: AcknowledgementReceiver,
         input_receiver: InputMessageReceiver,
         mix_sender: BatchMixMessageSender,
+        shutdown: ShutdownListener,
     ) {
         let controller_config = client_core::client::real_messages_control::Config::new(
             self.key_manager.ack_key(),
@@ -128,6 +134,7 @@ impl NymClient {
             mix_sender,
             topology_accessor,
             reply_key_storage,
+            shutdown,
         )
         .start();
     }
@@ -139,6 +146,7 @@ impl NymClient {
         query_receiver: ReceivedBufferRequestReceiver,
         mixnet_receiver: MixnetMessageReceiver,
         reply_key_storage: ReplyKeyStorage,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting received messages buffer controller...");
         ReceivedMessagesBufferController::new(
@@ -146,6 +154,7 @@ impl NymClient {
             query_receiver,
             mixnet_receiver,
             reply_key_storage,
+            shutdown,
         )
         .start()
     }
@@ -154,6 +163,7 @@ impl NymClient {
         &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() {
@@ -181,8 +191,6 @@ impl NymClient {
         let bandwidth_controller = BandwidthController::new(
             credential_storage::initialise_storage(self.config.get_base().get_database_path())
                 .await,
-            self.config.get_base().get_eth_endpoint(),
-            self.config.get_base().get_eth_private_key(),
         )
         .expect("Could not create bandwidth controller");
 
@@ -196,6 +204,7 @@ impl NymClient {
             ack_sender,
             self.config.get_base().get_gateway_response_timeout(),
             Some(bandwidth_controller),
+            Some(shutdown),
         );
 
         gateway_client
@@ -211,7 +220,11 @@ impl NymClient {
 
     // 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) {
+    async fn start_topology_refresher(
+        &mut self,
+        topology_accessor: TopologyAccessor,
+        shutdown: ShutdownListener,
+    ) {
         let topology_refresher_config = TopologyRefresherConfig::new(
             self.config.get_base().get_validator_api_endpoints(),
             self.config.get_base().get_topology_refresh_rate(),
@@ -233,7 +246,7 @@ impl NymClient {
         }
 
         info!("Starting topology refresher...");
-        topology_refresher.start();
+        topology_refresher.start(shutdown);
     }
 
     // controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -244,15 +257,17 @@ impl NymClient {
         &mut self,
         mix_rx: BatchMixMessageReceiver,
         gateway_client: GatewayClient,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting mix traffic controller...");
-        MixTrafficController::new(mix_rx, gateway_client).start();
+        MixTrafficController::new(mix_rx, gateway_client, shutdown).start();
     }
 
     fn start_socks5_listener(
         &self,
         buffer_requester: ReceivedBufferRequestSender,
         msg_input: InputMessageSender,
+        shutdown: ShutdownListener,
     ) {
         info!("Starting socks5 listener...");
         let auth_methods = vec![AuthenticationMethods::NoAuth as u8];
@@ -264,43 +279,57 @@ impl NymClient {
             authenticator,
             self.config.get_provider_mix_address(),
             self.as_mix_recipient(),
+            shutdown,
         );
         tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
     }
 
     /// blocking version of `start` method. Will run forever (or until SIGINT is sent)
     pub async fn run_forever(&mut self) {
-        self.start().await;
-        if let Err(e) = tokio::signal::ctrl_c().await {
-            error!(
-                "There was an error while capturing SIGINT - {:?}. We will terminate regardless",
-                e
-            );
-        }
+        let mut shutdown = self.start().await;
+        wait_for_signal().await;
 
-        println!(
-            "Received SIGINT - the client will terminate now (threads are not yet nicely stopped, if you see stack traces that's alright)."
-        );
+        log::info!("Sending shutdown");
+        client_core::client::SHUTDOWN_HAS_BEEN_SIGNALLED.store(true, Ordering::Relaxed);
+        shutdown.signal_shutdown().ok();
+
+        log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
+        shutdown.wait_for_shutdown().await;
+
+        log::info!("Stopping nym-socks5-client");
     }
 
     // Variant of `run_forever` that listends for remote control messages
     pub async fn run_and_listen(&mut self, mut receiver: Socks5ControlMessageReceiver) {
-        self.start().await;
+        let mut shutdown = self.start().await;
         tokio::select! {
             message = receiver.next() => {
                 log::debug!("Received message: {:?}", message);
                 match message {
                     Some(Socks5ControlMessage::Stop) => {
-                        log::info!("Shutting down");
-                        log::info!("Graceful shutdown of tasks not yet implemented, you might see (harmless) panics until then");
+                        log::info!("Received stop message");
+                    }
+                    None => {
+                        log::info!("Channel closed, stopping");
                     }
-                    None => log::debug!("None"),
                 }
             }
+            _ = tokio::signal::ctrl_c() => {
+                log::info!("Received SIGINT");
+            },
         }
+
+        log::info!("Sending shutdown");
+        client_core::client::SHUTDOWN_HAS_BEEN_SIGNALLED.store(true, Ordering::Relaxed);
+        shutdown.signal_shutdown().ok();
+
+        log::info!("Waiting for tasks to finish... (Press ctrl-c to force)");
+        shutdown.wait_for_shutdown().await;
+
+        log::info!("Stopping nym-socks5-client");
     }
 
-    pub async fn start(&mut self) {
+    pub async fn start(&mut self) -> ShutdownNotifier {
         info!("Starting nym client");
         // channels for inter-component communication
         // TODO: make the channels be internally created by the relevant components
@@ -330,33 +359,52 @@ impl NymClient {
             ReplyKeyStorage::load(self.config.get_base().get_reply_encryption_key_store_path())
                 .expect("Failed to load reply key storage!");
 
+        // 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())
+        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)
+            .start_gateway_client(mixnet_messages_sender, ack_sender, shutdown.subscribe())
             .await;
 
-        self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client);
+        self.start_mix_traffic_controller(
+            sphinx_message_receiver,
+            gateway_client,
+            shutdown.subscribe(),
+        );
         self.start_real_traffic_controller(
             shared_topology_accessor.clone(),
             reply_key_storage,
             ack_receiver,
             input_receiver,
             sphinx_message_sender.clone(),
+            shutdown.subscribe(),
         );
 
-        self.start_cover_traffic_stream(shared_topology_accessor, sphinx_message_sender);
-        self.start_socks5_listener(received_buffer_request_sender, input_sender);
+        self.start_cover_traffic_stream(
+            shared_topology_accessor,
+            sphinx_message_sender,
+            shutdown.subscribe(),
+        );
+        self.start_socks5_listener(
+            received_buffer_request_sender,
+            input_sender,
+            shutdown.subscribe(),
+        );
 
         info!("Client startup finished!");
         info!("The address of this client is: {}", self.as_mix_recipient());
+
+        shutdown
     }
 }

clients/socks5/src/commands/init.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use clap::Args;
-use client_core::config::GatewayEndpoint;
+use client_core::{config::GatewayEndpoint, error::ClientCoreError};
 use config::NymConfig;
 
 use crate::{
@@ -10,9 +10,6 @@ use crate::{
     commands::{override_config, OverrideConfig},
 };
 
-#[cfg(all(feature = "eth", not(feature = "coconut")))]
-use crate::commands::{DEFAULT_ETH_ENDPOINT, DEFAULT_ETH_PRIVATE_KEY};
-
 #[derive(Args, Clone)]
 pub(crate) struct Init {
     /// Id of the nym-mixnet-client we want to create config for.
@@ -47,27 +44,9 @@ pub(crate) struct Init {
 
     /// Set this client to work in a enabled credentials mode that would attempt to use gateway
     /// with bandwidth credential requirement.
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     #[clap(long)]
     enabled_credentials_mode: bool,
-
-    /// URL of an Ethereum full node that we want to use for getting bandwidth tokens from ERC20
-    /// tokens. If you don't want to set this value, use --enabled-credentials-mode instead
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(
-        long,
-        default_value_if("enabled-credentials-mode", None, Some(DEFAULT_ETH_ENDPOINT))
-    )]
-    eth_endpoint: String,
-
-    /// Ethereum private key used for obtaining bandwidth tokens from ERC20 tokens. If you don't
-    /// want to set this value, use --enabled-credentials-mode instead")
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(
-        long,
-        default_value_if("enabled-credentials-mode", None, Some(DEFAULT_ETH_PRIVATE_KEY))
-    )]
-    eth_private_key: String,
 }
 
 impl From<Init> for OverrideConfig {
@@ -76,15 +55,8 @@ impl From<Init> for OverrideConfig {
             validators: init_config.validators,
             port: init_config.port,
             fastmode: init_config.fastmode,
-
-            #[cfg(any(feature = "eth", feature = "coconut"))]
+            #[cfg(feature = "coconut")]
             enabled_credentials_mode: init_config.enabled_credentials_mode,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_private_key: Some(init_config.eth_private_key),
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_endpoint: Some(init_config.eth_endpoint),
         }
     }
 }
@@ -120,7 +92,12 @@ pub(crate) async fn execute(args: &Init) {
     let override_config_fields = OverrideConfig::from(args.clone());
     config = override_config(config, override_config_fields);
 
-    let gateway = setup_gateway(id, register_gateway, user_chosen_gateway_id, &config).await;
+    let gateway = setup_gateway(id, register_gateway, user_chosen_gateway_id, &config)
+        .await
+        .unwrap_or_else(|err| {
+            eprintln!("Failed to setup gateway\nError: {err}");
+            std::process::exit(1)
+        });
     config.get_base_mut().with_gateway_endpoint(gateway);
 
     let config_save_location = config.get_config_file_save_location();
@@ -138,7 +115,10 @@ pub(crate) async fn execute(args: &Init) {
     );
     println!("Client configuration completed.");
 
-    client_core::init::show_address(config.get_base());
+    client_core::init::show_address(config.get_base()).unwrap_or_else(|err| {
+        eprintln!("Failed to show address\nError: {err}");
+        std::process::exit(1)
+    });
 }
 
 async fn setup_gateway(
@@ -146,7 +126,7 @@ async fn setup_gateway(
     register: bool,
     user_chosen_gateway_id: Option<&str>,
     config: &Config,
-) -> GatewayEndpoint {
+) -> Result<GatewayEndpoint, 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.
@@ -155,16 +135,16 @@ async fn setup_gateway(
             config.get_base().get_validator_api_endpoints(),
             user_chosen_gateway_id,
         )
-        .await;
+        .await?;
         log::debug!("Querying gateway gives: {}", gateway);
 
         // Registering with gateway by setting up and writing shared keys to disk
         log::trace!("Registering gateway");
         client_core::init::register_with_gateway_and_store_keys(gateway.clone(), config.get_base())
-            .await;
+            .await?;
         println!("Saved all generated keys");
 
-        gateway.into()
+        Ok(gateway.into())
     } else if user_chosen_gateway_id.is_some() {
         // Just set the config, don't register or create any keys
         // This assumes that the user knows what they are doing, and that the existing keys are
@@ -174,22 +154,21 @@ async fn setup_gateway(
             config.get_base().get_validator_api_endpoints(),
             user_chosen_gateway_id,
         )
-        .await;
+        .await?;
         log::debug!("Querying gateway gives: {}", gateway);
-        gateway.into()
+        Ok(gateway.into())
     } else {
         println!("Not registering gateway, will reuse existing config and keys");
-        match Config::load_from_file(Some(id)) {
-            Ok(existing_config) => existing_config.get_base().get_gateway_endpoint().clone(),
-            Err(err) => {
-                panic!(
-                    "Unable to configure gateway: {err}. \n
-                    Seems like the client was already initialized but it was not possible to read \
-                    the existing configuration file. \n
-                    CAUTION: Consider backing up your gateway keys and try force gateway registration, or \
-                    removing the existing configuration and starting over."
-                )
-            }
-        }
+        let existing_config = Config::load_from_file(Some(id)).map_err(|err| {
+            log::error!(
+                "Unable to configure gateway: {err}. \n
+                Seems like the client was already initialized but it was not possible to read \
+                the existing configuration file. \n
+                CAUTION: Consider backing up your gateway keys and try force gateway registration, or \
+                removing the existing configuration and starting over."
+            );
+            ClientCoreError::CouldNotLoadExistingGatewayConfiguration(err)
+        })?;
+        Ok(existing_config.get_base().get_gateway_endpoint().clone())
     }
 }

clients/socks5/src/commands/mod.rs

@@ -9,13 +9,6 @@ pub mod init;
 pub(crate) mod run;
 pub(crate) mod upgrade;
 
-#[cfg(not(feature = "coconut"))]
-pub(crate) const DEFAULT_ETH_ENDPOINT: &str =
-    "https://rinkeby.infura.io/v3/00000000000000000000000000000000";
-#[cfg(not(feature = "coconut"))]
-pub(crate) const DEFAULT_ETH_PRIVATE_KEY: &str =
-    "0000000000000000000000000000000000000000000000000000000000000001";
-
 fn long_version() -> String {
     format!(
         r#"
@@ -78,14 +71,8 @@ pub(crate) struct OverrideConfig {
     port: Option<u16>,
     fastmode: bool,
 
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     enabled_credentials_mode: bool,
-
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    eth_private_key: Option<String>,
-
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    eth_endpoint: Option<String>,
 }
 
 pub(crate) async fn execute(args: &Cli) {
@@ -111,31 +98,12 @@ pub(crate) fn override_config(mut config: Config, args: OverrideConfig) -> Confi
         config = config.with_port(port);
     }
 
-    #[cfg(all(not(feature = "eth"), not(feature = "coconut")))]
-    {
-        config
-            .get_base_mut()
-            .with_eth_endpoint(DEFAULT_ETH_ENDPOINT.to_string());
-        config
-            .get_base_mut()
-            .with_eth_private_key(DEFAULT_ETH_PRIVATE_KEY.to_string());
-    }
-
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     {
         if args.enabled_credentials_mode {
             config.get_base_mut().with_disabled_credentials(false)
         }
     }
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    {
-        if let Some(eth_endpoint) = args.eth_endpoint {
-            config.get_base_mut().with_eth_endpoint(eth_endpoint);
-        }
-        if let Some(eth_private_key) = args.eth_private_key {
-            config.get_base_mut().with_eth_private_key(eth_private_key);
-        }
-    }
 
     if args.fastmode {
         config.get_base_mut().set_high_default_traffic_volume();

clients/socks5/src/commands/run.rs

@@ -40,21 +40,9 @@ pub(crate) struct Run {
 
     /// Set this client to work in a enabled credentials mode that would attempt to use gateway
     /// with bandwidth credential requirement.
-    #[cfg(any(feature = "eth", feature = "coconut"))]
+    #[cfg(feature = "coconut")]
     #[clap(long)]
     enabled_credentials_mode: bool,
-
-    /// URL of an Ethereum full node that we want to use for getting bandwidth tokens from ERC20
-    /// tokens. If you don't want to set this value, use --enabled-credentials-mode instead
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(long)]
-    eth_endpoint: Option<String>,
-
-    /// Ethereum private key used for obtaining bandwidth tokens from ERC20 tokens. If you don't
-    /// want to set this value, use --enabled-credentials-mode instead
-    #[cfg(all(feature = "eth", not(feature = "coconut")))]
-    #[clap(long)]
-    eth_private_key: Option<String>,
 }
 
 impl From<Run> for OverrideConfig {
@@ -64,14 +52,8 @@ impl From<Run> for OverrideConfig {
             port: run_config.port,
             fastmode: false,
 
-            #[cfg(any(feature = "eth", feature = "coconut"))]
+            #[cfg(feature = "coconut")]
             enabled_credentials_mode: run_config.enabled_credentials_mode,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_private_key: run_config.eth_private_key,
-
-            #[cfg(all(feature = "eth", not(feature = "coconut")))]
-            eth_endpoint: run_config.eth_endpoint,
         }
     }
 }

clients/socks5/src/socks/client.rs

@@ -20,6 +20,7 @@ use socks5_requests::{ConnectionId, Message, RemoteAddress, Request};
 use std::io;
 use std::net::SocketAddr;
 use std::pin::Pin;
+use task::ShutdownListener;
 use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
 use tokio::{self, net::TcpStream};
 
@@ -140,6 +141,7 @@ pub(crate) struct SocksClient {
     service_provider: Recipient,
     self_address: Recipient,
     started_proxy: bool,
+    shutdown_listener: ShutdownListener,
 }
 
 impl Drop for SocksClient {
@@ -163,6 +165,7 @@ impl SocksClient {
         service_provider: Recipient,
         controller_sender: ControllerSender,
         self_address: Recipient,
+        shutdown_listener: ShutdownListener,
     ) -> Self {
         let connection_id = Self::generate_random();
         SocksClient {
@@ -176,6 +179,7 @@ impl SocksClient {
             service_provider,
             self_address,
             started_proxy: false,
+            shutdown_listener,
         }
     }
 
@@ -250,6 +254,7 @@ impl SocksClient {
             conn_receiver,
             input_sender,
             connection_id,
+            self.shutdown_listener.clone(),
         )
         .run(move |conn_id, read_data, socket_closed| {
             let provider_request = Request::new_send(conn_id, read_data, socket_closed);

clients/socks5/src/socks/mixnet_responses.rs

@@ -1,16 +1,19 @@
-use client_core::client::received_buffer::ReconstructedMessagesReceiver;
-use client_core::client::received_buffer::{ReceivedBufferMessage, ReceivedBufferRequestSender};
 use futures::channel::mpsc;
 use futures::StreamExt;
 use log::*;
+
+use client_core::client::received_buffer::ReconstructedMessagesReceiver;
+use client_core::client::received_buffer::{ReceivedBufferMessage, ReceivedBufferRequestSender};
 use nymsphinx::receiver::ReconstructedMessage;
 use proxy_helpers::connection_controller::{ControllerCommand, ControllerSender};
 use socks5_requests::Message;
+use task::ShutdownListener;
 
 pub(crate) struct MixnetResponseListener {
     buffer_requester: ReceivedBufferRequestSender,
     mix_response_receiver: ReconstructedMessagesReceiver,
     controller_sender: ControllerSender,
+    shutdown: ShutdownListener,
 }
 
 impl Drop for MixnetResponseListener {
@@ -25,6 +28,7 @@ impl MixnetResponseListener {
     pub(crate) fn new(
         buffer_requester: ReceivedBufferRequestSender,
         controller_sender: ControllerSender,
+        shutdown: ShutdownListener,
     ) -> Self {
         let (mix_response_sender, mix_response_receiver) = mpsc::unbounded();
         buffer_requester
@@ -35,6 +39,7 @@ impl MixnetResponseListener {
             buffer_requester,
             mix_response_receiver,
             controller_sender,
+            shutdown,
         }
     }
 
@@ -73,11 +78,25 @@ impl MixnetResponseListener {
     }
 
     pub(crate) async fn run(&mut self) {
-        while let Some(received_responses) = self.mix_response_receiver.next().await {
-            for reconstructed_message in received_responses {
-                self.on_message(reconstructed_message).await;
+        while !self.shutdown.is_shutdown() {
+            tokio::select! {
+                received_responses = self.mix_response_receiver.next() => match received_responses {
+                    Some(received_responses) => {
+                        for reconstructed_message in received_responses {
+                            self.on_message(reconstructed_message).await;
+                        }
+                    },
+                    None => {
+                        log::trace!("MixnetResponseListener: Stopping since channel closed");
+                        break;
+                    }
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("MixnetResponseListener: Received shutdown");
+                }
             }
         }
-        error!("We should never see this message");
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("MixnetResponseListener: Exiting");
     }
 }

clients/socks5/src/socks/server.rs

@@ -11,6 +11,7 @@ use log::*;
 use nymsphinx::addressing::clients::Recipient;
 use proxy_helpers::connection_controller::Controller;
 use std::net::SocketAddr;
+use task::ShutdownListener;
 use tokio::net::TcpListener;
 
 /// A Socks5 server that listens for connections.
@@ -19,6 +20,7 @@ pub struct SphinxSocksServer {
     listening_address: SocketAddr,
     service_provider: Recipient,
     self_address: Recipient,
+    shutdown: ShutdownListener,
 }
 
 impl SphinxSocksServer {
@@ -28,6 +30,7 @@ impl SphinxSocksServer {
         authenticator: Authenticator,
         service_provider: Recipient,
         self_address: Recipient,
+        shutdown: ShutdownListener,
     ) -> Self {
         // hardcode ip as we (presumably) ONLY want to listen locally. If we change it, we can
         // just modify the config
@@ -38,6 +41,7 @@ impl SphinxSocksServer {
             listening_address: format!("{}:{}", ip, port).parse().unwrap(),
             service_provider,
             self_address,
+            shutdown,
         }
     }
 
@@ -52,62 +56,73 @@ impl SphinxSocksServer {
         info!("Serving Connections...");
 
         // controller for managing all active connections
-        let (mut active_streams_controller, controller_sender) = Controller::new();
+        let (mut active_streams_controller, controller_sender) =
+            Controller::new(self.shutdown.clone());
         tokio::spawn(async move {
             active_streams_controller.run().await;
         });
 
         // listener for mix messages
-        let mut mixnet_response_listener =
-            MixnetResponseListener::new(buffer_requester, controller_sender.clone());
-
+        let mut mixnet_response_listener = MixnetResponseListener::new(
+            buffer_requester,
+            controller_sender.clone(),
+            self.shutdown.clone(),
+        );
         tokio::spawn(async move {
             mixnet_response_listener.run().await;
         });
 
         loop {
-            if let Ok((stream, _remote)) = listener.accept().await {
-                // TODO Optimize this
-                let mut client = SocksClient::new(
-                    stream,
-                    self.authenticator.clone(),
-                    input_sender.clone(),
-                    self.service_provider,
-                    controller_sender.clone(),
-                    self.self_address,
-                );
+            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,
+                        controller_sender.clone(),
+                        self.self_address,
+                        self.shutdown.clone(),
+                    );
 
-                tokio::spawn(async move {
-                    {
-                        match client.run().await {
-                            Ok(_) => {}
-                            Err(error) => {
-                                error!("Error! {}", error);
-                                let error_text = format!("{}", error);
+                    tokio::spawn(async move {
+                        {
+                            match client.run().await {
+                                Ok(_) => {}
+                                Err(error) => {
+                                    error!("Error! {}", error);
+                                    let error_text = format!("{}", error);
 
-                                let response: ResponseCode;
+                                    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 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");
+                                    };
                                 }
-
-                                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
-                    }
-                });
+                            };
+                            // client gets dropped here
+                        }
+                    });
+                },
+                _ = self.shutdown.recv() => {
+                    log::trace!("SphinxSocksServer: Received shutdown");
+                    log::debug!("SphinxSocksServer: Exiting");
+                    return Ok(());
+                }
             }
         }
     }

clients/validator/tests/query/balance.test.ts

@@ -4,8 +4,8 @@ import expect from 'expect';
 describe('Query: balances', () => {
   it('can query for an account balance', async () => {
     const client = await ValidatorClient.connectForQuery(
-      'https://rpc.nyx.nodes.guru/', 'https://validator.nymtech.net/api/', 'n', 'n14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sjyvg3g', 'n1nc5tatafv6eyq7llkr2gv50ff9e22mnf70qgjlv737ktmt4eswrq73f2nw', 'nym');
+      'https://rpc.nymtech.net/', 'https://validator.nymtech.net/api/', 'n', 'n14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9sjyvg3g', 'n1nc5tatafv6eyq7llkr2gv50ff9e22mnf70qgjlv737ktmt4eswrq73f2nw', 'nym');
     const balance = await client.getBalance('n10yyd98e2tuwu0f7ypz9dy3hhjw7v772q6287gy');
     expect(Number.parseFloat(balance.amount)).toBeGreaterThan(0);
   }).timeout(5000);
-})
+})

clients/webassembly/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "nym-client-wasm"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>", "Jedrzej Stuczynski <andrew@nymtech.net>"]
-version = "1.0.1"
+version = "1.0.0"
 edition = "2021"
 keywords = ["nym", "sphinx", "wasm", "webassembly", "privacy", "client"]
 license = "Apache-2.0"
@@ -32,7 +32,7 @@ credentials = { path = "../../common/credentials", optional = true }
 crypto = { path = "../../common/crypto" }
 nymsphinx = { path = "../../common/nymsphinx" }
 topology = { path = "../../common/topology" }
-gateway-client = { path = "../../common/client-libs/gateway-client", default-features = false, features = ["wasm"] }
+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" }
 

clients/webassembly/README.md

@@ -11,6 +11,7 @@ This client is part of the [Nym](https://nymtech.net/docs) project. It's written
 ## Security Status 
 
 From a security point of view, this module is not yet complete. Key missing features include, but are not limited to: cover traffic, sending packets with delay according to Poisson distribution. 
+
 They should be implemented soon. You can build your applications, but don't rely on it for strong anonymity yet if your application needs cover traffic.
 
 ## Using it
@@ -37,6 +38,6 @@ To be clear, this is not something that most JS developers need to worry about,
 
 If you're a Nym platform developer who's made changes to the Rust (or JS) files and wants to re-publish the package to NPM, here's how you do it: 
 
-1. `wasm-pack build --scope nymproject` builds the wasm binaries into the `pkg` directory (not in source control)
-2. bump version numbers as necessary for SemVer
-3. `wasm-pack publish --access=public` will publish your changed package to NPM
+1. bump version numbers as necessary for SemVer
+2. `wasm-pack build --scope nymproject` builds the wasm binaries into the `pkg` directory (not in source control)
+3. `cd pkg && npm publish --access=public` will publish your changed package to NPM

clients/webassembly/js-example/index.js

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 import {
-   NymClient,
-   set_panic_hook
+    NymClient,
+    set_panic_hook
 } from "@nymproject/nym-client-wasm"
 
 // current limitation of rust-wasm for async stuff : (
@@ -25,7 +25,7 @@ async function main() {
     set_panic_hook();
 
     // validator server we will use to get topology from
-    const validator = "https://sandbox-validator.nymtech.net/api"; //"http://localhost:8081";
+    const validator = "https://validator.nymtech.net/api"; //"http://localhost:8081";
 
     client = new NymClient(validator);
 

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

@@ -36,6 +36,64 @@
         "node": ">=10.0.0"
       }
     },
+    "node_modules/@jridgewell/gen-mapping": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.2.tgz",
+      "integrity": "sha512-mh65xKQAzI6iBcFzwv28KVWSmCkdRBWoOh+bYQGW3+6OZvbbN3TqMGo5hqYxQniRcH9F2VZIoJCm4pa3BPDK/A==",
+      "dev": true,
+      "dependencies": {
+        "@jridgewell/set-array": "^1.0.1",
+        "@jridgewell/sourcemap-codec": "^1.4.10",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      },
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/resolve-uri": {
+      "version": "3.1.0",
+      "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.0.tgz",
+      "integrity": "sha512-F2msla3tad+Mfht5cJq7LSXcdudKTWCVYUgw6pLFOOHSTtZlj6SWNYAp+AhuqLmWdBO2X5hPrLcu8cVP8fy28w==",
+      "dev": true,
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/set-array": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/set-array/-/set-array-1.1.2.tgz",
+      "integrity": "sha512-xnkseuNADM0gt2bs+BvhO0p78Mk762YnZdsuzFV018NoG1Sj1SCQvpSqa7XUaTam5vAGasABV9qXASMKnFMwMw==",
+      "dev": true,
+      "engines": {
+        "node": ">=6.0.0"
+      }
+    },
+    "node_modules/@jridgewell/source-map": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/source-map/-/source-map-0.3.2.tgz",
+      "integrity": "sha512-m7O9o2uR8k2ObDysZYzdfhb08VuEml5oWGiosa1VdaPZ/A6QyPkAJuwN0Q1lhULOf6B7MtQmHENS743hWtCrgw==",
+      "dev": true,
+      "dependencies": {
+        "@jridgewell/gen-mapping": "^0.3.0",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "node_modules/@jridgewell/sourcemap-codec": {
+      "version": "1.4.14",
+      "resolved": "https://registry.npmjs.org/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.4.14.tgz",
+      "integrity": "sha512-XPSJHWmi394fuUuzDnGz1wiKqWfo1yXecHQMRf2l6hztTO+nPru658AyDngaBe7isIxEkRsPR3FZh+s7iVa4Uw==",
+      "dev": true
+    },
+    "node_modules/@jridgewell/trace-mapping": {
+      "version": "0.3.15",
+      "resolved": "https://registry.npmjs.org/@jridgewell/trace-mapping/-/trace-mapping-0.3.15.tgz",
+      "integrity": "sha512-oWZNOULl+UbhsgB51uuZzglikfIKSUBO/M9W2OfEjn7cmqoAiCgmv9lyACTUacZwBz0ITnJ2NqjU8Tx0DHL88g==",
+      "dev": true,
+      "dependencies": {
+        "@jridgewell/resolve-uri": "^3.0.3",
+        "@jridgewell/sourcemap-codec": "^1.4.10"
+      }
+    },
     "node_modules/@nodelib/fs.scandir": {
       "version": "2.1.5",
       "resolved": "https://registry.npmjs.org/@nodelib/fs.scandir/-/fs.scandir-2.1.5.tgz",
@@ -3272,14 +3330,14 @@
       }
     },
     "node_modules/terser": {
-      "version": "5.12.1",
-      "resolved": "https://registry.npmjs.org/terser/-/terser-5.12.1.tgz",
-      "integrity": "sha512-NXbs+7nisos5E+yXwAD+y7zrcTkMqb0dEJxIGtSKPdCBzopf7ni4odPul2aechpV7EXNvOudYOX2bb5tln1jbQ==",
+      "version": "5.15.0",
+      "resolved": "https://registry.npmjs.org/terser/-/terser-5.15.0.tgz",
+      "integrity": "sha512-L1BJiXVmheAQQy+as0oF3Pwtlo4s3Wi1X2zNZ2NxOB4wx9bdS9Vk67XQENLFdLYGCK/Z2di53mTj/hBafR+dTA==",
       "dev": true,
       "dependencies": {
+        "@jridgewell/source-map": "^0.3.2",
         "acorn": "^8.5.0",
         "commander": "^2.20.0",
-        "source-map": "~0.7.2",
         "source-map-support": "~0.5.20"
       },
       "bin": {
@@ -3372,15 +3430,6 @@
         "url": "https://opencollective.com/webpack"
       }
     },
-    "node_modules/terser/node_modules/source-map": {
-      "version": "0.7.3",
-      "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.7.3.tgz",
-      "integrity": "sha512-CkCj6giN3S+n9qrYiBTX5gystlENnRW5jZeNLHpe6aue+SrHcG5VYwujhW9s4dY31mEGsxBDrHR6oI69fTXsaQ==",
-      "dev": true,
-      "engines": {
-        "node": ">= 8"
-      }
-    },
     "node_modules/thunky": {
       "version": "1.1.0",
       "resolved": "https://registry.npmjs.org/thunky/-/thunky-1.1.0.tgz",
@@ -3828,6 +3877,55 @@
       "integrity": "sha512-dBVuXR082gk3jsFp7Rd/JI4kytwGHecnCoTtXFb7DB6CNHp4rg5k1bhg0nWdLGLnOV71lmDzGQaLMy8iPLY0pw==",
       "dev": true
     },
+    "@jridgewell/gen-mapping": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/gen-mapping/-/gen-mapping-0.3.2.tgz",
+      "integrity": "sha512-mh65xKQAzI6iBcFzwv28KVWSmCkdRBWoOh+bYQGW3+6OZvbbN3TqMGo5hqYxQniRcH9F2VZIoJCm4pa3BPDK/A==",
+      "dev": true,
+      "requires": {
+        "@jridgewell/set-array": "^1.0.1",
+        "@jridgewell/sourcemap-codec": "^1.4.10",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "@jridgewell/resolve-uri": {
+      "version": "3.1.0",
+      "resolved": "https://registry.npmjs.org/@jridgewell/resolve-uri/-/resolve-uri-3.1.0.tgz",
+      "integrity": "sha512-F2msla3tad+Mfht5cJq7LSXcdudKTWCVYUgw6pLFOOHSTtZlj6SWNYAp+AhuqLmWdBO2X5hPrLcu8cVP8fy28w==",
+      "dev": true
+    },
+    "@jridgewell/set-array": {
+      "version": "1.1.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/set-array/-/set-array-1.1.2.tgz",
+      "integrity": "sha512-xnkseuNADM0gt2bs+BvhO0p78Mk762YnZdsuzFV018NoG1Sj1SCQvpSqa7XUaTam5vAGasABV9qXASMKnFMwMw==",
+      "dev": true
+    },
+    "@jridgewell/source-map": {
+      "version": "0.3.2",
+      "resolved": "https://registry.npmjs.org/@jridgewell/source-map/-/source-map-0.3.2.tgz",
+      "integrity": "sha512-m7O9o2uR8k2ObDysZYzdfhb08VuEml5oWGiosa1VdaPZ/A6QyPkAJuwN0Q1lhULOf6B7MtQmHENS743hWtCrgw==",
+      "dev": true,
+      "requires": {
+        "@jridgewell/gen-mapping": "^0.3.0",
+        "@jridgewell/trace-mapping": "^0.3.9"
+      }
+    },
+    "@jridgewell/sourcemap-codec": {
+      "version": "1.4.14",
+      "resolved": "https://registry.npmjs.org/@jridgewell/sourcemap-codec/-/sourcemap-codec-1.4.14.tgz",
+      "integrity": "sha512-XPSJHWmi394fuUuzDnGz1wiKqWfo1yXecHQMRf2l6hztTO+nPru658AyDngaBe7isIxEkRsPR3FZh+s7iVa4Uw==",
+      "dev": true
+    },
+    "@jridgewell/trace-mapping": {
+      "version": "0.3.15",
+      "resolved": "https://registry.npmjs.org/@jridgewell/trace-mapping/-/trace-mapping-0.3.15.tgz",
+      "integrity": "sha512-oWZNOULl+UbhsgB51uuZzglikfIKSUBO/M9W2OfEjn7cmqoAiCgmv9lyACTUacZwBz0ITnJ2NqjU8Tx0DHL88g==",
+      "dev": true,
+      "requires": {
+        "@jridgewell/resolve-uri": "^3.0.3",
+        "@jridgewell/sourcemap-codec": "^1.4.10"
+      }
+    },
     "@nodelib/fs.scandir": {
       "version": "2.1.5",
       "resolved": "https://registry.npmjs.org/@nodelib/fs.scandir/-/fs.scandir-2.1.5.tgz",
@@ -6334,23 +6432,15 @@
       "dev": true
     },
     "terser": {
-      "version": "5.12.1",
-      "resolved": "https://registry.npmjs.org/terser/-/terser-5.12.1.tgz",
-      "integrity": "sha512-NXbs+7nisos5E+yXwAD+y7zrcTkMqb0dEJxIGtSKPdCBzopf7ni4odPul2aechpV7EXNvOudYOX2bb5tln1jbQ==",
+      "version": "5.15.0",
+      "resolved": "https://registry.npmjs.org/terser/-/terser-5.15.0.tgz",
+      "integrity": "sha512-L1BJiXVmheAQQy+as0oF3Pwtlo4s3Wi1X2zNZ2NxOB4wx9bdS9Vk67XQENLFdLYGCK/Z2di53mTj/hBafR+dTA==",
       "dev": true,
       "requires": {
+        "@jridgewell/source-map": "^0.3.2",
         "acorn": "^8.5.0",
         "commander": "^2.20.0",
-        "source-map": "~0.7.2",
         "source-map-support": "~0.5.20"
-      },
-      "dependencies": {
-        "source-map": {
-          "version": "0.7.3",
-          "resolved": "https://registry.npmjs.org/source-map/-/source-map-0.7.3.tgz",
-          "integrity": "sha512-CkCj6giN3S+n9qrYiBTX5gystlENnRW5jZeNLHpe6aue+SrHcG5VYwujhW9s4dY31mEGsxBDrHR6oI69fTXsaQ==",
-          "dev": true
-        }
       }
     },
     "terser-webpack-plugin": {

clients/webassembly/js-example/webpack.config.js

@@ -9,7 +9,7 @@ module.exports = {
   },
   mode: "development",
   plugins: [
-    new CopyWebpackPlugin({patterns: ['index.html']})
+    new CopyWebpackPlugin({ patterns: ['index.html'] })
   ],
-  experiments: { asyncWebAssembly: true }
+  experiments: { syncWebAssembly: true }
 };

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

@@ -10,23 +10,22 @@ edition = "2021"
 # TODO: (for this and other crates), similarly to 'tokio', import only required "futures" modules rather than
 # the entire crate
 futures = "0.3"
-json = "0.12.4"
 log = "0.4"
 thiserror = "1.0"
 url = "2.2"
 rand = { version = "0.7.3", features = ["wasm-bindgen"] }
-secp256k1 = "0.20.3"
-web3 = { version = "0.17.0", default-features = false }
+secp256k1 = { version = "0.20.3", optional = true }
+web3 = { version = "0.17.0", default-features = false, optional = true }
 async-trait = { version = "0.1.51" }
 
 # internal
+coconut-interface = { path = "../../coconut-interface", optional = true }
 credentials = { path = "../../credentials" }
 crypto = { path = "../../crypto" }
 gateway-requests = { path = "../../../gateway/gateway-requests" }
+network-defaults = { path = "../../network-defaults" }
 nymsphinx = { path = "../../nymsphinx" }
 pemstore = { path = "../../pemstore" }
-coconut-interface = { path = "../../coconut-interface", optional = true }
-network-defaults = { path = "../../network-defaults" }
 validator-client = { path = "../validator-client", optional = true }
 
 [dependencies.tungstenite]
@@ -38,12 +37,19 @@ default-features = false
 version = "1.19.1"
 features = ["macros", "rt", "net", "sync", "time"]
 
+[target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-stream]
+version = "0.1.9"
+features = ["net", "sync", "time"]
+
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies.tokio-tungstenite]
 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"
@@ -73,5 +79,5 @@ features = ["js"]
 
 [features]
 coconut = ["gateway-requests/coconut", "coconut-interface", "validator-client", "credentials/coconut"]
-wasm = ["web3/wasm", "web3/http", "web3/signing"]
-default = ["web3/default"]
+wasm = []
+default = ["web3/default", "secp256k1"]

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

@@ -2,7 +2,6 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::error::GatewayClientError;
-use std::str::FromStr;
 
 #[cfg(target_arch = "wasm32")]
 use crate::wasm_storage::Storage;
@@ -15,6 +14,8 @@ use crate::wasm_storage::StorageError;
 #[cfg(all(not(target_arch = "wasm32"), feature = "coconut"))]
 use credential_storage::error::StorageError;
 
+#[cfg(feature = "coconut")]
+use std::str::FromStr;
 #[cfg(feature = "coconut")]
 use {
     coconut_interface::Base58,
@@ -23,65 +24,12 @@ use {
     },
 };
 
-#[cfg(not(feature = "coconut"))]
-use {
-    credentials::token::bandwidth::TokenCredential,
-    crypto::asymmetric::identity,
-    network_defaults::{
-        eth_contract::ETH_ERC20_JSON_ABI, eth_contract::ETH_JSON_ABI, BANDWIDTH_VALUE,
-        ETH_BURN_FUNCTION_NAME, ETH_CONTRACT_ADDRESS, ETH_ERC20_APPROVE_FUNCTION_NAME,
-        ETH_ERC20_CONTRACT_ADDRESS, ETH_MIN_BLOCK_DEPTH, TOKENS_TO_BURN, UTOKENS_TO_BURN,
-    },
-    pemstore::traits::PemStorableKeyPair,
-    rand::rngs::OsRng,
-    secp256k1::SecretKey,
-    web3::{
-        contract::{Contract, Options},
-        ethabi::Token,
-        signing::{Key, SecretKeyRef},
-        transports::Http,
-        types::{Address, U256, U64},
-        Web3,
-    },
-};
-
-#[cfg(not(feature = "coconut"))]
-pub fn eth_contract(web3: Web3<Http>) -> Contract<Http> {
-    Contract::from_json(
-        web3.eth(),
-        Address::from(ETH_CONTRACT_ADDRESS),
-        json::parse(ETH_JSON_ABI)
-            .expect("Invalid json abi")
-            .dump()
-            .as_bytes(),
-    )
-    .expect("Invalid json abi")
-}
-
-#[cfg(not(feature = "coconut"))]
-pub fn eth_erc20_contract(web3: Web3<Http>) -> Contract<Http> {
-    Contract::from_json(
-        web3.eth(),
-        Address::from(ETH_ERC20_CONTRACT_ADDRESS),
-        json::parse(ETH_ERC20_JSON_ABI)
-            .expect("Invalid json abi")
-            .dump()
-            .as_bytes(),
-    )
-    .expect("Invalid json abi")
-}
-
 #[derive(Clone)]
 pub struct BandwidthController<St: Storage> {
+    #[allow(dead_code)]
     storage: St,
     #[cfg(feature = "coconut")]
     validator_endpoints: Vec<url::Url>,
-    #[cfg(not(feature = "coconut"))]
-    contract: Contract<Http>,
-    #[cfg(not(feature = "coconut"))]
-    erc20_contract: Contract<Http>,
-    #[cfg(not(feature = "coconut"))]
-    eth_private_key: SecretKey,
 }
 
 impl<St> BandwidthController<St>
@@ -97,60 +45,8 @@ where
     }
 
     #[cfg(not(feature = "coconut"))]
-    pub fn new(
-        storage: St,
-        eth_endpoint: String,
-        eth_private_key: String,
-    ) -> Result<Self, GatewayClientError> {
-        // Fail early, on invalid url
-        let transport =
-            Http::new(&eth_endpoint).map_err(|_| GatewayClientError::InvalidURL(eth_endpoint))?;
-        let web3 = web3::Web3::new(transport);
-        // Fail early, on invalid abi
-        let contract = eth_contract(web3.clone());
-        let erc20_contract = eth_erc20_contract(web3);
-        let eth_private_key = secp256k1::SecretKey::from_str(&eth_private_key)
-            .map_err(|_| GatewayClientError::InvalidEthereumPrivateKey)?;
-
-        Ok(BandwidthController {
-            storage,
-            contract,
-            erc20_contract,
-            eth_private_key,
-        })
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    async fn backup_keypair(&self, keypair: &identity::KeyPair) -> Result<(), GatewayClientError> {
-        self.storage
-            .insert_erc20_credential(
-                keypair.public_key().to_base58_string(),
-                keypair.private_key().to_base58_string(),
-            )
-            .await?;
-
-        Ok(())
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    async fn restore_keypair(&self) -> Result<identity::KeyPair, GatewayClientError> {
-        let data = self.storage.get_next_erc20_credential().await?;
-        let public_key = identity::PublicKey::from_base58_string(data.public_key).unwrap();
-        let private_key = identity::PrivateKey::from_base58_string(data.private_key).unwrap();
-
-        Ok(identity::KeyPair::from_keys(private_key, public_key))
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    async fn mark_keypair_as_spent(
-        &self,
-        keypair: &identity::KeyPair,
-    ) -> Result<(), GatewayClientError> {
-        self.storage
-            .consume_erc20_credential(keypair.public_key().to_base58_string())
-            .await?;
-
-        Ok(())
+    pub fn new(storage: St) -> Result<Self, GatewayClientError> {
+        Ok(BandwidthController { storage })
     }
 
     #[cfg(feature = "coconut")]
@@ -179,206 +75,4 @@ where
             &verification_key,
         )?)
     }
-
-    #[cfg(not(feature = "coconut"))]
-    pub async fn prepare_token_credential(
-        &self,
-        gateway_identity: identity::PublicKey,
-        gateway_owner: String,
-    ) -> Result<TokenCredential, GatewayClientError> {
-        let kp = match self.restore_keypair().await {
-            Ok(kp) => kp,
-            Err(_) => {
-                let mut rng = OsRng;
-                let kp = identity::KeyPair::new(&mut rng);
-                self.backup_keypair(&kp).await?;
-                kp
-            }
-        };
-
-        let verification_key = *kp.public_key();
-        let signed_verification_key = kp.private_key().sign(&verification_key.to_bytes());
-        self.buy_token_credential(verification_key, signed_verification_key, gateway_owner)
-            .await?;
-
-        self.mark_keypair_as_spent(&kp).await?;
-
-        let message: Vec<u8> = verification_key
-            .to_bytes()
-            .iter()
-            .chain(gateway_identity.to_bytes().iter())
-            .copied()
-            .collect();
-        let signature = kp.private_key().sign(&message);
-
-        Ok(TokenCredential::new(
-            verification_key,
-            gateway_identity,
-            BANDWIDTH_VALUE,
-            signature,
-        ))
-    }
-
-    #[cfg(not(feature = "coconut"))]
-    pub async fn buy_token_credential(
-        &self,
-        verification_key: identity::PublicKey,
-        signed_verification_key: identity::Signature,
-        gateway_owner: String,
-    ) -> Result<(), GatewayClientError> {
-        let confirmations = if cfg!(debug_assertions) {
-            1
-        } else {
-            ETH_MIN_BLOCK_DEPTH
-        };
-        // 15 seconds per confirmation block + 10 seconds of network overhead + 20 seconds of wait for kill
-        log::info!(
-            "Waiting for Ethereum transaction. This should take about {} seconds",
-            (confirmations + 1) * 15 + 30
-        );
-        let mut options = Options::default();
-        let estimation = self
-            .erc20_contract
-            .estimate_gas(
-                ETH_ERC20_APPROVE_FUNCTION_NAME,
-                (
-                    Token::Address(Address::from(ETH_CONTRACT_ADDRESS)),
-                    Token::Uint(U256::from(UTOKENS_TO_BURN)),
-                ),
-                SecretKeyRef::from(&self.eth_private_key).address(),
-                options.clone(),
-            )
-            .await?;
-        options.gas = Some(estimation);
-        log::info!("Calling ERC20 approve in 10 seconds with an estimated gas of {}. Kill the process if you want to abort", estimation);
-        #[cfg(not(target_arch = "wasm32"))]
-        tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
-        #[cfg(target_arch = "wasm32")]
-        if let Err(err) = fluvio_wasm_timer::Delay::new(std::time::Duration::from_secs(10)).await {
-            log::error!(
-                "the timer has gone away while waiting for possible kill! - {}",
-                err
-            );
-        }
-        let recipt = self
-            .erc20_contract
-            .signed_call_with_confirmations(
-                ETH_ERC20_APPROVE_FUNCTION_NAME,
-                (
-                    Token::Address(Address::from(ETH_CONTRACT_ADDRESS)),
-                    Token::Uint(U256::from(UTOKENS_TO_BURN)),
-                ),
-                options,
-                1, // One confirmation is enough, as we'll be consuming the approved token next anyway
-                &self.eth_private_key,
-            )
-            .await?;
-        if Some(U64::from(0u64)) == recipt.status {
-            return Err(GatewayClientError::BurnTokenError(
-                web3::Error::InvalidResponse(format!(
-                    "Approve transaction status is 0 (failure): {:?}",
-                    recipt.logs,
-                )),
-            ));
-        } else {
-            log::info!(
-                "Approved {} tokens for bandwidth use on Ethereum",
-                TOKENS_TO_BURN
-            );
-        }
-
-        let mut options = Options::default();
-        let estimation = self
-            .contract
-            .estimate_gas(
-                ETH_BURN_FUNCTION_NAME,
-                (
-                    Token::Uint(U256::from(UTOKENS_TO_BURN)),
-                    Token::Uint(U256::from(&verification_key.to_bytes())),
-                    Token::Bytes(signed_verification_key.to_bytes().to_vec()),
-                    Token::String(gateway_owner.clone()),
-                ),
-                SecretKeyRef::from(&self.eth_private_key).address(),
-                options.clone(),
-            )
-            .await?;
-        options.gas = Some(estimation);
-        log::info!("Generating bandwidth on ETH contract in 10 seconds with an estimated gas of {}. \
-         Kill the process if you want to abort. Keep in mind that if you abort now, you'll still have \
-         some tokens approved for bandwidth spending from the previous action. \
-         If you don't want that, you'll need to manually decreaseAllowance to revert the approval.", estimation);
-        #[cfg(not(target_arch = "wasm32"))]
-        tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
-        #[cfg(target_arch = "wasm32")]
-        if let Err(err) = fluvio_wasm_timer::Delay::new(std::time::Duration::from_secs(10)).await {
-            log::error!(
-                "the timer has gone away while waiting for possible kill! - {}",
-                err
-            );
-        }
-        let recipt = self
-            .contract
-            .signed_call_with_confirmations(
-                ETH_BURN_FUNCTION_NAME,
-                (
-                    Token::Uint(U256::from(UTOKENS_TO_BURN)),
-                    Token::Uint(U256::from(&verification_key.to_bytes())),
-                    Token::Bytes(signed_verification_key.to_bytes().to_vec()),
-                    Token::String(gateway_owner),
-                ),
-                options,
-                confirmations,
-                &self.eth_private_key,
-            )
-            .await?;
-        if Some(U64::from(0u64)) == recipt.status {
-            Err(GatewayClientError::BurnTokenError(
-                web3::Error::InvalidResponse(format!(
-                    "Transaction status is 0 (failure): {:?}",
-                    recipt.logs,
-                )),
-            ))
-        } else {
-            log::info!(
-                "Bought bandwidth on Ethereum: {} MB",
-                BANDWIDTH_VALUE / 1024 / 1024
-            );
-            Ok(())
-        }
-    }
-}
-
-#[cfg(not(feature = "coconut"))]
-#[cfg(test)]
-mod tests {
-    use network_defaults::ETH_EVENT_NAME;
-
-    use super::*;
-
-    #[test]
-    fn parse_contract() {
-        let transport =
-            Http::new("https://rinkeby.infura.io/v3/00000000000000000000000000000000").unwrap();
-        let web3 = web3::Web3::new(transport);
-        // test no panic occurs
-        eth_contract(web3);
-    }
-
-    #[test]
-    fn parse_erc20_contract() {
-        let transport =
-            Http::new("https://rinkeby.infura.io/v3/00000000000000000000000000000000").unwrap();
-        let web3 = web3::Web3::new(transport);
-        // test no panic occurs
-        eth_erc20_contract(web3);
-    }
-
-    #[test]
-    fn check_event_name_constant_against_abi() {
-        let transport =
-            Http::new("https://rinkeby.infura.io/v3/00000000000000000000000000000000").unwrap();
-        let web3 = web3::Web3::new(transport);
-        let contract = eth_contract(web3);
-        assert!(contract.abi().event(ETH_EVENT_NAME).is_ok());
-    }
 }

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

@@ -15,8 +15,6 @@ use crate::wasm_storage::PersistentStorage;
 use coconut_interface::Credential;
 #[cfg(not(target_arch = "wasm32"))]
 use credential_storage::PersistentStorage;
-#[cfg(not(feature = "coconut"))]
-use credentials::token::bandwidth::TokenCredential;
 use crypto::asymmetric::identity;
 use futures::{FutureExt, SinkExt, StreamExt};
 use gateway_requests::authentication::encrypted_address::EncryptedAddressBytes;
@@ -30,6 +28,8 @@ use rand::rngs::OsRng;
 use std::convert::TryFrom;
 use std::sync::Arc;
 use std::time::Duration;
+#[cfg(not(target_arch = "wasm32"))]
+use task::ShutdownListener;
 use tungstenite::protocol::Message;
 
 #[cfg(not(target_arch = "wasm32"))]
@@ -65,6 +65,10 @@ pub struct GatewayClient {
     reconnection_attempts: usize,
     /// Delay between each subsequent reconnection attempt.
     reconnection_backoff: Duration,
+
+    #[cfg(not(target_arch = "wasm32"))]
+    /// Listen to shutdown messages.
+    shutdown: Option<ShutdownListener>,
 }
 
 impl GatewayClient {
@@ -80,6 +84,7 @@ impl GatewayClient {
         ack_sender: AcknowledgementSender,
         response_timeout_duration: Duration,
         bandwidth_controller: Option<BandwidthController<PersistentStorage>>,
+        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
     ) -> Self {
         GatewayClient {
             authenticated: false,
@@ -91,12 +96,19 @@ impl GatewayClient {
             local_identity,
             shared_key,
             connection: SocketState::NotConnected,
-            packet_router: PacketRouter::new(ack_sender, mixnet_message_sender),
+            packet_router: PacketRouter::new(
+                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,
         }
     }
 
@@ -123,6 +135,7 @@ impl GatewayClient {
         gateway_owner: String,
         local_identity: Arc<identity::KeyPair>,
         response_timeout_duration: Duration,
+        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
     ) -> Self {
         use futures::channel::mpsc;
 
@@ -130,7 +143,12 @@ 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(ack_tx, mix_tx);
+        let packet_router = PacketRouter::new(
+            ack_tx,
+            mix_tx,
+            #[cfg(not(target_arch = "wasm32"))]
+            shutdown.clone(),
+        );
 
         GatewayClient {
             authenticated: false,
@@ -148,6 +166,8 @@ impl GatewayClient {
             should_reconnect_on_failure: false,
             reconnection_attempts: DEFAULT_RECONNECTION_ATTEMPTS,
             reconnection_backoff: DEFAULT_RECONNECTION_BACKOFF,
+            #[cfg(not(target_arch = "wasm32"))]
+            shutdown,
         }
     }
 
@@ -279,8 +299,33 @@ impl GatewayClient {
         let mut fused_timeout = timeout.fuse();
         let mut fused_stream = conn.fuse();
 
+        // Bit of an ugly workaround for selecting on an `Option` without having access to
+        // `tokio::select`
+        #[cfg(not(target_arch = "wasm32"))]
+        let shutdown = {
+            let m_shutdown = self.shutdown.clone();
+            async {
+                if let Some(mut s) = m_shutdown {
+                    s.recv().await
+                } else {
+                    std::future::pending::<()>().await
+                }
+            }
+            .fuse()
+        };
+        #[cfg(not(target_arch = "wasm32"))]
+        tokio::pin!(shutdown);
+
+        #[cfg(target_arch = "wasm32")]
+        let mut shutdown = std::future::pending::<()>().fuse();
+
         loop {
             futures::select! {
+                _ = shutdown => {
+                    log::trace!("GatewayClient control response: Received shutdown");
+                    log::debug!("GatewayClient control response: Exiting");
+                    break Err(GatewayClientError::ConnectionClosedGatewayShutdown);
+                }
                 _ = &mut fused_timeout => {
                     break Err(GatewayClientError::Timeout);
                 }
@@ -291,14 +336,16 @@ impl GatewayClient {
                     };
                     match ws_msg {
                         Message::Binary(bin_msg) => {
-                            self.packet_router.route_received(vec![bin_msg]);
+                            if let Err(err) = self.packet_router.route_received(vec![bin_msg]) {
+                                log::warn!("Route received failed: {:?}", err);
+                            }
                         }
                         Message::Text(txt_msg) => {
                             break ServerResponse::try_from(txt_msg).map_err(|_| GatewayClientError::MalformedResponse);
                         }
                         _ => (),
                     }
-               }
+                }
             }
         }
     }
@@ -505,30 +552,6 @@ impl GatewayClient {
         Ok(())
     }
 
-    #[cfg(not(feature = "coconut"))]
-    async fn claim_token_bandwidth(
-        &mut self,
-        credential: TokenCredential,
-    ) -> Result<(), GatewayClientError> {
-        let mut rng = OsRng;
-
-        let iv = IV::new_random(&mut rng);
-
-        let msg = ClientControlRequest::new_enc_token_bandwidth_credential(
-            &credential,
-            self.shared_key.as_ref().unwrap(),
-            iv,
-        )
-        .into();
-        self.bandwidth_remaining = match self.send_websocket_message(msg).await? {
-            ServerResponse::Bandwidth { available_total } => Ok(available_total),
-            ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
-            _ => Err(GatewayClientError::UnexpectedResponse),
-        }?;
-
-        Ok(())
-    }
-
     async fn try_claim_testnet_bandwidth(&mut self) -> Result<(), GatewayClientError> {
         let msg = ClientControlRequest::ClaimFreeTestnetBandwidth.into();
         self.bandwidth_remaining = match self.send_websocket_message(msg).await? {
@@ -567,17 +590,10 @@ impl GatewayClient {
             .prepare_coconut_credential()
             .await?;
         #[cfg(not(feature = "coconut"))]
-        let credential = self
-            .bandwidth_controller
-            .as_ref()
-            .unwrap()
-            .prepare_token_credential(self.gateway_identity, _gateway_owner)
-            .await?;
+        return self.try_claim_testnet_bandwidth().await;
 
         #[cfg(feature = "coconut")]
         return self.claim_coconut_bandwidth(credential).await;
-        #[cfg(not(feature = "coconut"))]
-        return self.claim_token_bandwidth(credential).await;
     }
 
     fn estimate_required_bandwidth(&self, packets: &[MixPacket]) -> i64 {
@@ -723,6 +739,8 @@ impl GatewayClient {
                                 .as_ref()
                                 .expect("no shared key present even though we're authenticated!"),
                         ),
+                        #[cfg(not(target_arch = "wasm32"))]
+                        self.shutdown.clone(),
                     )
                 }
                 _ => unreachable!(),

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

@@ -64,6 +64,9 @@ pub enum GatewayClientError {
     #[error("Connection was abruptly closed")]
     ConnectionAbruptlyClosed,
 
+    #[error("Connection was abruptly closed as gateway was stopped")]
+    ConnectionClosedGatewayShutdown,
+
     #[error("Received response was malformed")]
     MalformedResponse,
 
@@ -93,6 +96,9 @@ pub enum GatewayClientError {
 
     #[error("Timed out")]
     Timeout,
+
+    #[error("Failed to send mixnet message")]
+    MixnetMsgSenderFailedToSend,
 }
 
 impl GatewayClientError {

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

@@ -8,6 +8,10 @@ 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>>>;
@@ -19,20 +23,28 @@ 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>,
 }
 
 impl PacketRouter {
     pub fn new(
         ack_sender: AcknowledgementSender,
         mixnet_message_sender: MixnetMessageSender,
+        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
     ) -> Self {
         PacketRouter {
             ack_sender,
             mixnet_message_sender,
+            #[cfg(not(target_arch = "wasm32"))]
+            shutdown,
         }
     }
 
-    pub fn route_received(&self, unwrapped_packets: Vec<Vec<u8>>) {
+    pub fn route_received(
+        &mut self,
+        unwrapped_packets: Vec<Vec<u8>>,
+    ) -> Result<(), GatewayClientError> {
         let mut received_messages = Vec::new();
         let mut received_acks = Vec::new();
 
@@ -60,24 +72,29 @@ impl PacketRouter {
             }
         }
 
-        // due to how we are currently using it, those unwraps can't fail, but if we ever
-        // wanted to make `gateway-client` into some more generic library, we would probably need
-        // to catch that error or something.
         if !received_messages.is_empty() {
             trace!("routing 'real'");
-            self.mixnet_message_sender
-                .unbounded_send(received_messages)
-                .unwrap();
+            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
+                        log::warn!("Failed to send mixnet message due to receiver task shutdown");
+                        return Err(GatewayClientError::MixnetMsgSenderFailedToSend);
+                    }
+                }
+                // This should never happen during ordinary operation the way it's currently used.
+                // Abort to be on the safe side
+                panic!("Failed to send mixnet message: {:?}", err);
+            }
         }
 
         if !received_acks.is_empty() {
             trace!("routing acks");
-            match self.ack_sender.unbounded_send(received_acks) {
-                Ok(_) => {}
-                Err(e) => {
-                    error!("failed to send ack: {:?}", e);
-                }
+            if let Err(e) = self.ack_sender.unbounded_send(received_acks) {
+                error!("failed to send ack: {:?}", e);
             };
         }
+        Ok(())
     }
 }

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

@@ -11,6 +11,8 @@ use gateway_requests::registration::handshake::SharedKeys;
 use gateway_requests::BinaryResponse;
 use log::*;
 use std::sync::Arc;
+#[cfg(not(target_arch = "wasm32"))]
+use task::ShutdownListener;
 use tungstenite::Message;
 
 #[cfg(not(target_arch = "wasm32"))]
@@ -44,9 +46,9 @@ pub(crate) struct PartiallyDelegated {
 impl PartiallyDelegated {
     fn route_socket_message(
         ws_msg: Message,
-        packet_router: &PacketRouter,
+        packet_router: &mut PacketRouter,
         shared_key: &SharedKeys,
-    ) {
+    ) -> Result<(), GatewayClientError> {
         match ws_msg {
             Message::Binary(bin_msg) => {
                 // this function decrypts the request and checks the MAC
@@ -60,7 +62,7 @@ impl PartiallyDelegated {
                                 "message received from the gateway was malformed! - {:?}",
                                 err
                             );
-                            return;
+                            return Ok(());
                         }
                     };
 
@@ -74,18 +76,22 @@ impl PartiallyDelegated {
             // This would also require NOT discarding any text responses here.
 
             // TODO: those can return the "send confirmations" - perhaps it should be somehow worked around?
-            Message::Text(text) => trace!(
-                "received a text message - probably a response to some previous query! - {}",
-                text
-            ),
-            _ => (),
-        };
+            Message::Text(text) => {
+                trace!(
+                    "received a text message - probably a response to some previous query! - {}",
+                    text
+                );
+                Ok(())
+            }
+            _ => Ok(()),
+        }
     }
 
     pub(crate) fn split_and_listen_for_mixnet_messages(
         conn: WsConn,
         packet_router: PacketRouter,
         shared_key: Arc<SharedKeys>,
+        #[cfg(not(target_arch = "wasm32"))] shutdown: Option<ShutdownListener>,
     ) -> Self {
         // when called for, it NEEDS TO yield back the stream so that we could merge it and
         // read control request responses.
@@ -97,10 +103,36 @@ impl PartiallyDelegated {
         let mixnet_receiver_future = async move {
             let mut fused_receiver = notify_receiver.fuse();
             let mut fused_stream = (&mut stream).fuse();
+            let mut packet_router = packet_router;
+
+            // Bit of an ugly workaround for selecting on an `Option` without having access to
+            // `tokio::select`
+            #[cfg(not(target_arch = "wasm32"))]
+            let shutdown = {
+                let m_shutdown = shutdown.clone();
+                async {
+                    if let Some(mut s) = m_shutdown {
+                        s.recv().await
+                    } else {
+                        std::future::pending::<()>().await
+                    }
+                }
+                .fuse()
+            };
+            #[cfg(not(target_arch = "wasm32"))]
+            tokio::pin!(shutdown);
+
+            #[cfg(target_arch = "wasm32")]
+            let mut shutdown = std::future::pending::<()>().fuse();
 
             let ret_err = loop {
                 futures::select! {
-                    _ = fused_receiver => {
+                    _ = shutdown => {
+                        log::trace!("GatewayClient listener: Received shutdown");
+                        log::debug!("GatewayClient listener: Exiting");
+                        return;
+                    }
+                    _ = &mut fused_receiver => {
                         break Ok(());
                     }
                     msg = fused_stream.next() => {
@@ -108,7 +140,9 @@ impl PartiallyDelegated {
                             Err(err) => break Err(err),
                             Ok(msg) => msg
                         };
-                        Self::route_socket_message(ws_msg, &packet_router, shared_key.as_ref());
+                        if let Err(err) = Self::route_socket_message(ws_msg, &mut packet_router, shared_key.as_ref()) {
+                            log::warn!("Route socket message failed: {:?}", err);
+                        }
                     }
                 };
             };

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

@@ -26,13 +26,6 @@ pub struct CoconutCredential {
     pub signature: String,
 }
 
-pub struct ERC20Credential {
-    pub id: i64,
-    pub public_key: String,
-    pub private_key: String,
-    pub consumed: bool,
-}
-
 #[async_trait]
 pub trait Storage: Send + Sync {
     async fn insert_coconut_credential(
@@ -47,16 +40,6 @@ pub trait Storage: Send + Sync {
     async fn get_next_coconut_credential(&self) -> Result<CoconutCredential, StorageError>;
 
     async fn remove_coconut_credential(&self, id: i64) -> Result<(), StorageError>;
-
-    async fn insert_erc20_credential(
-        &self,
-        public_key: String,
-        private_key: String,
-    ) -> Result<(), StorageError>;
-
-    async fn get_next_erc20_credential(&self) -> Result<ERC20Credential, StorageError>;
-
-    async fn consume_erc20_credential(&self, public_key: String) -> Result<(), StorageError>;
 }
 
 #[async_trait]
@@ -79,20 +62,4 @@ impl Storage for PersistentStorage {
     async fn remove_coconut_credential(&self, _id: i64) -> Result<(), StorageError> {
         Err(StorageError::WasmNotSupported)
     }
-
-    async fn insert_erc20_credential(
-        &self,
-        _public_key: String,
-        _private_key: String,
-    ) -> Result<(), StorageError> {
-        Err(StorageError::WasmNotSupported)
-    }
-
-    async fn get_next_erc20_credential(&self) -> Result<ERC20Credential, StorageError> {
-        Err(StorageError::WasmNotSupported)
-    }
-
-    async fn consume_erc20_credential(&self, _public_key: String) -> Result<(), StorageError> {
-        Err(StorageError::WasmNotSupported)
-    }
 }

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

@@ -338,6 +338,13 @@ impl<C> NymdClient<C> {
         &self.client_address.as_ref().unwrap()[0]
     }
 
+    pub fn signer(&self) -> &DirectSecp256k1HdWallet
+    where
+        C: SigningCosmWasmClient,
+    {
+        self.client.signer()
+    }
+
     pub fn gas_price(&self) -> &GasPrice
     where
         C: SigningCosmWasmClient,

common/config/Cargo.toml

@@ -9,7 +9,6 @@ edition = "2021"
 [dependencies]
 cfg-if = "1.0.0"
 handlebars = "3.0.1"
-humantime-serde = "1.0"
 log = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 toml = "0.5.6"

common/cosmwasm-smart-contracts/mixnet-contract/Cargo.toml

@@ -18,7 +18,7 @@ fixed = { version = "1.1", features = ["serde"] }
 az = "1.1"
 log = "0.4.14"
 time = { version = "0.3.6", features = ["parsing", "formatting"] }
-ts-rs = "6.1.2"
+ts-rs = {version = "6.1.2", optional = true}
 
 contracts-common = { path = "../contracts-common" }
 
@@ -27,4 +27,4 @@ time = { version = "0.3.5", features = ["serde", "macros"] }
 
 [features]
 default = []
-generate-ts = []
+generate-ts = ['ts-rs']

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

@@ -166,6 +166,7 @@ pub mod fixed_U128_as_string {
     use super::U128;
     use serde::de::Error;
     use serde::Deserialize;
+    #[allow(unused_imports)]
     use std::str::FromStr;
 
     pub fn serialize<S>(val: &U128, serializer: S) -> Result<S::Ok, S::Error>

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

@@ -216,7 +216,6 @@ pub enum QueryMsg {
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
 pub struct MigrateMsg {
-    pub mixnet_denom: String,
     pub nodes_to_remove: Option<Vec<NodeToRemove>>,
 }
 

common/cosmwasm-smart-contracts/vesting-contract/Cargo.toml

@@ -3,16 +3,12 @@ name = "vesting-contract-common"
 version = "0.1.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
 cosmwasm-std = "1.0.0"
 mixnet-contract-common = { path = "../mixnet-contract" }
 serde = { version = "1.0", features = ["derive"] }
 schemars = "0.8"
-cw-storage-plus = "0.13.4"
-config = { path = "../../config" }
-ts-rs = "6.1.2"
+ts-rs = {version = "6.1.2", optional = true}
 
 [features]
-generate-ts = []
+generate-ts = ["ts-rs"]

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

@@ -12,9 +12,7 @@ pub struct InitMsg {
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
 #[serde(rename_all = "snake_case")]
-pub struct MigrateMsg {
-    pub mix_denom: String,
-}
+pub struct MigrateMsg {}
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema, Default)]
 pub struct VestingSpecification {
@@ -119,11 +117,6 @@ pub enum ExecuteMsg {
     UpdateLockedPledgeCap {
         amount: Uint128,
     },
-    MigrateHeightsToTimestamps {
-        account_id: u32,
-        mix_identity: String,
-        height_timestamp_map: Vec<(u64, u64)>,
-    },
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]

common/credential-storage/Cargo.toml

@@ -7,7 +7,6 @@ edition = "2021"
 
 [dependencies]
 async-trait = { version = "0.1.51" }
-nymcoconut = { path = "../nymcoconut" }
 
 log = "0.4"
 sqlx = { version = "0.5", features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"]}

common/credential-storage/src/erc20.rs

@@ -17,55 +17,4 @@ impl ERC20CredentialManager {
     pub(crate) fn new(connection_pool: sqlx::SqlitePool) -> Self {
         ERC20CredentialManager { connection_pool }
     }
-
-    /// Inserts provided signature into the database.
-    ///
-    /// # Arguments
-    ///
-    /// * `public_key`: Base58 representation of a public key.
-    /// * `private_key`: Base58 representation of a private key.
-    pub(crate) async fn insert_erc20_credential(
-        &self,
-        public_key: String,
-        private_key: String,
-    ) -> Result<(), sqlx::Error> {
-        sqlx::query!(
-            "INSERT INTO erc20_credentials(public_key, private_key, consumed) VALUES (?, ?, ?)",
-            public_key,
-            private_key,
-            false,
-        )
-        .execute(&self.connection_pool)
-        .await?;
-        Ok(())
-    }
-
-    /// Tries to retrieve one of the stored, unused credentials.
-    pub(crate) async fn get_next_erc20_credential(&self) -> Result<ERC20Credential, sqlx::Error> {
-        sqlx::query_as!(
-            ERC20Credential,
-            "SELECT * FROM erc20_credentials WHERE consumed = false"
-        )
-        .fetch_one(&self.connection_pool)
-        .await
-    }
-
-    /// Mark a credential as being consumed.
-    pub(crate) async fn consume_erc20_credential(
-        &self,
-        public_key: String,
-    ) -> Result<(), sqlx::Error> {
-        sqlx::query!(
-            r#"
-                UPDATE erc20_credentials
-                SET consumed = true
-                WHERE public_key = ?
-            "#,
-            public_key
-        )
-        .execute(&self.connection_pool)
-        .await?;
-
-        Ok(())
-    }
 }

common/credential-storage/src/lib.rs

@@ -4,18 +4,16 @@
  */
 
 use crate::coconut::CoconutCredentialManager;
-use crate::erc20::ERC20CredentialManager;
 use crate::error::StorageError;
 use crate::storage::Storage;
 
-use crate::models::{CoconutCredential, ERC20Credential};
+use crate::models::CoconutCredential;
 use async_trait::async_trait;
 use log::{debug, error};
 use sqlx::ConnectOptions;
 use std::path::{Path, PathBuf};
 
 mod coconut;
-mod erc20;
 pub mod error;
 mod models;
 pub mod storage;
@@ -24,7 +22,6 @@ pub mod storage;
 #[derive(Clone)]
 pub struct PersistentStorage {
     coconut_credential_manager: CoconutCredentialManager,
-    erc20_credential_manager: ERC20CredentialManager,
 }
 
 impl PersistentStorage {
@@ -60,7 +57,6 @@ impl PersistentStorage {
 
         Ok(PersistentStorage {
             coconut_credential_manager: CoconutCredentialManager::new(connection_pool.clone()),
-            erc20_credential_manager: ERC20CredentialManager::new(connection_pool),
         })
     }
 }
@@ -104,36 +100,6 @@ impl Storage for PersistentStorage {
 
         Ok(())
     }
-
-    async fn insert_erc20_credential(
-        &self,
-        public_key: String,
-        private_key: String,
-    ) -> Result<(), StorageError> {
-        self.erc20_credential_manager
-            .insert_erc20_credential(public_key, private_key)
-            .await?;
-
-        Ok(())
-    }
-
-    async fn get_next_erc20_credential(&self) -> Result<ERC20Credential, StorageError> {
-        let credential = self
-            .erc20_credential_manager
-            .get_next_erc20_credential()
-            .await?;
-
-        Ok(credential)
-    }
-
-    async fn consume_erc20_credential(&self, public_key: String) -> Result<(), StorageError> {
-        let credential = self
-            .erc20_credential_manager
-            .consume_erc20_credential(public_key)
-            .await?;
-
-        Ok(credential)
-    }
 }
 
 pub async fn initialise_storage(path: PathBuf) -> PersistentStorage {

common/credential-storage/src/models.rs

@@ -10,11 +10,3 @@ pub struct CoconutCredential {
     pub binding_number: String,
     pub signature: String,
 }
-
-pub struct ERC20Credential {
-    #[allow(dead_code)]
-    pub id: i64,
-    pub public_key: String,
-    pub private_key: String,
-    pub consumed: bool,
-}

common/credential-storage/src/storage.rs

@@ -3,7 +3,7 @@
 
 use async_trait::async_trait;
 
-use crate::models::{CoconutCredential, ERC20Credential};
+use crate::models::CoconutCredential;
 use crate::StorageError;
 
 #[async_trait]
@@ -31,22 +31,4 @@ pub trait Storage: Send + Sync {
     ///
     /// * `signature`: Coconut credential in the form of a signature.
     async fn remove_coconut_credential(&self, id: i64) -> Result<(), StorageError>;
-
-    /// Inserts provided signature into the database.
-    ///
-    /// # Arguments
-    ///
-    /// * `public_key`: Base58 representation of a public key.
-    /// * `private_key`: Base58 representation of a private key.
-    async fn insert_erc20_credential(
-        &self,
-        public_key: String,
-        private_key: String,
-    ) -> Result<(), StorageError>;
-
-    /// Tries to retrieve one of the stored, unused credential data.
-    async fn get_next_erc20_credential(&self) -> Result<ERC20Credential, StorageError>;
-
-    /// Mark a credential as being consumed.
-    async fn consume_erc20_credential(&self, public_key: String) -> Result<(), StorageError>;
 }

common/crypto/Cargo.toml

@@ -26,7 +26,6 @@ subtle-encoding = { version = "0.5", features =  ["bech32-preview"]}
 # internal
 nymsphinx-types = { path = "../nymsphinx/types" }
 pemstore = { path = "../../common/pemstore" }
-config = { path="../../common/config" }
 
 [dev-dependencies]
 rand_chacha = "0.2"

common/mixnode-common/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2021"
 
 [dependencies]
 bytes = "1.0"
-dashmap = "4.0"
 futures = "0.3"
 humantime-serde = "1.0"
 log = "0.4"
@@ -19,7 +18,6 @@ tokio-util = { version = "0.7.3", features = ["codec"] }
 url = "2.2"
 
 crypto =  { path = "../crypto" }
-nonexhaustive-delayqueue = { path = "../nonexhaustive-delayqueue" }
 nymsphinx-acknowledgements = { path = "../nymsphinx/acknowledgements" }
 nymsphinx-addressing = { path = "../nymsphinx/addressing" }
 nymsphinx-forwarding = { path = "../nymsphinx/forwarding" }

common/network-defaults/src/lib.rs

@@ -7,7 +7,6 @@ use std::{env::var, path::PathBuf};
 use url::Url;
 
 pub mod all;
-pub mod eth_contract;
 pub mod mainnet;
 pub mod qa;
 pub mod sandbox;

common/network-defaults/src/mainnet.rs

@@ -25,7 +25,7 @@ pub(crate) const _ETH_ERC20_CONTRACT_ADDRESS: [u8; 20] =
 pub(crate) const REWARDING_VALIDATOR_ADDRESS: &str = "n10yyd98e2tuwu0f7ypz9dy3hhjw7v772q6287gy";
 
 pub(crate) const STATISTICS_SERVICE_DOMAIN_ADDRESS: &str = "https://mainnet-stats.nymte.ch:8090/";
-pub const NYMD_VALIDATOR: &str = "https://rpc.nyx.nodes.guru/";
+pub const NYMD_VALIDATOR: &str = "https://rpc.nymtech.net";
 pub const API_VALIDATOR: &str = "https://validator.nymtech.net/api/";
 pub(crate) fn validators() -> Vec<ValidatorDetails> {
     vec![ValidatorDetails::new(NYMD_VALIDATOR, Some(API_VALIDATOR))]

common/socks5/ordered-buffer/README.md

@@ -1,5 +1,7 @@
 Ordered Buffer
 ==============
 
+**CURRENTLY UNUSED**
+
 This crate takes care of reliably and speedily turning bytes into a series of ordered message fragments on one side, and of reliably reassembling the fragments into the original message on the other. 
 

common/socks5/proxy-helpers/Cargo.toml

@@ -16,6 +16,7 @@ futures = "0.3"
 log = "0.4"
 socks5-requests = { path = "../requests" }
 ordered-buffer = { path = "../ordered-buffer" }
+task = { path = "../../task" }
 
 [dev-dependencies]
 tokio-test = "0.4.2"

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

@@ -7,6 +7,7 @@ use log::*;
 use ordered_buffer::{OrderedMessage, OrderedMessageBuffer};
 use socks5_requests::ConnectionId;
 use std::collections::{HashMap, HashSet};
+use task::ShutdownListener;
 
 /// A generic message produced after reading from a socket/connection. It includes data that was
 /// actually read alongside boolean indicating whether the connection got closed so that
@@ -74,10 +75,12 @@ pub struct Controller {
     // buffer for messages received before connection was established due to mixnet being able to
     // un-order messages. Note we don't ever expect to have more than 1-2 messages per connection here
     pending_messages: HashMap<ConnectionId, Vec<(Vec<u8>, bool)>>,
+
+    shutdown: ShutdownListener,
 }
 
 impl Controller {
-    pub fn new() -> (Self, ControllerSender) {
+    pub fn new(shutdown: ShutdownListener) -> (Self, ControllerSender) {
         let (sender, receiver) = mpsc::unbounded();
         (
             Controller {
@@ -85,6 +88,7 @@ impl Controller {
                 receiver,
                 recently_closed: HashSet::new(),
                 pending_messages: HashMap::new(),
+                shutdown,
             },
             sender,
         )
@@ -170,16 +174,24 @@ impl Controller {
     }
 
     pub async fn run(&mut self) {
-        while let Some(command) = self.receiver.next().await {
-            match command {
-                ControllerCommand::Send(conn_id, data, is_closed) => {
-                    self.send_to_connection(conn_id, data, is_closed)
+        loop {
+            tokio::select! {
+                command = self.receiver.next() => match command {
+                    Some(ControllerCommand::Send(conn_id, data, is_closed)) => {
+                        self.send_to_connection(conn_id, data, is_closed)
+                    }
+                    Some(ControllerCommand::Insert(conn_id, sender)) => {
+                        self.insert_connection(conn_id, sender)
+                    }
+                    Some(ControllerCommand::Remove(conn_id)) => self.remove_connection(conn_id),
+                    None => {
+                        log::trace!("SOCKS5 Controller: Stopping since channel closed");
+                        break;
+                    }
                 }
-                ControllerCommand::Insert(conn_id, sender) => {
-                    self.insert_connection(conn_id, sender)
-                }
-                ControllerCommand::Remove(conn_id) => self.remove_connection(conn_id),
             }
         }
+        assert!(self.shutdown.is_shutdown_poll());
+        log::debug!("SOCKS5 Controller: Exiting");
     }
 }

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

@@ -11,6 +11,7 @@ use log::*;
 use ordered_buffer::OrderedMessageSender;
 use socks5_requests::ConnectionId;
 use std::{io, sync::Arc};
+use task::ShutdownListener;
 use tokio::select;
 use tokio::{net::tcp::OwnedReadHalf, sync::Notify, time::sleep};
 
@@ -74,6 +75,7 @@ where
     is_finished
 }
 
+#[allow(clippy::too_many_arguments)]
 pub(super) async fn run_inbound<F, S>(
     mut reader: OwnedReadHalf,
     local_destination_address: String, // addresses are provided for better logging
@@ -82,6 +84,7 @@ pub(super) async fn run_inbound<F, S>(
     mix_sender: MixProxySender<S>,
     adapter_fn: F,
     shutdown_notify: Arc<Notify>,
+    mut shutdown_listener: ShutdownListener,
 ) -> OwnedReadHalf
 where
     F: Fn(ConnectionId, Vec<u8>, bool) -> S + Send + 'static,
@@ -106,6 +109,10 @@ where
                 send_empty_close(connection_id, &mut message_sender, &mix_sender, &adapter_fn);
                 break;
             }
+            _ = shutdown_listener.recv() => {
+                log::trace!("ProxyRunner inbound: Received shutdown");
+                break;
+            }
         }
     }
     trace!("{} - inbound closed", connection_id);

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

@@ -5,6 +5,7 @@ use crate::connection_controller::ConnectionReceiver;
 use futures::channel::mpsc;
 use socks5_requests::ConnectionId;
 use std::{sync::Arc, time::Duration};
+use task::ShutdownListener;
 use tokio::{net::TcpStream, sync::Notify};
 
 mod inbound;
@@ -44,6 +45,9 @@ pub struct ProxyRunner<S> {
     local_destination_address: String,
     remote_source_address: String,
     connection_id: ConnectionId,
+
+    // Listens to shutdown commands from higher up
+    shutdown_listener: ShutdownListener,
 }
 
 impl<S> ProxyRunner<S>
@@ -57,6 +61,7 @@ where
         mix_receiver: ConnectionReceiver,
         mix_sender: MixProxySender<S>,
         connection_id: ConnectionId,
+        shutdown_listener: ShutdownListener,
     ) -> Self {
         ProxyRunner {
             mix_receiver: Some(mix_receiver),
@@ -65,6 +70,7 @@ where
             local_destination_address,
             remote_source_address,
             connection_id,
+            shutdown_listener,
         }
     }
 
@@ -86,6 +92,7 @@ where
             self.mix_sender.clone(),
             adapter_fn,
             Arc::clone(&shutdown_notify),
+            self.shutdown_listener.clone(),
         );
 
         let outbound_future = outbound::run_outbound(
@@ -95,6 +102,7 @@ where
             self.mix_receiver.take().unwrap(),
             self.connection_id,
             shutdown_notify,
+            self.shutdown_listener.clone(),
         );
 
         // TODO: this shouldn't really have to spawn tasks inside "library" code, but

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

@@ -8,6 +8,7 @@ use futures::StreamExt;
 use log::*;
 use socks5_requests::ConnectionId;
 use std::{sync::Arc, time::Duration};
+use task::ShutdownListener;
 use tokio::io::AsyncWriteExt;
 use tokio::select;
 use tokio::{net::tcp::OwnedWriteHalf, sync::Notify, time::sleep, time::Instant};
@@ -50,6 +51,7 @@ pub(super) async fn run_outbound(
     mut mix_receiver: ConnectionReceiver,
     connection_id: ConnectionId,
     shutdown_notify: Arc<Notify>,
+    mut shutdown_listener: ShutdownListener,
 ) -> (OwnedWriteHalf, ConnectionReceiver) {
     let shutdown_future = shutdown_notify.notified().then(|_| sleep(SHUTDOWN_TIMEOUT));
     tokio::pin!(shutdown_future);
@@ -78,6 +80,10 @@ pub(super) async fn run_outbound(
                 debug!("closing outbound proxy after inbound was closed {:?} ago", SHUTDOWN_TIMEOUT);
                 break;
             }
+            _ = shutdown_listener.recv() => {
+                log::trace!("ProxyRunner outbound: Received shutdown");
+                break;
+            }
         }
     }
 

common/socks5/requests/Cargo.toml

@@ -8,5 +8,4 @@ edition = "2021"
 
 [dependencies]
 nymsphinx-addressing = { path = "../../../common/nymsphinx/addressing" }
-ordered-buffer = {path = "../ordered-buffer"}
 thiserror = "1"

common/task/Cargo.toml

@@ -8,7 +8,6 @@ edition = "2021"
 [dependencies]
 log = "0.4"
 tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal"] }
-tokio-util = { version = "0.7.3", features = ["codec"] }
 
 [dev-dependencies]
 tokio = { version = "1.19.1", features = ["rt-multi-thread", "net", "signal", "test-util", "macros"] }

common/task/src/lib.rs

@@ -2,5 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub mod shutdown;
+pub mod signal;
 
 pub use shutdown::{ShutdownListener, ShutdownNotifier};
+pub use signal::wait_for_signal;

common/task/src/shutdown.rs

@@ -74,7 +74,7 @@ pub struct ShutdownListener {
 }
 
 impl ShutdownListener {
-    pub fn new(notify: watch::Receiver<()>) -> ShutdownListener {
+    fn new(notify: watch::Receiver<()>) -> ShutdownListener {
         ShutdownListener {
             shutdown: false,
             notify,
@@ -92,6 +92,25 @@ impl ShutdownListener {
         let _ = self.notify.changed().await;
         self.shutdown = true;
     }
+
+    pub fn is_shutdown_poll(&mut self) -> bool {
+        if self.shutdown {
+            return true;
+        }
+        match self.notify.has_changed() {
+            Ok(has_changed) => {
+                if has_changed {
+                    self.shutdown = true;
+                }
+                has_changed
+            }
+            Err(err) => {
+                log::debug!("Polling shutdown failed: {err}");
+                log::debug!("Assuming this means we should shutdown...");
+                true
+            }
+        }
+    }
 }
 
 #[cfg(test)]

common/task/src/signal.rs

@@ -0,0 +1,27 @@
+#[cfg(unix)]
+pub async fn wait_for_signal() {
+    use tokio::signal::unix::{signal, SignalKind};
+    let mut sigterm = signal(SignalKind::terminate()).expect("Failed to setup SIGTERM channel");
+    let mut sigquit = signal(SignalKind::quit()).expect("Failed to setup SIGQUIT channel");
+
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            log::info!("Received SIGINT");
+        },
+        _ = sigterm.recv() => {
+            log::info!("Received SIGTERM");
+        }
+        _ = sigquit.recv() => {
+            log::info!("Received SIGQUIT");
+        }
+    }
+}
+
+#[cfg(not(unix))]
+pub async fn wait_for_signal() {
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            log::info!("Received SIGINT");
+        },
+    }
+}

contracts/CHANGELOG.md

@@ -1,4 +1,4 @@
-## Unreleased
+## [nym-contracts-v1.0.2](https://github.com/nymtech/nym/tree/nym-contracts-v1.0.2) (2022-09-13)
 
 ### Added
 
@@ -12,10 +12,12 @@
 
 - vesting-contract: the contract now correctly stores delegations with their timestamp as opposed to using block height ([#1544])
 - mixnet-contract: compounding delegator rewards is now possible even if the associated mixnode had already unbonded ([#1571])
+- mixnet-contract: fixed reward accumulation after claiming rewards ([#1613])
 
 [#1544]: https://github.com/nymtech/nym/pull/1544
 [#1569]: https://github.com/nymtech/nym/pull/1569
 [#1569]: https://github.com/nymtech/nym/pull/1571
+[#1613]: https://github.com/nymtech/nym/pull/1613
 
 ## [nym-contracts-v1.0.1](https://github.com/nymtech/nym/tree/nym-contracts-v1.0.1) (2022-06-22)
 

contracts/Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["bandwidth-claim", "coconut-bandwidth", "mixnet", "vesting", "multisig/cw3-flex-multisig", "multisig/cw4-group", "coconut-test"]
+members = ["coconut-bandwidth", "mixnet", "vesting", "multisig/cw3-flex-multisig", "multisig/cw4-group", "coconut-test"]
 
 [profile.release]
 opt-level = 3

contracts/bandwidth-claim/Cargo.toml

@@ -1,20 +0,0 @@
-[package]
-name = "bandwidth-claim"
-version = "1.0.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[lib]
-crate-type = ["cdylib", "rlib"]
-
-[dev-dependencies]
-[dependencies]
-bandwidth-claim-contract = { path = "../../common/bandwidth-claim-contract" }
-
-cosmwasm-std = "1.0.0"
-cosmwasm-storage = "1.0.0"
-
-schemars = "0.8"
-serde = { version = "1.0.103", default-features = false, features = ["derive"] }
-thiserror = "1.0.23"

contracts/bandwidth-claim/src/error.rs

@@ -1,27 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use cosmwasm_std::{StdError, VerificationError};
-use thiserror::Error;
-
-/// Custom errors for contract failure conditions.
-///
-/// Add any other custom errors you like here.
-/// Look at https://docs.rs/thiserror/1.0.21/thiserror/ for details.
-#[derive(Error, Debug, PartialEq)]
-pub enum ContractError {
-    #[error("{0}")]
-    Std(#[from] StdError),
-
-    #[error("Invalid size for signature items")]
-    InvalidSignatureSize,
-
-    #[error("This payment has already been claimed by someone")]
-    PaymentAlreadyClaimed,
-
-    #[error("Error while verifying ed25519 signature - {0}")]
-    VerificationError(#[from] VerificationError),
-
-    #[error("The payment is not properly signed")]
-    BadSignature,
-}

contracts/bandwidth-claim/src/lib.rs

@@ -1,103 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-mod error;
-mod queries;
-mod storage;
-mod support;
-mod transactions;
-
-use cosmwasm_std::{
-    entry_point, to_binary, Deps, DepsMut, Env, MessageInfo, QueryResponse, Response,
-};
-
-use crate::error::ContractError;
-use bandwidth_claim_contract::msg::{ExecuteMsg, InstantiateMsg, MigrateMsg, QueryMsg};
-
-/// Instantiate the contract.
-///
-/// `deps` contains Storage, API and Querier
-/// `env` contains block, message and contract info
-/// `msg` is the contract initialization message, sort of like a constructor call.
-#[entry_point]
-pub fn instantiate(
-    _deps: DepsMut<'_>,
-    _env: Env,
-    _info: MessageInfo,
-    _msg: InstantiateMsg,
-) -> Result<Response, ContractError> {
-    Ok(Response::default())
-}
-
-/// Handle an incoming message
-#[entry_point]
-pub fn execute(
-    deps: DepsMut<'_>,
-    env: Env,
-    info: MessageInfo,
-    msg: ExecuteMsg,
-) -> Result<Response, ContractError> {
-    match msg {
-        ExecuteMsg::LinkPayment { data } => transactions::link_payment(deps, env, info, data),
-    }
-}
-
-#[entry_point]
-pub fn query(deps: Deps<'_>, _env: Env, msg: QueryMsg) -> Result<QueryResponse, ContractError> {
-    let query_res = match msg {
-        QueryMsg::GetPayments { start_after, limit } => {
-            to_binary(&queries::query_payments_paged(deps, start_after, limit)?)
-        }
-    };
-
-    Ok(query_res?)
-}
-
-#[entry_point]
-pub fn migrate(_deps: DepsMut<'_>, _env: Env, _msg: MigrateMsg) -> Result<Response, ContractError> {
-    Ok(Default::default())
-}
-
-#[cfg(test)]
-pub mod tests {
-    use super::*;
-    use bandwidth_claim_contract::payment::PagedPaymentResponse;
-    use cosmwasm_std::testing::{mock_dependencies, mock_env, mock_info};
-    use cosmwasm_std::{coins, from_binary};
-
-    const TEST_MIX_DENOM: &str = "unym";
-
-    #[test]
-    fn initialize_contract() {
-        let mut deps = mock_dependencies();
-        let env = mock_env();
-        let msg = InstantiateMsg {};
-        let info = mock_info("creator", &[]);
-
-        let res = instantiate(deps.as_mut(), env.clone(), info, msg).unwrap();
-        assert_eq!(0, res.messages.len());
-
-        // payments should be empty after initialization
-        let res = query(
-            deps.as_ref(),
-            env.clone(),
-            QueryMsg::GetPayments {
-                start_after: None,
-                limit: Option::from(2),
-            },
-        )
-        .unwrap();
-        let page: PagedPaymentResponse = from_binary(&res).unwrap();
-        assert_eq!(0, page.payments.len()); // there are no payments in the list when it's just been initialized
-
-        // Contract balance should match what we initialized it as
-        assert_eq!(
-            coins(0, TEST_MIX_DENOM),
-            vec![deps
-                .as_ref()
-                .querier
-                .query_balance(env.contract.address, TEST_MIX_DENOM)
-                .unwrap()]
-        );
-    }
-}

contracts/bandwidth-claim/src/queries.rs

@@ -1,191 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use cosmwasm_std::{Deps, Order, StdResult};
-
-use crate::storage::payments_read;
-use bandwidth_claim_contract::keys::PublicKey;
-use bandwidth_claim_contract::payment::{PagedPaymentResponse, Payment};
-
-const PAYMENT_PAGE_MAX_LIMIT: u32 = 100;
-const PAYMENT_PAGE_DEFAULT_LIMIT: u32 = 50;
-
-/// Adds a 0 byte to terminate the `start_after` value given. This allows CosmWasm
-/// to get the succeeding key as the start of the next page.
-fn calculate_start_value<B: AsRef<[u8]>>(start_after: Option<B>) -> Option<Vec<u8>> {
-    start_after.as_ref().map(|identity| {
-        identity
-            .as_ref()
-            .iter()
-            .cloned()
-            .chain(std::iter::once(0))
-            .collect()
-    })
-}
-
-pub fn query_payments_paged(
-    deps: Deps<'_>,
-    start_after: Option<PublicKey>,
-    limit: Option<u32>,
-) -> StdResult<PagedPaymentResponse> {
-    let limit = limit
-        .unwrap_or(PAYMENT_PAGE_DEFAULT_LIMIT)
-        .min(PAYMENT_PAGE_MAX_LIMIT) as usize;
-    let start = calculate_start_value(start_after);
-
-    let payments = payments_read(deps.storage)
-        .range(start.as_deref(), None, Order::Ascending)
-        .take(limit)
-        .map(|res| res.map(|item| item.1))
-        .collect::<StdResult<Vec<Payment>>>()?;
-
-    let start_next_after = payments.last().map(|payment| payment.verification_key());
-
-    Ok(PagedPaymentResponse::new(payments, limit, start_next_after))
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::storage::payments;
-    use crate::support::tests::helpers;
-    use std::convert::TryInto;
-
-    #[test]
-    fn payments_empty_on_init() {
-        let deps = helpers::init_contract();
-        let response = query_payments_paged(deps.as_ref(), None, Option::from(2)).unwrap();
-        assert_eq!(0, response.payments.len());
-    }
-
-    #[test]
-    fn payments_paged_retrieval_obeys_limits() {
-        let mut deps = helpers::init_contract();
-        let storage = deps.as_mut().storage;
-        let limit = 2;
-        for n in 0u32..10000 {
-            let bytes: Vec<u8> = std::iter::repeat(n.to_be_bytes())
-                .take(8)
-                .flatten()
-                .collect();
-            let verification_key = PublicKey::new(bytes.try_into().unwrap());
-            let payment = helpers::payment_fixture();
-            payments(storage)
-                .save(&verification_key.to_bytes(), &payment)
-                .unwrap();
-        }
-
-        let page1 = query_payments_paged(deps.as_ref(), None, Option::from(limit)).unwrap();
-        assert_eq!(limit, page1.payments.len() as u32);
-    }
-
-    #[test]
-    fn payments_paged_retrieval_has_default_limit() {
-        let mut deps = helpers::init_contract();
-        let storage = deps.as_mut().storage;
-        for n in 0u32..100 {
-            let bytes: Vec<u8> = std::iter::repeat(n.to_be_bytes())
-                .take(8)
-                .flatten()
-                .collect();
-            let verification_key = PublicKey::new(bytes.try_into().unwrap());
-            let payment = helpers::payment_fixture();
-            payments(storage)
-                .save(&verification_key.to_bytes(), &payment)
-                .unwrap();
-        }
-
-        // query without explicitly setting a limit
-        let page1 = query_payments_paged(deps.as_ref(), None, None).unwrap();
-
-        assert_eq!(PAYMENT_PAGE_DEFAULT_LIMIT, page1.payments.len() as u32);
-    }
-
-    #[test]
-    fn payments_paged_retrieval_has_max_limit() {
-        let mut deps = helpers::init_contract();
-        let storage = deps.as_mut().storage;
-        for n in 0u32..10000 {
-            let bytes: Vec<u8> = std::iter::repeat(n.to_be_bytes())
-                .take(8)
-                .flatten()
-                .collect();
-            let verification_key = PublicKey::new(bytes.try_into().unwrap());
-            let payment = helpers::payment_fixture();
-            payments(storage)
-                .save(&verification_key.to_bytes(), &payment)
-                .unwrap();
-        }
-
-        // query with a crazily high limit in an attempt to use too many resources
-        let crazy_limit = 1000;
-        let page1 = query_payments_paged(deps.as_ref(), None, Option::from(crazy_limit)).unwrap();
-
-        // we default to a decent sized upper bound instead
-        assert_eq!(PAYMENT_PAGE_MAX_LIMIT, page1.payments.len() as u32);
-    }
-
-    #[test]
-    fn payments_pagination_works() {
-        let key1 = PublicKey::new([1; 32]);
-        let key2 = PublicKey::new([2; 32]);
-        let key3 = PublicKey::new([3; 32]);
-        let key4 = PublicKey::new([4; 32]);
-
-        let mut deps = helpers::init_contract();
-        let payment = helpers::payment_fixture();
-        payments(&mut deps.storage)
-            .save(&key1.to_bytes(), &payment)
-            .unwrap();
-
-        let per_page = 2;
-        let page1 = query_payments_paged(deps.as_ref(), None, Option::from(per_page)).unwrap();
-
-        // page should have 1 result on it
-        assert_eq!(1, page1.payments.len());
-
-        // save another
-        payments(&mut deps.storage)
-            .save(&key2.to_bytes(), &payment)
-            .unwrap();
-
-        // page1 should have 2 results on it
-        let page1 = query_payments_paged(deps.as_ref(), None, Option::from(per_page)).unwrap();
-        assert_eq!(2, page1.payments.len());
-
-        payments(&mut deps.storage)
-            .save(&key3.to_bytes(), &payment)
-            .unwrap();
-
-        // page1 still has 2 results
-        let page1 = query_payments_paged(deps.as_ref(), None, Option::from(per_page)).unwrap();
-        assert_eq!(2, page1.payments.len());
-
-        // retrieving the next page should start after the last key on this page
-        let start_after = key2;
-        let page2 = query_payments_paged(
-            deps.as_ref(),
-            Option::from(start_after),
-            Option::from(per_page),
-        )
-        .unwrap();
-
-        assert_eq!(1, page2.payments.len());
-
-        // save another one
-        payments(&mut deps.storage)
-            .save(&key4.to_bytes(), &payment)
-            .unwrap();
-
-        let start_after = key2;
-        let page2 = query_payments_paged(
-            deps.as_ref(),
-            Option::from(start_after),
-            Option::from(per_page),
-        )
-        .unwrap();
-
-        // now we have 2 pages, with 2 results on the second page
-        assert_eq!(2, page2.payments.len());
-    }
-}

contracts/bandwidth-claim/src/storage.rs

@@ -1,79 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use cosmwasm_std::Storage;
-use cosmwasm_storage::{bucket, bucket_read, Bucket, ReadonlyBucket};
-use schemars::JsonSchema;
-use serde::{Deserialize, Serialize};
-
-use bandwidth_claim_contract::payment::Payment;
-
-// buckets
-const PREFIX_PAYMENTS: &[u8] = b"payments";
-const PREFIX_STATUS: &[u8] = b"status";
-
-#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize, JsonSchema)]
-pub enum Status {
-    Unchecked,
-    Checked,
-    Spent,
-}
-
-pub fn payments(storage: &mut dyn Storage) -> Bucket<'_, Payment> {
-    bucket(storage, PREFIX_PAYMENTS)
-}
-
-pub fn payments_read(storage: &dyn Storage) -> ReadonlyBucket<'_, Payment> {
-    bucket_read(storage, PREFIX_PAYMENTS)
-}
-
-pub fn status(storage: &mut dyn Storage) -> Bucket<'_, Status> {
-    bucket(storage, PREFIX_STATUS)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::support::tests::helpers;
-    use bandwidth_claim_contract::keys::PublicKey;
-    use cosmwasm_std::testing::MockStorage;
-
-    #[test]
-    fn payments_single_read_retrieval() {
-        let mut storage = MockStorage::new();
-        let key1 = PublicKey::new([1; 32]);
-        let key2 = PublicKey::new([2; 32]);
-        let payment1 = helpers::payment_fixture();
-        let payment2 = helpers::payment_fixture();
-        payments(&mut storage)
-            .save(key1.as_ref(), &payment1)
-            .unwrap();
-        payments(&mut storage)
-            .save(key2.as_ref(), &payment2)
-            .unwrap();
-
-        let res1 = payments_read(&storage).load(key1.as_ref()).unwrap();
-        let res2 = payments_read(&storage).load(key2.as_ref()).unwrap();
-        assert_eq!(payment1, res1);
-        assert_eq!(payment2, res2);
-    }
-
-    #[test]
-    fn status_single_read_retrieval() {
-        let mut storage = MockStorage::new();
-        let key1 = PublicKey::new([1; 32]);
-        let key2 = PublicKey::new([2; 32]);
-        let status_value = Status::Unchecked;
-        status(&mut storage)
-            .save(key1.as_ref(), &status_value)
-            .unwrap();
-        status(&mut storage)
-            .save(key2.as_ref(), &status_value)
-            .unwrap();
-
-        let res1 = status(&mut storage).load(key1.as_ref()).unwrap();
-        assert_eq!(status_value, res1);
-        let res2 = status(&mut storage).load(key2.as_ref()).unwrap();
-        assert_eq!(status_value, res2);
-    }
-}

contracts/bandwidth-claim/src/support/mod.rs

@@ -1,3 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-pub mod tests;

contracts/bandwidth-claim/src/support/tests.rs

@@ -1,28 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-#[cfg(test)]
-pub mod helpers {
-    use crate::instantiate;
-    use bandwidth_claim_contract::keys::PublicKey;
-    use bandwidth_claim_contract::msg::InstantiateMsg;
-    use bandwidth_claim_contract::payment::Payment;
-    use cosmwasm_std::testing::{mock_dependencies, mock_env, mock_info, MockApi, MockQuerier};
-    use cosmwasm_std::{Empty, MemoryStorage, OwnedDeps};
-
-    pub fn init_contract() -> OwnedDeps<MemoryStorage, MockApi, MockQuerier<Empty>> {
-        let mut deps = mock_dependencies();
-        let msg = InstantiateMsg {};
-        let env = mock_env();
-        let info = mock_info("creator", &[]);
-        instantiate(deps.as_mut(), env.clone(), info, msg).unwrap();
-        return deps;
-    }
-
-    pub fn payment_fixture() -> Payment {
-        let public_key = PublicKey::new([1; 32]);
-        let gateway_identity = PublicKey::new([2; 32]);
-        let bandwidth = 42;
-        Payment::new(public_key, gateway_identity, bandwidth)
-    }
-}

contracts/bandwidth-claim/src/transactions.rs

@@ -1,146 +0,0 @@
-// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: Apache-2.0
-
-use cosmwasm_std::{DepsMut, Env, MessageInfo, Response};
-
-use crate::error::ContractError;
-use crate::storage::{payments, status, Status};
-use bandwidth_claim_contract::payment::{LinkPaymentData, Payment};
-
-pub(crate) fn link_payment(
-    deps: DepsMut<'_>,
-    _env: Env,
-    _info: MessageInfo,
-    data: LinkPaymentData,
-) -> Result<Response, ContractError> {
-    let mut status_bucket = status(deps.storage);
-
-    let verification_key = data.verification_key.to_bytes();
-    let gateway_identity = data.gateway_identity.to_bytes();
-    let message: Vec<u8> = verification_key
-        .iter()
-        .chain(gateway_identity.iter())
-        .copied()
-        .collect();
-    let signature = data.signature.to_bytes();
-
-    if let Ok(Some(_)) = status_bucket.may_load(&verification_key) {
-        return Err(ContractError::PaymentAlreadyClaimed);
-    }
-
-    if !deps
-        .api
-        .ed25519_verify(&message, &signature, &verification_key)?
-    {
-        return Err(ContractError::BadSignature);
-    }
-
-    status_bucket.save(&verification_key, &Status::Unchecked)?;
-    payments(deps.storage).save(
-        &verification_key,
-        &Payment::new(data.verification_key, data.gateway_identity, data.bandwidth),
-    )?;
-
-    Ok(Response::default())
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::storage::payments_read;
-    use crate::support::tests::helpers;
-    use bandwidth_claim_contract::keys::PublicKey;
-    use cosmwasm_std::testing::{mock_env, mock_info};
-
-    #[test]
-    fn bad_signature_payment() {
-        let mut deps = helpers::init_contract();
-        let env = mock_env();
-        let info = mock_info("owner", &[]);
-
-        let payment_data = LinkPaymentData::new([1; 32], [2; 32], 42, [3; 64]);
-
-        assert_eq!(
-            link_payment(deps.as_mut(), env, info, payment_data),
-            Err(ContractError::BadSignature)
-        );
-    }
-
-    #[test]
-    fn good_payment() {
-        let mut deps = helpers::init_contract();
-        let env = mock_env();
-        let info = mock_info("owner", &[]);
-
-        let verification_key = [
-            78, 142, 213, 13, 39, 169, 76, 205, 242, 206, 129, 208, 190, 51, 139, 206, 245, 199,
-            120, 151, 181, 250, 192, 153, 123, 104, 129, 139, 60, 254, 243, 98,
-        ];
-        let gateway_identity = [
-            106, 76, 76, 238, 214, 177, 233, 112, 56, 33, 21, 201, 89, 42, 69, 196, 175, 56, 6,
-            110, 184, 167, 203, 63, 1, 167, 134, 102, 165, 215, 3, 212,
-        ];
-        let bandwidth = 42;
-        let signature = [
-            200, 134, 156, 198, 113, 180, 129, 90, 70, 28, 176, 201, 35, 208, 145, 28, 15, 16, 9,
-            110, 148, 188, 193, 75, 157, 201, 206, 211, 128, 215, 66, 207, 175, 155, 48, 24, 171,
-            254, 9, 37, 108, 205, 143, 37, 77, 189, 162, 52, 44, 130, 173, 60, 220, 22, 193, 3,
-            111, 90, 123, 147, 206, 8, 137, 1,
-        ];
-
-        let payment_data =
-            LinkPaymentData::new(verification_key, gateway_identity, bandwidth, signature);
-
-        assert!(link_payment(deps.as_mut(), env, info, payment_data).is_ok());
-
-        assert_eq!(
-            payments_read(&deps.storage)
-                .load(&verification_key)
-                .unwrap(),
-            Payment::new(
-                PublicKey::new(verification_key),
-                PublicKey::new(gateway_identity),
-                bandwidth
-            )
-        );
-        assert_eq!(
-            status(&mut deps.storage).load(&verification_key).unwrap(),
-            Status::Unchecked
-        )
-    }
-
-    #[test]
-    fn double_spend_protection() {
-        let mut deps = helpers::init_contract();
-        let env = mock_env();
-        let info = mock_info("owner", &[]);
-
-        let verification_key = [
-            78, 142, 213, 13, 39, 169, 76, 205, 242, 206, 129, 208, 190, 51, 139, 206, 245, 199,
-            120, 151, 181, 250, 192, 153, 123, 104, 129, 139, 60, 254, 243, 98,
-        ];
-        let gateway_identity = [
-            106, 76, 76, 238, 214, 177, 233, 112, 56, 33, 21, 201, 89, 42, 69, 196, 175, 56, 6,
-            110, 184, 167, 203, 63, 1, 167, 134, 102, 165, 215, 3, 212,
-        ];
-        let bandwidth = 42;
-        let signature = [
-            200, 134, 156, 198, 113, 180, 129, 90, 70, 28, 176, 201, 35, 208, 145, 28, 15, 16, 9,
-            110, 148, 188, 193, 75, 157, 201, 206, 211, 128, 215, 66, 207, 175, 155, 48, 24, 171,
-            254, 9, 37, 108, 205, 143, 37, 77, 189, 162, 52, 44, 130, 173, 60, 220, 22, 193, 3,
-            111, 90, 123, 147, 206, 8, 137, 1,
-        ];
-
-        let payment_data =
-            LinkPaymentData::new(verification_key, gateway_identity, bandwidth, signature);
-
-        link_payment(deps.as_mut(), env.clone(), info.clone(), payment_data).unwrap();
-
-        // Only the verification key is used for double spending protection, the other data is irrelevant
-        let second_payment_data = LinkPaymentData::new(verification_key, [1; 32], 10, [2; 64]);
-        assert_eq!(
-            link_payment(deps.as_mut(), env, info, second_payment_data),
-            Err(ContractError::PaymentAlreadyClaimed)
-        )
-    }
-}

contracts/basic-bandwidth-generation/.gitignore

@@ -1,7 +0,0 @@
-node_modules
-../.env
-.to_do.md
-
-#Hardhat files
-cache
-artifacts

contracts/basic-bandwidth-generation/.solcover.js

@@ -1,6 +0,0 @@
-module.exports = {
-  skipFiles: [
-    'CosmosToken.sol', 
-    'Gravity.sol'
-  ]
-};

contracts/basic-bandwidth-generation/README.md

@@ -1,17 +0,0 @@
-# Basic Bandwidth Credential Generator 
-
-This directory contains the contract and unit tests for the `BandwidthGenerator` smart contract. 
-
-This contract allows users to generate Basic Bandwidth Credentials (BBCs) on the Nym cosmos blockchain using ERC20 representations of NYM as payment, utilising the Cosmos Gravity Bridge for cross-chain payment. 
-         
-BBCs are credentials that will be presented to Gateways by a Nym Client, and represent a certain amount of bandwidth which can be sent through the Nym Mixnet. 
-
-By default 1 NYM = 1 GB of bandwidth. The ratio of NYM - bandwidth is denominated in bytes, and represented in the smart contract by the `BytesPerToken` variable. This variable can be adjusted by the contract owner. 
-  
-The amount of bandwidth bought is calculated according to the following formula: 
-`(Token amount in 'wei' * BytesPerToken) / 10**18`
-
-## Usage 
-* `npm install`
-* `npx hardhat compile`
-* `npx hardhat test` 

contracts/basic-bandwidth-generation/artifacts/@openzeppelin/contracts/token/ERC20/ERC20.sol/ERC20.dbg.json

@@ -1,4 +0,0 @@
-{
-  "_format": "hh-sol-dbg-1",
-  "buildInfo": "../../../../../build-info/1f95016aa87f8376998977fedfcae017.json"
-}

contracts/basic-bandwidth-generation/artifacts/@openzeppelin/contracts/token/ERC20/IERC20.sol/IERC20.dbg.json

@@ -1,4 +0,0 @@
-{
-  "_format": "hh-sol-dbg-1",
-  "buildInfo": "../../../../../build-info/1f95016aa87f8376998977fedfcae017.json"
-}

contracts/basic-bandwidth-generation/artifacts/@openzeppelin/contracts/token/ERC20/IERC20.sol/IERC20.json

@@ -1,194 +0,0 @@
-{
-  "_format": "hh-sol-artifact-1",
-  "contractName": "IERC20",
-  "sourceName": "@openzeppelin/contracts/token/ERC20/IERC20.sol",
-  "abi": [
-    {
-      "anonymous": false,
-      "inputs": [
-        {
-          "indexed": true,
-          "internalType": "address",
-          "name": "owner",
-          "type": "address"
-        },
-        {
-          "indexed": true,
-          "internalType": "address",
-          "name": "spender",
-          "type": "address"
-        },
-        {
-          "indexed": false,
-          "internalType": "uint256",
-          "name": "value",
-          "type": "uint256"
-        }
-      ],
-      "name": "Approval",
-      "type": "event"
-    },
-    {
-      "anonymous": false,
-      "inputs": [
-        {
-          "indexed": true,
-          "internalType": "address",
-          "name": "from",
-          "type": "address"
-        },
-        {
-          "indexed": true,
-          "internalType": "address",
-          "name": "to",
-          "type": "address"
-        },
-        {
-          "indexed": false,
-          "internalType": "uint256",
-          "name": "value",
-          "type": "uint256"
-        }
-      ],
-      "name": "Transfer",
-      "type": "event"
-    },
-    {
-      "inputs": [
-        {
-          "internalType": "address",
-          "name": "owner",
-          "type": "address"
-        },
-        {
-          "internalType": "address",
-          "name": "spender",
-          "type": "address"
-        }
-      ],
-      "name": "allowance",
-      "outputs": [
-        {
-          "internalType": "uint256",
-          "name": "",
-          "type": "uint256"
-        }
-      ],
-      "stateMutability": "view",
-      "type": "function"
-    },
-    {
-      "inputs": [
-        {
-          "internalType": "address",
-          "name": "spender",
-          "type": "address"
-        },
-        {
-          "internalType": "uint256",
-          "name": "amount",
-          "type": "uint256"
-        }
-      ],
-      "name": "approve",
-      "outputs": [
-        {
-          "internalType": "bool",
-          "name": "",
-          "type": "bool"
-        }
-      ],
-      "stateMutability": "nonpayable",
-      "type": "function"
-    },
-    {
-      "inputs": [
-        {
-          "internalType": "address",
-          "name": "account",
-          "type": "address"
-        }
-      ],
-      "name": "balanceOf",
-      "outputs": [
-        {
-          "internalType": "uint256",
-          "name": "",
-          "type": "uint256"
-        }
-      ],
-      "stateMutability": "view",
-      "type": "function"
-    },
-    {
-      "inputs": [],
-      "name": "totalSupply",
-      "outputs": [
-        {
-          "internalType": "uint256",
-          "name": "",
-          "type": "uint256"
-        }
-      ],
-      "stateMutability": "view",
-      "type": "function"
-    },
-    {
-      "inputs": [
-        {
-          "internalType": "address",
-          "name": "recipient",
-          "type": "address"
-        },
-        {
-          "internalType": "uint256",
-          "name": "amount",
-          "type": "uint256"
-        }
-      ],
-      "name": "transfer",
-      "outputs": [
-        {
-          "internalType": "bool",
-          "name": "",
-          "type": "bool"
-        }
-      ],
-      "stateMutability": "nonpayable",
-      "type": "function"
-    },
-    {
-      "inputs": [
-        {
-          "internalType": "address",
-          "name": "sender",
-          "type": "address"
-        },
-        {
-          "internalType": "address",
-          "name": "recipient",
-          "type": "address"
-        },
-        {
-          "internalType": "uint256",
-          "name": "amount",
-          "type": "uint256"
-        }
-      ],
-      "name": "transferFrom",
-      "outputs": [
-        {
-          "internalType": "bool",
-          "name": "",
-          "type": "bool"
-        }
-      ],
-      "stateMutability": "nonpayable",
-      "type": "function"
-    }
-  ],
-  "bytecode": "0x",
-  "deployedBytecode": "0x",
-  "linkReferences": {},
-  "deployedLinkReferences": {}
-}

contracts/basic-bandwidth-generation/artifacts/@openzeppelin/contracts/token/ERC20/extensions/IERC20Metadata.sol/IERC20Metadata.dbg.json

@@ -1,4 +0,0 @@
-{
-  "_format": "hh-sol-dbg-1",
-  "buildInfo": "../../../../../../build-info/1f95016aa87f8376998977fedfcae017.json"
-}