Mixnet and vesting contracts v1.0.2

* Fix claim -> undelegate

* Fix ordering

* Changelog

.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:

.github/workflows/nightly_build_on_dispatch.yml

@@ -166,8 +166,8 @@ jobs:
           GIT_BRANCH: "${GITHUB_REF##*/}"
           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_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+          KEYBASE_NYM_CHANNEL: "ci-nightly"
           IS_SUCCESS: "${{ env.WORKFLOW_CONCLUSION == 'success' }}"
         uses: docker://keybaseio/client:stable-node
         with:

.github/workflows/notification_on_dispatch.yml

@@ -0,0 +1,26 @@
+name: Nightly builds on dispatch
+
+on: workflow_dispatch
+jobs:
+  notification:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository code
+        uses: actions/checkout@v2
+      - name: Keybase - Node Install
+        run: npm install
+        working-directory: .github/workflows/support-files
+      - name: Keybase - Send Notification
+        env:
+          NYM_NOTIFICATION_KIND: nightly
+          NYM_PROJECT_NAME: "Notification test"
+          GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
+          GIT_COMMIT_MESSAGE: "${{ github.event.head_commit.message }}"
+          GIT_BRANCH: "${GITHUB_REF##*/}"
+          KEYBASE_NYMBOT_USERNAME: "${{ secrets.KEYBASE_NYMBOT_USERNAME }}"
+          KEYBASE_NYMBOT_PAPERKEY: "${{ secrets.KEYBASE_NYMBOT_PAPERKEY }}"
+          KEYBASE_NYMBOT_TEAM: "${{ secrets.KEYBASE_NYMBOT_TEAM }}"
+          KEYBASE_NYM_CHANNEL: "ci-nightly"
+        uses: docker://keybaseio/client:stable-node
+        with:
+          args: .github/workflows/support-files/notifications/entry_point.sh

CHANGELOG.md

@@ -12,10 +12,12 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 ### 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])
 
 [#1541]: https://github.com/nymtech/nym/pull/1541
 [#1558]: https://github.com/nymtech/nym/pull/1558
 [#1577]: https://github.com/nymtech/nym/pull/1577
+[#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.lock

@@ -667,7 +667,10 @@ dependencies = [
  "rand 0.7.3",
  "serde",
  "sled",
+ "tap",
+ "task",
  "tempfile",
+ "thiserror",
  "tokio",
  "topology",
  "url",
@@ -2052,8 +2055,10 @@ dependencies = [
  "pemstore",
  "rand 0.7.3",
  "secp256k1",
+ "task",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "tokio-tungstenite 0.14.0",
  "tungstenite 0.13.0",
  "url",
@@ -3253,6 +3258,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sled",
+ "task",
  "tokio",
  "tokio-tungstenite 0.14.0",
  "topology",
@@ -3373,6 +3379,7 @@ dependencies = [
  "socks5-requests",
  "sqlx 0.6.1",
  "statistics-common",
+ "task",
  "thiserror",
  "tokio",
  "tokio-tungstenite 0.17.2",
@@ -3422,6 +3429,7 @@ dependencies = [
  "serde",
  "snafu",
  "socks5-requests",
+ "task",
  "tokio",
  "topology",
  "url",
@@ -4256,6 +4264,7 @@ dependencies = [
  "log",
  "ordered-buffer",
  "socks5-requests",
+ "task",
  "tokio",
  "tokio-test",
  "tokio-util 0.7.3",
@@ -6136,6 +6145,7 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util 0.7.3",
 ]
 
 [[package]]

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/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,19 +33,20 @@ 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"]

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() {
@@ -197,6 +205,7 @@ impl NymClient {
             ack_sender,
             self.config.get_base().get_gateway_response_timeout(),
             Some(bandwidth_controller),
+            Some(shutdown),
         );
 
         gateway_client
@@ -212,7 +221,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 +247,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 +258,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 +322,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 +371,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 +432,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::{
@@ -120,7 +120,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 +143,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 +154,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 +163,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 +182,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/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/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() {
@@ -196,6 +206,7 @@ impl NymClient {
             ack_sender,
             self.config.get_base().get_gateway_response_timeout(),
             Some(bandwidth_controller),
+            Some(shutdown),
         );
 
         gateway_client
@@ -211,7 +222,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 +248,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 +259,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 +281,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 +361,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::{
@@ -120,7 +120,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 +143,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 +154,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 +163,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 +182,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/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);
-})
+})

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

@@ -20,13 +20,13 @@ web3 = { version = "0.17.0", default-features = false }
 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 +38,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"

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

@@ -30,6 +30,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 +67,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 +86,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 +98,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 +137,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 +145,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 +168,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 +301,31 @@ 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
+                }
+            }
+            .fuse()
+        };
+        #[cfg(not(target_arch = "wasm32"))]
+        tokio::pin!(shutdown);
+
+        #[cfg(target_arch = "wasm32")]
+        let mut shutdown = Box::pin(async {}.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);
                         }
                         _ => (),
                     }
-               }
+                }
             }
         }
     }
@@ -723,6 +770,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,34 @@ 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
+                    }
+                }
+                .fuse()
+            };
+            #[cfg(not(target_arch = "wasm32"))]
+            tokio::pin!(shutdown);
+
+            #[cfg(target_arch = "wasm32")]
+            let mut shutdown = Box::pin(async {}.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 +138,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/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/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/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/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/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/mixnet/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "mixnet-contract"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["Dave Hrycyszyn <futurechimp@users.noreply.github.com>"]
 edition = "2021"
 

contracts/mixnet/src/contract.rs

@@ -471,6 +471,7 @@ pub fn query(deps: Deps<'_>, env: Env, msg: QueryMsg) -> Result<QueryResponse, C
     Ok(query_res?)
 }
 
+#[allow(unused)]
 fn blacklist_malicious_node(storage: &mut dyn Storage, owner: &Addr) -> Result<(), ContractError> {
     let mixnode_bond = match crate::mixnodes::storage::mixnodes()
         .idx
@@ -491,6 +492,7 @@ fn blacklist_malicious_node(storage: &mut dyn Storage, owner: &Addr) -> Result<(
 }
 
 // Removes nodes we've deemed malicious, returns the pledge to the owners, but does not send any rewards
+#[allow(unused)]
 fn remove_malicious_node(
     storage: &mut dyn Storage,
     api: &dyn Api,
@@ -507,17 +509,8 @@ fn remove_malicious_node(
 }
 
 #[entry_point]
-pub fn migrate(deps: DepsMut<'_>, env: Env, msg: MigrateMsg) -> Result<Response, ContractError> {
-    let mut response = Response::new();
-    for node in msg.nodes_to_remove().iter() {
-        let mut sub_response = remove_malicious_node(deps.storage, deps.api, &env, node)
-            .unwrap_or_else(|_| panic!("Could not remove node: {:?}", node));
-        response.messages.append(&mut sub_response.messages);
-        response.attributes.append(&mut sub_response.attributes);
-        response.events.append(&mut sub_response.events);
-    }
-
-    Ok(response)
+pub fn migrate(_deps: DepsMut<'_>, _env: Env, _msg: MigrateMsg) -> Result<Response, ContractError> {
+    Ok(Response::default())
 }
 
 #[cfg(test)]

contracts/mixnet/src/rewards/transactions.rs

@@ -547,16 +547,9 @@ pub fn calculate_delegator_reward(
         .load(storage, (key.clone(), mix_identity.to_string()))
         .unwrap_or(0);
 
-    // Get delegations newer then last_claimed_height, it would be nice to also fold this into the iteration bellow but it should be ok for now, as
-    // I doubt folks refresh their delegations often
     let mut delegations = delegations_storage::delegations()
         .prefix((mix_identity.to_string(), key))
-        .range(
-            storage,
-            Some(Bound::inclusive(last_claimed_height)),
-            None,
-            Order::Descending,
-        )
+        .range(storage, None, None, Order::Ascending)
         .filter_map(|record| record.ok())
         .map(|(_, delegation)| delegation)
         .collect::<Vec<Delegation>>();

contracts/vesting/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vesting-contract"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["Drazen Urch <durch@users.noreply.github.com>"]
 edition = "2021"
 
@@ -22,4 +22,4 @@ cw-storage-plus = { version = "0.13.4", features = ["iterator"] }
 
 schemars = "0.8"
 serde = { version = "1.0", default-features = false, features = ["derive"] }
-thiserror = { version = "1.0" }
+thiserror = { version = "1.0" }

contracts/vesting/src/contract.rs

@@ -1,8 +1,7 @@
 use crate::errors::ContractError;
 use crate::storage::{
-    account_from_address, locked_pledge_cap, remove_delegation, save_delegation,
-    update_locked_pledge_cap, BlockTimestampSecs, ADMIN, DELEGATIONS, MIXNET_CONTRACT_ADDRESS,
-    MIX_DENOM,
+    account_from_address, locked_pledge_cap, update_locked_pledge_cap, BlockTimestampSecs, ADMIN,
+    DELEGATIONS, MIXNET_CONTRACT_ADDRESS, MIX_DENOM,
 };
 use crate::traits::{
     DelegatingAccount, GatewayBondingAccount, MixnodeBondingAccount, VestingAccount,
@@ -30,6 +29,7 @@ use vesting_contract_common::{
 
 pub const INITIAL_LOCKED_PLEDGE_CAP: Uint128 = Uint128::new(100_000_000_000);
 
+/// Instantiate the contract
 #[entry_point]
 pub fn instantiate(
     deps: DepsMut<'_>,
@@ -37,7 +37,7 @@ pub fn instantiate(
     info: MessageInfo,
     msg: InitMsg,
 ) -> Result<Response, ContractError> {
-    // ADMIN is set to the address that instantiated the contract, TODO: make this updatable
+    //! ADMIN is set to the address that instantiated the contract
     ADMIN.save(deps.storage, &info.sender.to_string())?;
     MIXNET_CONTRACT_ADDRESS.save(deps.storage, &msg.mixnet_contract_address)?;
     MIX_DENOM.save(deps.storage, &msg.mix_denom)?;
@@ -128,20 +128,12 @@ pub fn execute(
         ExecuteMsg::UpdateStakingAddress { to_address } => {
             try_update_staking_address(to_address, info, deps)
         }
-        ExecuteMsg::MigrateHeightsToTimestamps {
-            account_id,
-            mix_identity,
-            height_timestamp_map,
-        } => try_migrate_heights_to_timestamps(
-            account_id,
-            mix_identity,
-            height_timestamp_map,
-            info,
-            deps,
-        ),
     }
 }
 
+/// Update locked_pledge_cap, the hard cap for staking/bonding with unvested tokens.
+///
+/// Callable by ADMIN only, see [instantiate].
 pub fn try_update_locked_pledge_cap(
     amount: Uint128,
     info: MessageInfo,
@@ -154,6 +146,7 @@ pub fn try_update_locked_pledge_cap(
     Ok(Response::default())
 }
 
+/// Update config for a mixnode bonded with vesting account, sends [mixnet_contract_common::ExecuteMsg::UpdateMixnodeConfig] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_update_mixnode_config(
     profit_margin_percent: u8,
     info: MessageInfo,
@@ -163,7 +156,9 @@ pub fn try_update_mixnode_config(
     account.try_update_mixnode_config(profit_margin_percent, deps.storage)
 }
 
-// Only contract admin, set at init
+/// Updates mixnet contract address, for cases when a new mixnet contract is deployed.
+///
+/// Callable by ADMIN only, see [instantiate].
 pub fn try_update_mixnet_address(
     address: String,
     info: MessageInfo,
@@ -176,7 +171,7 @@ pub fn try_update_mixnet_address(
     Ok(Response::default())
 }
 
-// Only contract owner of vesting account
+/// Withdraw already vested coins.
 pub fn try_withdraw_vested_coins(
     amount: Coin,
     env: Env,
@@ -217,6 +212,7 @@ pub fn try_withdraw_vested_coins(
     }
 }
 
+/// Transfer ownership of the entire vesting account.
 fn try_transfer_ownership(
     to_address: String,
     info: MessageInfo,
@@ -233,6 +229,7 @@ fn try_transfer_ownership(
     }
 }
 
+/// Set or update staking address for a vesting account.
 fn try_update_staking_address(
     to_address: Option<String>,
     info: MessageInfo,
@@ -250,31 +247,7 @@ fn try_update_staking_address(
     }
 }
 
-pub fn try_migrate_heights_to_timestamps(
-    account_id: u32,
-    mix_identity: String,
-    height_timestamp_map: Vec<(u64, u64)>,
-    info: MessageInfo,
-    deps: DepsMut<'_>,
-) -> Result<Response, ContractError> {
-    if info.sender != ADMIN.load(deps.storage)? {
-        return Err(ContractError::NotAdmin(info.sender.as_str().to_string()));
-    }
-
-    for (height, timestamp) in height_timestamp_map {
-        let amount = DELEGATIONS.load(deps.storage, (account_id, mix_identity.clone(), height))?;
-        remove_delegation((account_id, mix_identity.clone(), height), deps.storage)?;
-        save_delegation(
-            (account_id, mix_identity.clone(), timestamp),
-            amount,
-            deps.storage,
-        )?;
-    }
-
-    Ok(Response::default())
-}
-
-// Owner or staking
+/// Bond a gateway, sends [mixnet_contract_common::ExecuteMsg::BondGatewayOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_bond_gateway(
     gateway: Gateway,
     owner_signature: String,
@@ -289,11 +262,13 @@ pub fn try_bond_gateway(
     account.try_bond_gateway(gateway, owner_signature, pledge, &env, deps.storage)
 }
 
+/// Unbond a gateway, sends [mixnet_contract_common::ExecuteMsg::UnbondGatewayOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_unbond_gateway(info: MessageInfo, deps: DepsMut<'_>) -> Result<Response, ContractError> {
     let account = account_from_address(info.sender.as_str(), deps.storage, deps.api)?;
     account.try_unbond_gateway(deps.storage)
 }
 
+/// Track gateway unbonding, invoked by the mixnet contract after succesful unbonding, message containes coins returned including any accrued rewards.
 pub fn try_track_unbond_gateway(
     owner: &str,
     amount: Coin,
@@ -308,6 +283,7 @@ pub fn try_track_unbond_gateway(
     Ok(Response::new().add_event(new_track_gateway_unbond_event()))
 }
 
+/// Compound operator reward, sends [mixnet_contract_common::ExecuteMsg::CompoundOperatorRewardOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS], adds available rewards to the existing bond
 pub fn try_compound_operator_reward(
     info: MessageInfo,
     deps: DepsMut<'_>,
@@ -316,6 +292,7 @@ pub fn try_compound_operator_reward(
     account.try_compound_operator_reward(deps.storage)
 }
 
+/// Bond a mixnode, sends [mixnet_contract_common::ExecuteMsg::BondMixnodeOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_bond_mixnode(
     mix_node: MixNode,
     owner_signature: String,
@@ -330,11 +307,13 @@ pub fn try_bond_mixnode(
     account.try_bond_mixnode(mix_node, owner_signature, pledge, &env, deps.storage)
 }
 
+/// Unbond a mixnode, sends [mixnet_contract_common::ExecuteMsg::UnbondMixnodeOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 pub fn try_unbond_mixnode(info: MessageInfo, deps: DepsMut<'_>) -> Result<Response, ContractError> {
     let account = account_from_address(info.sender.as_str(), deps.storage, deps.api)?;
     account.try_unbond_mixnode(deps.storage)
 }
 
+/// Track mixnode unbonding, invoked by the mixnet contract after succesful unbonding, message containes coins returned including any accrued rewards.
 pub fn try_track_unbond_mixnode(
     owner: &str,
     amount: Coin,
@@ -349,6 +328,7 @@ pub fn try_track_unbond_mixnode(
     Ok(Response::new().add_event(new_track_mixnode_unbond_event()))
 }
 
+/// Track reward collection, invoked by the mixnert contract after sucessful reward compounding or claiming
 fn try_track_reward(
     deps: DepsMut<'_>,
     info: MessageInfo,
@@ -363,6 +343,7 @@ fn try_track_reward(
     Ok(Response::new().add_event(new_track_reward_event()))
 }
 
+/// Track undelegation, invoked by the mixnet contract after sucessful undelegation, message contains coins returned with any accrued rewards.
 fn try_track_undelegation(
     address: &str,
     mix_identity: IdentityKey,
@@ -378,6 +359,7 @@ fn try_track_undelegation(
     Ok(Response::new().add_event(new_track_undelegation_event()))
 }
 
+/// Delegate to mixnode, sends [mixnet_contract_common::ExecuteMsg::DelegateToMixnodeOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS]..
 fn try_delegate_to_mixnode(
     mix_identity: IdentityKey,
     amount: Coin,
@@ -391,6 +373,7 @@ fn try_delegate_to_mixnode(
     account.try_delegate_to_mixnode(mix_identity, amount, &env, deps.storage)
 }
 
+/// Compounds deleagtor reward, ie adds it to the existing delegations for a node, sends [mixnet_contract_common::ExecuteMsg::CompoundDelegatorRewardOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 fn try_compound_delegator_reward(
     mix_identity: IdentityKey,
     info: MessageInfo,
@@ -400,6 +383,7 @@ fn try_compound_delegator_reward(
     account.try_compound_delegator_reward(mix_identity, deps.storage)
 }
 
+/// Claims operator reward, sends [mixnet_contract_common::ExecuteMsg::ClaimOperatorRewardOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 fn try_claim_operator_reward(
     deps: DepsMut<'_>,
     info: MessageInfo,
@@ -408,6 +392,7 @@ fn try_claim_operator_reward(
     account.try_claim_operator_reward(deps.storage)
 }
 
+/// Claims delegator reward, sends [mixnet_contract_common::ExecuteMsg::ClaimDelegatorRewardOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 fn try_claim_delegator_reward(
     deps: DepsMut<'_>,
     info: MessageInfo,
@@ -417,6 +402,7 @@ fn try_claim_delegator_reward(
     account.try_claim_delegator_reward(mix_identity, deps.storage)
 }
 
+/// Undelegates from a mixnode, sends [mixnet_contract_common::ExecuteMsg::UndelegateFromMixnodeOnBehalf] to [crate::storage::MIXNET_CONTRACT_ADDRESS].
 fn try_undelegate_from_mixnode(
     mix_identity: IdentityKey,
     info: MessageInfo,
@@ -426,6 +412,9 @@ fn try_undelegate_from_mixnode(
     account.try_undelegate_from_mixnode(mix_identity, deps.storage)
 }
 
+/// Creates a new periodic vesting account, and deposits funds to vest into the contract.
+///
+/// Callable by ADMIN only, see [instantiate].
 fn try_create_periodic_vesting_account(
     owner_address: &str,
     staking_address: Option<String>,
@@ -568,10 +557,12 @@ pub fn query(deps: Deps<'_>, env: Env, msg: QueryMsg) -> Result<QueryResponse, C
     Ok(query_res?)
 }
 
+/// Get locked_pledge_cap, the hard cap for staking/bonding with unvested tokens.
 pub fn get_locked_pledge_cap(deps: Deps<'_>) -> Uint128 {
     locked_pledge_cap(deps.storage)
 }
 
+/// Get current vesting period for a given [crate::vesting::Account].
 pub fn try_get_current_vesting_period(
     address: &str,
     deps: Deps<'_>,
@@ -581,11 +572,13 @@ pub fn try_get_current_vesting_period(
     Ok(account.get_current_vesting_period(env.block.time))
 }
 
+/// Loads mixnode bond from vesting contract storage.
 pub fn try_get_mixnode(address: &str, deps: Deps<'_>) -> Result<Option<PledgeData>, ContractError> {
     let account = account_from_address(address, deps.storage, deps.api)?;
     account.load_mixnode_pledge(deps.storage)
 }
 
+/// Loads gateway bond from vesting contract storage.
 pub fn try_get_gateway(address: &str, deps: Deps<'_>) -> Result<Option<PledgeData>, ContractError> {
     let account = account_from_address(address, deps.storage, deps.api)?;
     account.load_gateway_pledge(deps.storage)
@@ -595,6 +588,7 @@ pub fn try_get_account(address: &str, deps: Deps<'_>) -> Result<Account, Contrac
     account_from_address(address, deps.storage, deps.api)
 }
 
+/// Gets currently locked coins, see [crate::traits::VestingAccount::locked_coins]
 pub fn try_get_locked_coins(
     vesting_account_address: &str,
     block_time: Option<Timestamp>,
@@ -605,6 +599,7 @@ pub fn try_get_locked_coins(
     account.locked_coins(block_time, &env, deps.storage)
 }
 
+/// Returns currently locked coins, see [crate::traits::VestingAccount::spendable_coins]
 pub fn try_get_spendable_coins(
     vesting_account_address: &str,
     block_time: Option<Timestamp>,
@@ -615,6 +610,7 @@ pub fn try_get_spendable_coins(
     account.spendable_coins(block_time, &env, deps.storage)
 }
 
+/// Returns coins that have vested, see [crate::traits::VestingAccount::get_vested_coins]
 pub fn try_get_vested_coins(
     vesting_account_address: &str,
     block_time: Option<Timestamp>,
@@ -625,6 +621,7 @@ pub fn try_get_vested_coins(
     account.get_vested_coins(block_time, &env, deps.storage)
 }
 
+/// Returns coins that are vesting, see [crate::traits::VestingAccount::get_vesting_coins]
 pub fn try_get_vesting_coins(
     vesting_account_address: &str,
     block_time: Option<Timestamp>,
@@ -635,6 +632,7 @@ pub fn try_get_vesting_coins(
     account.get_vesting_coins(block_time, &env, deps.storage)
 }
 
+/// See [crate::traits::VestingAccount::get_start_time]
 pub fn try_get_start_time(
     vesting_account_address: &str,
     deps: Deps<'_>,
@@ -643,6 +641,7 @@ pub fn try_get_start_time(
     Ok(account.get_start_time())
 }
 
+/// See [crate::traits::VestingAccount::get_end_time]
 pub fn try_get_end_time(
     vesting_account_address: &str,
     deps: Deps<'_>,
@@ -651,6 +650,7 @@ pub fn try_get_end_time(
     Ok(account.get_end_time())
 }
 
+/// See [crate::traits::VestingAccount::get_original_vesting]
 pub fn try_get_original_vesting(
     vesting_account_address: &str,
     deps: Deps<'_>,
@@ -659,6 +659,7 @@ pub fn try_get_original_vesting(
     Ok(account.get_original_vesting())
 }
 
+/// See [crate::traits::VestingAccount::get_delegated_free]
 pub fn try_get_delegated_free(
     block_time: Option<Timestamp>,
     vesting_account_address: &str,
@@ -669,6 +670,7 @@ pub fn try_get_delegated_free(
     account.get_delegated_free(block_time, &env, deps.storage)
 }
 
+/// See [crate::traits::VestingAccount::get_delegated_vesting]
 pub fn try_get_delegated_vesting(
     block_time: Option<Timestamp>,
     vesting_account_address: &str,
@@ -679,6 +681,7 @@ pub fn try_get_delegated_vesting(
     account.get_delegated_vesting(block_time, &env, deps.storage)
 }
 
+/// Returns timestamps at which delegations were made
 pub fn try_get_delegation_times(
     deps: Deps<'_>,
     vesting_account_address: &str,

contracts/vesting/src/lib.rs

@@ -1,3 +1,6 @@
+#![allow(rustdoc::private_intra_doc_links)]
+//! Nym vesting contract, providing vesting accounts with ability to stake unvested tokens
+
 pub mod contract;
 mod errors;
 mod queued_migrations;

contracts/vesting/src/traits/vesting_account.rs

@@ -9,12 +9,13 @@ pub trait VestingAccount {
         env: &Env,
     ) -> Result<Uint128, ContractError>;
 
-    // locked_coins returns the set of coins that are not spendable (can still be delegated tough) (i.e. locked),
-    // defined as the vesting coins that are not delegated or pledged.
-    //
-    // To get spendable coins of a vesting account, first the total balance must
-    // be retrieved and the locked tokens can be subtracted from the total balance.
-    // Note, the spendable balance can be negative.
+    /// Returns the set of coins that are not spendable (can still be delegated tough) (i.e. locked),
+    /// defined as the vesting coins that are not delegated or pledged.
+    ///
+    /// To get spendable coins of a vesting account, first the total balance must
+    /// be retrieved and the locked tokens can be subtracted from the total balance.
+    /// Note, the spendable balance can be negative.
+    /// See [/vesting-contract/struct.Account.html/method.locked_coins] for impl
     fn locked_coins(
         &self,
         block_time: Option<Timestamp>,
@@ -22,7 +23,8 @@ pub trait VestingAccount {
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
 
-    // Calculates the total spendable balance that can be sent to other accounts.
+    /// Calculated as current_balance minus [crate::traits::VestingAccount::locked_coins]
+    /// See [/vesting-contract/struct.Account.html/method.spendable_coins] for impl
     fn spendable_coins(
         &self,
         block_time: Option<Timestamp>,
@@ -30,12 +32,15 @@ pub trait VestingAccount {
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
 
+    /// See [/vesting-contract/struct.Account.html/method.get_vested_coins] for impl
     fn get_vested_coins(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
+
+    /// See [/vesting-contract/struct.Account.html/method.get_vesting_coins] for impl
     fn get_vesting_coins(
         &self,
         block_time: Option<Timestamp>,
@@ -43,39 +48,52 @@ pub trait VestingAccount {
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
 
+    /// See [/vesting-contract/struct.Account.html/method.get_start_time] for impl
     fn get_start_time(&self) -> Timestamp;
+    /// See [/vesting-contract/struct.Account.html/method.get_end_time] for impl
     fn get_end_time(&self) -> Timestamp;
 
+    /// Returns amount of coins set at account creation
+    /// See [/vesting-contract/struct.Account.html/method.get_original_vesting] for impl
     fn get_original_vesting(&self) -> OriginalVestingResponse;
+
+    /// See [/vesting-contract/struct.Account.html/method.get_delegated_free] for impl
     fn get_delegated_free(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
+
+    /// See [/vesting-contract/struct.Account.html/method.get_delegated_vesting] for impl
     fn get_delegated_vesting(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
+
+    /// See [/vesting-contract/struct.Account.html/method.get_pledged_free] for impl
     fn get_pledged_free(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
+    /// See [/vesting-contract/struct.Account.html/method.get_pledged_vesting] for impl
     fn get_pledged_vesting(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError>;
+    /// See [/vesting-contract/struct.Account.html/method.transfer_ownership] for impl
     fn transfer_ownership(
         &mut self,
         to_address: &Addr,
         storage: &mut dyn Storage,
     ) -> Result<(), ContractError>;
+    /// See [/vesting-contract/struct.Account.html/method.update_staking_address] for impl
     fn update_staking_address(
         &mut self,
         to_address: Option<Addr>,

contracts/vesting/src/vesting/account/vesting_account.rs

@@ -16,13 +16,14 @@ impl VestingAccount for Account {
             + self.get_pledged_vesting(None, env, storage)?.amount)
     }
 
+    /// See [VestingAccount::locked_coins] for documentation.
+    /// Returns 0 in case of underflow. Which is fine, as the amount of pledged and delegated tokens can be larger then vesting_coins due to rewards and vesting periods expiring
     fn locked_coins(
         &self,
         block_time: Option<Timestamp>,
         env: &Env,
         storage: &dyn Storage,
     ) -> Result<Coin, ContractError> {
-        // Returns 0 in case of underflow. Which is fine, as the amount of pledged and delegated tokens can be larger then vesting_coins due to rewards and vesting periods expiring
         Ok(Coin {
             amount: Uint128::new(
                 self.get_vesting_coins(block_time, env, storage)?

contracts/vesting/src/vesting/mod.rs

@@ -925,97 +925,4 @@ mod tests {
         // the 50M delegation wasn't a thing here for VESTING tokens either
         assert_eq!(delegated_vesting.amount, Uint128::zero());
     }
-
-    #[test]
-    fn migrate_heights_to_timestamps() {
-        let mut deps = init_contract();
-        let mut env = mock_env();
-
-        let account = vesting_account_new_fixture(&mut deps.storage, &env);
-        let mix_identity = String::from("identity");
-        let mut curr_block = env.block.clone();
-        let mut delegation_blocks = std::iter::from_fn(move || {
-            curr_block.height += 1;
-            curr_block.time = curr_block.time.plus_seconds(5);
-            Some(curr_block.clone())
-        })
-        .take(100)
-        .collect::<Vec<_>>();
-
-        for block in delegation_blocks.iter() {
-            DELEGATIONS
-                .save(
-                    &mut deps.storage,
-                    (account.storage_key(), mix_identity.clone(), block.height),
-                    &Uint128::new(90_000_000_000),
-                )
-                .unwrap();
-        }
-
-        let delegations = try_get_delegation_times(
-            deps.as_ref(),
-            account.owner_address().as_str(),
-            mix_identity.clone(),
-        )
-        .unwrap();
-        assert_eq!(
-            delegations.delegation_timestamps.len(),
-            delegation_blocks.len()
-        );
-        for (heights, delegation_block) in delegations
-            .delegation_timestamps
-            .iter()
-            .zip(delegation_blocks.iter())
-        {
-            assert_eq!(*heights, delegation_block.height);
-        }
-
-        let height_timestamp_map = delegation_blocks
-            .iter()
-            .map(|block| (block.height, block.time.seconds()))
-            .collect();
-        let admin = ADMIN.load(&deps.storage).unwrap();
-        try_migrate_heights_to_timestamps(
-            account.storage_key(),
-            mix_identity.clone(),
-            height_timestamp_map,
-            mock_info(&admin, &[]),
-            deps.as_mut(),
-        )
-        .unwrap();
-
-        // Some new delegation appears during migration
-        env.block.height += 200;
-        env.block.time = env.block.time.plus_seconds(1000);
-        delegation_blocks.push(env.block.clone());
-        account
-            .try_delegate_to_mixnode(
-                String::from("identity"),
-                Coin {
-                    amount: Uint128::new(90_000_000_000),
-                    denom: TEST_COIN_DENOM.to_string(),
-                },
-                &env,
-                &mut deps.storage,
-            )
-            .unwrap();
-
-        let delegations = try_get_delegation_times(
-            deps.as_ref(),
-            account.owner_address().as_str(),
-            mix_identity.clone(),
-        )
-        .unwrap();
-        assert_eq!(
-            delegations.delegation_timestamps.len(),
-            delegation_blocks.len()
-        );
-        for (timestamp, delegation_block) in delegations
-            .delegation_timestamps
-            .iter()
-            .zip(delegation_blocks.iter())
-        {
-            assert_eq!(*timestamp, delegation_block.time.seconds());
-        }
-    }
 }

mixnode/src/node/mod.rs

@@ -26,7 +26,7 @@ use rand::thread_rng;
 use std::net::SocketAddr;
 use std::process;
 use std::sync::Arc;
-use task::{ShutdownListener, ShutdownNotifier};
+use task::{wait_for_signal, ShutdownListener, ShutdownNotifier};
 use version_checker::parse_version;
 
 mod http;
@@ -334,31 +334,3 @@ impl MixNode {
         self.wait_for_interrupt(shutdown).await
     }
 }
-
-#[cfg(unix)]
-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))]
-async fn wait_for_signal() {
-    tokio::select! {
-        _ = tokio::signal::ctrl_c() => {
-            log::info!("Received SIGINT");
-        },
-    }
-}

nym-connect/Cargo.lock

@@ -629,6 +629,9 @@ dependencies = [
  "rand 0.7.3",
  "serde",
  "sled",
+ "tap",
+ "task",
+ "thiserror",
  "tokio",
  "topology",
  "url",
@@ -1919,8 +1922,10 @@ dependencies = [
  "pemstore",
  "rand 0.7.3",
  "secp256k1",
+ "task",
  "thiserror",
  "tokio",
+ "tokio-stream",
  "tokio-tungstenite",
  "tungstenite",
  "url",
@@ -3459,6 +3464,7 @@ dependencies = [
  "serde",
  "snafu",
  "socks5-requests",
+ "task",
  "tokio",
  "topology",
  "url",
@@ -4266,6 +4272,7 @@ dependencies = [
  "log",
  "ordered-buffer",
  "socks5-requests",
+ "task",
  "tokio",
  "tokio-util 0.7.3",
 ]
@@ -5624,6 +5631,15 @@ dependencies = [
  "xattr",
 ]
 
+[[package]]
+name = "task"
+version = "0.1.0"
+dependencies = [
+ "log",
+ "tokio",
+ "tokio-util 0.7.3",
+]
+
 [[package]]
 name = "tauri"
 version = "1.0.3"
@@ -5944,18 +5960,18 @@ checksum = "8eaa81235c7058867fa8c0e7314f33dcce9c215f535d1913822a2b3f5e289f3c"
 
 [[package]]
 name = "thiserror"
-version = "1.0.31"
+version = "1.0.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd829fe32373d27f76265620b5309d0340cb8550f523c1dda251d6298069069a"
+checksum = "8c1b05ca9d106ba7d2e31a9dab4a64e7be2cce415321966ea3132c49a656e252"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "1.0.31"
+version = "1.0.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0396bc89e626244658bef819e22d0cc459e795a5ebe878e6ec336d1674a8d79a"
+checksum = "e8f2591983642de85c921015f3f070c665a197ed69e417af436115e3a1407487"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -6070,6 +6086,7 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util 0.7.3",
 ]
 
 [[package]]

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

@@ -172,7 +172,7 @@ pub async fn init_socks5_config(provider_address: String, chosen_gateway_id: Str
     );
     log::info!("Client configuration completed.");
 
-    client_core::init::show_address(config.get_base());
+    client_core::init::show_address(config.get_base())?;
     Ok(())
 }
 
@@ -191,13 +191,13 @@ 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");
 
         Ok(gateway.into())
@@ -210,23 +210,21 @@ async fn setup_gateway(
             config.get_base().get_validator_api_endpoints(),
             user_chosen_gateway_id,
         )
-        .await;
+        .await?;
         log::debug!("Querying gateway gives: {}", gateway);
         Ok(gateway.into())
     } else {
         println!("Not registering gateway, will reuse existing config and keys");
-        match Socks5Config::load_from_file(Some(id)) {
-            Ok(existing_config) => Ok(existing_config.get_base().get_gateway_endpoint().clone()),
-            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."
-                );
-                Err(BackendError::CouldNotLoadExistingGatewayConfiguration(err))
-            }
-        }
+        let existing_config = Socks5Config::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."
+            );
+            BackendError::CouldNotLoadExistingGatewayConfiguration(err)
+        })?;
+        Ok(existing_config.get_base().get_gateway_endpoint().clone())
     }
 }

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

@@ -1,3 +1,4 @@
+use client_core::error::ClientCoreError;
 use serde::{Serialize, Serializer};
 use thiserror::Error;
 
@@ -29,6 +30,11 @@ pub enum BackendError {
         #[from]
         source: serde_json::Error,
     },
+    #[error("{source}")]
+    ClientCoreError {
+        #[from]
+        source: ClientCoreError,
+    },
 
     #[error("State error")]
     StateError,

nym-wallet/CHANGELOG.md

@@ -1,3 +1,7 @@
+## [nym-wallet-v1.0.9](https://github.com/nymtech/nym/releases/tag/nym-wallet-v1.0.8) (2022-09-08)
+
+- wallet: change default `nymd` URL to https://rpc.nymtech.net
+
 ## [nym-wallet-v1.0.8](https://github.com/nymtech/nym/releases/tag/nym-wallet-v1.0.8) (2022-08-11)
 
 - wallet: new bonding flow and screen for bonded node

nym-wallet/Cargo.lock

@@ -645,16 +645,16 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "3.1.18"
+version = "3.2.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d2dbdf4bdacb33466e854ce889eee8dfd5729abf7ccd7664d0a2d60cd384440b"
+checksum = "23b71c3ce99b7611011217b366d923f1d0a7e07a92bb2dbf1e84508c673ca3bd"
 dependencies = [
  "atty",
  "bitflags",
  "clap_derive",
  "clap_lex",
  "indexmap",
- "lazy_static",
+ "once_cell",
  "strsim 0.10.0",
  "termcolor",
  "textwrap",
@@ -662,9 +662,9 @@ dependencies = [
 
 [[package]]
 name = "clap_derive"
-version = "3.1.18"
+version = "3.2.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "25320346e922cffe59c0bbc5410c8d8784509efb321488971081313cb1e1a33c"
+checksum = "ea0c8bce528c4be4da13ea6fead8965e95b6073585a2f05204bd8f4119f82a65"
 dependencies = [
  "heck 0.4.0",
  "proc-macro-error",
@@ -675,9 +675,9 @@ dependencies = [
 
 [[package]]
 name = "clap_lex"
-version = "0.2.0"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a37c35f1112dad5e6e0b1adaff798507497a18fceeb30cceb3bae7d1427b9213"
+checksum = "2850f2f5a82cbf437dd5af4d49848fbdfc27c157c3d010345776f952765261c5"
 dependencies = [
  "os_str_bytes",
 ]
@@ -756,9 +756,9 @@ dependencies = [
 
 [[package]]
 name = "concurrent-queue"
-version = "1.2.2"
+version = "1.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "30ed07550be01594c6026cff2a1d7fe9c8f683caa798e12b68694ac9e88286a3"
+checksum = "af4780a44ab5696ea9e28294517f1fffb421a83a25af521333c838635509db9c"
 dependencies = [
  "cache-padded",
 ]
@@ -1456,9 +1456,9 @@ dependencies = [
 
 [[package]]
 name = "enumflags2"
-version = "0.7.4"
+version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1b3ab37dc79652c9d85f1f7b6070d77d321d2467f5fe7b00d6b7a86c57b092ae"
+checksum = "e75d4cd21b95383444831539909fbb14b9dc3fdceb2a6f5d36577329a1f55ccb"
 dependencies = [
  "enumflags2_derive",
  "serde",
@@ -1490,9 +1490,9 @@ dependencies = [
 
 [[package]]
 name = "event-listener"
-version = "2.5.2"
+version = "2.5.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "77f3309417938f28bf8228fcff79a4a37103981e3e186d2ccd19c74b38f4eb71"
+checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0"
 
 [[package]]
 name = "execute"
@@ -3055,7 +3055,7 @@ dependencies = [
 
 [[package]]
 name = "nym_wallet"
-version = "1.0.8"
+version = "1.0.9"
 dependencies = [
  "aes-gcm",
  "argon2 0.3.4",
@@ -3073,6 +3073,7 @@ dependencies = [
  "eyre",
  "futures",
  "itertools",
+ "k256",
  "log",
  "mixnet-contract-common",
  "nym-types",
@@ -3150,9 +3151,9 @@ dependencies = [
 
 [[package]]
 name = "once_cell"
-version = "1.9.0"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da32515d9f6e6e489d7bc9d84c71b060db7247dc035bbe44eac88cf87486d8d5"
+checksum = "2f7254b99e31cad77da24b08ebf628882739a608578bb1bcdfc1f9c21260d7c0"
 
 [[package]]
 name = "opaque-debug"
@@ -3605,10 +3606,11 @@ dependencies = [
 
 [[package]]
 name = "polling"
-version = "2.2.0"
+version = "2.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "685404d509889fade3e86fe3a5803bca2ec09b0c0778d5ada6ec8bf7a8de5259"
+checksum = "899b00b9c8ab553c743b3e11e87c5c7d423b2a2de229ba95b24a756344748011"
 dependencies = [
+ "autocfg 1.1.0",
  "cfg-if",
  "libc",
  "log",

nym-wallet/src-tauri/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "nym_wallet"
-version = "1.0.8"
+version = "1.0.9"
 description = "Nym Native Wallet"
 authors = ["Nym Technologies SA"]
 license = ""
@@ -42,6 +42,7 @@ thiserror = "1.0"
 tokio = { version = "1.10", features = ["full"] }
 toml = "0.5.8"
 url = "2.2"
+k256 = { version = "0.10", features = ["ecdsa", "sha256"] }
 
 aes-gcm = "0.9.4"
 argon2 = { version = "0.3.2", features = ["std"] }

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

@@ -552,7 +552,7 @@ api_url = 'https://baz/api'
             .next()
             .map(|v| v.nymd_url)
             .unwrap();
-        assert_eq!(nymd_url.as_ref(), "https://rpc.nyx.nodes.guru/");
+        assert_eq!(nymd_url.as_ref(), "https://rpc.nymtech.net/");
 
         let api_url = config
             .get_base_validators(WalletNetwork::MAINNET)

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

@@ -69,6 +69,11 @@ pub enum BackendError {
         #[from]
         source: reqwest::Error,
     },
+    #[error("{source}")]
+    K256Error {
+        #[from]
+        source: k256::ecdsa::Error,
+    },
     #[error("failed to encrypt the given data with the provided password")]
     EncryptionError,
     #[error("failed to decrypt the given data with the provided password")]
@@ -111,6 +116,8 @@ pub enum BackendError {
     UnknownCoinDenom(String),
     #[error("Network {network} doesn't have any associated registered coin denoms")]
     NoCoinsRegistered { network: Network },
+    #[error("Signature error {0}")]
+    SignatureError(String),
 }
 
 impl Serialize for BackendError {

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

@@ -18,6 +18,7 @@ mod wallet_storage;
 
 use crate::menu::AddDefaultSubmenus;
 use crate::operations::mixnet;
+use crate::operations::signatures;
 use crate::operations::simulate;
 use crate::operations::validator_api;
 use crate::operations::vesting;
@@ -146,6 +147,8 @@ fn main() {
             simulate::mixnet::simulate_claim_operator_reward,
             simulate::mixnet::simulate_compound_operator_reward,
             simulate::mixnet::simulate_compound_delegator_reward,
+            signatures::sign::sign,
+            signatures::sign::verify,
         ])
         .menu(Menu::new().add_default_app_submenu_if_macos())
         .run(tauri::generate_context!())

nym-wallet/src-tauri/src/operations/mod.rs

@@ -1,4 +1,5 @@
 pub mod mixnet;
+pub mod signatures;
 pub mod simulate;
 pub mod validator_api;
 pub mod vesting;

nym-wallet/src-tauri/src/operations/signatures/mod.rs

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

nym-wallet/src-tauri/src/operations/signatures/sign.rs

@@ -0,0 +1,140 @@
+use std::str::FromStr;
+
+use crate::error::BackendError;
+use crate::state::WalletState;
+use cosmrs::crypto::secp256k1::{Signature, VerifyingKey};
+use cosmrs::crypto::PublicKey;
+use cosmrs::AccountId;
+use k256::ecdsa::signature::Verifier;
+use serde::Serialize;
+use serde_json::json;
+
+#[derive(Debug, Serialize)]
+pub struct SignatureOutputJson {
+    pub account_id: String,
+    pub public_key: PublicKey,
+    pub signature: String,
+}
+
+#[tauri::command]
+pub async fn sign(
+    message: String,
+    state: tauri::State<'_, WalletState>,
+) -> Result<String, BackendError> {
+    let guard = state.read().await;
+    let client = guard.current_client()?;
+    let wallet = client.nymd.signer();
+    let derived_accounts = wallet.try_derive_accounts()?;
+    let account = derived_accounts.first().ok_or_else(|| {
+        log::error!(">>> Unable to derive account");
+        BackendError::SignatureError("unable to derive account".to_string())
+    })?;
+
+    log::info!("<<< Signing message");
+    let signature = wallet.sign_raw_with_account(account, message.as_bytes())?;
+    let signature_as_hex_string = signature.to_string();
+    let output = SignatureOutputJson {
+        account_id: account.address().to_string(),
+        public_key: account.public_key(),
+        signature: signature_as_hex_string.to_string(),
+    };
+    log::info!(">>> Signing data {}", json!(output),);
+    Ok(signature_as_hex_string)
+}
+
+async fn get_pubkey_from_account_address(
+    address: &AccountId,
+    state: &tauri::State<'_, WalletState>,
+) -> Result<PublicKey, BackendError> {
+    log::info!("Getting public key for address {} from chain...", address);
+    let guard = state.read().await;
+    let client = guard.current_client()?;
+    let account = client
+        .nymd
+        .get_account_details(address)
+        .await?
+        .ok_or_else(|| {
+            log::error!("No account associated with address {}", address);
+            BackendError::SignatureError(format!("No account associated with address {}", address))
+        })?;
+    let base_account = account.try_get_base_account()?;
+
+    base_account.pubkey.ok_or_else(|| {
+        log::error!("No pubkey found for address {}", address);
+        BackendError::SignatureError(format!("No pubkey found for address {}", address))
+    })
+}
+
+enum VerifyInputKind {
+    PublicKey(PublicKey),
+    AccountAddress(String),
+    CurrentAccountAddress,
+}
+
+impl TryFrom<Option<String>> for VerifyInputKind {
+    type Error = BackendError;
+
+    fn try_from(value: Option<String>) -> Result<Self, Self::Error> {
+        let key = match value {
+            Some(key) => key,
+            None => return Ok(VerifyInputKind::CurrentAccountAddress),
+        };
+        if key.trim().is_empty() {
+            return Err(BackendError::SignatureError(
+                "Please ensure the public key or address is not empty or whitespace".to_string(),
+            ));
+        }
+        let account_id = AccountId::from_str(&key);
+        let key_from_json = PublicKey::from_json(&key);
+        if account_id.is_err() && key_from_json.is_err() {
+            return Err(BackendError::SignatureError(
+                "Please ensure the public key or address is valid".to_string(),
+            ));
+        }
+        if let Ok(k) = key_from_json {
+            Ok(VerifyInputKind::PublicKey(k))
+        } else {
+            Ok(VerifyInputKind::AccountAddress(key))
+        }
+    }
+}
+
+#[tauri::command]
+pub async fn verify(
+    public_key_as_json_or_account_address: Option<String>,
+    signature_as_hex: String,
+    message: String,
+    state: tauri::State<'_, WalletState>,
+) -> Result<(), BackendError> {
+    let public_key = match VerifyInputKind::try_from(public_key_as_json_or_account_address)? {
+        VerifyInputKind::PublicKey(key) => key,
+        VerifyInputKind::AccountAddress(address) => {
+            // get public key from the given account address
+            get_pubkey_from_account_address(&AccountId::from_str(&address)?, &state).await?
+        }
+        VerifyInputKind::CurrentAccountAddress => {
+            // get public key from current account address
+            let guard = state.read().await;
+            let client = guard.current_client()?;
+            let address = client.nymd.address();
+            get_pubkey_from_account_address(address, &state).await?
+        }
+    };
+
+    if public_key.type_url() != PublicKey::SECP256K1_TYPE_URL {
+        return Err(BackendError::SignatureError(
+            "Sorry, we only support secp256k1 public keys at the moment".to_string(),
+        ));
+    }
+
+    log::info!("<<< Verifying signature [{}]", signature_as_hex);
+    let verifying_key = VerifyingKey::from_sec1_bytes(&public_key.to_bytes())?;
+    let signature = Signature::from_str(&signature_as_hex)?;
+    let message_as_bytes = message.into_bytes();
+    Ok(verifying_key
+        .verify(&message_as_bytes, &signature)
+        .map_err(|e| {
+            log::error!(">>> Verification failed, wrong signature");
+            e
+        })?)
+}

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

@@ -516,7 +516,7 @@ mod tests {
                 "http://nymd_url.com/".parse().unwrap(),
                 "http://foo.com".parse().unwrap(),
                 "http://bar.com".parse().unwrap(),
-                "https://rpc.nyx.nodes.guru".parse().unwrap(),
+                "https://rpc.nymtech.net".parse().unwrap(),
             ],
         );
         assert_eq!(
@@ -538,7 +538,7 @@ mod tests {
                 "http://nymd_url.com/".parse().unwrap(),
                 "http://foo.com".parse().unwrap(),
                 "http://bar.com".parse().unwrap(),
-                "https://rpc.nyx.nodes.guru".parse().unwrap(),
+                "https://rpc.nymtech.net".parse().unwrap(),
             ],
         )
     }

nym-wallet/src-tauri/tauri.conf.json

@@ -1,7 +1,7 @@
 {
   "package": {
     "productName": "nym-wallet",
-    "version": "1.0.8"
+    "version": "1.0.9"
   },
   "build": {
     "distDir": "../dist",

nym-wallet/src/requests/index.ts

@@ -8,3 +8,4 @@ export * from './utils';
 export * from './delegation';
 export * from './rewards';
 export * from './simulate';
+export * from './signature';

nym-wallet/src/requests/signature.ts

@@ -0,0 +1,9 @@
+import { invokeWrapper } from './wrapper';
+
+export const sign = async (message: string): Promise<string> => invokeWrapper<string>('sign', { message });
+
+export const verify = async (
+  signatureAsHex: string,
+  message: string,
+  publicKeyAsJsonOrAccountAddress?: string | null,
+): Promise<string> => invokeWrapper<string>('verify', { publicKeyAsJsonOrAccountAddress, signatureAsHex, message });

nym-wallet/wallet-recovery-cli/Cargo.toml

@@ -11,7 +11,7 @@ anyhow = "1.0"
 argon2 = "0.4"
 base64 = "0.13"
 bip39 = "1.0"
-clap = { version = "3.0", features = ["derive"] }
+clap = { version = "3.2.20", features = ["derive"] }
 log = "0.4"
 pretty_env_logger = "0.4"
 serde_json = "1.0.0"

nym-wallet/wallet-recovery-cli/src/main.rs

@@ -32,15 +32,15 @@ enum DecryptedAccount {
 struct Args {
     /// Password used to attempt to decrypt the logins found in the file. The option can be
     /// provided multiple times for files that require multiple passwords.
-    #[clap(short, long, min_values(1), required = true)]
+    #[clap(short, long, value_parser, required = true)]
     password: Vec<String>,
 
     /// Path to the wallet file that will be decrypted.
-    #[clap(short, long)]
+    #[clap(short, long, value_parser)]
     file: String,
 
     /// Raw mode. Skips trying to parse the decrypted content.
-    #[clap(short, long)]
+    #[clap(short, long, action)]
     raw: bool,
 }
 

service-providers/network-requester/Cargo.toml

@@ -34,4 +34,5 @@ ordered-buffer = {path = "../../common/socks5/ordered-buffer"}
 proxy-helpers = { path = "../../common/socks5/proxy-helpers" }
 socks5-requests = { path = "../../common/socks5/requests" }
 statistics-common = { path = "../../common/statistics" }
+task = { path = "../../common/task" }
 websocket-requests = { path = "../../clients/native/websocket-requests" }

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

@@ -7,6 +7,7 @@ use proxy_helpers::connection_controller::ConnectionReceiver;
 use proxy_helpers::proxy_runner::ProxyRunner;
 use socks5_requests::{ConnectionId, Message as Socks5Message, RemoteAddress, Response};
 use std::io;
+use task::ShutdownListener;
 use tokio::net::TcpStream;
 
 /// A TCP connection between the Socks5 service provider, which makes
@@ -40,6 +41,7 @@ impl Connection {
         &mut self,
         mix_receiver: ConnectionReceiver,
         mix_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>,
+        shutdown: ShutdownListener,
     ) {
         let stream = self.conn.take().unwrap();
         let remote_source_address = "???".to_string(); // we don't know ip address of requester
@@ -52,6 +54,7 @@ impl Connection {
             mix_receiver,
             mix_sender,
             connection_id,
+            shutdown,
         )
         .run(move |conn_id, read_data, socket_closed| {
             (

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

@@ -19,6 +19,7 @@ use socks5_requests::{
 use statistics_common::collector::StatisticsSender;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use task::ShutdownListener;
 use tokio_tungstenite::tungstenite::protocol::Message;
 use websocket::WebsocketConnectionError;
 use websocket_requests::{requests::ClientRequest, responses::ServerResponse};
@@ -134,6 +135,7 @@ impl ServiceProvider {
         return_address: Recipient,
         controller_sender: ControllerSender,
         mix_input_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>,
+        shutdown: ShutdownListener,
     ) {
         let mut conn = match Connection::new(conn_id, remote_addr.clone(), return_address).await {
             Ok(conn) => conn,
@@ -170,7 +172,8 @@ impl ServiceProvider {
         );
 
         // run the proxy on the connection
-        conn.run_proxy(mix_receiver, mix_input_sender).await;
+        conn.run_proxy(mix_receiver, mix_input_sender, shutdown)
+            .await;
 
         // proxy is done - remove the access channel from the controller
         controller_sender
@@ -192,6 +195,7 @@ impl ServiceProvider {
         conn_id: ConnectionId,
         remote_addr: String,
         return_address: Recipient,
+        shutdown: ShutdownListener,
     ) {
         if !self.open_proxy && !self.outbound_request_filter.check(&remote_addr) {
             let log_msg = format!("Domain {:?} failed filter check", remote_addr);
@@ -218,6 +222,7 @@ impl ServiceProvider {
                 return_address,
                 controller_sender_clone,
                 mix_input_sender_clone,
+                shutdown,
             )
             .await
         });
@@ -241,6 +246,7 @@ impl ServiceProvider {
         controller_sender: &mut ControllerSender,
         mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>,
         stats_collector: Option<ServiceStatisticsCollector>,
+        shutdown: ShutdownListener,
     ) {
         let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) {
             Ok(msg) => msg,
@@ -265,6 +271,7 @@ impl ServiceProvider {
                         req.conn_id,
                         req.remote_addr,
                         req.return_address,
+                        shutdown,
                     )
                 }
 
@@ -302,8 +309,12 @@ impl ServiceProvider {
         let (mix_input_sender, mix_input_receiver) =
             mpsc::unbounded::<(Socks5Message, Recipient)>();
 
-        // controller for managing all active connections
-        let (mut active_connections_controller, mut controller_sender) = Controller::new();
+        // Controller for managing all active connections.
+        // We provide it with a ShutdownListener since it requires it, even though for the network
+        // requester shutdown signalling is not yet fully implemented.
+        let shutdown = task::ShutdownNotifier::default();
+        let (mut active_connections_controller, mut controller_sender) =
+            Controller::new(shutdown.subscribe());
         tokio::spawn(async move {
             active_connections_controller.run().await;
         });
@@ -353,6 +364,7 @@ impl ServiceProvider {
                 &mut controller_sender,
                 &mix_input_sender,
                 stats_collector.clone(),
+                shutdown.subscribe(),
             )
             .await;
         }

validator-api/src/network_monitor/monitor/sender.rs

@@ -209,6 +209,7 @@ impl PacketSender {
         // currently we do not care about acks at all, but we must keep the channel alive
         // so that the gateway client would not crash
         let (ack_sender, ack_receiver) = mpsc::unbounded();
+
         let mut gateway_client = GatewayClient::new(
             address,
             Arc::clone(&fresh_gateway_client_data.local_identity),
@@ -219,6 +220,7 @@ impl PacketSender {
             ack_sender,
             fresh_gateway_client_data.gateway_response_timeout,
             Some(fresh_gateway_client_data.bandwidth_controller.clone()),
+            None,
         );
 
         gateway_client