wasm fix

* Idea for transparent ShutdownManager use

* Tracker hierarchies

* Fix wasm shutdown, convinience shutdown method

.github/workflows/ci-check-ns-api-version.yml

@@ -16,7 +16,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/ci-check-nym-stats-api-version.yml

@@ -16,7 +16,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/greetings.yml

@@ -6,7 +6,7 @@ jobs:
   greeting:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/first-interaction@v1
+    - uses: actions/first-interaction@v3
       with:
         repo-token: ${{ secrets.GITHUB_TOKEN }}
         issue-message: 'Thank you for raising this issue'

.github/workflows/nightly-security-audit.yml

@@ -31,7 +31,7 @@ jobs:
       - name: Check out repository code
         uses: actions/checkout@v4
       - name: Download report from previous job
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v5
         with:
           name: report
           path: .github/workflows/support-files/notifications

.github/workflows/publish-nym-binaries.yml

@@ -30,13 +30,11 @@ jobs:
       release_date: ${{ fromJSON(steps.create-release.outputs.assets)[0].published_at }}
       client_hash: ${{ steps.binary-hashes.outputs.client_hash }}
       nymvisor_hash: ${{ steps.binary-hashes.outputs.nymvisor_hash }}
-      nymnode_hash: ${{ steps.binary-hashes.outputs.nymnode_hash }}
       socks5_hash: ${{ steps.binary-hashes.outputs.socks5_hash }}
       netreq_hash: ${{ steps.binary-hashes.outputs.netreq_hash }}
       cli_hash: ${{ steps.binary-hashes.outputs.cli_hash }}
       client_version: ${{ steps.binary-versions.outputs.client_version }}
       nymvisor_version: ${{ steps.binary-versions.outputs.nymvisor_version }}
-      nymnode_version: ${{ steps.binary-versions.outputs.nymnode_version }}
       socks5_version: ${{ steps.binary-versions.outputs.socks5_version }}
       netreq_version: ${{ steps.binary-versions.outputs.netreq_version }}
       cli_version: ${{ steps.binary-versions.outputs.cli_version }}
@@ -56,7 +54,7 @@ jobs:
       - name: Install Rust stable
         uses: actions-rs/toolchain@v1
         with:
-          toolchain: 1.86.0
+          toolchain: 1.88.0
           override: true
 
       - name: Build all binaries
@@ -76,7 +74,6 @@ jobs:
             target/release/nym-network-requester
             target/release/nym-cli
             target/release/nymvisor
-            target/release/nym-node
           retention-days: 30
     
       - id: create-release
@@ -91,7 +88,6 @@ jobs:
             target/release/nym-network-requester
             target/release/nym-cli
             target/release/nymvisor
-            target/release/nym-node
 
   push-release-data-client:
     if: ${{ (startsWith(github.ref, 'refs/tags/nym-binaries-') && github.event_name == 'release') || github.event_name == 'workflow_dispatch' }}

.github/workflows/publish-nyms5-android-apk.yml

@@ -25,7 +25,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Install Java
-        uses: actions/setup-java@v4
+        uses: actions/setup-java@v5
         with:
           distribution: "temurin"
           java-version: "17"
@@ -91,7 +91,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Download binary artifact
-        uses: actions/download-artifact@v4
+        uses: actions/download-artifact@v5
         with:
           name: nyms5-apk-arch64
           path: apk

.github/workflows/push-credential-proxy.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/nym-credential-proxy/Cargo.toml

.github/workflows/push-data-observatory.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/push-network-monitor.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/nym-network-monitor/Cargo.toml

.github/workflows/push-node-status-api.yaml

@@ -34,18 +34,22 @@ jobs:
       - name: Get version from cargo.toml
         id: get_version
         run: |
-          yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml
+          VERSION=$(yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml)
+          echo "result=$VERSION" >> $GITHUB_OUTPUT
 
       - name: Set GIT_TAG variable
         run: echo "GIT_TAG=${{ env.WORKING_DIRECTORY }}-${{ steps.get_version.outputs.result }}" >> $GITHUB_ENV
 
-      - name: Set RELEASE_TAG variable
+      - name: Initialise RELEASE_TAG
+        run: echo "RELEASE_TAG=" >> $GITHUB_ENV
+      
+      - name: Set RELEASE_TAG for release
         if: github.event.inputs.release_image == 'true'
         run: echo "RELEASE_TAG=golden-" >> $GITHUB_ENV
-
+      
       - name: Set IMAGE_NAME_AND_TAGS variable
         run: echo "IMAGE_NAME_AND_TAGS=${{ env.CONTAINER_NAME }}:${{ env.RELEASE_TAG }}${{ steps.get_version.outputs.result }}" >> $GITHUB_ENV
-
+        
       - name: New env vars
         run: echo "RELEASE_TAG='$RELEASE_TAG' GIT_TAG='$GIT_TAG' IMAGE_NAME_AND_TAGS='$IMAGE_NAME_AND_TAGS'"        
 
@@ -65,6 +69,6 @@ jobs:
 
       - name: BuildAndPushImageOnHarbor
         run: |
-          docker build -f ${{ env.WORKING_DIRECTORY }}/Dockerfile . -t harbor.nymte.ch/nym/${{ env.IMAGE_NAME_AND_TAGS }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
+          docker build -f ${{ env.WORKING_DIRECTORY }}/Dockerfile-pg . -t harbor.nymte.ch/nym/${{ env.IMAGE_NAME_AND_TAGS }} -t harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }}:latest
           docker push harbor.nymte.ch/nym/${{ env.CONTAINER_NAME }} --all-tags
 

.github/workflows/push-nym-api.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/nym-api/Cargo.toml

.github/workflows/push-nym-node.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/push-nym-statistics-api.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/push-nyx-chain-watcher.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

.github/workflows/push-validator-rewarder.yaml

@@ -26,7 +26,7 @@ jobs:
           git config --global user.name "Lawrence Stalder"
 
       - name: Get version from cargo.toml
-        uses: mikefarah/yq@v4.45.4
+        uses: mikefarah/yq@v4.47.1
         id: get_version
         with:
           cmd: yq -oy '.package.version' ${{ env.WORKING_DIRECTORY }}/Cargo.toml

CHANGELOG.md

@@ -4,6 +4,48 @@ Post 1.0.0 release, the changelog format is based on [Keep a Changelog](https://
 
 ## [Unreleased]
 
+## [2025.16-halloumi] (2025-09-16)
+
+- Backport metadata endpoint ([#6010])
+- bugfix: make sure tables are removed in correct order to not trigger FK constraint issue ([#5987])
+- chore: move authenticator into gateway crate ([#5982])
+- Fix the ns api ci workflow ([#5981])
+- Remove freshness check on testrun submit ([#5977])
+- Update sysinfo to the latest ([#5976])
+- bugfix: manually calculate per node work on rewarded set changes ([#5972])
+- fixing the ci for ns agent ([#5965])
+- Feature/testing utils ([#5963])
+- bugfix: fix ci-build for linux (and use updated runner) ([#5958])
+- chore: updated refs to cheddar rev of nym repo ([#5955])
+- http api client adjustment ([#5953])
+- chore: fix rust 1.89 clippy issues ([#5944])
+- Wireguard metadata client library ([#5943])
+- chore: remove unused import ([#5942])
+- feat: introduce additional checks when attempting to send to bounded channels ([#5941])
+- Move credential verifier in peer controller ([#5938])
+- change PK/FK on expiration date signatures tables ([#5934])
+- Wireguard private metadata ([#5915])
+
+[#6010]: https://github.com/nymtech/nym/pull/6010
+[#5987]: https://github.com/nymtech/nym/pull/5987
+[#5982]: https://github.com/nymtech/nym/pull/5982
+[#5981]: https://github.com/nymtech/nym/pull/5981
+[#5977]: https://github.com/nymtech/nym/pull/5977
+[#5976]: https://github.com/nymtech/nym/pull/5976
+[#5972]: https://github.com/nymtech/nym/pull/5972
+[#5965]: https://github.com/nymtech/nym/pull/5965
+[#5963]: https://github.com/nymtech/nym/pull/5963
+[#5958]: https://github.com/nymtech/nym/pull/5958
+[#5955]: https://github.com/nymtech/nym/pull/5955
+[#5953]: https://github.com/nymtech/nym/pull/5953
+[#5944]: https://github.com/nymtech/nym/pull/5944
+[#5943]: https://github.com/nymtech/nym/pull/5943
+[#5942]: https://github.com/nymtech/nym/pull/5942
+[#5941]: https://github.com/nymtech/nym/pull/5941
+[#5938]: https://github.com/nymtech/nym/pull/5938
+[#5934]: https://github.com/nymtech/nym/pull/5934
+[#5915]: https://github.com/nymtech/nym/pull/5915
+
 ## [2025.15-gruyere] (2025-08-20)
 
 - Migrate strum to 0.27.2 ([#5960])

Cargo.toml

@@ -43,6 +43,7 @@ members = [
     "common/cosmwasm-smart-contracts/nym-performance-contract",
     "common/cosmwasm-smart-contracts/nym-pool-contract",
     "common/cosmwasm-smart-contracts/vesting-contract",
+    "common/credential-proxy",
     "common/credential-storage",
     "common/credential-utils",
     "common/credential-verification",
@@ -53,12 +54,11 @@ members = [
     "common/ecash-signer-check",
     "common/ecash-signer-check-types",
     "common/ecash-time",
-    "common/execute",
     "common/exit-policy",
     "common/gateway-requests",
     "common/gateway-stats-storage",
     "common/gateway-storage",
-    "common/http-api-client",
+    "common/http-api-client", "common/http-api-client-macro",
     "common/http-api-common",
     "common/inclusion-probability",
     "common/ip-packet-requests",
@@ -96,7 +96,7 @@ members = [
     "common/ticketbooks-merkle",
     "common/topology",
     "common/tun",
-    "common/types",
+    "common/types", "common/upgrade-mode-check",
     "common/verloc",
     "common/wasm/client-core",
     "common/wasm/storage",
@@ -112,10 +112,12 @@ members = [
     "gateway",
     "nym-api",
     "nym-api/nym-api-requests",
+    "nym-authenticator-client",
     "nym-browser-extension/storage",
     "nym-credential-proxy/nym-credential-proxy",
     "nym-credential-proxy/nym-credential-proxy-requests",
     "nym-credential-proxy/vpn-api-lib-wasm",
+    "nym-ip-packet-client",
     "nym-network-monitor",
     "nym-node",
     "nym-node-status-api/nym-node-status-agent",
@@ -123,9 +125,10 @@ members = [
     "nym-node-status-api/nym-node-status-client",
     "nym-node/nym-node-metrics",
     "nym-node/nym-node-requests",
-    "nym-outfox",
+    "nym-outfox", "nym-signers-monitor",
     "nym-statistics-api",
     "nym-validator-rewarder",
+    "nym-wg-gateway-client",
     "nyx-chain-watcher",
     "sdk/ffi/cpp",
     "sdk/ffi/go",
@@ -223,7 +226,7 @@ clap_complete = "4.5"
 clap_complete_fig = "4.5"
 colored = "2.2"
 comfy-table = "7.1.4"
-console = "0.15.11"
+console = "0.16.0"
 console-subscriber = "0.4.1"
 console_error_panic_hook = "0.1"
 const-str = "0.5.6"
@@ -270,11 +273,13 @@ humantime = "2.2.0"
 humantime-serde = "1.1.1"
 hyper = "1.6.0"
 hyper-util = "0.1"
-indicatif = "0.17.11"
+indicatif = "0.18.0"
 inquire = "0.6.2"
+inventory = "0.3.21"
 ip_network = "0.4.1"
 ipnetwork = "0.20"
 itertools = "0.14.0"
+jwt-simple = { version = "0.12.12", default-features = false, features = ["pure-rust"] }
 k256 = "0.13"
 lazy_static = "1.5.0"
 ledger-transport = "0.10.0"
@@ -318,6 +323,7 @@ serde_json_path = "0.7.2"
 serde_repr = "0.1"
 serde_with = "3.9.0"
 serde_yaml = "0.9.25"
+serde_plain = "1.0.2"
 sha2 = "0.10.9"
 si-scale = "0.2.3"
 snow = "0.9.6"
@@ -326,14 +332,15 @@ sqlx = "0.8.6"
 strum = "0.27.2"
 strum_macros = "0.27.2"
 subtle-encoding = "0.5"
-syn = "1"
+syn = "2"
 sysinfo = "0.37.0"
 tap = "1.0.1"
 tar = "0.4.44"
+test-with = { version = "0.15.4", default-features = false }
 tempfile = "3.20"
 thiserror = "2.0"
 time = "0.3.41"
-tokio = "1.45"
+tokio = "1.47"
 tokio-postgres = "0.7"
 tokio-stream = "0.1.17"
 tokio-test = "0.4.4"

clients/native/Cargo.toml

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

clients/native/src/client/mod.rs

@@ -11,7 +11,7 @@ use nym_client_core::client::base_client::{
     BaseClientBuilder, ClientInput, ClientOutput, ClientState,
 };
 use nym_sphinx::params::PacketType;
-use nym_task::TaskHandle;
+use nym_task::ShutdownManager;
 use nym_validator_client::QueryHttpRpcNyxdClient;
 use std::error::Error;
 use std::path::PathBuf;
@@ -29,6 +29,8 @@ pub struct SocketClient {
 
     /// Optional path to a .json file containing standalone network details.
     custom_mixnet: Option<PathBuf>,
+
+    shutdown_manager: ShutdownManager,
 }
 
 impl SocketClient {
@@ -40,6 +42,7 @@ impl SocketClient {
         SocketClient {
             config,
             custom_mixnet,
+            shutdown_manager: Default::default(),
         }
     }
 
@@ -49,7 +52,7 @@ impl SocketClient {
         client_output: ClientOutput,
         client_state: ClientState,
         self_address: &Recipient,
-        task_client: nym_task::TaskClient,
+        shutdown_token: nym_task::ShutdownToken,
         packet_type: PacketType,
     ) {
         info!("Starting websocket listener...");
@@ -77,24 +80,24 @@ impl SocketClient {
             shared_lane_queue_lengths,
             reply_controller_sender,
             Some(packet_type),
-            task_client.fork("websocket_handler"),
+            shutdown_token.clone(),
         );
 
         websocket::Listener::new(
             config.socket.host,
             config.socket.listening_port,
-            task_client.with_suffix("websocket_listener"),
+            shutdown_token.child_token(),
         )
         .start(websocket_handler);
     }
 
     /// blocking version of `start_socket` method. Will run forever (or until SIGINT is sent)
     pub async fn run_socket_forever(self) -> Result<(), Box<dyn Error + Send + Sync>> {
-        let shutdown = self.start_socket().await?;
+        let mut shutdown = self.start_socket().await?;
 
-        let res = shutdown.wait_for_shutdown().await;
+        shutdown.run_until_shutdown().await;
         log::info!("Stopping nym-client");
-        res
+        Ok(())
     }
 
     async fn initialise_storage(&self) -> Result<OnDiskPersistent, ClientError> {
@@ -119,6 +122,7 @@ impl SocketClient {
 
         let mut base_client =
             BaseClientBuilder::new(self.config().base(), storage, dkg_query_client)
+                .with_shutdown(self.shutdown_manager.shutdown_tracker_owned())
                 .with_user_agent(user_agent);
 
         if let Some(custom_mixnet) = &self.custom_mixnet {
@@ -128,7 +132,7 @@ impl SocketClient {
         Ok(base_client)
     }
 
-    pub async fn start_socket(self) -> Result<TaskHandle, ClientError> {
+    pub async fn start_socket(self) -> Result<ShutdownManager, ClientError> {
         if !self.config.socket.socket_type.is_websocket() {
             return Err(ClientError::InvalidSocketMode);
         }
@@ -147,13 +151,13 @@ impl SocketClient {
             client_output,
             client_state,
             &self_address,
-            started_client.task_handle.get_handle(),
+            self.shutdown_manager.child_shutdown_token(),
             packet_type,
         );
 
         info!("Client startup finished!");
         info!("The address of this client is: {self_address}");
 
-        Ok(started_client.task_handle)
+        Ok(self.shutdown_manager)
     }
 }

clients/native/src/websocket/handler.rs

@@ -19,7 +19,7 @@ use nym_sphinx::receiver::ReconstructedMessage;
 use nym_task::connections::{
     ConnectionCommand, ConnectionCommandSender, ConnectionId, LaneQueueLengths, TransmissionLane,
 };
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use std::time::Duration;
 use tokio::net::TcpStream;
 use tokio::time::Instant;
@@ -44,7 +44,7 @@ pub(crate) struct HandlerBuilder {
     lane_queue_lengths: LaneQueueLengths,
     reply_controller_sender: ReplyControllerSender,
     packet_type: Option<PacketType>,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl HandlerBuilder {
@@ -57,7 +57,7 @@ impl HandlerBuilder {
         lane_queue_lengths: LaneQueueLengths,
         reply_controller_sender: ReplyControllerSender,
         packet_type: Option<PacketType>,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         Self {
             msg_input,
@@ -67,14 +67,13 @@ impl HandlerBuilder {
             lane_queue_lengths,
             reply_controller_sender,
             packet_type,
-            task_client,
+            shutdown_token,
         }
     }
 
     // TODO: make sure we only ever have one active handler
     pub fn create_active_handler(&self) -> Handler {
-        let mut task_client = self.task_client.fork("active_handler");
-        task_client.disarm();
+        let shutdown_token = self.shutdown_token.clone();
         Handler {
             msg_input: self.msg_input.clone(),
             client_connection_tx: self.client_connection_tx.clone(),
@@ -85,7 +84,7 @@ impl HandlerBuilder {
             lane_queue_lengths: self.lane_queue_lengths.clone(),
             reply_controller_sender: self.reply_controller_sender.clone(),
             packet_type: self.packet_type,
-            task_client,
+            shutdown_token,
         }
     }
 }
@@ -100,19 +99,14 @@ pub(crate) struct Handler {
     lane_queue_lengths: LaneQueueLengths,
     reply_controller_sender: ReplyControllerSender,
     packet_type: Option<PacketType>,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl Drop for Handler {
     fn drop(&mut self) {
-        if let Err(err) = self
+        let _ = self
             .buffer_requester
-            .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect)
-        {
-            if !self.task_client.is_shutdown_poll() {
-                error!("failed to disconnect the receiver from the buffer: {err}");
-            }
-        }
+            .unbounded_send(ReceivedBufferMessage::ReceiverDisconnect);
     }
 }
 
@@ -142,7 +136,7 @@ impl Handler {
         {
             Ok(length) => length,
             Err(err) => {
-                if !self.task_client.is_shutdown_poll() {
+                if !self.shutdown_token.is_cancelled() {
                     error!(
                         "Failed to get reply queue length for connection {connection_id}: {err}"
                     );
@@ -192,7 +186,7 @@ impl Handler {
         // the ack control is now responsible for chunking, etc.
         let input_msg = InputMessage::new_regular(recipient, message, lane, self.packet_type);
         if let Err(err) = self.msg_input.send(input_msg).await {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send message to the input buffer: {err}");
             }
         }
@@ -225,7 +219,7 @@ impl Handler {
         let input_msg =
             InputMessage::new_anonymous(recipient, message, reply_surbs, lane, self.packet_type);
         if let Err(err) = self.msg_input.send(input_msg).await {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send anonymous message to the input buffer: {err}");
             }
         }
@@ -253,7 +247,7 @@ impl Handler {
 
         let input_msg = InputMessage::new_reply(recipient_tag, message, lane, self.packet_type);
         if let Err(err) = self.msg_input.send(input_msg).await {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send reply message to the input buffer: {err}");
             }
         }
@@ -275,7 +269,7 @@ impl Handler {
             .client_connection_tx
             .unbounded_send(ConnectionCommand::Close(connection_id))
         {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send close connection command: {err}");
             }
         }
@@ -394,11 +388,14 @@ impl Handler {
     }
 
     async fn listen_for_requests(&mut self, mut msg_receiver: ReconstructedMessagesReceiver) {
-        let mut task_client = self.task_client.fork("select");
-        task_client.disarm();
+        let shutdown_token = self.shutdown_token.clone();
 
-        while !task_client.is_shutdown() {
+        loop {
             tokio::select! {
+                  _ = shutdown_token.cancelled() => {
+                    log::trace!("Websocket handler: Received shutdown");
+                    break;
+                }
                 // we can either get a client request from the websocket
                 socket_msg = self.next_websocket_request() => {
                     if socket_msg.is_none() {
@@ -436,9 +433,6 @@ impl Handler {
                         break;
                     }
                 }
-                _ = task_client.recv() => {
-                    log::trace!("Websocket handler: Received shutdown");
-                }
             }
         }
         log::debug!("Websocket handler: Exiting");
@@ -464,7 +458,7 @@ impl Handler {
                     reconstructed_sender,
                 ))
         {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("failed to announce the receiver to the buffer: {err}");
             }
         }

clients/native/src/websocket/listener.rs

@@ -3,7 +3,7 @@
 
 use super::handler::HandlerBuilder;
 use log::*;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use std::net::IpAddr;
 use std::{net::SocketAddr, process, sync::Arc};
 use tokio::io::AsyncWriteExt;
@@ -23,15 +23,15 @@ impl State {
 pub(crate) struct Listener {
     address: SocketAddr,
     state: State,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl Listener {
-    pub(crate) fn new(host: IpAddr, port: u16, task_client: TaskClient) -> Self {
+    pub(crate) fn new(host: IpAddr, port: u16, shutdown_token: ShutdownToken) -> Self {
         Listener {
             address: SocketAddr::new(host, port),
             state: State::AwaitingConnection,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -46,11 +46,11 @@ impl Listener {
 
         let notify = Arc::new(Notify::new());
 
-        while !self.task_client.is_shutdown() {
+        while !self.shutdown_token.is_cancelled() {
             tokio::select! {
                 // When the handler finishes we check if shutdown is signalled
                 _ = notify.notified() => {
-                    if self.task_client.is_shutdown() {
+                    if self.shutdown_token.is_cancelled() {
                         log::trace!("Websocket listener: detected shutdown after connection closed");
                         break;
                     }
@@ -59,7 +59,7 @@ impl Listener {
                 }
                 // ... but when there is no connected client at the time of shutdown being
                 // signalled, we handle it here.
-                _ = self.task_client.recv() => {
+                _ = self.shutdown_token.cancelled() => {
                     if !self.state.is_connected() {
                         log::trace!("Not connected: shutting down");
                         break;

clients/socks5/Cargo.toml

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

common/bandwidth-controller/src/utils.rs

@@ -13,7 +13,7 @@ use nym_credentials_interface::{
 };
 use nym_ecash_time::Date;
 use nym_validator_client::coconut::all_ecash_api_clients;
-use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
 use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
 use nym_validator_client::EcashApiClient;
 use rand::prelude::SliceRandom;

common/client-core/Cargo.toml

@@ -53,6 +53,7 @@ nym-client-core-config-types = { path = "./config-types", features = [
 nym-client-core-surb-storage = { path = "./surb-storage" }
 nym-client-core-gateways-storage = { path = "./gateways-storage" }
 nym-ecash-time = { path = "../ecash-time" }
+nym-mixnet-contract-common = { path = "../cosmwasm-smart-contracts/mixnet-contract" }
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies]
 nym-mixnet-client = { path = "../client-libs/mixnet-client", default-features = false }

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

@@ -27,13 +27,13 @@ use crate::client::topology_control::nym_api_provider::NymApiTopologyProvider;
 use crate::client::topology_control::{
     TopologyAccessor, TopologyRefresher, TopologyRefresherConfig,
 };
+use crate::config;
 use crate::config::{Config, DebugConfig};
 use crate::error::ClientCoreError;
 use crate::init::{
     setup_gateway,
     types::{GatewaySetup, InitialisationResult},
 };
-use crate::{config, spawn_future};
 use futures::channel::mpsc;
 use nym_bandwidth_controller::BandwidthController;
 use nym_client_core_config_types::{ForgetMe, RememberMe};
@@ -48,16 +48,15 @@ use nym_gateway_client::{
 use nym_sphinx::acknowledgements::AckKey;
 use nym_sphinx::addressing::clients::Recipient;
 use nym_sphinx::addressing::nodes::NodeIdentity;
-use nym_sphinx::params::PacketType;
 use nym_sphinx::receiver::{ReconstructedMessage, SphinxMessageReceiver};
 use nym_statistics_common::clients::ClientStatsSender;
 use nym_statistics_common::generate_client_stats_id;
 use nym_task::connections::{ConnectionCommandReceiver, ConnectionCommandSender, LaneQueueLengths};
-use nym_task::{TaskClient, TaskHandle};
+use nym_task::ShutdownTracker;
 use nym_topology::provider_trait::TopologyProvider;
 use nym_topology::HardcodedTopologyProvider;
 use nym_validator_client::nym_api::NymApiClientExt;
-use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, NymApiClient, UserAgent};
+use nym_validator_client::{nyxd::contract_traits::DkgQueryClient, UserAgent};
 use rand::prelude::SliceRandom;
 use rand::rngs::OsRng;
 use rand::thread_rng;
@@ -95,7 +94,6 @@ impl ClientInput {
     }
 }
 
-#[derive(Clone)]
 pub struct ClientOutput {
     pub received_buffer_request_sender: ReceivedBufferRequestSender,
 }
@@ -195,7 +193,7 @@ pub struct BaseClientBuilder<C, S: MixnetClientStorage> {
     wait_for_gateway: bool,
     custom_topology_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
     custom_gateway_transceiver: Option<Box<dyn GatewayTransceiver + Send>>,
-    shutdown: Option<TaskClient>,
+    shutdown: Option<ShutdownTracker>,
     user_agent: Option<UserAgent>,
 
     setup_method: GatewaySetup,
@@ -281,7 +279,7 @@ where
     }
 
     #[must_use]
-    pub fn with_shutdown(mut self, shutdown: TaskClient) -> Self {
+    pub fn with_shutdown(mut self, shutdown: ShutdownTracker) -> Self {
         self.shutdown = Some(shutdown);
         self
     }
@@ -325,11 +323,11 @@ where
         topology_accessor: TopologyAccessor,
         mix_tx: BatchMixMessageSender,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) {
         info!("Starting loop cover traffic stream...");
 
-        let stream = LoopCoverTrafficStream::new(
+        let mut stream = LoopCoverTrafficStream::new(
             ack_key,
             debug_config.acknowledgements.average_ack_delay,
             mix_tx,
@@ -338,10 +336,9 @@ where
             debug_config.traffic,
             debug_config.cover_traffic,
             stats_tx,
-            task_client,
         );
-
-        stream.start();
+        shutdown_tracker
+            .try_spawn_named_with_shutdown(async move { stream.run().await }, "CoverTrafficStream");
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -357,13 +354,12 @@ where
         reply_controller_receiver: ReplyControllerReceiver,
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
-        task_client: TaskClient,
-        packet_type: PacketType,
         stats_tx: ClientStatsSender,
+        shutdown_tracker: &ShutdownTracker,
     ) {
         info!("Starting real traffic stream...");
 
-        RealMessagesController::new(
+        let real_messages_controller = RealMessagesController::new(
             controller_config,
             key_rotation_config,
             ack_receiver,
@@ -376,9 +372,63 @@ where
             lane_queue_lengths,
             client_connection_rx,
             stats_tx,
-            task_client,
-        )
-        .start(packet_type);
+            shutdown_tracker.clone_shutdown_token(),
+        );
+
+        // break out all the subtasks
+        let (mut out_queue_control, mut reply_controller, ack_controller) =
+            real_messages_controller.into_tasks();
+        let (
+            mut ack_listener,
+            mut input_listener,
+            mut retransmission_listener,
+            mut sent_notification_listener,
+            mut ack_action_controller,
+        ) = ack_controller.into_tasks();
+
+        shutdown_tracker.try_spawn_named(
+            async move { out_queue_control.run().await },
+            "RealMessagesController::OutQueueControl",
+        );
+
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move { reply_controller.run(shutdown_token).await },
+            "RealMessagesController::ReplyController",
+        );
+
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move { ack_listener.run(shutdown_token).await },
+            "AcknowledgementController::AcknowledgementListener",
+        );
+
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move { input_listener.run(shutdown_token).await },
+            "AcknowledgementController::InputMessageListener",
+        );
+
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move { retransmission_listener.run(shutdown_token).await },
+            "AcknowledgementController::RetransmissionRequestListener",
+        );
+
+        shutdown_tracker.try_spawn_named_with_shutdown(
+            async move {
+                sent_notification_listener.run().await;
+            },
+            "AcknowledgementController::SentNotificationListener",
+        );
+
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move { ack_action_controller.run(shutdown_token).await },
+            "AcknowledgementController::ActionController",
+        );
+
+        // .start(packet_type);
     }
 
     // buffer controlling all messages fetched from provider
@@ -389,21 +439,29 @@ where
         mixnet_receiver: MixnetMessageReceiver,
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
-        shutdown: TaskClient,
         metrics_reporter: ClientStatsSender,
+        shutdown_tracker: &ShutdownTracker,
     ) {
         info!("Starting received messages buffer controller...");
-        let controller: ReceivedMessagesBufferController<SphinxMessageReceiver> =
-            ReceivedMessagesBufferController::new(
-                local_encryption_keypair,
-                query_receiver,
-                mixnet_receiver,
-                reply_key_storage,
-                reply_controller_sender,
-                metrics_reporter,
-                shutdown,
-            );
-        controller.start()
+        let controller = ReceivedMessagesBufferController::<SphinxMessageReceiver>::new(
+            local_encryption_keypair,
+            query_receiver,
+            mixnet_receiver,
+            reply_key_storage,
+            reply_controller_sender,
+            metrics_reporter,
+            shutdown_tracker.clone_shutdown_token(),
+        );
+        let (mut msg_receiver, mut req_receiver) = controller.into_tasks();
+
+        shutdown_tracker.try_spawn_named(
+            async move { msg_receiver.run().await },
+            "ReceivedMessagesBufferController::FragmentedMessageReceiver",
+        );
+        shutdown_tracker.try_spawn_named(
+            async move { req_receiver.run().await },
+            "ReceivedMessagesBufferController::RequestReceiver",
+        );
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -415,7 +473,7 @@ where
         packet_router: PacketRouter,
         stats_reporter: ClientStatsSender,
         #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
-        shutdown: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> Result<GatewayClient<C, S::CredentialStore>, ClientCoreError>
     where
         <S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -434,7 +492,7 @@ where
                     packet_router,
                     bandwidth_controller,
                     stats_reporter,
-                    shutdown,
+                    shutdown_tracker.clone_shutdown_token(),
                 )
             } else {
                 let cfg = GatewayConfig::new(
@@ -459,7 +517,7 @@ where
                     stats_reporter,
                     #[cfg(unix)]
                     connection_fd_callback,
-                    shutdown,
+                    shutdown_tracker.clone_shutdown_token(),
                 )
             };
 
@@ -522,7 +580,7 @@ where
         packet_router: PacketRouter,
         stats_reporter: ClientStatsSender,
         #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
-        mut shutdown: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> Result<Box<dyn GatewayTransceiver + Send>, ClientCoreError>
     where
         <S::KeyStore as KeyStore>::StorageError: Send + Sync + 'static,
@@ -539,7 +597,6 @@ where
                 Err(ClientCoreError::CustomGatewaySelectionExpected)
             } else {
                 // and make sure to invalidate the task client, so we wouldn't cause premature shutdown
-                shutdown.disarm();
                 custom_gateway_transceiver.set_packet_router(packet_router)?;
                 Ok(custom_gateway_transceiver)
             };
@@ -555,7 +612,7 @@ where
             stats_reporter,
             #[cfg(unix)]
             connection_fd_callback,
-            shutdown,
+            shutdown_tracker,
         )
         .await?;
 
@@ -566,7 +623,7 @@ where
         custom_provider: Option<Box<dyn TopologyProvider + Send + Sync>>,
         config_topology: config::Topology,
         nym_api_urls: Vec<Url>,
-        nym_api_client: NymApiClient,
+        nym_api_client: nym_http_api_client::Client,
     ) -> Box<dyn TopologyProvider + Send + Sync> {
         // if no custom provider was ... provided ..., create one using nym-api
         custom_provider.unwrap_or_else(|| {
@@ -586,22 +643,20 @@ where
         topology_accessor: TopologyAccessor,
         local_gateway: NodeIdentity,
         wait_for_gateway: bool,
-        mut task_client: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> Result<(), ClientCoreError> {
         let topology_refresher_config =
             TopologyRefresherConfig::new(topology_config.topology_refresh_rate);
 
         if topology_config.disable_refreshing {
             // if we're not spawning the refresher, don't cause shutdown immediately
-            info!("The background topology refesher is not going to be started");
-            task_client.disarm();
+            info!("The background topology refresher is not going to be started");
         }
 
         let mut topology_refresher = TopologyRefresher::new(
             topology_refresher_config,
             topology_accessor,
             topology_provider,
-            task_client,
         );
         // before returning, block entire runtime to refresh the current network view so that any
         // components depending on topology would see a non-empty view
@@ -646,7 +701,10 @@ where
             // don't spawn the refresher if we don't want to be refreshing the topology.
             // only use the initial values obtained
             info!("Starting topology refresher...");
-            topology_refresher.start();
+            shutdown_tracker.try_spawn_named_with_shutdown(
+                async move { topology_refresher.run().await },
+                "TopologyRefresher",
+            );
         }
 
         Ok(())
@@ -657,7 +715,7 @@ where
         user_agent: Option<UserAgent>,
         client_stats_id: String,
         input_sender: Sender<InputMessage>,
-        task_client: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> ClientStatsSender {
         info!("Starting statistics control...");
         StatisticsControl::create_and_start(
@@ -667,18 +725,23 @@ where
                 .unwrap_or("unknown".to_string()),
             client_stats_id,
             input_sender.clone(),
-            task_client,
+            shutdown_tracker,
         )
     }
 
     fn start_mix_traffic_controller(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
-        shutdown: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> (BatchMixMessageSender, ClientRequestSender) {
         info!("Starting mix traffic controller...");
-        let (mix_traffic_controller, mix_tx, client_tx) =
-            MixTrafficController::new(gateway_transceiver, shutdown);
-        mix_traffic_controller.start();
+        let (mut mix_traffic_controller, mix_tx, client_tx) =
+            MixTrafficController::new(gateway_transceiver, shutdown_tracker.clone_shutdown_token());
+
+        shutdown_tracker.try_spawn_named(
+            async move { mix_traffic_controller.run().await },
+            "MixTrafficController",
+        );
+
         (mix_tx, client_tx)
     }
 
@@ -686,7 +749,7 @@ where
     async fn setup_persistent_reply_storage(
         backend: S::ReplyStore,
         key_rotation_config: KeyRotationConfig,
-        shutdown: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> Result<CombinedReplyStorage, ClientCoreError>
     where
         <S::ReplyStore as ReplyStorageBackend>::StorageError: Sync + Send,
@@ -711,13 +774,14 @@ where
             })?;
 
         let store_clone = mem_store.clone();
-        spawn_future!(
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
             async move {
                 persistent_storage
-                    .flush_on_shutdown(store_clone, shutdown)
+                    .flush_on_shutdown(store_clone, shutdown_token)
                     .await
             },
-            "PersistentReplyStorage::flush_on_shutdown"
+            "PersistentReplyStorage::flush_on_shutdown",
         );
 
         Ok(mem_store)
@@ -749,21 +813,29 @@ where
         setup_gateway(setup_method, key_store, details_store).await
     }
 
-    fn construct_nym_api_client(config: &Config, user_agent: Option<UserAgent>) -> NymApiClient {
+    fn construct_nym_api_client(
+        config: &Config,
+        user_agent: Option<UserAgent>,
+    ) -> Result<nym_http_api_client::Client, ClientCoreError> {
         let mut nym_api_urls = config.get_nym_api_endpoints();
         nym_api_urls.shuffle(&mut thread_rng());
 
+        let mut builder = nym_http_api_client::Client::builder(nym_api_urls[0].clone())
+            .map_err(ClientCoreError::from)?;
+
         if let Some(user_agent) = user_agent {
-            NymApiClient::new_with_user_agent(nym_api_urls[0].clone(), user_agent)
-        } else {
-            NymApiClient::new(nym_api_urls[0].clone())
+            builder = builder.with_user_agent(user_agent);
         }
+
+        builder = builder.with_bincode();
+
+        builder.build().map_err(ClientCoreError::from)
     }
 
     async fn determine_key_rotation_state(
-        client: &NymApiClient,
+        client: &nym_http_api_client::Client,
     ) -> Result<KeyRotationConfig, ClientCoreError> {
-        Ok(client.nym_api.get_key_rotation_info().await?.into())
+        Ok(client.get_key_rotation_info().await?.into())
     }
 
     pub async fn start_base(mut self) -> Result<BaseClient, ClientCoreError>
@@ -808,12 +880,12 @@ where
         let shared_topology_accessor =
             TopologyAccessor::new(self.config.debug.topology.ignore_egress_epoch_role);
 
-        // Shutdown notifier for signalling tasks to stop
-        let shutdown = self
-            .shutdown
-            .map(Into::<TaskHandle>::into)
-            .unwrap_or_default()
-            .name_if_unnamed("BaseNymClient");
+        // Create a shutdown tracker for this client - either as a child of provided tracker
+        // or get one from the registry
+        let shutdown_tracker = match self.shutdown {
+            Some(parent_tracker) => parent_tracker.child_tracker(),
+            None => nym_task::get_sdk_shutdown_tracker()?,
+        };
 
         // channels responsible for dealing with reply-related fun
         let (reply_controller_sender, reply_controller_receiver) =
@@ -830,7 +902,7 @@ where
             .dkg_query_client
             .map(|client| BandwidthController::new(credential_store, client));
 
-        let nym_api_client = Self::construct_nym_api_client(&self.config, self.user_agent.clone());
+        let nym_api_client = Self::construct_nym_api_client(&self.config, self.user_agent.clone())?;
         let key_rotation_config = Self::determine_key_rotation_state(&nym_api_client).await?;
 
         let topology_provider = Self::setup_topology_provider(
@@ -845,7 +917,7 @@ where
             self.user_agent.clone(),
             generate_client_stats_id(*self_address.identity()),
             input_sender.clone(),
-            shutdown.fork("statistics_control"),
+            &shutdown_tracker.child_tracker(),
         );
 
         // needs to be started as the first thing to block if required waiting for the gateway
@@ -855,14 +927,14 @@ where
             shared_topology_accessor.clone(),
             self_address.gateway(),
             self.wait_for_gateway,
-            shutdown.fork("topology_refresher"),
+            &shutdown_tracker.child_tracker(),
         )
         .await?;
 
         let gateway_packet_router = PacketRouter::new(
             ack_sender,
             mixnet_messages_sender,
-            shutdown.get_handle().named("gateway-packet-router"),
+            shutdown_tracker.clone_shutdown_token(),
         );
 
         let gateway_transceiver = Self::setup_gateway_transceiver(
@@ -875,7 +947,7 @@ where
             stats_reporter.clone(),
             #[cfg(unix)]
             self.connection_fd_callback,
-            shutdown.fork("gateway_transceiver"),
+            &shutdown_tracker.child_tracker(),
         )
         .await?;
         let gateway_ws_fd = gateway_transceiver.ws_fd();
@@ -883,7 +955,7 @@ where
         let reply_storage = Self::setup_persistent_reply_storage(
             reply_storage_backend,
             key_rotation_config,
-            shutdown.fork("persistent_reply_storage"),
+            &shutdown_tracker.child_tracker(),
         )
         .await?;
 
@@ -893,8 +965,8 @@ where
             mixnet_messages_receiver,
             reply_storage.key_storage(),
             reply_controller_sender.clone(),
-            shutdown.fork("received_messages_buffer"),
             stats_reporter.clone(),
+            &shutdown_tracker.child_tracker(),
         );
 
         // The message_sender is the transmitter for any component generating sphinx packets
@@ -904,7 +976,7 @@ where
 
         let (message_sender, client_request_sender) = Self::start_mix_traffic_controller(
             gateway_transceiver,
-            shutdown.fork("mix_traffic_controller"),
+            &shutdown_tracker.child_tracker(),
         );
 
         // Channels that the websocket listener can use to signal downstream to the real traffic
@@ -933,9 +1005,8 @@ where
             reply_controller_receiver,
             shared_lane_queue_lengths.clone(),
             client_connection_rx,
-            shutdown.fork("real_traffic_controller"),
-            self.config.debug.traffic.packet_type,
             stats_reporter.clone(),
+            &shutdown_tracker.child_tracker(),
         );
 
         if !self
@@ -951,7 +1022,7 @@ where
                 shared_topology_accessor.clone(),
                 message_sender,
                 stats_reporter.clone(),
-                shutdown.fork("cover_traffic_stream"),
+                &shutdown_tracker.child_tracker(),
             );
         }
 
@@ -979,7 +1050,7 @@ where
                 gateway_connection: GatewayConnection { gateway_ws_fd },
             },
             stats_reporter,
-            task_handle: shutdown,
+            shutdown_handle: shutdown_tracker, // The primary tracker for this client
             client_request_sender,
             forget_me: self.config.debug.forget_me,
             remember_me: self.config.debug.remember_me,
@@ -995,7 +1066,7 @@ pub struct BaseClient {
     pub client_state: ClientState,
     pub stats_reporter: ClientStatsSender,
     pub client_request_sender: ClientRequestSender,
-    pub task_handle: TaskHandle,
+    pub shutdown_handle: ShutdownTracker,
     pub forget_me: ForgetMe,
     pub remember_me: RememberMe,
 }

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

@@ -3,7 +3,7 @@
 
 use crate::client::mix_traffic::BatchMixMessageSender;
 use crate::client::topology_control::TopologyAccessor;
-use crate::{config, spawn_future};
+use crate::config;
 use futures::task::{Context, Poll};
 use futures::{Future, Stream, StreamExt};
 use nym_sphinx::acknowledgements::AckKey;
@@ -12,7 +12,6 @@ use nym_sphinx::cover::generate_loop_cover_packet;
 use nym_sphinx::params::{PacketSize, PacketType};
 use nym_sphinx::utils::sample_poisson_duration;
 use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
-use nym_task::TaskClient;
 use rand::{rngs::OsRng, CryptoRng, Rng};
 use std::pin::Pin;
 use std::sync::Arc;
@@ -69,8 +68,6 @@ where
     packet_type: PacketType,
 
     stats_tx: ClientStatsSender,
-
-    task_client: TaskClient,
 }
 
 impl<R> Stream for LoopCoverTrafficStream<R>
@@ -117,7 +114,6 @@ impl LoopCoverTrafficStream<OsRng> {
         traffic_config: config::Traffic,
         cover_config: config::CoverTraffic,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
     ) -> Self {
         let rng = OsRng;
 
@@ -137,7 +133,6 @@ impl LoopCoverTrafficStream<OsRng> {
             use_legacy_sphinx_format: traffic_config.use_legacy_sphinx_format,
             packet_type: traffic_config.packet_type,
             stats_tx,
-            task_client,
         }
     }
 
@@ -235,12 +230,13 @@ impl LoopCoverTrafficStream<OsRng> {
         tokio::task::yield_now().await;
     }
 
+    // it's fine if cover traffic stream task gets killed whilst processing next message
     #[allow(clippy::panic)]
-    pub fn start(mut self) {
+    pub async fn run(&mut self) {
         if self.cover_traffic.disable_loop_cover_traffic_stream {
             // we should have never got here in the first place - the task should have never been created to begin with
             // so panic and review the code that lead to this branch
-            panic!("attempted to start LoopCoverTrafficStream while config explicitly disabled it.")
+            panic!("attempted to run LoopCoverTrafficStream while config explicitly disabled it.")
         }
 
         // we should set initial delay only when we actually start the stream
@@ -250,32 +246,11 @@ impl LoopCoverTrafficStream<OsRng> {
         );
         self.set_next_delay(sampled);
 
-        let mut shutdown = self.task_client.fork("select");
+        while self.next().await.is_some() {
+            self.on_new_message().await;
+        }
 
-        spawn_future!(
-            async move {
-                debug!("Started LoopCoverTrafficStream with graceful shutdown support");
-
-                while !shutdown.is_shutdown() {
-                    tokio::select! {
-                        biased;
-                        _ = shutdown.recv() => {
-                            tracing::trace!("LoopCoverTrafficStream: Received shutdown");
-                        }
-                        next = self.next() => {
-                            if next.is_some() {
-                                self.on_new_message().await;
-                            } else {
-                                tracing::trace!("LoopCoverTrafficStream: Stopping since channel closed");
-                                break;
-                            }
-                        }
-                    }
-                }
-                shutdown.recv_timeout().await;
-                tracing::debug!("LoopCoverTrafficStream: Exiting");
-            },
-            "LoopCoverTrafficStream"
-        )
+        // this should never get triggered
+        error!("cover traffic stream has been exhausted!")
     }
 }

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

@@ -2,11 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::client::mix_traffic::transceiver::GatewayTransceiver;
-use crate::error::ClientCoreError;
-use crate::spawn_future;
 use nym_gateway_requests::ClientRequest;
 use nym_sphinx::forwarding::packet::MixPacket;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use tracing::*;
 use transceiver::ErasedGatewayError;
 
@@ -34,13 +32,13 @@ pub struct MixTrafficController {
     // in long run `gateway_client` will be moved away from `MixTrafficController` anyway.
     consecutive_gateway_failure_count: usize,
 
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl MixTrafficController {
     pub fn new<T>(
         gateway_transceiver: T,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> (
         MixTrafficController,
         BatchMixMessageSender,
@@ -60,7 +58,7 @@ impl MixTrafficController {
                 mix_rx: message_receiver,
                 client_rx: client_receiver,
                 consecutive_gateway_failure_count: 0,
-                task_client,
+                shutdown_token,
             },
             message_sender,
             client_sender,
@@ -69,7 +67,7 @@ impl MixTrafficController {
 
     pub fn new_dynamic(
         gateway_transceiver: Box<dyn GatewayTransceiver + Send>,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> (
         MixTrafficController,
         BatchMixMessageSender,
@@ -84,7 +82,7 @@ impl MixTrafficController {
                 mix_rx: message_receiver,
                 client_rx: client_receiver,
                 consecutive_gateway_failure_count: 0,
-                task_client,
+                shutdown_token,
             },
             message_sender,
             client_sender,
@@ -107,7 +105,7 @@ impl MixTrafficController {
 
         tokio::select! {
             biased;
-            _ = self.task_client.recv() => {
+            _ = self.shutdown_token.cancelled() => {
                 trace!("received shutdown while handling messages");
                 Ok(())
             }
@@ -127,7 +125,7 @@ impl MixTrafficController {
     async fn on_client_request(&mut self, client_request: ClientRequest) {
         tokio::select! {
             biased;
-             _ = self.task_client.recv() => {
+             _ = self.shutdown_token.cancelled() => {
                 trace!("received shutdown while handling client request");
             }
             result = self.gateway_transceiver.send_client_request(client_request) => {
@@ -138,51 +136,44 @@ impl MixTrafficController {
         }
     }
 
-    pub fn start(mut self) {
-        spawn_future!(
-            async move {
-                debug!("Started MixTrafficController with graceful shutdown support");
-                while !self.task_client.is_shutdown() {
-                    tokio::select! {
-                        biased;
-                        _ = self.task_client.recv() => {
-                            tracing::trace!("MixTrafficController: Received shutdown");
-                            break;
-                        }
-                        mix_packets = self.mix_rx.recv() => match mix_packets {
-                            Some(mix_packets) => {
-                                if let Err(err) = self.on_messages(mix_packets).await {
-                                    error!("Failed to send sphinx packet(s) to the gateway: {err}");
-                                    if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
-                                        // Disconnect from the gateway. If we should try to re-connect
-                                        // is handled at a higher layer.
-                                        error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
-                                        // Do we need to handle the embedded mixnet client case
-                                        // separately?
-                                        self.task_client.send_we_stopped(Box::new(ClientCoreError::GatewayFailedToForwardMessages));
-                                        break;
-                                    }
-                                }
-                            },
-                            None => {
-                                tracing::trace!("MixTrafficController: Stopping since channel closed");
+    pub async fn run(&mut self) {
+        debug!("Started MixTrafficController with graceful shutdown support");
+        loop {
+            tokio::select! {
+                biased;
+                _ = self.shutdown_token.cancelled() => {
+                    trace!("MixTrafficController: Received shutdown");
+                    break;
+                }
+                mix_packets = self.mix_rx.recv() => match mix_packets {
+                    Some(mix_packets) => {
+                        if let Err(err) = self.on_messages(mix_packets).await {
+                            error!("Failed to send sphinx packet(s) to the gateway: {err}");
+                            if self.consecutive_gateway_failure_count == MAX_FAILURE_COUNT {
+                                // Disconnect from the gateway. If we should try to re-connect
+                                // is handled at a higher layer.
+                                error!("Failed to send sphinx packet to the gateway {MAX_FAILURE_COUNT} times in a row - assuming the gateway is dead");
+                                // Do we need to handle the embedded mixnet client case
+                                // separately?
                                 break;
                             }
-                        },
-                        client_request = self.client_rx.recv() => match client_request {
-                            Some(client_request) => {
-                                self.on_client_request(client_request).await;
-                            },
-                            None => {
-                                tracing::trace!("MixTrafficController, client request channel closed");
-                            }
-                        },
+                        }
+                    },
+                    None => {
+                        trace!("MixTrafficController: Stopping since channel closed");
+                        break;
                     }
-                }
-                self.task_client.recv_timeout().await;
-                tracing::debug!("MixTrafficController: Exiting");
-            },
-            "MixTrafficController"
-        );
+                },
+                client_request = self.client_rx.recv() => match client_request {
+                    Some(client_request) => {
+                        self.on_client_request(client_request).await;
+                    },
+                    None => {
+                        trace!("MixTrafficController, client request channel closed");
+                    break}
+                },
+            }
+        }
+        debug!("MixTrafficController: Exiting");
     }
 }

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

@@ -10,18 +10,17 @@ use nym_sphinx::{
     acknowledgements::{identifier::recover_identifier, AckKey},
     chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID},
 };
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use std::sync::Arc;
 use tracing::*;
 
 /// Module responsible for listening for any data resembling acknowledgements from the network
 /// and firing actions to remove them from the 'Pending' state.
-pub(super) struct AcknowledgementListener {
+pub(crate) struct AcknowledgementListener {
     ack_key: Arc<AckKey>,
     ack_receiver: AcknowledgementReceiver,
     action_sender: AckActionSender,
     stats_tx: ClientStatsSender,
-    task_client: TaskClient,
 }
 
 impl AcknowledgementListener {
@@ -30,14 +29,12 @@ impl AcknowledgementListener {
         ack_receiver: AcknowledgementReceiver,
         action_sender: AckActionSender,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
     ) -> Self {
         AcknowledgementListener {
             ack_key,
             ack_receiver,
             action_sender,
             stats_tx,
-            task_client,
         }
     }
 
@@ -68,14 +65,9 @@ impl AcknowledgementListener {
         trace!("Received {frag_id} from the mix network");
         self.stats_tx
             .report(PacketStatisticsEvent::RealAckReceived(ack_content.len()).into());
-        if let Err(err) = self
+        let _ = self
             .action_sender
-            .unbounded_send(Action::new_remove(frag_id))
-        {
-            if !self.task_client.is_shutdown_poll() {
-                error!("Failed to send remove action to action controller: {err}");
-            }
-        }
+            .unbounded_send(Action::new_remove(frag_id));
     }
 
     async fn handle_ack_receiver_item(&mut self, item: Vec<Vec<u8>>) {
@@ -85,11 +77,16 @@ impl AcknowledgementListener {
         }
     }
 
-    pub(super) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
         debug!("Started AcknowledgementListener with graceful shutdown support");
 
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
+                biased;
+                _ = shutdown_token.cancelled() => {
+                    tracing::trace!("AcknowledgementListener: Received shutdown");
+                    break;
+                }
                 acks = self.ack_receiver.next() => match acks {
                     Some(acks) => self.handle_ack_receiver_item(acks).await,
                     None => {
@@ -97,12 +94,9 @@ impl AcknowledgementListener {
                         break;
                     }
                 },
-                _ = self.task_client.recv() => {
-                    tracing::trace!("AcknowledgementListener: Received shutdown");
-                }
+
             }
         }
-        self.task_client.recv_timeout().await;
         tracing::debug!("AcknowledgementListener: Exiting");
     }
 }

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

@@ -8,7 +8,7 @@ use futures::StreamExt;
 use nym_nonexhaustive_delayqueue::{Expired, NonExhaustiveDelayQueue, QueueKey};
 use nym_sphinx::chunking::fragment::FragmentIdentifier;
 use nym_sphinx::Delay as SphinxDelay;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
@@ -82,7 +82,7 @@ impl Config {
     }
 }
 
-pub(super) struct ActionController {
+pub(crate) struct ActionController {
     /// Configurable parameters of the `ActionController`
     config: Config,
 
@@ -102,8 +102,6 @@ pub(super) struct ActionController {
 
     /// Channel for notifying `RetransmissionRequestListener` about expired acknowledgements.
     retransmission_sender: RetransmissionRequestSender,
-
-    task_client: TaskClient,
 }
 
 impl ActionController {
@@ -111,7 +109,6 @@ impl ActionController {
         config: Config,
         retransmission_sender: RetransmissionRequestSender,
         incoming_actions: AckActionReceiver,
-        task_client: TaskClient,
     ) -> Self {
         ActionController {
             config,
@@ -119,7 +116,6 @@ impl ActionController {
             pending_acks_timers: NonExhaustiveDelayQueue::new(),
             incoming_actions,
             retransmission_sender,
-            task_client,
         }
     }
 
@@ -226,14 +222,9 @@ impl ActionController {
             // downgrading an arc and then upgrading vs cloning is difference of 30ns vs 15ns
             // so it's literally a NO difference while it might prevent us from unnecessarily
             // resending data (in maybe 1 in 1 million cases, but it's something)
-            if let Err(err) = self
+            let _ = self
                 .retransmission_sender
-                .unbounded_send(Arc::downgrade(pending_ack_data))
-            {
-                if !self.task_client.is_shutdown_poll() {
-                    tracing::error!("Failed to send pending ack for retransmission: {err}");
-                }
-            }
+                .unbounded_send(Arc::downgrade(pending_ack_data));
         } else {
             // this shouldn't cause any issues but shouldn't have happened to begin with!
             error!("An already removed pending ack has expired")
@@ -251,11 +242,16 @@ impl ActionController {
         }
     }
 
-    pub(super) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
         debug!("Started ActionController with graceful shutdown support");
 
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
+                biased;
+                _ = shutdown_token.cancelled() => {
+                    tracing::trace!("ActionController: Received shutdown");
+                    break;
+                }
                 action = self.incoming_actions.next() => match action {
                     Some(action) => self.process_action(action),
                     None => {
@@ -272,13 +268,8 @@ impl ActionController {
                         break;
                     }
                 },
-                _ = self.task_client.recv() => {
-                    tracing::trace!("ActionController: Received shutdown");
-                    break;
-                }
             }
         }
-        self.task_client.recv_timeout().await;
         tracing::debug!("ActionController: Exiting");
     }
 }

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

@@ -10,21 +10,20 @@ use nym_sphinx::anonymous_replies::requests::AnonymousSenderTag;
 use nym_sphinx::forwarding::packet::MixPacket;
 use nym_sphinx::params::PacketType;
 use nym_task::connections::TransmissionLane;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use rand::{CryptoRng, Rng};
 use tracing::*;
 
 /// Module responsible for dealing with the received messages: splitting them, creating acknowledgements,
 /// putting everything into sphinx packets, etc.
 /// It also makes an initial sending attempt for said messages.
-pub(super) struct InputMessageListener<R>
+pub(crate) struct InputMessageListener<R>
 where
     R: CryptoRng + Rng,
 {
     input_receiver: InputMessageReceiver,
     message_handler: MessageHandler<R>,
     reply_controller_sender: ReplyControllerSender,
-    task_client: TaskClient,
 }
 
 impl<R> InputMessageListener<R>
@@ -38,13 +37,11 @@ where
         input_receiver: InputMessageReceiver,
         message_handler: MessageHandler<R>,
         reply_controller_sender: ReplyControllerSender,
-        task_client: TaskClient,
     ) -> Self {
         InputMessageListener {
             input_receiver,
             message_handler,
             reply_controller_sender,
-            task_client,
         }
     }
 
@@ -68,14 +65,9 @@ where
         max_retransmissions: Option<u32>,
     ) {
         // offload reply handling to the dedicated task
-        if let Err(err) =
+        let _ =
             self.reply_controller_sender
-                .send_reply(recipient_tag, data, lane, max_retransmissions)
-        {
-            if !self.task_client.is_shutdown_poll() {
-                error!("failed to send a reply - {err}");
-            }
-        }
+                .send_reply(recipient_tag, data, lane, max_retransmissions);
     }
 
     async fn handle_plain_message(
@@ -221,13 +213,13 @@ where
         };
     }
 
-    pub(super) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
         debug!("Started InputMessageListener with graceful shutdown support");
 
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                _ = self.task_client.recv() => {
+                _ = shutdown_token.cancelled() => {
                     tracing::trace!("InputMessageListener: Received shutdown");
                     break;
                 }
@@ -243,7 +235,6 @@ where
 
             }
         }
-        self.task_client.recv_timeout().await;
         tracing::debug!("InputMessageListener: Exiting");
     }
 }

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

@@ -10,7 +10,6 @@ use self::{
 use crate::client::inbound_messages::InputMessageReceiver;
 use crate::client::real_messages_control::message_handler::MessageHandler;
 use crate::client::replies::reply_controller::ReplyControllerSender;
-use crate::spawn_future;
 use action_controller::AckActionReceiver;
 use futures::channel::mpsc;
 use nym_gateway_client::AcknowledgementReceiver;
@@ -23,13 +22,11 @@ use nym_sphinx::{
     Delay as SphinxDelay,
 };
 use nym_statistics_common::clients::ClientStatsSender;
-use nym_task::TaskClient;
 use rand::{CryptoRng, Rng};
 use std::{
     sync::{Arc, Weak},
     time::Duration,
 };
-use tracing::*;
 
 pub(crate) use action_controller::{AckActionSender, Action};
 
@@ -190,6 +187,9 @@ pub(super) struct Config {
 
     /// Predefined packet size used for the encapsulated messages.
     packet_size: PacketSize,
+
+    /// Type of packets used for retransmissions
+    packet_type: PacketType,
 }
 
 impl Config {
@@ -197,12 +197,14 @@ impl Config {
         maximum_retransmissions: Option<u32>,
         ack_wait_addition: Duration,
         ack_wait_multiplier: f64,
+        packet_type: PacketType,
     ) -> Self {
         Config {
             maximum_retransmissions,
             ack_wait_addition,
             ack_wait_multiplier,
             packet_size: Default::default(),
+            packet_type,
         }
     }
 
@@ -212,7 +214,7 @@ impl Config {
     }
 }
 
-pub(super) struct AcknowledgementController<R>
+pub(crate) struct AcknowledgementController<R>
 where
     R: CryptoRng + Rng,
 {
@@ -234,7 +236,6 @@ where
         message_handler: MessageHandler<R>,
         reply_controller_sender: ReplyControllerSender,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
     ) -> Self {
         let (retransmission_tx, retransmission_rx) = mpsc::unbounded();
 
@@ -244,7 +245,6 @@ where
             action_config,
             retransmission_tx,
             connectors.ack_action_receiver,
-            task_client.fork("action_controller"),
         );
 
         // will listen for any acks coming from the network
@@ -253,7 +253,6 @@ where
             connectors.ack_receiver,
             connectors.ack_action_sender.clone(),
             stats_tx,
-            task_client.fork("acknowledgement_listener"),
         );
 
         // will listen for any new messages from the client
@@ -261,7 +260,6 @@ where
             connectors.input_receiver,
             message_handler.clone(),
             reply_controller_sender.clone(),
-            task_client.fork("input_message_listener"),
         );
 
         // will listen for any ack timeouts and trigger retransmission
@@ -271,16 +269,13 @@ where
             message_handler,
             retransmission_rx,
             reply_controller_sender,
-            task_client.fork("retransmission_request_listener"),
+            config.packet_type,
         );
 
         // 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,
-            connectors.ack_action_sender,
-            task_client.with_suffix("sent_notification_listener"),
-        );
+        let sent_notification_listener =
+            SentNotificationListener::new(connectors.sent_notifier, connectors.ack_action_sender);
 
         AcknowledgementController {
             acknowledgement_listener,
@@ -291,51 +286,21 @@ where
         }
     }
 
-    pub(super) fn start(self, packet_type: PacketType) {
-        let mut acknowledgement_listener = self.acknowledgement_listener;
-        let mut input_message_listener = self.input_message_listener;
-        let mut retransmission_request_listener = self.retransmission_request_listener;
-        let mut sent_notification_listener = self.sent_notification_listener;
-        let mut action_controller = self.action_controller;
-
-        spawn_future!(
-            async move {
-                acknowledgement_listener.run().await;
-                debug!("The acknowledgement listener has finished execution!");
-            },
-            "AcknowledgementController::AcknowledgementListener"
-        );
-
-        spawn_future!(
-            async move {
-                input_message_listener.run().await;
-                debug!("The input listener has finished execution!");
-            },
-            "AcknowledgementController::InputMessageListener"
-        );
-
-        spawn_future!(
-            async move {
-                retransmission_request_listener.run(packet_type).await;
-                debug!("The retransmission request listener has finished execution!");
-            },
-            "AcknowledgementController::RetransmissionRequestListener"
-        );
-
-        spawn_future!(
-            async move {
-                sent_notification_listener.run().await;
-                debug!("The sent notification listener has finished execution!");
-            },
-            "AcknowledgementController::SentNotificationListener"
-        );
-
-        spawn_future!(
-            async move {
-                action_controller.run().await;
-                debug!("The controller has finished execution!");
-            },
-            "AcknowledgementController::ActionController"
-        );
+    pub(crate) fn into_tasks(
+        self,
+    ) -> (
+        AcknowledgementListener,
+        InputMessageListener<R>,
+        RetransmissionRequestListener<R>,
+        SentNotificationListener,
+        ActionController,
+    ) {
+        (
+            self.acknowledgement_listener,
+            self.input_message_listener,
+            self.retransmission_request_listener,
+            self.sent_notification_listener,
+            self.action_controller,
+        )
     }
 }

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

@@ -13,19 +13,19 @@ use futures::StreamExt;
 use nym_sphinx::chunking::fragment::Fragment;
 use nym_sphinx::preparer::PreparedFragment;
 use nym_sphinx::{addressing::clients::Recipient, params::PacketType};
-use nym_task::{connections::TransmissionLane, TaskClient};
+use nym_task::{connections::TransmissionLane, ShutdownToken};
 use rand::{CryptoRng, Rng};
 use std::sync::{Arc, Weak};
 use tracing::*;
 
 // responsible for packet retransmission upon fired timer
-pub(super) struct RetransmissionRequestListener<R> {
+pub(crate) struct RetransmissionRequestListener<R> {
     maximum_retransmissions: Option<u32>,
     action_sender: AckActionSender,
     message_handler: MessageHandler<R>,
     request_receiver: RetransmissionRequestReceiver,
     reply_controller_sender: ReplyControllerSender,
-    task_client: TaskClient,
+    packet_type: PacketType,
 }
 
 impl<R> RetransmissionRequestListener<R>
@@ -38,7 +38,7 @@ where
         message_handler: MessageHandler<R>,
         request_receiver: RetransmissionRequestReceiver,
         reply_controller_sender: ReplyControllerSender,
-        task_client: TaskClient,
+        packet_type: PacketType,
     ) -> Self {
         RetransmissionRequestListener {
             maximum_retransmissions,
@@ -46,7 +46,7 @@ where
             message_handler,
             request_receiver,
             reply_controller_sender,
-            task_client,
+            packet_type,
         }
     }
 
@@ -67,7 +67,6 @@ where
     async fn on_retransmission_request(
         &mut self,
         weak_timed_out_ack: Weak<PendingAcknowledgement>,
-        packet_type: PacketType,
     ) {
         let timed_out_ack = match weak_timed_out_ack.upgrade() {
             Some(timed_out_ack) => timed_out_ack,
@@ -97,22 +96,18 @@ where
             } => {
                 // if this is retransmission for reply, offload it to the dedicated task
                 // that deals with all the surbs
-                if let Err(err) = self.reply_controller_sender.send_retransmission_data(
+                let _ = self.reply_controller_sender.send_retransmission_data(
                     *recipient_tag,
                     weak_timed_out_ack,
                     *extra_surb_request,
-                ) {
-                    if !self.task_client.is_shutdown_poll() {
-                        error!("Failed to send retransmission data to the reply controller: {err}");
-                    }
-                }
+                );
                 return;
             }
             PacketDestination::KnownRecipient(recipient) => {
                 self.prepare_normal_retransmission_chunk(
                     **recipient,
                     timed_out_ack.message_chunk.clone(),
-                    packet_type,
+                    self.packet_type,
                 )
                 .await
             }
@@ -153,14 +148,9 @@ where
         // is sent to the `OutQueueControl` and has gone through its internal queue
         // with the additional poisson delay.
         // And since Actions are executed in order `UpdateTimer` will HAVE TO be executed before `StartTimer`
-        if let Err(err) = self
+        let _ = self
             .action_sender
-            .unbounded_send(Action::new_update_pending_ack(frag_id, new_delay))
-        {
-            if !self.task_client.is_shutdown_poll() {
-                error!("Failed to send update pending ack action to the controller: {err}");
-            }
-        }
+            .unbounded_send(Action::new_update_pending_ack(frag_id, new_delay));
 
         // send to `OutQueueControl` to eventually send to the mix network
         self.message_handler
@@ -174,18 +164,18 @@ where
             .await
     }
 
-    pub(super) async fn run(&mut self, packet_type: PacketType) {
+    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
         debug!("Started RetransmissionRequestListener with graceful shutdown support");
 
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                 _ = self.task_client.recv() => {
+                _ = shutdown_token.cancelled() => {
                     tracing::trace!("RetransmissionRequestListener: Received shutdown");
                     break;
                 }
                 timed_out_ack = self.request_receiver.next() => match timed_out_ack {
-                    Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack, packet_type).await,
+                    Some(timed_out_ack) => self.on_retransmission_request(timed_out_ack).await,
                     None => {
                         tracing::trace!("RetransmissionRequestListener: Stopping since channel closed");
                         break;
@@ -194,7 +184,6 @@ where
 
             }
         }
-        self.task_client.recv_timeout().await;
         tracing::debug!("RetransmissionRequestListener: Exiting");
     }
 }

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

@@ -5,29 +5,25 @@ use super::action_controller::{AckActionSender, Action};
 use super::SentPacketNotificationReceiver;
 use futures::StreamExt;
 use nym_sphinx::chunking::fragment::{FragmentIdentifier, COVER_FRAG_ID};
-use nym_task::TaskClient;
 use tracing::*;
 
 /// Module responsible for starting up retransmission timers.
 /// It is required because when we send our packet to the `real traffic stream` controlled
 /// by a poisson timer, there's no guarantee the message will be sent immediately, so we might
 /// accidentally fire retransmission way quicker than we should have.
-pub(super) struct SentNotificationListener {
+pub(crate) struct SentNotificationListener {
     sent_notifier: SentPacketNotificationReceiver,
     action_sender: AckActionSender,
-    task_client: TaskClient,
 }
 
 impl SentNotificationListener {
     pub(super) fn new(
         sent_notifier: SentPacketNotificationReceiver,
         action_sender: AckActionSender,
-        task_client: TaskClient,
     ) -> Self {
         SentNotificationListener {
             sent_notifier,
             action_sender,
-            task_client,
         }
     }
 
@@ -36,37 +32,18 @@ impl SentNotificationListener {
             trace!("sent off a cover message - no need to start retransmission timer!");
             return;
         }
-        if let Err(err) = self
+        let _ = self
             .action_sender
-            .unbounded_send(Action::new_start_timer(frag_id))
-        {
-            if !self.task_client.is_shutdown_poll() {
-                error!("Failed to send start timer action to action controller: {err}");
-            }
-        }
+            .unbounded_send(Action::new_start_timer(frag_id));
     }
 
-    pub(super) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self) {
         debug!("Started SentNotificationListener with graceful shutdown support");
 
-        while !self.task_client.is_shutdown() {
-            tokio::select! {
-                frag_id = self.sent_notifier.next() => match frag_id {
-                    Some(frag_id) => {
-                        self.on_sent_message(frag_id).await;
-                    }
-                    None => {
-                        tracing::trace!("SentNotificationListener: Stopping since channel closed");
-                        break;
-                    }
-                },
-                _ = self.task_client.recv() => {
-                    tracing::trace!("SentNotificationListener: Received shutdown");
-                    break;
-                }
-            }
+        while let Some(frag_id) = self.sent_notifier.next().await {
+            self.on_sent_message(frag_id).await;
         }
-        assert!(self.task_client.is_shutdown_poll());
+
         tracing::debug!("SentNotificationListener: Exiting");
     }
 }

common/client-core/src/client/real_messages_control/message_handler.rs

@@ -20,7 +20,7 @@ use nym_sphinx::params::{PacketSize, PacketType};
 use nym_sphinx::preparer::{MessagePreparer, PreparedFragment};
 use nym_sphinx::Delay;
 use nym_task::connections::TransmissionLane;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use nym_topology::{NymRouteProvider, NymTopologyError};
 use rand::{CryptoRng, Rng};
 use std::collections::HashMap;
@@ -189,7 +189,7 @@ pub(crate) struct MessageHandler<R> {
     topology_access: TopologyAccessor,
     reply_key_storage: SentReplyKeys,
     tag_storage: UsedSenderTags,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl<R> MessageHandler<R>
@@ -205,7 +205,7 @@ where
         topology_access: TopologyAccessor,
         reply_key_storage: SentReplyKeys,
         tag_storage: UsedSenderTags,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self
     where
         R: Copy,
@@ -228,7 +228,7 @@ where
             topology_access,
             reply_key_storage,
             tag_storage,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -712,7 +712,7 @@ where
             .action_sender
             .unbounded_send(Action::UpdatePendingAck(id, new_delay))
         {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send update action to the controller: {err}");
             }
         }
@@ -723,7 +723,7 @@ where
             .action_sender
             .unbounded_send(Action::new_insert(pending_acks))
         {
-            if !self.task_client.is_shutdown_poll() {
+            if !self.shutdown_token.is_cancelled() {
                 error!("Failed to send insert action to the controller: {err}");
             }
         }
@@ -737,7 +737,7 @@ where
     ) {
         tokio::select! {
             biased;
-            _ = self.task_client.recv() => {
+            _ = self.shutdown_token.cancelled() => {
                 trace!("received shutdown while attempting to forward mixnet messages");
             }
             sending_res = self.real_message_sender.send((messages, transmission_lane)) => {

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

@@ -14,26 +14,21 @@ use crate::client::replies::reply_controller::{
     ReplyController, ReplyControllerReceiver, ReplyControllerSender,
 };
 use crate::client::replies::reply_storage::CombinedReplyStorage;
-use crate::config;
-use crate::{
-    client::{
-        inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
-        real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
-        topology_control::TopologyAccessor,
-    },
-    spawn_future,
+use crate::client::{
+    inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender,
+    real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors,
+    topology_control::TopologyAccessor,
 };
+use crate::config;
 use futures::channel::mpsc;
 use nym_gateway_client::AcknowledgementReceiver;
 use nym_sphinx::acknowledgements::AckKey;
 use nym_sphinx::addressing::clients::Recipient;
-use nym_sphinx::params::PacketType;
 use nym_statistics_common::clients::ClientStatsSender;
 use nym_task::connections::{ConnectionCommandReceiver, LaneQueueLengths};
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use rand::{rngs::OsRng, CryptoRng, Rng};
 use std::sync::Arc;
-use tracing::*;
 
 use crate::client::replies::reply_controller::key_rotation_helpers::KeyRotationConfig;
 pub(crate) use acknowledgement_control::{AckActionSender, Action};
@@ -69,6 +64,7 @@ impl<'a> From<&'a Config> for acknowledgement_control::Config {
             cfg.traffic.maximum_number_of_retransmissions,
             cfg.acks.ack_wait_addition,
             cfg.acks.ack_wait_multiplier,
+            cfg.traffic.packet_type,
         )
         .with_custom_packet_size(cfg.traffic.primary_packet_size)
     }
@@ -146,7 +142,7 @@ impl RealMessagesController<OsRng> {
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         let rng = OsRng;
 
@@ -178,7 +174,7 @@ impl RealMessagesController<OsRng> {
             topology_access.clone(),
             reply_storage.key_storage(),
             reply_storage.tags_storage(),
-            task_client.fork("message_handler"),
+            shutdown_token.clone(),
         );
 
         let ack_control = AcknowledgementController::new(
@@ -188,7 +184,6 @@ impl RealMessagesController<OsRng> {
             message_handler.clone(),
             reply_controller_sender,
             stats_tx.clone(),
-            task_client.fork("ack_control"),
         );
 
         let reply_control = ReplyController::new(
@@ -196,7 +191,6 @@ impl RealMessagesController<OsRng> {
             message_handler,
             reply_storage,
             reply_controller_receiver,
-            task_client.fork("reply_controller"),
         );
 
         let out_queue_control = OutQueueControl::new(
@@ -209,7 +203,7 @@ impl RealMessagesController<OsRng> {
             lane_queue_lengths,
             client_connection_rx,
             stats_tx,
-            task_client.with_suffix("out_queue_control"),
+            shutdown_token.clone(),
         );
 
         RealMessagesController {
@@ -219,26 +213,13 @@ impl RealMessagesController<OsRng> {
         }
     }
 
-    pub fn start(self, packet_type: PacketType) {
-        let mut out_queue_control = self.out_queue_control;
-        let ack_control = self.ack_control;
-        let mut reply_control = self.reply_control;
-
-        spawn_future!(
-            async move {
-                out_queue_control.run().await;
-                debug!("The out queue controller has finished execution!");
-            },
-            "RealMessagesController::OutQueueControl)"
-        );
-        spawn_future!(
-            async move {
-                reply_control.run().await;
-                debug!("The reply controller has finished execution!");
-            },
-            "RealMessagesController::ReplyController"
-        );
-
-        ack_control.start(packet_type);
+    pub fn into_tasks(
+        self,
+    ) -> (
+        OutQueueControl<OsRng>,
+        ReplyController<OsRng>,
+        AcknowledgementController<OsRng>,
+    ) {
+        (self.out_queue_control, self.reply_control, self.ack_control)
     }
 }

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

@@ -21,7 +21,7 @@ use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, C
 use nym_task::connections::{
     ConnectionCommand, ConnectionCommandReceiver, ConnectionId, LaneQueueLengths, TransmissionLane,
 };
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use rand::{CryptoRng, Rng};
 use std::pin::Pin;
 use std::sync::Arc;
@@ -119,7 +119,7 @@ where
     /// Channel used for sending metrics events (specifically `PacketStatistics` events) to the metrics tracker.
     stats_tx: ClientStatsSender,
 
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 #[derive(Debug)]
@@ -179,7 +179,7 @@ where
         lane_queue_lengths: LaneQueueLengths,
         client_connection_rx: ConnectionCommandReceiver,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         OutQueueControl {
             config,
@@ -194,7 +194,7 @@ where
             client_connection_rx,
             lane_queue_lengths,
             stats_tx,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -282,7 +282,7 @@ where
 
         let sending_res = tokio::select! {
             biased;
-            _ = self.task_client.recv() => {
+            _ = self.shutdown_token.cancelled() => {
                 trace!("received shutdown signal while attempting to send mix message");
                 return
             }
@@ -293,7 +293,7 @@ where
 
         match sending_res {
             Err(_) => {
-                if !self.task_client.is_shutdown_poll() {
+                if !self.shutdown_token.is_cancelled() {
                     tracing::error!(
                         "failed to send mixnet packet due to closed channel (outside of shutdown!)"
                     );
@@ -536,9 +536,7 @@ where
     }
 
     #[cfg(not(target_arch = "wasm32"))]
-    fn log_status(&self, shutdown: &mut TaskClient) {
-        use crate::error::ClientCoreStatusMessage;
-
+    fn log_status(&self) {
         let packets = self.transmission_buffer.total_size();
         let lanes = self.transmission_buffer.lanes();
         let mult = self.sending_delay_controller.current_multiplier();
@@ -567,32 +565,33 @@ where
             tracing::debug!("{status_str}");
         }
 
-        // Send status message to whoever is listening (possibly UI)
-        if mult == self.sending_delay_controller.max_multiplier() {
-            shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsVerySlow));
-        } else if mult > self.sending_delay_controller.min_multiplier() {
-            shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsSlow));
-        }
+        // leave the code commented in case somebody wanted to restore this logic with a different channel
+        // // Send status message to whoever is listening (possibly UI)
+        // if mult == self.sending_delay_controller.max_multiplier() {
+        //     shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsVerySlow));
+        // } else if mult > self.sending_delay_controller.min_multiplier() {
+        //     shutdown.send_status_msg(Box::new(ClientCoreStatusMessage::GatewayIsSlow));
+        // }
     }
 
-    pub(super) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self) {
         debug!("Started OutQueueControl with graceful shutdown support");
 
-        let mut shutdown = self.task_client.fork("select");
-
+        // avoid borrow on self
+        let shutdown_token = self.shutdown_token.clone();
         #[cfg(not(target_arch = "wasm32"))]
         {
             let mut status_timer = tokio::time::interval(Duration::from_secs(5));
 
-            while !shutdown.is_shutdown() {
+            loop {
                 tokio::select! {
                     biased;
-                    _ = shutdown.recv() => {
+                    _ = shutdown_token.cancelled() => {
                         tracing::trace!("OutQueueControl: Received shutdown");
                         break;
                     }
                     _ = status_timer.tick() => {
-                        self.log_status(&mut shutdown);
+                        self.log_status();
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
                         self.on_message(next_message).await;
@@ -602,16 +601,16 @@ where
                     }
                 }
             }
-            shutdown.recv_timeout().await;
         }
 
         #[cfg(target_arch = "wasm32")]
         {
-            while !shutdown.is_shutdown() {
+            loop {
                 tokio::select! {
                     biased;
-                    _ = shutdown.recv() => {
+                    _ = shutdown_token.cancelled() => {
                         tracing::trace!("OutQueueControl: Received shutdown");
+                        break;
                     }
                     next_message = self.next() => if let Some(next_message) = next_message {
                         self.on_message(next_message).await;

common/client-core/src/client/real_messages_control/real_traffic_stream/sending_delay_controller.rs

@@ -83,11 +83,13 @@ impl SendingDelayController {
         self.current_multiplier
     }
 
+    #[allow(dead_code)]
     #[cfg(not(target_arch = "wasm32"))]
     pub(crate) fn min_multiplier(&self) -> u32 {
         self.lower_bound
     }
 
+    #[allow(dead_code)]
     #[cfg(not(target_arch = "wasm32"))]
     pub(crate) fn max_multiplier(&self) -> u32 {
         self.upper_bound

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

@@ -1,10 +1,10 @@
 // Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::client::helpers::get_time_now;
 use crate::client::replies::{
     reply_controller::ReplyControllerSender, reply_storage::SentReplyKeys,
 };
-use crate::spawn_future;
 use futures::channel::mpsc;
 use futures::lock::Mutex;
 use futures::StreamExt;
@@ -19,10 +19,10 @@ use nym_sphinx::message::{NymMessage, PlainMessage};
 use nym_sphinx::params::ReplySurbKeyDigestAlgorithm;
 use nym_sphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage};
 use nym_statistics_common::clients::{packet_statistics::PacketStatisticsEvent, ClientStatsSender};
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use std::collections::HashSet;
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 use tracing::*;
 
 // The interval at which we check for stale buffers
@@ -54,7 +54,7 @@ struct ReceivedMessagesBufferInner<R: MessageReceiver> {
     stats_tx: ClientStatsSender,
 
     // Periodically check for stale buffers to clean up
-    last_stale_check: Instant,
+    last_stale_check: crate::client::helpers::Instant,
 }
 
 impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
@@ -154,7 +154,7 @@ impl<R: MessageReceiver> ReceivedMessagesBufferInner<R> {
     }
 
     fn cleanup_stale_buffers(&mut self) {
-        let now = Instant::now();
+        let now = get_time_now();
         if now - self.last_stale_check > STALE_BUFFER_CHECK_INTERVAL {
             self.last_stale_check = now;
             self.message_receiver
@@ -171,7 +171,7 @@ struct ReceivedMessagesBuffer<R: MessageReceiver> {
     inner: Arc<Mutex<ReceivedMessagesBufferInner<R>>>,
     reply_key_storage: SentReplyKeys,
     reply_controller_sender: ReplyControllerSender,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
@@ -180,7 +180,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
         stats_tx: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         ReceivedMessagesBuffer {
             inner: Arc::new(Mutex::new(ReceivedMessagesBufferInner {
@@ -190,11 +190,11 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                 message_sender: None,
                 recently_reconstructed: HashSet::new(),
                 stats_tx,
-                last_stale_check: Instant::now(),
+                last_stale_check: get_time_now(),
             })),
             reply_key_storage,
             reply_controller_sender,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -315,7 +315,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                     reply_surbs,
                     from_surb_request,
                 ) {
-                    if !self.task_client.is_shutdown_poll() {
+                    if !self.shutdown_token.is_cancelled() {
                         error!("{err}");
                     }
                 }
@@ -338,7 +338,7 @@ impl<R: MessageReceiver> ReceivedMessagesBuffer<R> {
                         .reply_controller_sender
                         .send_additional_surbs_request(*recipient, amount)
                     {
-                        if !self.task_client.is_shutdown_poll() {
+                        if !self.shutdown_token.is_cancelled() {
                             error!("{err}");
                         }
                     }
@@ -465,22 +465,22 @@ pub enum ReceivedBufferMessage {
     ReceiverDisconnect,
 }
 
-struct RequestReceiver<R: MessageReceiver> {
+pub(crate) struct RequestReceiver<R: MessageReceiver> {
     received_buffer: ReceivedMessagesBuffer<R>,
     query_receiver: ReceivedBufferRequestReceiver,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl<R: MessageReceiver> RequestReceiver<R> {
     fn new(
         received_buffer: ReceivedMessagesBuffer<R>,
         query_receiver: ReceivedBufferRequestReceiver,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         RequestReceiver {
             received_buffer,
             query_receiver,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -495,66 +495,70 @@ impl<R: MessageReceiver> RequestReceiver<R> {
         }
     }
 
-    async fn run(&mut self) {
+    pub(crate) async fn run(&mut self) {
         debug!("Started RequestReceiver with graceful shutdown support");
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                _ = self.task_client.recv() => {
+                _ = self.shutdown_token.cancelled() => {
                     tracing::trace!("RequestReceiver: Received shutdown");
+                    break;
                 }
                 request = self.query_receiver.next() => {
                     if let Some(message) = request {
                         self.handle_message(message).await
                     } else {
                         tracing::trace!("RequestReceiver: Stopping since channel closed");
+                        self.shutdown_token.cancelled().await;
                         break;
                     }
                 },
             }
         }
-        self.task_client.recv().await;
         tracing::debug!("RequestReceiver: Exiting");
     }
 }
 
-struct FragmentedMessageReceiver<R: MessageReceiver> {
+pub(crate) struct FragmentedMessageReceiver<R: MessageReceiver> {
     received_buffer: ReceivedMessagesBuffer<R>,
     mixnet_packet_receiver: MixnetMessageReceiver,
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl<R: MessageReceiver> FragmentedMessageReceiver<R> {
     fn new(
         received_buffer: ReceivedMessagesBuffer<R>,
         mixnet_packet_receiver: MixnetMessageReceiver,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         FragmentedMessageReceiver {
             received_buffer,
             mixnet_packet_receiver,
-            task_client,
+            shutdown_token,
         }
     }
 
-    async fn run(&mut self) -> Result<(), MessageRecoveryError> {
+    pub(crate) async fn run(&mut self) -> Result<(), MessageRecoveryError> {
         debug!("Started FragmentedMessageReceiver with graceful shutdown support");
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
+                biased;
+                _ = self.shutdown_token.cancelled() => {
+                    tracing::trace!("FragmentedMessageReceiver: Received shutdown");
+                    break;
+                }
                 new_messages = self.mixnet_packet_receiver.next() => {
                     if let Some(new_messages) = new_messages {
                         self.received_buffer.handle_new_received(new_messages).await?;
                     } else {
                         tracing::trace!("FragmentedMessageReceiver: Stopping since channel closed");
+                        self.shutdown_token.cancelled().await;
                         break;
                     }
                 },
-                _ = self.task_client.recv_with_delay() => {
-                    tracing::trace!("FragmentedMessageReceiver: Received shutdown");
-                }
+
             }
         }
-        self.task_client.recv_timeout().await;
         tracing::debug!("FragmentedMessageReceiver: Exiting");
         Ok(())
     }
@@ -573,48 +577,31 @@ impl<R: MessageReceiver + Clone + Send + 'static> ReceivedMessagesBufferControll
         reply_key_storage: SentReplyKeys,
         reply_controller_sender: ReplyControllerSender,
         metrics_reporter: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         let received_buffer = ReceivedMessagesBuffer::new(
             local_encryption_keypair,
             reply_key_storage,
             reply_controller_sender,
             metrics_reporter,
-            task_client.fork("received_messages_buffer"),
+            shutdown_token.clone(),
         );
 
         ReceivedMessagesBufferController {
             fragmented_message_receiver: FragmentedMessageReceiver::new(
                 received_buffer.clone(),
                 mixnet_packet_receiver,
-                task_client.fork("fragmented_message_receiver"),
+                shutdown_token.clone(),
             ),
             request_receiver: RequestReceiver::new(
                 received_buffer,
                 query_receiver,
-                task_client.with_suffix("request_receiver"),
+                shutdown_token.clone(),
             ),
         }
     }
 
-    pub fn start(self) {
-        let mut fragmented_message_receiver = self.fragmented_message_receiver;
-        let mut request_receiver = self.request_receiver;
-
-        spawn_future!(
-            async move {
-                match fragmented_message_receiver.run().await {
-                    Ok(_) => {}
-                    Err(e) => error!("{e}"),
-                }
-            },
-            "ReceivedMessagesBufferController::FragmentedMessageReceiver"
-        );
-        spawn_future!(
-            async move {
-                request_receiver.run().await;
-            },
-            "ReceivedMessagesBufferController::RequestReceiver"
-        );
+    pub(crate) fn into_tasks(self) -> (FragmentedMessageReceiver<R>, RequestReceiver<R>) {
+        (self.fragmented_message_receiver, self.request_receiver)
     }
 }

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

@@ -7,7 +7,7 @@ use crate::client::replies::reply_controller::key_rotation_helpers::KeyRotationC
 use crate::client::replies::reply_storage::CombinedReplyStorage;
 use crate::config;
 use futures::StreamExt;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use rand::rngs::OsRng;
 use rand::{CryptoRng, Rng};
 use std::time::Duration;
@@ -60,9 +60,6 @@ pub struct ReplyController<R> {
     receiver_controller: ReceiverReplyController<R>,
 
     request_receiver: ReplyControllerReceiver,
-
-    // Listen for shutdown signals
-    task_client: TaskClient,
 }
 
 impl ReplyController<OsRng> {
@@ -71,7 +68,6 @@ impl ReplyController<OsRng> {
         message_handler: MessageHandler<OsRng>,
         full_reply_storage: CombinedReplyStorage,
         request_receiver: ReplyControllerReceiver,
-        task_client: TaskClient,
     ) -> Self {
         ReplyController {
             config,
@@ -86,7 +82,6 @@ impl ReplyController<OsRng> {
                 message_handler,
             ),
             request_receiver,
-            task_client,
         }
     }
 }
@@ -148,22 +143,21 @@ where
         self.sender_controller.inspect_and_clear_stale_data(now)
     }
 
-    pub(crate) async fn run(&mut self) {
+    pub(crate) async fn run(&mut self, shutdown_token: ShutdownToken) {
         debug!("Started ReplyController with graceful shutdown support");
 
-        let mut shutdown = self.task_client.fork("reply-controller");
-
         let polling_rate = Duration::from_secs(5);
         let mut stale_inspection = new_interval_stream(polling_rate);
 
         let polling_rate = self.config.key_rotation.epoch_duration / 8;
         let mut invalidation_inspection = new_interval_stream(polling_rate);
 
-        while !shutdown.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                _ = shutdown.recv() => {
+                _ = shutdown_token.cancelled() => {
                     tracing::trace!("ReplyController: Received shutdown");
+                    break;
                 },
                 req = self.request_receiver.next() => match req {
                     Some(req) => self.handle_request(req).await,
@@ -181,7 +175,6 @@ where
                 }
             }
         }
-        assert!(shutdown.is_shutdown_poll());
         tracing::debug!("ReplyController: Exiting");
     }
 }

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

@@ -16,21 +16,17 @@
 #![warn(clippy::todo)]
 #![warn(clippy::dbg_macro)]
 
+use crate::client::inbound_messages::{InputMessage, InputMessageSender};
 use futures::StreamExt;
 use nym_client_core_config_types::StatsReporting;
 use nym_sphinx::addressing::Recipient;
 use nym_statistics_common::clients::{
     ClientStatsController, ClientStatsReceiver, ClientStatsSender,
 };
-use nym_task::{connections::TransmissionLane, TaskClient};
+use nym_task::{connections::TransmissionLane, ShutdownToken, ShutdownTracker};
 use std::time::Duration;
 
-use crate::{
-    client::inbound_messages::{InputMessage, InputMessageSender},
-    spawn_future,
-};
-
-/// Time interval between reporting statistics locally (logging/task_client)
+/// Time interval between reporting statistics locally (logging/shutdown_token)
 const LOCAL_REPORT_INTERVAL: Duration = Duration::from_secs(2);
 /// Interval for taking snapshots of the statistics
 const SNAPSHOT_INTERVAL: Duration = Duration::from_millis(500);
@@ -51,9 +47,6 @@ pub(crate) struct StatisticsControl {
 
     /// Config for stats reporting (enabled, address, interval)
     reporting_config: StatsReporting,
-
-    /// Task client for listening for shutdown
-    task_client: TaskClient,
 }
 
 impl StatisticsControl {
@@ -62,24 +55,20 @@ impl StatisticsControl {
         client_type: String,
         client_stats_id: String,
         report_tx: InputMessageSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> (Self, ClientStatsSender) {
         let (stats_tx, stats_rx) = tokio::sync::mpsc::unbounded_channel();
 
         let stats = ClientStatsController::new(client_stats_id, client_type);
 
-        let mut task_client_stats_sender = task_client.fork("stats_sender");
-        task_client_stats_sender.disarm();
-
         (
             StatisticsControl {
                 stats,
                 stats_rx,
                 report_tx,
                 reporting_config,
-                task_client,
             },
-            ClientStatsSender::new(Some(stats_tx), task_client_stats_sender),
+            ClientStatsSender::new(Some(stats_tx), shutdown_token),
         )
     }
 
@@ -99,7 +88,8 @@ impl StatisticsControl {
         }
     }
 
-    async fn run(&mut self) {
+    // manually control the shutdown mechanism as we don't want to get interrupted mid-snapshot
+    pub async fn run(&mut self, shutdown_token: ShutdownToken) {
         tracing::debug!("Started StatisticsControl with graceful shutdown support");
 
         #[cfg(not(target_arch = "wasm32"))]
@@ -129,10 +119,10 @@ impl StatisticsControl {
         let mut snapshot_interval =
             gloo_timers::future::IntervalStream::new(SNAPSHOT_INTERVAL.as_millis() as u32);
 
-        while !self.task_client.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                _ = self.task_client.recv() => {
+                _ = shutdown_token.cancelled() => {
                     tracing::trace!("StatisticsControl: Received shutdown");
                     break;
                 },
@@ -157,37 +147,34 @@ impl StatisticsControl {
                 }
 
                 _ = local_report_interval.next() => {
-                    self.stats.local_report(&mut self.task_client);
+                    self.stats.local_report();
                 }
             }
         }
         tracing::debug!("StatisticsControl: Exiting");
     }
 
-    pub(crate) fn start(mut self) {
-        spawn_future!(
-            async move {
-                self.run().await;
-            },
-            "StatisticsControl"
-        )
-    }
-
     pub(crate) fn create_and_start(
         reporting_config: StatsReporting,
         client_type: String,
         client_stats_id: String,
         report_tx: InputMessageSender,
-        task_client: TaskClient,
+        shutdown_tracker: &ShutdownTracker,
     ) -> ClientStatsSender {
-        let (controller, sender) = Self::create(
+        let (mut controller, sender) = Self::create(
             reporting_config,
             client_type,
             client_stats_id,
             report_tx,
-            task_client,
+            shutdown_tracker.child_shutdown_token(),
+        );
+        let shutdown_token = shutdown_tracker.clone_shutdown_token();
+        shutdown_tracker.try_spawn_named(
+            async move {
+                controller.run(shutdown_token).await;
+            },
+            "StatisticsControl",
         );
-        controller.start();
         sender
     }
 }

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

@@ -1,11 +1,9 @@
 // Copyright 2021-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::spawn_future;
 pub(crate) use accessor::{TopologyAccessor, TopologyReadPermit};
 use futures::StreamExt;
 use nym_sphinx::addressing::nodes::NodeIdentity;
-use nym_task::TaskClient;
 use nym_topology::NymTopologyError;
 use std::time::Duration;
 use tracing::*;
@@ -41,8 +39,6 @@ pub struct TopologyRefresher {
 
     refresh_rate: Duration,
     consecutive_failure_count: usize,
-
-    task_client: TaskClient,
 }
 
 impl TopologyRefresher {
@@ -50,14 +46,12 @@ impl TopologyRefresher {
         cfg: TopologyRefresherConfig,
         topology_accessor: TopologyAccessor,
         topology_provider: Box<dyn TopologyProvider + Send + Sync>,
-        task_client: TaskClient,
     ) -> Self {
         TopologyRefresher {
             topology_provider,
             topology_accessor,
             refresh_rate: cfg.refresh_rate,
             consecutive_failure_count: 0,
-            task_client,
         }
     }
 
@@ -144,40 +138,30 @@ impl TopologyRefresher {
         }
     }
 
-    pub fn start(mut self) {
-        spawn_future!(
-            async move {
-                debug!("Started TopologyRefresher with graceful shutdown support");
+    // it's perfectly fine if task is interrupted mid-refresh
+    // there's no data to persist or send over
+    pub async fn run(&mut self) {
+        debug!("Started TopologyRefresher with graceful shutdown support");
 
-                #[cfg(not(target_arch = "wasm32"))]
-                let mut interval = tokio_stream::wrappers::IntervalStream::new(
-                    tokio::time::interval(self.refresh_rate),
-                );
+        #[cfg(not(target_arch = "wasm32"))]
+        let mut interval =
+            tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(self.refresh_rate));
 
-                #[cfg(target_arch = "wasm32")]
-                let mut interval =
-                    gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
+        #[cfg(target_arch = "wasm32")]
+        let mut interval =
+            gloo_timers::future::IntervalStream::new(self.refresh_rate.as_millis() as u32);
 
-                // We already have an initial topology, so no need to refresh it immediately.
-                // My understanding is that js setInterval does not fire immediately, so it's not
-                // needed there.
-                #[cfg(not(target_arch = "wasm32"))]
-                interval.next().await;
+        // We already have an initial topology, so no need to refresh it immediately.
+        // My understanding is that js setInterval does not fire immediately, so it's not
+        // needed there.
+        #[cfg(not(target_arch = "wasm32"))]
+        interval.next().await;
 
-                while !self.task_client.is_shutdown() {
-                    tokio::select! {
-                        _ = interval.next() => {
-                            self.try_refresh().await;
-                        },
-                        _ = self.task_client.recv() => {
-                            tracing::trace!("TopologyRefresher: Received shutdown");
-                        },
-                    }
-                }
-                self.task_client.recv_timeout().await;
-                tracing::debug!("TopologyRefresher: Exiting");
-            },
-            "TopologyRefresher"
-        )
+        while interval.next().await.is_some() {
+            self.try_refresh().await;
+        }
+
+        // this should never get triggered
+        error!("topology refresher interval has been exhausted!")
     }
 }

common/client-core/src/client/topology_control/nym_api_provider.rs

@@ -2,8 +2,10 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use async_trait::async_trait;
+use nym_mixnet_contract_common::EpochRewardedSet;
 use nym_topology::provider_trait::{ToTopologyMetadata, TopologyProvider};
 use nym_topology::NymTopology;
+use nym_validator_client::nym_api::NymApiClientExt;
 use rand::prelude::SliceRandom;
 use rand::thread_rng;
 use std::cmp::min;
@@ -39,30 +41,43 @@ impl Config {
 pub struct NymApiTopologyProvider {
     config: Config,
 
-    validator_client: nym_validator_client::client::NymApiClient,
+    validator_client: nym_http_api_client::Client,
     nym_api_urls: Vec<Url>,
     currently_used_api: usize,
+    use_bincode: bool,
 }
 
 impl NymApiTopologyProvider {
     pub fn new(
         config: impl Into<Config>,
         mut nym_api_urls: Vec<Url>,
-        mut validator_client: nym_validator_client::client::NymApiClient,
+        validator_client: nym_http_api_client::Client,
     ) -> Self {
         nym_api_urls.shuffle(&mut thread_rng());
-        validator_client.change_nym_api(nym_api_urls[0].clone());
-
-        NymApiTopologyProvider {
+        let mut provider = NymApiTopologyProvider {
             config: config.into(),
             validator_client,
             nym_api_urls,
             currently_used_api: 0,
-        }
+            use_bincode: true,
+        };
+        // Set all API URLs - the client will try them in order with automatic failover
+        provider.validator_client.change_base_urls(
+            provider
+                .nym_api_urls
+                .iter()
+                .map(|u| u.clone().into())
+                .collect(),
+        );
+        provider
     }
 
     pub fn disable_bincode(&mut self) {
-        self.validator_client.use_bincode = false;
+        self.use_bincode = false;
+        // Note: The unified client doesn't support toggling bincode after creation.
+        // This would require recreating the client without bincode.
+        // For now, we'll track the preference but it won't take effect.
+        warn!("Disabling bincode on existing client is not currently supported");
     }
 
     fn use_next_nym_api(&mut self) {
@@ -72,8 +87,19 @@ impl NymApiTopologyProvider {
         }
 
         self.currently_used_api = (self.currently_used_api + 1) % self.nym_api_urls.len();
-        self.validator_client
-            .change_nym_api(self.nym_api_urls[self.currently_used_api].clone())
+
+        // Provide all URLs starting from the next one in rotation order
+        // This enables automatic failover to other endpoints
+        let rotated_urls: Vec<_> = self
+            .nym_api_urls
+            .iter()
+            .cycle()
+            .skip(self.currently_used_api)
+            .take(self.nym_api_urls.len())
+            .map(|u| u.clone().into())
+            .collect();
+
+        self.validator_client.change_base_urls(rotated_urls)
     }
 
     async fn get_current_compatible_topology(&mut self) -> Option<NymTopology> {
@@ -99,8 +125,13 @@ impl NymApiTopologyProvider {
                 .filter(|n| n.performance.round_to_integer() >= self.config.min_node_performance())
                 .collect::<Vec<_>>();
 
-            NymTopology::new(metadata.to_topology_metadata(), rewarded_set, Vec::new())
-                .with_skimmed_nodes(&nodes_filtered)
+            let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
+            NymTopology::new(
+                metadata.to_topology_metadata(),
+                epoch_rewarded_set,
+                Vec::new(),
+            )
+            .with_skimmed_nodes(&nodes_filtered)
         } else {
             // if we're not using extended topology, we're only getting active set mixnodes and gateways
 
@@ -148,8 +179,13 @@ impl NymApiTopologyProvider {
                 }
             }
 
-            NymTopology::new(metadata.to_topology_metadata(), rewarded_set, Vec::new())
-                .with_skimmed_nodes(&nodes)
+            let epoch_rewarded_set: EpochRewardedSet = rewarded_set.into();
+            NymTopology::new(
+                metadata.to_topology_metadata(),
+                epoch_rewarded_set,
+                Vec::new(),
+            )
+            .with_skimmed_nodes(&nodes)
         };
 
         if !topology.is_minimally_routable() {

common/client-core/src/error.rs

@@ -4,6 +4,7 @@
 use crate::client::mix_traffic::transceiver::ErasedGatewayError;
 use nym_crypto::asymmetric::ed25519::Ed25519RecoveryError;
 use nym_gateway_client::error::GatewayClientError;
+use nym_task::RegistryAccessError;
 use nym_topology::node::RoutingNodeError;
 use nym_topology::{NodeId, NymTopologyError};
 use nym_validator_client::nym_api::error::NymAPIError;
@@ -56,10 +57,7 @@ pub enum ClientCoreError {
     ListOfNymApisIsEmpty,
 
     #[error("failed to resolve a query to nym API: {source}")]
-    NymApiQueryFailure {
-        #[from]
-        source: NymAPIError,
-    },
+    NymApiQueryFailure { source: Box<NymAPIError> },
 
     #[error(
         "the current network topology seem to be insufficient to route any packets through:\n\t{0}"
@@ -245,6 +243,9 @@ pub enum ClientCoreError {
 
     #[error("failed to select valid gateway due to incomputable latency")]
     GatewaySelectionFailure { source: WeightedError },
+
+    #[error("Could not access task registry, {0}")]
+    RegistryAccess(#[from] RegistryAccessError),
 }
 
 impl From<tungstenite::Error> for ClientCoreError {
@@ -255,6 +256,14 @@ impl From<tungstenite::Error> for ClientCoreError {
     }
 }
 
+impl From<NymAPIError> for ClientCoreError {
+    fn from(err: NymAPIError) -> ClientCoreError {
+        ClientCoreError::NymApiQueryFailure {
+            source: Box::new(err),
+        }
+    }
+}
+
 /// Set of messages that the client can send to listeners via the task manager
 #[derive(Debug)]
 pub enum ClientCoreStatusMessage {

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

@@ -7,7 +7,8 @@ use futures::{SinkExt, StreamExt};
 use nym_crypto::asymmetric::ed25519;
 use nym_gateway_client::GatewayClient;
 use nym_topology::node::RoutingNode;
-use nym_validator_client::client::IdentityKeyRef;
+use nym_validator_client::client::{IdentityKeyRef, NymApiClientExt};
+use nym_validator_client::nym_nodes::SkimmedNodesWithMetadata;
 use nym_validator_client::UserAgent;
 use rand::{seq::SliceRandom, Rng};
 #[cfg(unix)]
@@ -83,6 +84,48 @@ struct GatewayWithLatency<'a, G: ConnectableGateway> {
     latency: Duration,
 }
 
+// Helper to collect all pages of entry nodes - replicates NymApiClient's convenience method
+async fn get_all_basic_entry_nodes_with_metadata(
+    client: &nym_http_api_client::Client,
+    use_bincode: bool,
+) -> Result<SkimmedNodesWithMetadata, ClientCoreError> {
+    // Get first page to obtain metadata
+    let mut page = 0;
+    let res = client
+        .get_basic_entry_assigned_nodes_v2(false, Some(page), None, use_bincode)
+        .await?;
+    let mut nodes = res.nodes.data;
+    let metadata = res.metadata;
+
+    if res.nodes.pagination.total == nodes.len() {
+        return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
+    }
+
+    page += 1;
+
+    // Collect remaining pages
+    loop {
+        let mut res = client
+            .get_basic_entry_assigned_nodes_v2(false, Some(page), None, use_bincode)
+            .await?;
+
+        if !metadata.consistency_check(&res.metadata) {
+            return Err(ClientCoreError::ValidatorClientError(
+                nym_validator_client::ValidatorClientError::InconsistentPagedMetadata,
+            ));
+        }
+
+        nodes.append(&mut res.nodes.data);
+        if nodes.len() < res.nodes.pagination.total {
+            page += 1
+        } else {
+            break;
+        }
+    }
+
+    Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
+}
+
 impl<'a, G: ConnectableGateway> GatewayWithLatency<'a, G> {
     fn new(gateway: &'a G, latency: Duration) -> Self {
         GatewayWithLatency { gateway, latency }
@@ -99,16 +142,28 @@ pub async fn gateways_for_init<R: Rng>(
     let nym_api = nym_apis
         .choose(rng)
         .ok_or(ClientCoreError::ListOfNymApisIsEmpty)?;
-    let client = if let Some(user_agent) = user_agent {
-        nym_validator_client::client::NymApiClient::new_with_user_agent(nym_api.clone(), user_agent)
-    } else {
-        nym_validator_client::client::NymApiClient::new(nym_api.clone())
-    };
+
+    // Use the unified HTTP client directly with optional user agent
+    let mut builder = nym_http_api_client::Client::builder(nym_api.clone())
+        .map_err(|e| {
+            ClientCoreError::ValidatorClientError(nym_validator_client::ValidatorClientError::from(
+                e,
+            ))
+        })?
+        .with_bincode(); // Use bincode for better performance
+
+    if let Some(user_agent) = user_agent {
+        builder = builder.with_user_agent(user_agent);
+    }
+
+    let client = builder.build().map_err(|e| {
+        ClientCoreError::ValidatorClientError(nym_validator_client::ValidatorClientError::from(e))
+    })?;
 
     tracing::debug!("Fetching list of gateways from: {nym_api}");
 
-    let gateways = client
-        .get_all_basic_entry_assigned_nodes_with_metadata()
+    // Use our helper to handle pagination
+    let gateways = get_all_basic_entry_nodes_with_metadata(&client, true)
         .await?
         .nodes;
     info!("nym api reports {} gateways", gateways.len());

common/client-core/src/lib.rs

@@ -17,7 +17,9 @@ pub use nym_topology::{
     HardcodedTopologyProvider, NymRouteProvider, NymTopology, NymTopologyError, TopologyProvider,
 };
 
+#[deprecated(note = "use spawn_future from nym_task crate instead")]
 #[cfg(target_arch = "wasm32")]
+#[track_caller]
 pub fn spawn_future<F>(future: F)
 where
     F: Future<Output = ()> + 'static,
@@ -25,9 +27,7 @@ where
     wasm_bindgen_futures::spawn_local(future);
 }
 
-// TODO: expose similar API to the rest of the codebase,
-// perhaps with some simple trait for a task to define its name
-
+#[deprecated(note = "use spawn_future from nym_task crate instead")]
 #[cfg(not(target_arch = "wasm32"))]
 #[track_caller]
 pub fn spawn_future<F>(future: F)
@@ -37,35 +37,3 @@ where
 {
     tokio::spawn(future);
 }
-
-#[cfg(not(target_arch = "wasm32"))]
-#[track_caller]
-pub fn spawn_named_future<F>(future: F, name: &str)
-where
-    F: Future + Send + 'static,
-    F::Output: Send + 'static,
-{
-    cfg_if::cfg_if! {if #[cfg(tokio_unstable)] {
-        #[allow(clippy::expect_used)]
-        tokio::task::Builder::new().name(name).spawn(future).expect("failed to spawn future");
-    } else {
-        let _ = name;
-        tracing::debug!(r#"the underlying binary hasn't been built with `RUSTFLAGS="--cfg tokio_unstable"` - the future naming won't do anything"#);
-        spawn_future(future);
-    }}
-}
-
-#[macro_export]
-macro_rules! spawn_future {
-    ($future:expr) => {{
-        $crate::spawn_future($future)
-    }};
-    ($future:expr, $name:expr) => {{
-        cfg_if::cfg_if! {if #[cfg(not(target_arch = "wasm32"))] {
-            $crate::spawn_named_future($future, $name)
-        } else {
-            let _ = $name;
-            $crate::spawn_future($future)
-        }}
-    }};
-}

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

@@ -40,7 +40,7 @@ where
     pub async fn flush_on_shutdown(
         mut self,
         mem_state: CombinedReplyStorage,
-        mut shutdown: nym_task::TaskClient,
+        shutdown: nym_task::ShutdownToken,
     ) {
         use tracing::{debug, error, info};
 
@@ -50,7 +50,7 @@ where
             return;
         }
 
-        shutdown.recv().await;
+        shutdown.cancelled().await;
 
         info!("PersistentReplyStorage is flushing all reply-related data to underlying storage");
         if let Err(err) = self.backend.flush_surb_storage(&mem_state).await {

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

@@ -12,7 +12,7 @@ use crate::socket_state::{ws_fd, PartiallyDelegatedHandle, SocketState};
 use crate::traits::GatewayPacketRouter;
 use crate::{cleanup_socket_message, try_decrypt_binary_message};
 use futures::{SinkExt, StreamExt};
-use nym_bandwidth_controller::{BandwidthController, BandwidthStatusMessage};
+use nym_bandwidth_controller::BandwidthController;
 use nym_credential_storage::ephemeral_storage::EphemeralStorage as EphemeralCredentialStorage;
 use nym_credential_storage::storage::Storage as CredentialStorage;
 use nym_credentials::CredentialSpendingData;
@@ -27,7 +27,7 @@ use nym_gateway_requests::{
 use nym_sphinx::forwarding::packet::MixPacket;
 use nym_statistics_common::clients::connection::ConnectionStatsEvent;
 use nym_statistics_common::clients::ClientStatsSender;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use nym_validator_client::nyxd::contract_traits::DkgQueryClient;
 use rand::rngs::OsRng;
 use std::sync::Arc;
@@ -109,7 +109,7 @@ pub struct GatewayClient<C, St = EphemeralCredentialStorage> {
     connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
 
     /// Listen to shutdown messages and send notifications back to the task manager
-    task_client: TaskClient,
+    shutdown_token: ShutdownToken,
 }
 
 impl<C, St> GatewayClient<C, St> {
@@ -124,7 +124,7 @@ impl<C, St> GatewayClient<C, St> {
         bandwidth_controller: Option<BandwidthController<C, St>>,
         stats_reporter: ClientStatsSender,
         #[cfg(unix)] connection_fd_callback: Option<Arc<dyn Fn(RawFd) + Send + Sync>>,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> Self {
         GatewayClient {
             cfg,
@@ -141,7 +141,7 @@ impl<C, St> GatewayClient<C, St> {
             negotiated_protocol: None,
             #[cfg(unix)]
             connection_fd_callback,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -293,7 +293,7 @@ impl<C, St> GatewayClient<C, St> {
 
         loop {
             tokio::select! {
-                _ = self.task_client.recv() => {
+                _ = self.shutdown_token.cancelled() => {
                     log::trace!("GatewayClient control response: Received shutdown");
                     log::debug!("GatewayClient control response: Exiting");
                     break Err(GatewayClientError::ConnectionClosedGatewayShutdown);
@@ -514,7 +514,7 @@ impl<C, St> GatewayClient<C, St> {
                 self.cfg.bandwidth.require_tickets,
                 derive_aes256_gcm_siv_key,
                 #[cfg(not(target_arch = "wasm32"))]
-                self.task_client.clone(),
+                self.shutdown_token.clone(),
             )
             .await
             .map_err(GatewayClientError::RegistrationFailure),
@@ -631,9 +631,6 @@ impl<C, St> GatewayClient<C, St> {
                 self.negotiated_protocol = protocol_version;
                 log::debug!("authenticated: {status}, bandwidth remaining: {bandwidth_remaining}");
 
-                self.task_client.send_status_msg(Box::new(
-                    BandwidthStatusMessage::RemainingBandwidth(bandwidth_remaining),
-                ));
                 Ok(())
             }
             ServerResponse::Error { message } => Err(GatewayClientError::GatewayError(message)),
@@ -1069,7 +1066,7 @@ impl<C, St> GatewayClient<C, St> {
                                 .expect("no shared key present even though we're authenticated!"),
                         ),
                         self.bandwidth.clone(),
-                        self.task_client.clone(),
+                        self.shutdown_token.clone(),
                     )
                 }
                 _ => unreachable!(),
@@ -1143,8 +1140,8 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
         // perfectly fine here, because it's not meant to be used
         let (ack_tx, _) = mpsc::unbounded();
         let (mix_tx, _) = mpsc::unbounded();
-        let task_client = TaskClient::dummy();
-        let packet_router = PacketRouter::new(ack_tx, mix_tx, task_client.clone());
+        let shutdown_token = ShutdownToken::default();
+        let packet_router = PacketRouter::new(ack_tx, mix_tx, shutdown_token.clone());
 
         GatewayClient {
             cfg: GatewayClientConfig::default().with_disabled_credentials_mode(true),
@@ -1157,11 +1154,11 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
             connection: SocketState::NotConnected,
             packet_router,
             bandwidth_controller: None,
-            stats_reporter: ClientStatsSender::new(None, task_client.clone()),
+            stats_reporter: ClientStatsSender::new(None, shutdown_token.clone()),
             negotiated_protocol: None,
             #[cfg(unix)]
             connection_fd_callback,
-            task_client,
+            shutdown_token,
         }
     }
 
@@ -1170,7 +1167,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
         packet_router: PacketRouter,
         bandwidth_controller: Option<BandwidthController<C, St>>,
         stats_reporter: ClientStatsSender,
-        task_client: TaskClient,
+        shutdown_token: ShutdownToken,
     ) -> GatewayClient<C, St> {
         // invariants that can't be broken
         // (unless somebody decided to expose some field that wasn't meant to be exposed)
@@ -1193,7 +1190,7 @@ impl GatewayClient<InitOnly, EphemeralCredentialStorage> {
             negotiated_protocol: self.negotiated_protocol,
             #[cfg(unix)]
             connection_fd_callback: self.connection_fd_callback,
-            task_client,
+            shutdown_token,
         }
     }
 }

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

@@ -7,7 +7,7 @@
 use crate::error::GatewayClientError;
 use crate::GatewayPacketRouter;
 use futures::channel::mpsc;
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 
 pub type MixnetMessageSender = mpsc::UnboundedSender<Vec<Vec<u8>>>;
 pub type MixnetMessageReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
@@ -19,14 +19,14 @@ pub type AcknowledgementReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
 pub struct PacketRouter {
     ack_sender: AcknowledgementSender,
     mixnet_message_sender: MixnetMessageSender,
-    shutdown: TaskClient,
+    shutdown: ShutdownToken,
 }
 
 impl PacketRouter {
     pub fn new(
         ack_sender: AcknowledgementSender,
         mixnet_message_sender: MixnetMessageSender,
-        shutdown: TaskClient,
+        shutdown: ShutdownToken,
     ) -> Self {
         PacketRouter {
             ack_sender,
@@ -42,7 +42,7 @@ impl PacketRouter {
         if let Err(err) = self.mixnet_message_sender.unbounded_send(received_messages) {
             // check if the failure is due to the shutdown being in progress and thus the receiver channel
             // having already been dropped
-            if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
+            if self.shutdown.is_cancelled() {
                 // This should ideally not happen, but it's ok
                 tracing::warn!("Failed to send mixnet messages due to receiver task shutdown");
                 return Err(GatewayClientError::ShutdownInProgress);
@@ -58,7 +58,7 @@ impl PacketRouter {
         if let Err(err) = self.ack_sender.unbounded_send(received_acks) {
             // check if the failure is due to the shutdown being in progress and thus the receiver channel
             // having already been dropped
-            if self.shutdown.is_shutdown_poll() || self.shutdown.is_dummy() {
+            if self.shutdown.is_cancelled() {
                 // This should ideally not happen, but it's ok
                 tracing::warn!("Failed to send acks due to receiver task shutdown");
                 return Err(GatewayClientError::ShutdownInProgress);
@@ -69,10 +69,6 @@ impl PacketRouter {
         }
         Ok(())
     }
-
-    pub fn disarm(&mut self) {
-        self.shutdown.disarm();
-    }
 }
 
 impl GatewayPacketRouter for PacketRouter {

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

@@ -11,7 +11,7 @@ use futures::stream::{SplitSink, SplitStream};
 use futures::{SinkExt, StreamExt};
 use nym_gateway_requests::shared_key::SharedGatewayKey;
 use nym_gateway_requests::{SensitiveServerResponse, ServerResponse, SimpleGatewayRequestsError};
-use nym_task::TaskClient;
+use nym_task::ShutdownToken;
 use si_scale::helpers::bibytes2;
 use std::os::raw::c_int as RawFd;
 use std::sync::Arc;
@@ -87,13 +87,13 @@ impl PartiallyDelegatedRouter {
         }
     }
 
-    async fn run(mut self, mut split_stream: SplitStream<WsConn>, mut task_client: TaskClient) {
+    async fn run(mut self, mut split_stream: SplitStream<WsConn>, shutdown_token: ShutdownToken) {
         let mut chunked_stream = (&mut split_stream).ready_chunks(8);
         let ret: Result<_, GatewayClientError> = loop {
             tokio::select! {
                 biased;
                 // received system-wide shutdown
-                _ = task_client.recv() => {
+                _ = shutdown_token.cancelled() => {
                     log::trace!("GatewayClient listener: Received shutdown");
                     log::debug!("GatewayClient listener: Exiting");
                     return;
@@ -118,11 +118,7 @@ impl PartiallyDelegatedRouter {
 
         let return_res = match ret {
             Err(err) => self.stream_return.send(Err(err)),
-            Ok(_) => {
-                self.packet_router.disarm();
-                task_client.disarm();
-                self.stream_return.send(Ok(split_stream))
-            }
+            Ok(_) => self.stream_return.send(Ok(split_stream)),
         };
 
         if return_res.is_err() {
@@ -266,8 +262,8 @@ impl PartiallyDelegatedRouter {
         Ok(plaintexts)
     }
 
-    fn spawn(self, split_stream: SplitStream<WsConn>, task_client: TaskClient) {
-        let fut = async move { self.run(split_stream, task_client).await };
+    fn spawn(self, split_stream: SplitStream<WsConn>, shutdown_token: ShutdownToken) {
+        let fut = async move { self.run(split_stream, shutdown_token).await };
 
         #[cfg(target_arch = "wasm32")]
         wasm_bindgen_futures::spawn_local(fut);
@@ -283,7 +279,7 @@ impl PartiallyDelegatedHandle {
         packet_router: PacketRouter,
         shared_key: Arc<SharedGatewayKey>,
         client_bandwidth: ClientBandwidth,
-        shutdown: TaskClient,
+        shutdown: ShutdownToken,
     ) -> Self {
         // when called for, it NEEDS TO yield back the stream so that we could merge it and
         // read control request responses.

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

@@ -5,8 +5,8 @@ use crate::nyxd::{self, NyxdClient};
 use crate::signing::direct_wallet::DirectSecp256k1HdWallet;
 use crate::signing::signer::{NoSigner, OfflineSigner};
 use crate::{
-    nym_api, DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient,
-    ReqwestRpcClient, ValidatorClientError,
+    DirectSigningReqwestRpcValidatorClient, QueryReqwestRpcValidatorClient, ReqwestRpcClient,
+    ValidatorClientError,
 };
 use nym_api_requests::ecash::models::{
     AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -20,11 +20,9 @@ use nym_api_requests::ecash::{
     PartialExpirationDateSignatureResponse, VerificationKeyResponse,
 };
 use nym_api_requests::models::{
-    ApiHealthResponse, GatewayBondAnnotated, GatewayCoreStatusResponse,
-    HistoricalPerformanceResponse, MixnodeCoreStatusResponse, MixnodeStatusResponse,
-    NymNodeDescription, RewardEstimationResponse, StakeSaturationResponse,
+    ApiHealthResponse, GatewayCoreStatusResponse, HistoricalPerformanceResponse,
+    MixnodeCoreStatusResponse, NymNodeDescription,
 };
-use nym_api_requests::models::{LegacyDescribedGateway, MixNodeBondAnnotated};
 use nym_api_requests::nym_nodes::{
     NodesByAddressesResponse, SemiSkimmedNodesWithMetadata, SkimmedNode, SkimmedNodesWithMetadata,
 };
@@ -153,7 +151,7 @@ impl Config {
 pub struct Client<C, S = NoSigner> {
     // ideally they would have been read-only, but unfortunately rust doesn't have such features
     // #[deprecated(note = "please use `nym_api_client` instead")]
-    pub nym_api: nym_api::Client,
+    pub nym_api: nym_http_api_client::Client,
     // pub nym_api_client: NymApiClient,
     pub nyxd: NyxdClient<C, S>,
 }
@@ -214,7 +212,7 @@ impl Client<ReqwestRpcClient> {
 
 impl<C> Client<C> {
     pub fn new_with_rpc_client(config: Config, rpc_client: C) -> Self {
-        let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
+        let nym_api_client = nym_http_api_client::Client::new(config.api_url.clone(), None);
 
         Client {
             nym_api: nym_api_client,
@@ -228,7 +226,7 @@ impl<C, S> Client<C, S> {
     where
         S: OfflineSigner,
     {
-        let nym_api_client = nym_api::Client::new(config.api_url.clone(), None);
+        let nym_api_client = nym_http_api_client::Client::new(config.api_url.clone(), None);
 
         Client {
             nym_api: nym_api_client,
@@ -249,65 +247,6 @@ impl<C, S> Client<C, S> {
         self.nym_api.change_base_urls(vec![new_endpoint.into()])
     }
 
-    #[deprecated]
-    pub async fn get_cached_mixnodes(&self) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_mixnodes_detailed(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnodes_detailed().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_mixnodes_detailed_unfiltered(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnodes_detailed_unfiltered().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_rewarded_mixnodes(
-        &self,
-    ) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_rewarded_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_rewarded_mixnodes_detailed(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
-        Ok(self.nym_api.get_rewarded_mixnodes_detailed().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_active_mixnodes(
-        &self,
-    ) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_active_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_active_mixnodes_detailed(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, ValidatorClientError> {
-        Ok(self.nym_api.get_active_mixnodes_detailed().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_gateways(&self) -> Result<Vec<GatewayBond>, ValidatorClientError> {
-        Ok(self.nym_api.get_gateways().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_gateways_detailed_unfiltered(
-        &self,
-    ) -> Result<Vec<GatewayBondAnnotated>, ValidatorClientError> {
-        Ok(self.nym_api.get_gateways_detailed_unfiltered().await?)
-    }
-
     pub async fn get_full_node_performance_history(
         &self,
         node_id: NodeId,
@@ -385,38 +324,25 @@ impl<C, S> Client<C, S> {
     }
 }
 
+/// DEPRECATED: Use nym_http_api_client::Client with from_network() or with_bincode() instead
+#[deprecated(
+    since = "1.2.0",
+    note = "Use nym_http_api_client::Client::from_network() or ClientBuilder::with_bincode() instead"
+)]
 #[derive(Clone)]
 pub struct NymApiClient {
     pub use_bincode: bool,
-    pub nym_api: nym_api::Client,
+    pub nym_api: nym_http_api_client::Client,
     // TODO: perhaps if we really need it at some (currently I don't see any reasons for it)
     // we could re-implement the communication with the REST API on port 1317
 }
 
-impl From<nym_api::Client> for NymApiClient {
-    fn from(nym_api: nym_api::Client) -> Self {
-        NymApiClient {
-            use_bincode: false,
-            nym_api,
-        }
-    }
-}
-
 // we have to allow the use of deprecated method here as they're calling the deprecated trait methods
 #[allow(deprecated)]
 impl NymApiClient {
-    pub fn new(api_url: Url) -> Self {
-        let nym_api = nym_api::Client::new(api_url, None);
-
-        NymApiClient {
-            use_bincode: true,
-            nym_api,
-        }
-    }
-
     #[cfg(not(target_arch = "wasm32"))]
     pub fn new_with_timeout(api_url: Url, timeout: std::time::Duration) -> Self {
-        let nym_api = nym_api::Client::new(api_url, Some(timeout));
+        let nym_api = nym_http_api_client::Client::new(api_url, Some(timeout));
 
         NymApiClient {
             use_bincode: true,
@@ -431,10 +357,10 @@ impl NymApiClient {
     }
 
     pub fn new_with_user_agent(api_url: Url, user_agent: impl Into<UserAgent>) -> Self {
-        let nym_api = nym_api::Client::builder::<_, ValidatorClientError>(api_url)
+        let nym_api = nym_http_api_client::Client::builder(api_url)
             .expect("invalid api url")
             .with_user_agent(user_agent.into())
-            .build::<ValidatorClientError>()
+            .build()
             .expect("failed to build nym api client");
 
         NymApiClient {
@@ -571,37 +497,6 @@ impl NymApiClient {
         Ok(self.nym_api.health().await?)
     }
 
-    #[deprecated]
-    pub async fn get_cached_active_mixnodes(
-        &self,
-    ) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_active_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_rewarded_mixnodes(
-        &self,
-    ) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_rewarded_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_mixnodes(&self) -> Result<Vec<MixNodeDetails>, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnodes().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_gateways(&self) -> Result<Vec<GatewayBond>, ValidatorClientError> {
-        Ok(self.nym_api.get_gateways().await?)
-    }
-
-    #[deprecated]
-    pub async fn get_cached_described_gateways(
-        &self,
-    ) -> Result<Vec<LegacyDescribedGateway>, ValidatorClientError> {
-        Ok(self.nym_api.get_gateways_described().await?)
-    }
-
     pub async fn get_all_described_nodes(
         &self,
     ) -> Result<Vec<NymNodeDescription>, ValidatorClientError> {
@@ -668,30 +563,6 @@ impl NymApiClient {
             .await?)
     }
 
-    #[deprecated]
-    pub async fn get_mixnode_status(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<MixnodeStatusResponse, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnode_status(mix_id).await?)
-    }
-
-    #[deprecated]
-    pub async fn get_mixnode_reward_estimation(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<RewardEstimationResponse, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnode_reward_estimation(mix_id).await?)
-    }
-
-    #[deprecated]
-    pub async fn get_mixnode_stake_saturation(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<StakeSaturationResponse, ValidatorClientError> {
-        Ok(self.nym_api.get_mixnode_stake_saturation(mix_id).await?)
-    }
-
     pub async fn blind_sign(
         &self,
         request_body: &BlindSignRequestBody,

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

@@ -3,7 +3,6 @@
 
 use crate::nyxd::contract_traits::{DkgQueryClient, PagedDkgQueryClient};
 use crate::nyxd::error::NyxdError;
-use crate::NymApiClient;
 use nym_coconut_dkg_common::types::{EpochId, NodeIndex};
 use nym_coconut_dkg_common::verification_key::ContractVKShare;
 use nym_compact_ecash::error::CompactEcashError;
@@ -15,7 +14,7 @@ use url::Url;
 // TODO: it really doesn't feel like this should live in this crate.
 #[derive(Clone)]
 pub struct EcashApiClient {
-    pub api_client: NymApiClient,
+    pub api_client: nym_http_api_client::Client,
     pub verification_key: VerificationKeyAuth,
     pub node_id: NodeIndex,
     pub cosmos_address: cosmrs::AccountId,
@@ -25,10 +24,15 @@ impl Display for EcashApiClient {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "[id: {}] {} @ {}",
+            "[id: {}] {} @ ({})",
             self.node_id,
             self.cosmos_address,
-            self.api_client.api_url()
+            self.api_client
+                .base_urls()
+                .iter()
+                .map(|url| url.to_string())
+                .collect::<Vec<String>>()
+                .join(", ")
         )
     }
 }
@@ -60,6 +64,9 @@ pub enum EcashApiError {
         source: CompactEcashError,
     },
 
+    #[error("failed to create API client: {0}")]
+    ClientError(String),
+
     #[error("the provided account address is malformed: {source}")]
     MalformedAccountAddress {
         #[from]
@@ -89,8 +96,13 @@ impl TryFrom<ContractVKShare> for EcashApiClient {
         // In non-client applications this resolver can cause warning logs about H2 connection
         // failure. This indicates that the long lived https connection was closed by the remote
         // peer and the resolver will have to reconnect. It should not impact actual functionality
+        let api_client = nym_http_api_client::Client::builder(url_address)
+            .map_err(|e| EcashApiError::ClientError(e.to_string()))?
+            .build()
+            .map_err(|e| EcashApiError::ClientError(e.to_string()))?;
+
         Ok(EcashApiClient {
-            api_client: NymApiClient::new(url_address),
+            api_client,
             verification_key: VerificationKeyAuth::try_from_bs58(&share.share)?,
             node_id: share.node_index,
             cosmos_address: share.owner.as_str().parse()?,

common/client-libs/validator-client/src/connection_tester.rs

@@ -1,7 +1,8 @@
+use crate::nym_api::NymApiClientExt;
 use crate::nyxd::contract_traits::MixnetQueryClient;
 use crate::nyxd::error::NyxdError;
 use crate::nyxd::Config as ClientConfig;
-use crate::{NymApiClient, QueryHttpRpcNyxdClient, ValidatorClientError};
+use crate::{QueryHttpRpcNyxdClient, ValidatorClientError};
 use colored::Colorize;
 use core::fmt;
 use itertools::Itertools;
@@ -87,8 +88,17 @@ fn setup_connection_tests<H: BuildHasher + 'static>(
         }
     });
 
-    let api_connection_test_clients = api_urls.map(|(network, url)| {
-        ClientForConnectionTest::Api(network, url.clone(), NymApiClient::new(url))
+    let api_connection_test_clients = api_urls.filter_map(|(network, url)| {
+        match nym_http_api_client::Client::builder(url.clone()).and_then(|b| b.build()) {
+            Ok(client) => Some(ClientForConnectionTest::Api(network, url, client)),
+            Err(err) => {
+                eprintln!(
+                    "Failed to create API client for {}: {err}",
+                    network.network_name
+                );
+                None
+            }
+        }
     });
 
     nyxd_connection_test_clients.chain(api_connection_test_clients)
@@ -160,7 +170,7 @@ async fn test_nyxd_connection(
 async fn test_nym_api_connection(
     network: NymNetworkDetails,
     url: &Url,
-    client: &NymApiClient,
+    client: &nym_http_api_client::Client,
 ) -> ConnectionResult {
     let result = match timeout(
         Duration::from_secs(CONNECTION_TEST_TIMEOUT_SEC),
@@ -186,7 +196,7 @@ async fn test_nym_api_connection(
 
 enum ClientForConnectionTest {
     Nyxd(NymNetworkDetails, Url, Box<QueryHttpRpcNyxdClient>),
-    Api(NymNetworkDetails, Url, NymApiClient),
+    Api(NymNetworkDetails, Url, nym_http_api_client::Client),
 }
 
 impl ClientForConnectionTest {

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

@@ -9,8 +9,7 @@ use thiserror::Error;
 pub enum ValidatorClientError {
     #[error("nym api request failed: {source}")]
     NymAPIError {
-        #[from]
-        source: nym_api::error::NymAPIError,
+        source: Box<nym_api::error::NymAPIError>,
     },
 
     #[error("Tendermint RPC request failure: {0}")]
@@ -28,3 +27,11 @@ pub enum ValidatorClientError {
     #[error("No validator API url has been provided")]
     NoAPIUrlAvailable,
 }
+
+impl From<nym_api::error::NymAPIError> for ValidatorClientError {
+    fn from(source: nym_api::error::NymAPIError) -> Self {
+        ValidatorClientError::NymAPIError {
+            source: Box::new(source),
+        }
+    }
+}

common/client-libs/validator-client/src/lib.rs

@@ -14,7 +14,6 @@ pub mod signing;
 pub use crate::error::ValidatorClientError;
 pub use crate::rpc::reqwest::ReqwestRpcClient;
 pub use crate::signing::direct_wallet::DirectSecp256k1HdWallet;
-pub use client::NymApiClient;
 pub use client::{Client, Config, EcashApiClient};
 pub use nym_api_requests::*;
 pub use nym_http_api_client::UserAgent;

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

@@ -1,7 +1,6 @@
 // Copyright 2022-2023 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: Apache-2.0
 
-use nym_api_requests::models::RequestError;
 use nym_http_api_client::HttpClientError;
 
-pub type NymAPIError = HttpClientError<RequestError>;
+pub type NymAPIError = HttpClientError;

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

@@ -3,6 +3,7 @@
 
 use crate::nym_api::error::NymAPIError;
 use crate::nym_api::routes::{ecash, CORE_STATUS_COUNT, SINCE_ARG};
+use crate::nym_nodes::SkimmedNodesWithMetadata;
 use async_trait::async_trait;
 use nym_api_requests::ecash::models::{
     AggregatedCoinIndicesSignatureResponse, AggregatedExpirationDateSignatureResponse,
@@ -15,9 +16,8 @@ use nym_api_requests::ecash::models::{
 use nym_api_requests::ecash::VerificationKeyResponse;
 use nym_api_requests::models::{
     AnnotationResponse, ApiHealthResponse, BinaryBuildInformationOwned, ChainBlocksStatusResponse,
-    ChainStatusResponse, KeyRotationInfoResponse, LegacyDescribedMixNode, NodePerformanceResponse,
-    NodeRefreshBody, NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse,
-    SignerInformationResponse,
+    ChainStatusResponse, KeyRotationInfoResponse, NodePerformanceResponse, NodeRefreshBody,
+    NymNodeDescription, PerformanceHistoryResponse, RewardedSetResponse, SignerInformationResponse,
 };
 use nym_api_requests::nym_nodes::{
     NodesByAddressesRequestBody, NodesByAddressesResponse, PaginatedCachedNodesResponseV1,
@@ -31,26 +31,22 @@ pub use nym_api_requests::{
         VerifyEcashCredentialBody,
     },
     models::{
-        ComputeRewardEstParam, GatewayBondAnnotated, GatewayCoreStatusResponse,
-        GatewayStatusReportResponse, GatewayUptimeHistoryResponse, LegacyDescribedGateway,
-        MixNodeBondAnnotated, MixnodeCoreStatusResponse, MixnodeStatusReportResponse,
-        MixnodeStatusResponse, MixnodeUptimeHistoryResponse, RewardEstimationResponse,
-        StakeSaturationResponse, UptimeResponse,
+        GatewayCoreStatusResponse, GatewayStatusReportResponse, GatewayUptimeHistoryResponse,
+        MixnodeCoreStatusResponse, MixnodeStatusReportResponse, MixnodeStatusResponse,
+        MixnodeUptimeHistoryResponse, StakeSaturationResponse, UptimeResponse,
     },
-    nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SkimmedNode},
+    nym_nodes::{CachedNodesResponse, SemiSkimmedNode, SemiSkimmedNodesWithMetadata, SkimmedNode},
     NymNetworkDetailsResponse,
 };
-use nym_contracts_common::IdentityKey;
 use nym_http_api_client::{ApiClient, NO_PARAMS};
-use nym_mixnet_contract_common::mixnode::MixNodeDetails;
-use nym_mixnet_contract_common::{GatewayBond, IdentityKeyRef, NodeId, NymNodeDetails};
+use nym_mixnet_contract_common::{IdentityKeyRef, NodeId, NymNodeDetails};
 use std::net::IpAddr;
 use time::format_description::BorrowedFormatItem;
 use time::Date;
 use tracing::instrument;
 
+use crate::ValidatorClientError;
 pub use nym_coconut_dkg_common::types::EpochId;
-pub use nym_http_api_client::Client;
 
 pub mod error;
 pub mod routes;
@@ -62,6 +58,9 @@ pub fn rfc_3339_date() -> Vec<BorrowedFormatItem<'static>> {
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
 pub trait NymApiClientExt: ApiClient {
+    /// Get the current API URL being used by the client
+    fn api_url(&self) -> &url::Url;
+
     async fn health(&self) -> Result<ApiHealthResponse, NymAPIError> {
         self.get_json(
             &[
@@ -87,104 +86,6 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
-        self.get_json(&[routes::V1_API_VERSION, routes::MIXNODES], NO_PARAMS)
-            .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnodes_detailed(&self) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::MIXNODES,
-                routes::DETAILED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_gateways_detailed(&self) -> Result<Vec<GatewayBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::GATEWAYS,
-                routes::DETAILED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_gateways_detailed_unfiltered(
-        &self,
-    ) -> Result<Vec<GatewayBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::GATEWAYS,
-                routes::DETAILED_UNFILTERED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnodes_detailed_unfiltered(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::MIXNODES,
-                routes::DETAILED_UNFILTERED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_gateways(&self) -> Result<Vec<GatewayBond>, NymAPIError> {
-        self.get_json(&[routes::V1_API_VERSION, routes::GATEWAYS], NO_PARAMS)
-            .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_gateways_described(&self) -> Result<Vec<LegacyDescribedGateway>, NymAPIError> {
-        self.get_json(
-            &[routes::V1_API_VERSION, routes::GATEWAYS, routes::DESCRIBED],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnodes_described(&self) -> Result<Vec<LegacyDescribedMixNode>, NymAPIError> {
-        self.get_json(
-            &[routes::V1_API_VERSION, routes::MIXNODES, routes::DESCRIBED],
-            NO_PARAMS,
-        )
-        .await
-    }
-
     #[tracing::instrument(level = "debug", skip_all)]
     async fn get_node_performance_history(
         &self,
@@ -241,6 +142,156 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
+    async fn get_current_rewarded_set(&self) -> Result<RewardedSetResponse, NymAPIError> {
+        self.get_rewarded_set().await
+    }
+
+    async fn get_all_basic_nodes_with_metadata(
+        &self,
+    ) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
+        // unroll first loop iteration in order to obtain the metadata
+        let mut page = 0;
+        let res = self
+            .get_basic_nodes_v2(false, Some(page), None, true)
+            .await?;
+        let mut nodes = res.nodes.data;
+        let metadata = res.metadata;
+
+        if res.nodes.pagination.total == nodes.len() {
+            return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
+        }
+
+        page += 1;
+
+        loop {
+            let mut res = self
+                .get_basic_nodes_v2(false, Some(page), None, true)
+                .await?;
+
+            if !metadata.consistency_check(&res.metadata) {
+                // Create a custom error for inconsistent metadata
+                return Err(NymAPIError::InternalResponseInconsistency {
+                    url: self.api_url().clone(),
+                    details: "Inconsistent paged metadata".to_string(),
+                });
+            }
+
+            nodes.append(&mut res.nodes.data);
+            if nodes.len() >= res.nodes.pagination.total {
+                break;
+            } else {
+                page += 1
+            }
+        }
+
+        Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
+    }
+
+    async fn get_all_basic_active_mixing_assigned_nodes_with_metadata(
+        &self,
+    ) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
+        // Get all mixing nodes that are in the active/rewarded set
+        let mut page = 0;
+        let res = self
+            .get_basic_active_mixing_assigned_nodes_v2(false, Some(page), None, false)
+            .await?;
+
+        let metadata = res.metadata;
+        let mut nodes = res.nodes.data;
+
+        if res.nodes.pagination.total == nodes.len() {
+            return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
+        }
+
+        page += 1;
+
+        loop {
+            let res = self
+                .get_basic_active_mixing_assigned_nodes_v2(false, Some(page), None, false)
+                .await?;
+
+            if !metadata.consistency_check(&res.metadata) {
+                return Err(NymAPIError::InternalResponseInconsistency {
+                    url: self.api_url().clone(),
+                    details: "Inconsistent paged metadata".to_string(),
+                });
+            }
+
+            nodes.append(&mut res.nodes.data.clone());
+
+            // Check if we've got all nodes
+            if nodes.len() >= res.nodes.pagination.total {
+                break;
+            } else {
+                page += 1;
+            }
+        }
+
+        Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
+    }
+
+    async fn get_all_basic_entry_assigned_nodes_with_metadata(
+        &self,
+    ) -> Result<SkimmedNodesWithMetadata, NymAPIError> {
+        // Get all nodes that can act as entry gateways
+        let mut page = 0;
+        let res = self
+            .get_basic_entry_assigned_nodes_v2(false, Some(page), None, false)
+            .await?;
+
+        let metadata = res.metadata;
+        let mut nodes = res.nodes.data;
+
+        if res.nodes.pagination.total == nodes.len() {
+            return Ok(SkimmedNodesWithMetadata::new(nodes, metadata));
+        }
+
+        page += 1;
+
+        loop {
+            let res = self
+                .get_basic_entry_assigned_nodes_v2(false, Some(page), None, false)
+                .await?;
+
+            if !metadata.consistency_check(&res.metadata) {
+                return Err(NymAPIError::InternalResponseInconsistency {
+                    url: self.api_url().clone(),
+                    details: "Inconsistent paged metadata".to_string(),
+                });
+            }
+
+            nodes.append(&mut res.nodes.data.clone());
+
+            // Check if we've got all nodes
+            if nodes.len() >= res.nodes.pagination.total {
+                break;
+            } else {
+                page += 1;
+            }
+        }
+
+        Ok(SkimmedNodesWithMetadata::new(nodes, metadata))
+    }
+
+    async fn get_all_described_nodes(&self) -> Result<Vec<NymNodeDescription>, NymAPIError> {
+        // TODO: deal with paging in macro or some helper function or something, because it's the same pattern everywhere
+        let mut page = 0;
+        let mut descriptions = Vec::new();
+
+        loop {
+            let mut res = self.get_nodes_described(Some(page), None).await?;
+
+            descriptions.append(&mut res.data);
+            if descriptions.len() < res.pagination.total {
+                page += 1
+            } else {
+                break;
+            }
+        }
+
+        Ok(descriptions)
+    }
+
     #[tracing::instrument(level = "debug", skip_all)]
     async fn get_nym_nodes(
         &self,
@@ -268,6 +319,25 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
+    async fn get_all_bonded_nym_nodes(&self) -> Result<Vec<NymNodeDetails>, ValidatorClientError> {
+        // TODO: deal with paging in macro or some helper function or something, because it's the same pattern everywhere
+        let mut page = 0;
+        let mut bonds = Vec::new();
+
+        loop {
+            let mut res = self.get_nym_nodes(Some(page), None).await?;
+
+            bonds.append(&mut res.data);
+            if bonds.len() < res.pagination.total {
+                page += 1
+            } else {
+                break;
+            }
+        }
+
+        Ok(bonds)
+    }
+
     #[deprecated]
     #[tracing::instrument(level = "debug", skip_all)]
     async fn get_basic_mixnodes(&self) -> Result<CachedNodesResponse<SkimmedNode>, NymAPIError> {
@@ -677,42 +747,6 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_active_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
-        self.get_json(
-            &[routes::V1_API_VERSION, routes::MIXNODES, routes::ACTIVE],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_active_mixnodes_detailed(&self) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::MIXNODES,
-                routes::ACTIVE,
-                routes::DETAILED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_rewarded_mixnodes(&self) -> Result<Vec<MixNodeDetails>, NymAPIError> {
-        self.get_json(
-            &[routes::V1_API_VERSION, routes::MIXNODES, routes::REWARDED],
-            NO_PARAMS,
-        )
-        .await
-    }
-
     #[deprecated]
     #[instrument(level = "debug", skip(self))]
     async fn get_mixnode_report(
@@ -789,24 +823,6 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_rewarded_mixnodes_detailed(
-        &self,
-    ) -> Result<Vec<MixNodeBondAnnotated>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS,
-                routes::MIXNODES,
-                routes::REWARDED,
-                routes::DETAILED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
     #[deprecated]
     #[instrument(level = "debug", skip(self))]
     async fn get_gateway_core_status_count(
@@ -874,104 +890,6 @@ pub trait NymApiClientExt: ApiClient {
         }
     }
 
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnode_status(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<MixnodeStatusResponse, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS_ROUTES,
-                routes::MIXNODE,
-                &mix_id.to_string(),
-                routes::STATUS,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnode_reward_estimation(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<RewardEstimationResponse, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS_ROUTES,
-                routes::MIXNODE,
-                &mix_id.to_string(),
-                routes::REWARD_ESTIMATION,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn compute_mixnode_reward_estimation(
-        &self,
-        mix_id: NodeId,
-        request_body: &ComputeRewardEstParam,
-    ) -> Result<RewardEstimationResponse, NymAPIError> {
-        self.post_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS_ROUTES,
-                routes::MIXNODE,
-                &mix_id.to_string(),
-                routes::COMPUTE_REWARD_ESTIMATION,
-            ],
-            NO_PARAMS,
-            request_body,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnode_stake_saturation(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<StakeSaturationResponse, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS_ROUTES,
-                routes::MIXNODE,
-                &mix_id.to_string(),
-                routes::STAKE_SATURATION,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[allow(deprecated)]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnode_inclusion_probability(
-        &self,
-        mix_id: NodeId,
-    ) -> Result<nym_api_requests::models::InclusionProbabilityResponse, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::STATUS_ROUTES,
-                routes::MIXNODE,
-                &mix_id.to_string(),
-                routes::INCLUSION_CHANCE,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
     #[instrument(level = "debug", skip(self))]
     async fn get_current_node_performance(
         &self,
@@ -1020,34 +938,6 @@ pub trait NymApiClientExt: ApiClient {
         .await
     }
 
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_mixnodes_blacklisted(&self) -> Result<Vec<NodeId>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::MIXNODES,
-                routes::BLACKLISTED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
-    #[deprecated]
-    #[instrument(level = "debug", skip(self))]
-    async fn get_gateways_blacklisted(&self) -> Result<Vec<IdentityKey>, NymAPIError> {
-        self.get_json(
-            &[
-                routes::V1_API_VERSION,
-                routes::GATEWAYS,
-                routes::BLACKLISTED,
-            ],
-            NO_PARAMS,
-        )
-        .await
-    }
-
     #[instrument(level = "debug", skip(self, request_body))]
     async fn blind_sign(
         &self,
@@ -1371,8 +1261,49 @@ pub trait NymApiClientExt: ApiClient {
         )
         .await
     }
+
+    /// Method to change the base API URLs being used by the client
+    fn change_base_urls(&mut self, urls: Vec<url::Url>);
+
+    /// Retrieve expanded information for all bonded nodes on the network
+    async fn get_all_expanded_nodes(&self) -> Result<SemiSkimmedNodesWithMetadata, NymAPIError> {
+        // Unroll the first iteration to get the metadata
+        let mut page = 0;
+
+        let res = self.get_expanded_nodes(false, Some(page), None).await?;
+        let mut nodes = res.nodes.data;
+        let metadata = res.metadata;
+
+        if res.nodes.pagination.total == nodes.len() {
+            return Ok(SemiSkimmedNodesWithMetadata::new(nodes, metadata));
+        }
+
+        page += 1;
+
+        loop {
+            let mut res = self.get_expanded_nodes(false, Some(page), None).await?;
+
+            nodes.append(&mut res.nodes.data);
+            if nodes.len() < res.nodes.pagination.total {
+                page += 1
+            } else {
+                break;
+            }
+        }
+
+        Ok(SemiSkimmedNodesWithMetadata::new(nodes, metadata))
+    }
 }
 
+// Client is already nym_http_api_client::Client (re-exported above), so just one impl needed
 #[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
 #[cfg_attr(not(target_arch = "wasm32"), async_trait)]
-impl NymApiClientExt for Client {}
+impl NymApiClientExt for nym_http_api_client::Client {
+    fn api_url(&self) -> &url::Url {
+        self.current_url().as_ref()
+    }
+
+    fn change_base_urls(&mut self, urls: Vec<url::Url>) {
+        self.change_base_urls(urls.into_iter().map(|u| u.into()).collect());
+    }
+}

common/client-libs/validator-client/src/nyxd/cosmwasm_client/client_traits/query_client.rs

@@ -28,7 +28,7 @@ use cosmrs::proto::cosmwasm::wasm::v1::{
     QueryRawContractStateResponse, QuerySmartContractStateRequest, QuerySmartContractStateResponse,
 };
 use cosmrs::tendermint::{block, chain, Hash};
-use cosmrs::{AccountId, Coin as CosmosCoin, Tx};
+use cosmrs::{AccountId, Coin as CosmosCoin};
 use prost::Message;
 use serde::{Deserialize, Serialize};
 
@@ -556,23 +556,12 @@ pub trait CosmWasmClient: TendermintRpcClient {
         Ok(serde_json::from_slice(&res.data)?)
     }
 
-    // deprecation warning is due to the fact the protobuf files built were based on cosmos-sdk 0.44,
-    // where they prefer using tx_bytes directly. However, in 0.42, which we are using at the time
-    // of writing this, the option does not work
-    // TODO: we should really stop using the `tx` argument here and use `tx_bytes` exlusively,
-    // however, at the time of writing this update, while our QA and mainnet networks do support it,
-    // sandbox is still running old version of wasmd that lacks support for `tx_bytes`
-    #[allow(deprecated)]
-    async fn query_simulate(
-        &self,
-        tx: Option<Tx>,
-        tx_bytes: Vec<u8>,
-    ) -> Result<SimulateResponse, NyxdError> {
+    async fn query_simulate(&self, tx_bytes: Vec<u8>) -> Result<SimulateResponse, NyxdError> {
         let path = Some("/cosmos.tx.v1beta1.Service/Simulate".to_owned());
 
         let req = SimulateRequest {
-            tx: tx.map(Into::into),
             tx_bytes,
+            ..Default::default()
         };
 
         let res = self

common/client-libs/validator-client/src/nyxd/cosmwasm_client/client_traits/signing_client.rs

@@ -81,17 +81,14 @@ where
             auth_info: single_unspecified_signer_auth(public_key, sequence_response.sequence),
             signatures: vec![Vec::new()],
         };
-        self.query_simulate(Some(partial_tx), Vec::new()).await
 
-        // for completion sake, once we're able to transition into using `tx_bytes`,
-        // we might want to use something like this instead:
-        // let tx_raw: tx::Raw = cosmrs::proto::cosmos::tx::v1beta1::TxRaw {
-        //     body_bytes: partial_tx.body.into_bytes().unwrap(),
-        //     auth_info_bytes: partial_tx.auth_info.into_bytes().unwrap(),
-        //     signatures: partial_tx.signatures,
-        // }
-        // .into();
-        // self.query_simulate(None, tx_raw.to_bytes().unwrap()).await
+        let tx_raw: tx::Raw = cosmrs::proto::cosmos::tx::v1beta1::TxRaw {
+            body_bytes: partial_tx.body.into_bytes()?,
+            auth_info_bytes: partial_tx.auth_info.into_bytes()?,
+            signatures: partial_tx.signatures,
+        }
+        .into();
+        self.query_simulate(tx_raw.to_bytes()?).await
     }
 
     async fn upload(

common/commands/Cargo.toml

@@ -38,6 +38,7 @@ cosmrs = { workspace = true }
 cosmwasm-std = { workspace = true }
 
 nym-validator-client = { path = "../client-libs/validator-client" }
+nym-http-api-client = { path = "../http-api-client" }
 nym-bin-common = { path = "../../common/bin-common", features = ["output_format"] }
 nym-crypto = { path = "../../common/crypto", features = ["asymmetric"] }
 nym-network-defaults = { path = "../network-defaults" }

common/commands/src/context/mod.rs

@@ -2,12 +2,12 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::context::errors::ContextError;
+pub use nym_http_api_client::Client as NymApiClient;
 use nym_network_defaults::{
     setup_env,
     var_names::{MIXNET_CONTRACT_ADDRESS, NYM_API, NYXD, VESTING_CONTRACT_ADDRESS},
     NymNetworkDetails,
 };
-pub use nym_validator_client::nym_api::Client as NymApiClient;
 use nym_validator_client::nyxd::{self, AccountId, NyxdClient};
 use nym_validator_client::{
     DirectSigningHttpRpcNyxdClient, DirectSigningHttpRpcValidatorClient, QueryHttpRpcNyxdClient,

common/cosmwasm-smart-contracts/mixnet-contract/bindings/ts-packages/types/src/types/rust/Role.ts

@@ -1,3 +1,3 @@
 // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually.
 
-export type Role = "EntryGateway" | "Layer1" | "Layer2" | "Layer3" | "ExitGateway" | "Standby";
+export type Role = "entry_gateway" | "layer1" | "layer2" | "layer3" | "exit_gateway" | "standby";

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

@@ -23,6 +23,7 @@ use serde_repr::{Deserialize_repr, Serialize_repr};
 
 /// Full details associated with given mixnode.
 #[cw_serde]
+#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
 pub struct MixNodeDetails {
     /// Basic bond information of this mixnode, such as owner address, original pledge, etc.
     pub bond_information: MixNodeBond,
@@ -695,6 +696,7 @@ impl From<LegacyMixLayer> for u8 {
     Copy,
 )]
 #[schemars(crate = "::cosmwasm_schema::schemars")]
+#[cfg_attr(feature = "utoipa", derive(utoipa::ToSchema))]
 pub struct PendingMixNodeChanges {
     pub pledge_change: Option<EpochEventId>,
 

common/credential-proxy/Cargo.toml

@@ -0,0 +1,55 @@
+[package]
+name = "nym-credential-proxy-lib"
+version = "0.1.0"
+authors.workspace = true
+repository.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+edition.workspace = true
+license.workspace = true
+rust-version.workspace = true
+readme.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+axum = { workspace = true }
+bip39 = { workspace = true, features = ["zeroize"] }
+bs58 = { workspace = true }
+futures = { workspace = true }
+humantime = { workspace = true }
+rand = { workspace = true }
+reqwest = { workspace = true, features = ["rustls-tls"] }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+strum = { workspace = true, features = ["derive"] }
+strum_macros = { workspace = true }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate", "time"] }
+time = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["sync"] }
+tokio-util = { workspace = true, features = ["rt"] }
+tracing = { workspace = true }
+uuid = { workspace = true, features = ["serde"] }
+url = { workspace = true }
+zeroize = { workspace = true }
+
+nym-credentials = { path = "../credentials" }
+nym-crypto = { path = "../crypto", features = ["asymmetric", "rand", "serde"] }
+nym-credentials-interface = { path = "../credentials-interface" }
+nym-credential-proxy-requests = { path = "../../nym-credential-proxy/nym-credential-proxy-requests" }
+nym-ecash-signer-check = { path = "../ecash-signer-check" }
+nym-ecash-contract-common = { path = "../cosmwasm-smart-contracts/ecash-contract" }
+nym-compact-ecash = { path = "../nym_offline_compact_ecash" }
+nym-validator-client = { path = "../client-libs/validator-client" }
+nym-network-defaults = { path = "../network-defaults" }
+
+[dev-dependencies]
+tempfile = { workspace = true }
+
+[build-dependencies]
+anyhow = { workspace = true }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
+sqlx = { workspace = true, features = ["runtime-tokio-rustls", "sqlite", "macros", "migrate"] }
+
+[lints]
+workspace = true

common/credential-proxy/src/deposits_buffer/helpers.rs

@@ -10,11 +10,11 @@ use nym_validator_client::nyxd::{Coin, Hash};
 use time::OffsetDateTime;
 use zeroize::Zeroizing;
 
-pub(crate) struct BufferedDeposit {
-    pub(crate) deposit_id: u32,
+pub struct BufferedDeposit {
+    pub deposit_id: u32,
 
     // note: this type implements `ZeroizeOnDrop`
-    pub(crate) ed25519_private_key: ed25519::PrivateKey,
+    pub ed25519_private_key: ed25519::PrivateKey,
 }
 
 impl TryFrom<StorableEcashDeposit> for BufferedDeposit {
@@ -36,14 +36,14 @@ impl TryFrom<StorableEcashDeposit> for BufferedDeposit {
 }
 
 impl BufferedDeposit {
-    pub(crate) fn new(deposit_id: u32, ed25519_private_key: ed25519::PrivateKey) -> Self {
+    pub fn new(deposit_id: u32, ed25519_private_key: ed25519::PrivateKey) -> Self {
         BufferedDeposit {
             deposit_id,
             ed25519_private_key,
         }
     }
 
-    pub(crate) fn sign_ticketbook_plaintext(
+    pub fn sign_ticketbook_plaintext(
         &self,
         withdrawal_request: &WithdrawalRequest,
     ) -> ed25519::Signature {
@@ -52,13 +52,13 @@ impl BufferedDeposit {
     }
 }
 
-pub(crate) struct PerformedDeposits {
-    pub(crate) deposits_data: Vec<BufferedDeposit>,
+pub struct PerformedDeposits {
+    pub deposits_data: Vec<BufferedDeposit>,
 
     // shared by all performed deposits as they were included in the same tx
-    pub(crate) tx_hash: Hash,
-    pub(crate) requested_on: OffsetDateTime,
-    pub(crate) deposit_amount: Coin,
+    pub tx_hash: Hash,
+    pub requested_on: OffsetDateTime,
+    pub deposit_amount: Coin,
 }
 
 impl PerformedDeposits {

common/credential-proxy/src/deposits_buffer/mod.rs

@@ -1,11 +1,11 @@
 // Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use crate::deposits_buffer::helpers::{request_sizes, BufferedDeposit, PerformedDeposits};
+use crate::deposits_buffer::helpers::request_sizes;
 use crate::deposits_buffer::refill_task::RefillTask;
 use crate::error::CredentialProxyError;
-use crate::http::state::required_deposit_cache::RequiredDepositCache;
-use crate::http::state::ChainClient;
+use crate::shared_state::nyxd_client::ChainClient;
+use crate::shared_state::required_deposit_cache::RequiredDepositCache;
 use crate::storage::CredentialProxyStorage;
 use nym_compact_ecash::PublicKeyUser;
 use nym_crypto::asymmetric::ed25519;
@@ -21,6 +21,8 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, instrument, warn};
 use uuid::Uuid;
 
+pub use helpers::{BufferedDeposit, PerformedDeposits};
+
 pub(crate) mod helpers;
 mod refill_task;
 
@@ -43,12 +45,12 @@ struct DepositsBufferInner {
 }
 
 #[derive(Clone)]
-pub(crate) struct DepositsBuffer {
+pub struct DepositsBuffer {
     inner: Arc<DepositsBufferInner>,
 }
 
 impl DepositsBuffer {
-    pub(crate) async fn new(
+    pub async fn new(
         storage: CredentialProxyStorage,
         client: ChainClient,
         required_deposit_cache: RequiredDepositCache,
@@ -250,7 +252,7 @@ impl DepositsBuffer {
         }
     }
 
-    pub(crate) async fn get_valid_deposit(
+    pub async fn get_valid_deposit(
         &self,
         request_uuid: Uuid,
         requested_on: OffsetDateTime,
@@ -290,7 +292,7 @@ impl DepositsBuffer {
         }
     }
 
-    pub(crate) async fn wait_for_shutdown(&self) {
+    pub async fn wait_for_shutdown(&self) {
         let task_handle = self.inner.deposits_refill_task.take_task_join_handle();
         if let Some(task_handle) = task_handle {
             if !task_handle.is_finished() {

common/credential-proxy/src/error.rs

@@ -1,9 +1,9 @@
-// Copyright 2024 Nym Technologies SA <contact@nymtech.net>
+// Copyright 2025 Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
 use nym_ecash_signer_check::SignerCheckError;
 use nym_validator_client::coconut::EcashApiError;
-use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::nym_api::{error::NymAPIError, EpochId};
 use nym_validator_client::nyxd::error::NyxdError;
 use std::io;
 use std::net::SocketAddr;
@@ -71,6 +71,9 @@ pub enum CredentialProxyError {
         source: EcashApiError,
     },
 
+    #[error("Nym API request failed: {source}")]
+    NymApiFailure { source: Box<NymAPIError> },
+
     #[error("Compact ecash internal error: {0}")]
     CompactEcashInternalError(#[from] nym_compact_ecash::error::CompactEcashError),
 
@@ -128,6 +131,38 @@ pub enum CredentialProxyError {
         #[from]
         source: SignerCheckError,
     },
+
+    #[error(
+        "this operation couldn't be completed as the program is in the process of shutting down"
+    )]
+    ShutdownInProgress,
+
+    #[error("failed to obtain wallet shares with id {id}: {message}")]
+    ShareByIdLoadError { message: String, id: i64 },
+
+    #[error("failed to obtain wallet shares with device_id {device_id} and credential_id: {credential_id}: {message}")]
+    ShareByDeviceLoadError {
+        message: String,
+        device_id: String,
+        credential_id: String,
+    },
+
+    #[error("could not find shares with id {id}")]
+    SharesByIdNotFound { id: i64 },
+
+    #[error("could not find shares with device_id {device_id} and credential_id: {credential_id}")]
+    SharesByDeviceNotFound {
+        device_id: String,
+        credential_id: String,
+    },
+}
+
+impl From<NymAPIError> for CredentialProxyError {
+    fn from(source: NymAPIError) -> Self {
+        CredentialProxyError::NymApiFailure {
+            source: Box::new(source),
+        }
+    }
 }
 
 impl CredentialProxyError {

common/credential-proxy/src/helpers.rs

@@ -0,0 +1,67 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use rand::rngs::OsRng;
+use rand::RngCore;
+use time::OffsetDateTime;
+use tracing::{debug, info, warn};
+use uuid::Uuid;
+
+pub fn random_uuid() -> Uuid {
+    let mut bytes = [0u8; 16];
+    let mut rng = OsRng;
+    rng.fill_bytes(&mut bytes);
+    Uuid::from_bytes(bytes)
+}
+
+pub struct LockTimer {
+    created: OffsetDateTime,
+    message: String,
+}
+
+impl LockTimer {
+    pub fn new<S: Into<String>>(message: S) -> Self {
+        LockTimer {
+            message: message.into(),
+            ..Default::default()
+        }
+    }
+}
+
+impl Drop for LockTimer {
+    fn drop(&mut self) {
+        let time_taken = OffsetDateTime::now_utc() - self.created;
+        let time_taken_formatted = humantime::format_duration(time_taken.unsigned_abs());
+        if time_taken > time::Duration::SECOND * 10 {
+            warn!(time_taken = %time_taken_formatted, "{}", self.message)
+        } else if time_taken > time::Duration::SECOND * 5 {
+            info!(time_taken = %time_taken_formatted, "{}", self.message)
+        } else {
+            debug!(time_taken = %time_taken_formatted, "{}", self.message)
+        };
+    }
+}
+
+impl Default for LockTimer {
+    fn default() -> Self {
+        LockTimer {
+            created: OffsetDateTime::now_utc(),
+            message: "released the lock".to_string(),
+        }
+    }
+}
+
+// #[allow(clippy::panic)]
+// fn build_sha_short() -> &'static str {
+//     let bin_info = bin_info!();
+//     if bin_info.commit_sha.len() < 7 {
+//         panic!("unavailable build commit sha")
+//     }
+//
+//     if bin_info.commit_sha == "VERGEN_IDEMPOTENT_OUTPUT" {
+//         error!("the binary hasn't been built correctly. it doesn't have a commit sha information");
+//         return "unknown";
+//     }
+//
+//     &bin_info.commit_sha[..7]
+// }

common/credential-proxy/src/http_helpers.rs

@@ -1,11 +1,12 @@
-// Copyright 2024 Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
 
 use crate::error::CredentialProxyError;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
 use axum::Json;
 use nym_credential_proxy_requests::api::v1::ErrorResponse;
+use tracing::warn;
 use uuid::Uuid;
 
 #[derive(Debug, Clone)]
@@ -35,6 +36,10 @@ impl RequestError {
         }
     }
 
+    pub fn new_plain_error(err: CredentialProxyError) -> Self {
+        Self::from_err(err, StatusCode::INTERNAL_SERVER_ERROR)
+    }
+
     pub fn new_server_error(err: CredentialProxyError, uuid: Uuid) -> Self {
         RequestError::new_with_uuid(err.to_string(), uuid, StatusCode::INTERNAL_SERVER_ERROR)
     }
@@ -59,3 +64,12 @@ impl IntoResponse for RequestError {
         (self.status, Json(self.inner)).into_response()
     }
 }
+
+pub fn db_failure<T>(err: CredentialProxyError, uuid: Uuid) -> Result<T, RequestError> {
+    warn!("db failure: {err}");
+    Err(RequestError::new_with_uuid(
+        format!("oh no, something went wrong {err}"),
+        uuid,
+        StatusCode::INTERNAL_SERVER_ERROR,
+    ))
+}

common/credential-proxy/src/lib.rs

@@ -0,0 +1,13 @@
+// Copyright 2025 Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+pub mod deposits_buffer;
+pub mod error;
+pub mod helpers;
+pub mod http_helpers;
+pub mod nym_api_helpers;
+pub mod quorum_checker;
+pub mod shared_state;
+pub mod storage;
+pub mod ticketbook_manager;
+pub mod webhook;

common/credential-proxy/src/nym_api_helpers.rs

@@ -19,9 +19,9 @@ use time::{Date, OffsetDateTime};
 use tokio::sync::{Mutex, RwLock, RwLockReadGuard};
 use tracing::warn;
 
-pub(crate) struct CachedEpoch {
+pub struct CachedEpoch {
     valid_until: OffsetDateTime,
-    pub(crate) current_epoch: Epoch,
+    pub current_epoch: Epoch,
 }
 
 impl Default for CachedEpoch {
@@ -34,11 +34,11 @@ impl Default for CachedEpoch {
 }
 
 impl CachedEpoch {
-    pub(crate) fn is_valid(&self) -> bool {
+    pub fn is_valid(&self) -> bool {
         self.valid_until > OffsetDateTime::now_utc()
     }
 
-    pub(crate) fn update(&mut self, epoch: Epoch) {
+    pub fn update(&mut self, epoch: Epoch) {
         let now = OffsetDateTime::now_utc();
 
         let validity_duration = if let Some(epoch_finish) = epoch.deadline {
@@ -58,13 +58,13 @@ impl CachedEpoch {
 }
 
 // a map of items that never change for given key
-pub(crate) struct CachedImmutableItems<K, V> {
+pub struct CachedImmutableItems<K, V> {
     // I wonder if there's a more efficient structure with OnceLock or OnceCell or something
     inner: RwLock<HashMap<K, V>>,
 }
 
 // an item that stays constant throughout given epoch
-pub(crate) type CachedImmutableEpochItem<T> = CachedImmutableItems<EpochId, T>;
+pub type CachedImmutableEpochItem<T> = CachedImmutableItems<EpochId, T>;
 
 impl<K, V> Default for CachedImmutableItems<K, V> {
     fn default() -> Self {
@@ -86,11 +86,7 @@ impl<K, V> CachedImmutableItems<K, V>
 where
     K: Eq + Hash,
 {
-    pub(crate) async fn get_or_init<F, U, E>(
-        &self,
-        key: K,
-        f: F,
-    ) -> Result<RwLockReadGuard<'_, V>, E>
+    pub async fn get_or_init<F, U, E>(&self, key: K, f: F) -> Result<RwLockReadGuard<'_, V>, E>
     where
         F: FnOnce() -> U,
         U: Future<Output = Result<V, E>>,
@@ -129,9 +125,7 @@ where
     }
 }
 
-pub(crate) fn ensure_sane_expiration_date(
-    expiration_date: Date,
-) -> Result<(), CredentialProxyError> {
+pub fn ensure_sane_expiration_date(expiration_date: Date) -> Result<(), CredentialProxyError> {
     let today = ecash_today();
 
     if expiration_date < today.date() {
@@ -146,7 +140,7 @@ pub(crate) fn ensure_sane_expiration_date(
     Ok(())
 }
 
-pub(crate) async fn query_all_threshold_apis<F, T, U>(
+pub async fn query_all_threshold_apis<F, T, U>(
     all_apis: Vec<EcashApiClient>,
     threshold: u64,
     f: F,

common/credential-proxy/src/quorum_checker.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: GPL-3.0-only
 
 use crate::error::CredentialProxyError;
-use crate::http::state::ChainClient;
+use crate::shared_state::nyxd_client::ChainClient;
 use nym_ecash_signer_check::{check_known_dealers, dkg_details_with_client};
 use std::ops::Deref;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -12,17 +12,17 @@ use tokio_util::sync::CancellationToken;
 use tracing::{error, info, warn};
 
 #[derive(Clone)]
-pub(crate) struct QuorumState {
+pub struct QuorumState {
     available: Arc<AtomicBool>,
 }
 
 impl QuorumState {
-    pub(crate) fn available(&self) -> bool {
+    pub fn available(&self) -> bool {
         self.available.load(Ordering::Acquire)
     }
 }
 
-pub(crate) struct QuorumStateChecker {
+pub struct QuorumStateChecker {
     client: ChainClient,
     cancellation_token: CancellationToken,
     check_interval: Duration,

common/credential-proxy/src/shared_state/ecash_state.rs

@@ -0,0 +1,49 @@
+// Copyright 2025 Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::nym_api_helpers::{CachedEpoch, CachedImmutableEpochItem, CachedImmutableItems};
+use crate::quorum_checker::QuorumState;
+use crate::shared_state::required_deposit_cache::RequiredDepositCache;
+use nym_compact_ecash::VerificationKeyAuth;
+use nym_credentials::{AggregatedCoinIndicesSignatures, AggregatedExpirationDateSignatures};
+use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::EcashApiClient;
+use time::Date;
+use tokio::sync::RwLock;
+
+pub struct EcashState {
+    pub required_deposit_cache: RequiredDepositCache,
+
+    pub quorum_state: QuorumState,
+
+    pub cached_epoch: RwLock<CachedEpoch>,
+
+    pub master_verification_key: CachedImmutableEpochItem<VerificationKeyAuth>,
+
+    pub threshold_values: CachedImmutableEpochItem<u64>,
+
+    pub epoch_clients: CachedImmutableEpochItem<Vec<EcashApiClient>>,
+
+    pub coin_index_signatures: CachedImmutableEpochItem<AggregatedCoinIndicesSignatures>,
+
+    pub expiration_date_signatures:
+        CachedImmutableItems<(EpochId, Date), AggregatedExpirationDateSignatures>,
+}
+
+impl EcashState {
+    pub fn new(
+        required_deposit_cache: RequiredDepositCache,
+        quorum_state: QuorumState,
+    ) -> EcashState {
+        EcashState {
+            required_deposit_cache,
+            quorum_state,
+            cached_epoch: Default::default(),
+            master_verification_key: Default::default(),
+            threshold_values: Default::default(),
+            epoch_clients: Default::default(),
+            coin_index_signatures: Default::default(),
+            expiration_date_signatures: Default::default(),
+        }
+    }
+}

common/credential-proxy/src/shared_state/mod.rs

@@ -0,0 +1,496 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::deposits_buffer::{BufferedDeposit, DepositsBuffer};
+use crate::error::CredentialProxyError;
+use crate::nym_api_helpers::{ensure_sane_expiration_date, query_all_threshold_apis};
+use crate::shared_state::ecash_state::EcashState;
+use crate::shared_state::nyxd_client::ChainClient;
+use crate::storage::CredentialProxyStorage;
+use nym_compact_ecash::scheme::coin_indices_signatures::{
+    aggregate_annotated_indices_signatures, CoinIndexSignatureShare,
+};
+use nym_compact_ecash::scheme::expiration_date_signatures::{
+    aggregate_annotated_expiration_signatures, ExpirationDateSignatureShare,
+};
+use nym_compact_ecash::{Base58, PublicKeyUser, VerificationKeyAuth};
+use nym_credential_proxy_requests::api::v1::ticketbook::models::{
+    AggregatedCoinIndicesSignaturesResponse, AggregatedExpirationDateSignaturesResponse,
+    GlobalDataParams, MasterVerificationKeyResponse,
+};
+use nym_credentials::ecash::utils::EcashTime;
+use nym_credentials::{
+    AggregatedCoinIndicesSignatures, AggregatedExpirationDateSignatures, EpochVerificationKey,
+};
+use nym_ecash_contract_common::deposit::DepositId;
+use nym_validator_client::client::NymApiClientExt;
+use nym_validator_client::coconut::EcashApiError;
+use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::nyxd::contract_traits::dkg_query_client::Epoch;
+use nym_validator_client::nyxd::contract_traits::{DkgQueryClient, PagedDkgQueryClient};
+use nym_validator_client::nyxd::Coin;
+use nym_validator_client::{DirectSigningHttpRpcNyxdClient, EcashApiClient};
+use std::sync::Arc;
+use std::time::Duration;
+use time::{Date, OffsetDateTime};
+use tokio::sync::RwLockReadGuard;
+use tokio::time::Instant;
+use tracing::{debug, error, info, warn};
+use uuid::Uuid;
+
+pub mod ecash_state;
+pub mod nyxd_client;
+pub mod required_deposit_cache;
+
+#[derive(Clone)]
+pub struct CredentialProxyState {
+    inner: Arc<CredentialProxyStateInner>,
+}
+
+impl CredentialProxyState {
+    pub fn new(
+        storage: CredentialProxyStorage,
+        client: ChainClient,
+        deposits_buffer: DepositsBuffer,
+        ecash_state: EcashState,
+    ) -> Self {
+        CredentialProxyState {
+            inner: Arc::new(CredentialProxyStateInner {
+                storage,
+                client,
+                deposits_buffer,
+                ecash_state,
+            }),
+        }
+    }
+
+    pub fn storage(&self) -> &CredentialProxyStorage {
+        &self.inner.storage
+    }
+
+    pub async fn deposit_amount(&self) -> Result<Coin, CredentialProxyError> {
+        self.ecash_state()
+            .required_deposit_cache
+            .get_or_update(self.client())
+            .await
+    }
+
+    pub fn client(&self) -> &ChainClient {
+        &self.inner.client
+    }
+
+    pub fn deposits_buffer(&self) -> &DepositsBuffer {
+        &self.inner.deposits_buffer
+    }
+
+    pub fn ecash_state(&self) -> &EcashState {
+        &self.inner.ecash_state
+    }
+
+    pub(crate) async fn query_chain(&self) -> RwLockReadGuard<'_, DirectSigningHttpRpcNyxdClient> {
+        self.inner.client.query_chain().await
+    }
+
+    pub async fn ensure_credentials_issuable(&self) -> Result<(), CredentialProxyError> {
+        let epoch = self.current_epoch().await?;
+
+        if epoch.state.is_final() {
+            Ok(())
+        } else if let Some(final_timestamp) = epoch.final_timestamp_secs() {
+            // SAFETY: the timestamp values in our DKG contract should be valid timestamps,
+            // otherwise it means the chain is seriously misbehaving
+            #[allow(clippy::unwrap_used)]
+            let finish_dt = OffsetDateTime::from_unix_timestamp(final_timestamp as i64).unwrap();
+
+            Err(CredentialProxyError::CredentialsNotYetIssuable {
+                availability: finish_dt,
+            })
+        } else if epoch.state.is_waiting_initialisation() {
+            Err(CredentialProxyError::UninitialisedDkg)
+        } else {
+            Err(CredentialProxyError::UnknownEcashFailure)
+        }
+    }
+
+    pub async fn get_deposit(
+        &self,
+        request_uuid: Uuid,
+        requested_on: OffsetDateTime,
+        client_pubkey: PublicKeyUser,
+    ) -> Result<BufferedDeposit, CredentialProxyError> {
+        let start = Instant::now();
+        let deposit = self
+            .deposits_buffer()
+            .get_valid_deposit(request_uuid, requested_on, client_pubkey)
+            .await;
+
+        let time_taken = start.elapsed();
+        let formatted = humantime::format_duration(time_taken);
+        if time_taken > Duration::from_secs(10) {
+            warn!("attempting to get buffered deposit took {formatted}. perhaps the buffer is too small or the process/chain is overloaded?")
+        } else {
+            debug!("attempting to get buffered deposit took {formatted}")
+        };
+
+        deposit
+    }
+
+    pub async fn insert_deposit_usage_error(&self, deposit_id: DepositId, error: String) {
+        if let Err(err) = self
+            .storage()
+            .insert_deposit_usage_error(deposit_id, error)
+            .await
+        {
+            error!("failed to insert information about deposit (id: {deposit_id}) usage failure: {err}")
+        }
+    }
+
+    pub async fn current_epoch_id(&self) -> Result<EpochId, CredentialProxyError> {
+        let read_guard = self.inner.ecash_state.cached_epoch.read().await;
+        if read_guard.is_valid() {
+            return Ok(read_guard.current_epoch.epoch_id);
+        }
+
+        // update cache
+        drop(read_guard);
+        let mut write_guard = self.inner.ecash_state.cached_epoch.write().await;
+        let epoch = self.query_chain().await.get_current_epoch().await?;
+
+        write_guard.update(epoch);
+        Ok(epoch.epoch_id)
+    }
+
+    pub async fn current_epoch(&self) -> Result<Epoch, CredentialProxyError> {
+        let read_guard = self.ecash_state().cached_epoch.read().await;
+        if read_guard.is_valid() {
+            return Ok(read_guard.current_epoch);
+        }
+
+        // update cache
+        drop(read_guard);
+        let mut write_guard = self.ecash_state().cached_epoch.write().await;
+        let epoch = self.query_chain().await.get_current_epoch().await?;
+
+        write_guard.update(epoch);
+        Ok(epoch)
+    }
+
+    pub async fn global_data(
+        &self,
+        global_data: GlobalDataParams,
+        epoch_id: EpochId,
+        expiration_date: Date,
+    ) -> Result<
+        (
+            Option<MasterVerificationKeyResponse>,
+            Option<AggregatedExpirationDateSignaturesResponse>,
+            Option<AggregatedCoinIndicesSignaturesResponse>,
+        ),
+        CredentialProxyError,
+    > {
+        let master_verification_key = if global_data.include_master_verification_key {
+            debug!("including master verification key in the response");
+            Some(
+                self.master_verification_key(Some(epoch_id))
+                    .await
+                    .map(|key| MasterVerificationKeyResponse {
+                        epoch_id,
+                        bs58_encoded_key: key.to_bs58(),
+                    })
+                    .inspect_err(|err| warn!("request failure: {err}"))?,
+            )
+        } else {
+            None
+        };
+
+        let aggregated_expiration_date_signatures =
+            if global_data.include_expiration_date_signatures {
+                debug!("including expiration date signatures in the response");
+                Some(
+                    self.master_expiration_date_signatures(epoch_id, expiration_date)
+                        .await
+                        .map(|signatures| AggregatedExpirationDateSignaturesResponse {
+                            signatures: signatures.clone(),
+                        })
+                        .inspect_err(|err| warn!("request failure: {err}"))?,
+                )
+            } else {
+                None
+            };
+
+        let aggregated_coin_index_signatures = if global_data.include_coin_index_signatures {
+            debug!("including coin index signatures in the response");
+            Some(
+                self.master_coin_index_signatures(Some(epoch_id))
+                    .await
+                    .map(|signatures| AggregatedCoinIndicesSignaturesResponse {
+                        signatures: signatures.clone(),
+                    })
+                    .inspect_err(|err| warn!("request failure: {err}"))?,
+            )
+        } else {
+            None
+        };
+
+        Ok((
+            master_verification_key,
+            aggregated_expiration_date_signatures,
+            aggregated_coin_index_signatures,
+        ))
+    }
+
+    pub async fn master_verification_key(
+        &self,
+        epoch_id: Option<EpochId>,
+    ) -> Result<RwLockReadGuard<'_, VerificationKeyAuth>, CredentialProxyError> {
+        let epoch_id = match epoch_id {
+            Some(id) => id,
+            None => self.current_epoch_id().await?,
+        };
+
+        self.inner
+            .ecash_state
+            .master_verification_key
+            .get_or_init(epoch_id, || async {
+                // 1. check the storage
+                if let Some(stored) = self
+                    .inner
+                    .storage
+                    .get_master_verification_key(epoch_id)
+                    .await?
+                {
+                    return Ok(stored.key);
+                }
+
+                info!("attempting to establish master verification key for epoch {epoch_id}...");
+
+                // 2. perform actual aggregation
+                let all_apis = self.ecash_clients(epoch_id).await?;
+                let threshold = self.ecash_threshold(epoch_id).await?;
+
+                if all_apis.len() < threshold as usize {
+                    return Err(CredentialProxyError::InsufficientNumberOfSigners {
+                        threshold,
+                        available: all_apis.len(),
+                    });
+                }
+
+                let master_key = nym_credentials::aggregate_verification_keys(&all_apis)?;
+
+                let epoch = EpochVerificationKey {
+                    epoch_id,
+                    key: master_key,
+                };
+
+                // 3. save the key in the storage for when we reboot
+                self.inner
+                    .storage
+                    .insert_master_verification_key(&epoch)
+                    .await?;
+
+                Ok(epoch.key)
+            })
+            .await
+    }
+
+    pub async fn master_coin_index_signatures(
+        &self,
+        epoch_id: Option<EpochId>,
+    ) -> Result<RwLockReadGuard<'_, AggregatedCoinIndicesSignatures>, CredentialProxyError> {
+        let epoch_id = match epoch_id {
+            Some(id) => id,
+            None => self.current_epoch_id().await?,
+        };
+
+        self.inner
+            .ecash_state
+            .coin_index_signatures
+            .get_or_init(epoch_id, || async {
+                // 1. check the storage
+                if let Some(master_sigs) = self
+                    .inner
+                    .storage
+                    .get_master_coin_index_signatures(epoch_id)
+                    .await?
+                {
+                    return Ok(master_sigs);
+                }
+
+                info!(
+                    "attempting to establish master coin index signatures for epoch {epoch_id}..."
+                );
+
+                // 2. go around APIs and attempt to aggregate the data
+                let master_vk = self.master_verification_key(Some(epoch_id)).await?;
+                let all_apis = self.ecash_clients(epoch_id).await?;
+                let threshold = self.ecash_threshold(epoch_id).await?;
+
+                let get_partial_signatures = |api: EcashApiClient| async {
+                    // move the api into the closure
+                    let api = api;
+                    let node_index = api.node_id;
+                    let partial_vk = api.verification_key;
+
+                    let partial = api
+                        .api_client
+                        .partial_coin_indices_signatures(Some(epoch_id))
+                        .await?
+                        .signatures;
+                    Ok(CoinIndexSignatureShare {
+                        index: node_index,
+                        key: partial_vk,
+                        signatures: partial,
+                    })
+                };
+
+                let shares =
+                    query_all_threshold_apis(all_apis.clone(), threshold, get_partial_signatures)
+                        .await?;
+
+                let aggregated = aggregate_annotated_indices_signatures(
+                    nym_credentials_interface::ecash_parameters(),
+                    &master_vk,
+                    &shares,
+                )?;
+
+                let sigs = AggregatedCoinIndicesSignatures {
+                    epoch_id,
+                    signatures: aggregated,
+                };
+
+                // 3. save the signatures in the storage for when we reboot
+                self.inner
+                    .storage
+                    .insert_master_coin_index_signatures(&sigs)
+                    .await?;
+
+                Ok(sigs)
+            })
+            .await
+    }
+
+    pub async fn master_expiration_date_signatures(
+        &self,
+        epoch_id: EpochId,
+        expiration_date: Date,
+    ) -> Result<RwLockReadGuard<'_, AggregatedExpirationDateSignatures>, CredentialProxyError> {
+        self.inner.ecash_state
+            .expiration_date_signatures
+            .get_or_init((epoch_id, expiration_date), || async {
+                // 1. sanity check to see if the expiration_date is not nonsense
+                ensure_sane_expiration_date(expiration_date)?;
+
+                // 2. check the storage
+                if let Some(master_sigs) = self
+                    .storage()
+                    .get_master_expiration_date_signatures(expiration_date, epoch_id)
+                    .await?
+                {
+                    return Ok(master_sigs);
+                }
+
+
+                info!(
+                    "attempting to establish master expiration date signatures for {expiration_date} and epoch {epoch_id}..."
+                );
+
+                // 3. go around APIs and attempt to aggregate the data
+                let epoch_id = self.current_epoch_id().await?;
+                let master_vk = self.master_verification_key(Some(epoch_id)).await?;
+                let all_apis = self.ecash_clients(epoch_id).await?;
+                let threshold = self.ecash_threshold(epoch_id).await?;
+
+                let get_partial_signatures = |api: EcashApiClient| async {
+                    // move the api into the closure
+                    let api = api;
+                    let node_index = api.node_id;
+                    let partial_vk = api.verification_key;
+
+                    let partial = api
+                        .api_client
+                        .partial_expiration_date_signatures(Some(expiration_date), Some(epoch_id))
+                        .await?
+                        .signatures;
+                    Ok(ExpirationDateSignatureShare {
+                        index: node_index,
+                        key: partial_vk,
+                        signatures: partial,
+                    })
+                };
+
+                let shares =
+                    query_all_threshold_apis(all_apis.clone(), threshold, get_partial_signatures)
+                        .await?;
+
+                let aggregated = aggregate_annotated_expiration_signatures(
+                    &master_vk,
+                    expiration_date.ecash_unix_timestamp(),
+                    &shares,
+                )?;
+
+                let sigs = AggregatedExpirationDateSignatures {
+                    epoch_id,
+                    expiration_date,
+                    signatures: aggregated,
+                };
+
+                // 4. save the signatures in the storage for when we reboot
+                self.inner.storage
+                    .insert_master_expiration_date_signatures(&sigs)
+                    .await?;
+
+                Ok(sigs)
+            })
+            .await
+    }
+
+    pub async fn ecash_clients(
+        &self,
+        epoch_id: EpochId,
+    ) -> Result<RwLockReadGuard<'_, Vec<EcashApiClient>>, CredentialProxyError> {
+        self.inner
+            .ecash_state
+            .epoch_clients
+            .get_or_init(epoch_id, || async {
+                Ok(self
+                    .query_chain()
+                    .await
+                    .get_all_verification_key_shares(epoch_id)
+                    .await?
+                    .into_iter()
+                    .map(TryInto::try_into)
+                    .collect::<anyhow::Result<Vec<_>, EcashApiError>>()?)
+            })
+            .await
+    }
+
+    pub async fn ecash_threshold(&self, epoch_id: EpochId) -> Result<u64, CredentialProxyError> {
+        self.inner
+            .ecash_state
+            .threshold_values
+            .get_or_init(epoch_id, || async {
+                if let Some(threshold) = self
+                    .query_chain()
+                    .await
+                    .get_epoch_threshold(epoch_id)
+                    .await?
+                {
+                    Ok(threshold)
+                } else {
+                    Err(CredentialProxyError::UnavailableThreshold { epoch_id })
+                }
+            })
+            .await
+            .map(|t| *t)
+    }
+}
+
+struct CredentialProxyStateInner {
+    storage: CredentialProxyStorage,
+
+    client: ChainClient,
+
+    deposits_buffer: DepositsBuffer,
+
+    ecash_state: EcashState,
+}

common/credential-proxy/src/shared_state/nyxd_client.rs

@@ -0,0 +1,126 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::CredentialProxyError;
+use crate::helpers::LockTimer;
+use nym_ecash_contract_common::msg::ExecuteMsg;
+use nym_validator_client::nyxd::contract_traits::NymContractsProvider;
+use nym_validator_client::nyxd::cosmwasm_client::types::ExecuteResult;
+use nym_validator_client::nyxd::{Coin, CosmWasmClient, NyxdClient};
+use nym_validator_client::{nyxd, DirectSigningHttpRpcNyxdClient};
+use std::ops::Deref;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
+use tracing::{instrument, warn};
+
+#[derive(Clone)]
+pub struct ChainClient(Arc<RwLock<DirectSigningHttpRpcNyxdClient>>);
+
+impl ChainClient {
+    pub fn new(mnemonic: bip39::Mnemonic) -> Result<Self, CredentialProxyError> {
+        let network_details = nym_network_defaults::NymNetworkDetails::new_from_env();
+        let client_config = nyxd::Config::try_from_nym_network_details(&network_details)?;
+
+        let nyxd_url = network_details
+            .endpoints
+            .first()
+            .ok_or_else(|| CredentialProxyError::NoNyxEndpointsAvailable)?
+            .nyxd_url
+            .as_str();
+
+        let client = NyxdClient::connect_with_mnemonic(client_config, nyxd_url, mnemonic)?;
+
+        if client.ecash_contract_address().is_none() {
+            return Err(CredentialProxyError::UnavailableEcashContract);
+        }
+
+        if client.dkg_contract_address().is_none() {
+            return Err(CredentialProxyError::UnavailableDKGContract);
+        }
+
+        Ok(ChainClient(Arc::new(RwLock::new(client))))
+    }
+
+    pub async fn query_chain(&self) -> ChainReadPermit<'_> {
+        let _acquire_timer = LockTimer::new("acquire chain query permit");
+        self.0.read().await
+    }
+
+    pub async fn start_chain_tx(&self) -> ChainWritePermit<'_> {
+        let _acquire_timer = LockTimer::new("acquire exclusive chain write permit");
+
+        ChainWritePermit {
+            lock_timer: LockTimer::new("exclusive chain access permit"),
+            inner: self.0.write().await,
+        }
+    }
+}
+
+pub type ChainReadPermit<'a> = RwLockReadGuard<'a, DirectSigningHttpRpcNyxdClient>;
+
+// explicitly wrap the WriteGuard for extra information regarding time taken
+pub struct ChainWritePermit<'a> {
+    // it's not really dead, we only care about it being dropped
+    #[allow(dead_code)]
+    lock_timer: LockTimer,
+    inner: RwLockWriteGuard<'a, DirectSigningHttpRpcNyxdClient>,
+}
+
+impl ChainWritePermit<'_> {
+    #[instrument(skip(self, short_sha, info), err(Display))]
+    pub async fn make_deposits(
+        self,
+        short_sha: &'static str,
+        info: Vec<(String, Coin)>,
+    ) -> Result<ExecuteResult, CredentialProxyError> {
+        let address = self.inner.address();
+        let starting_sequence = self.inner.get_sequence(&address).await?.sequence;
+
+        let deposits = info.len();
+
+        let ecash_contract = self
+            .inner
+            .ecash_contract_address()
+            .ok_or(CredentialProxyError::UnavailableEcashContract)?;
+        let deposit_messages = info
+            .into_iter()
+            .map(|(identity_key, amount)| {
+                (
+                    ExecuteMsg::DepositTicketBookFunds { identity_key },
+                    vec![amount],
+                )
+            })
+            .collect::<Vec<_>>();
+
+        let res = self
+            .inner
+            .execute_multiple(
+                ecash_contract,
+                deposit_messages,
+                None,
+                format!("cp-{short_sha}: performing {deposits} deposits"),
+            )
+            .await?;
+
+        loop {
+            let updated_sequence = self.inner.get_sequence(&address).await?.sequence;
+
+            if updated_sequence > starting_sequence {
+                break;
+            }
+            warn!("wrong sequence number... waiting before releasing chain lock");
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+
+        Ok(res)
+    }
+}
+
+impl Deref for ChainWritePermit<'_> {
+    type Target = DirectSigningHttpRpcNyxdClient;
+
+    fn deref(&self) -> &Self::Target {
+        self.inner.deref()
+    }
+}

common/credential-proxy/src/shared_state/required_deposit_cache.rs

@@ -2,14 +2,14 @@
 // SPDX-License-Identifier: GPL-3.0-only
 
 use crate::error::CredentialProxyError;
-use crate::http::state::ChainClient;
+use crate::shared_state::nyxd_client::ChainClient;
 use nym_validator_client::nyxd::contract_traits::EcashQueryClient;
 use nym_validator_client::nyxd::Coin;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::sync::RwLock;
 
-pub(crate) struct CachedDeposit {
+pub struct CachedDeposit {
     valid_until: OffsetDateTime,
     required_amount: Coin,
 }
@@ -40,12 +40,12 @@ impl Default for CachedDeposit {
 }
 
 #[derive(Clone, Default)]
-pub(crate) struct RequiredDepositCache {
+pub struct RequiredDepositCache {
     inner: Arc<RwLock<CachedDeposit>>,
 }
 
 impl RequiredDepositCache {
-    pub(crate) async fn get_or_update(
+    pub async fn get_or_update(
         &self,
         chain_client: &ChainClient,
     ) -> Result<Coin, CredentialProxyError> {

common/credential-proxy/src/storage/mod.rs

@@ -1,7 +1,6 @@
 // Copyright 2024 - Nym Technologies SA <contact@nymtech.net>
 // SPDX-License-Identifier: GPL-3.0-only
 
-use crate::credentials::ticketbook::NodeId;
 use crate::deposits_buffer::helpers::{BufferedDeposit, PerformedDeposits};
 use crate::error::CredentialProxyError;
 use crate::storage::manager::SqliteStorageManager;
@@ -26,6 +25,10 @@ use uuid::Uuid;
 
 mod manager;
 pub mod models;
+pub(crate) mod pruner;
+
+// TODO: proper import
+type NodeId = u64;
 
 #[derive(Clone)]
 pub struct CredentialProxyStorage {
@@ -74,7 +77,7 @@ impl CredentialProxyStorage {
     }
 
     #[allow(dead_code)]
-    pub(crate) async fn load_blinded_shares_status_by_shares_id(
+    pub async fn load_blinded_shares_status_by_shares_id(
         &self,
         id: i64,
     ) -> Result<Option<BlindedShares>, CredentialProxyError> {
@@ -84,7 +87,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn load_wallet_shares_by_shares_id(
+    pub async fn load_wallet_shares_by_shares_id(
         &self,
         id: i64,
     ) -> Result<Vec<MinimalWalletShare>, CredentialProxyError> {
@@ -94,7 +97,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn load_shares_error_by_shares_id(
+    pub async fn load_shares_error_by_shares_id(
         &self,
         id: i64,
     ) -> Result<Option<String>, CredentialProxyError> {
@@ -105,7 +108,7 @@ impl CredentialProxyStorage {
     }
 
     #[allow(dead_code)]
-    pub(crate) async fn load_blinded_shares_status_by_device_and_credential_id(
+    pub async fn load_blinded_shares_status_by_device_and_credential_id(
         &self,
         device_id: &str,
         credential_id: &str,
@@ -116,7 +119,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn load_wallet_shares_by_device_and_credential_id(
+    pub async fn load_wallet_shares_by_device_and_credential_id(
         &self,
         device_id: &str,
         credential_id: &str,
@@ -127,7 +130,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn load_shares_error_by_device_and_credential_id(
+    pub async fn load_shares_error_by_device_and_credential_id(
         &self,
         device_id: &str,
         credential_id: &str,
@@ -138,7 +141,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn insert_new_pending_async_shares_request(
+    pub async fn insert_new_pending_async_shares_request(
         &self,
         request: Uuid,
         device_id: &str,
@@ -150,7 +153,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn update_pending_async_blinded_shares_issued(
+    pub async fn update_pending_async_blinded_shares_issued(
         &self,
         available_shares: usize,
         device_id: &str,
@@ -166,7 +169,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn update_pending_async_blinded_shares_error(
+    pub async fn update_pending_async_blinded_shares_error(
         &self,
         available_shares: usize,
         device_id: &str,
@@ -184,7 +187,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn prune_old_blinded_shares(&self) -> Result<(), CredentialProxyError> {
+    pub async fn prune_old_blinded_shares(&self) -> Result<(), CredentialProxyError> {
         let max_age = OffsetDateTime::now_utc() - time::Duration::days(31);
 
         self.storage_manager
@@ -199,7 +202,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn insert_new_deposits(
+    pub async fn insert_new_deposits(
         &self,
         deposits: &PerformedDeposits,
     ) -> Result<(), CredentialProxyError> {
@@ -211,9 +214,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn load_unused_deposits(
-        &self,
-    ) -> Result<Vec<BufferedDeposit>, CredentialProxyError> {
+    pub async fn load_unused_deposits(&self) -> Result<Vec<BufferedDeposit>, CredentialProxyError> {
         self.storage_manager
             .load_unused_deposits()
             .await?
@@ -222,7 +223,7 @@ impl CredentialProxyStorage {
             .collect()
     }
 
-    pub(crate) async fn insert_deposit_usage(
+    pub async fn insert_deposit_usage(
         &self,
         deposit_id: DepositId,
         requested_on: OffsetDateTime,
@@ -240,7 +241,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn insert_deposit_usage_error(
+    pub async fn insert_deposit_usage_error(
         &self,
         deposit_id: DepositId,
         error: String,
@@ -251,7 +252,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn insert_partial_wallet_share(
+    pub async fn insert_partial_wallet_share(
         &self,
         deposit_id: DepositId,
         epoch_id: EpochId,
@@ -291,7 +292,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn get_master_verification_key(
+    pub async fn get_master_verification_key(
         &self,
         epoch_id: EpochId,
     ) -> Result<Option<EpochVerificationKey>, CredentialProxyError> {
@@ -309,7 +310,7 @@ impl CredentialProxyStorage {
         Ok(Some(deserialised))
     }
 
-    pub(crate) async fn insert_master_verification_key(
+    pub async fn insert_master_verification_key(
         &self,
         key: &EpochVerificationKey,
     ) -> Result<(), CredentialProxyError> {
@@ -320,7 +321,7 @@ impl CredentialProxyStorage {
             .await?)
     }
 
-    pub(crate) async fn get_master_coin_index_signatures(
+    pub async fn get_master_coin_index_signatures(
         &self,
         epoch_id: EpochId,
     ) -> Result<Option<AggregatedCoinIndicesSignatures>, CredentialProxyError> {
@@ -340,7 +341,7 @@ impl CredentialProxyStorage {
         Ok(Some(deserialised))
     }
 
-    pub(crate) async fn insert_master_coin_index_signatures(
+    pub async fn insert_master_coin_index_signatures(
         &self,
         signatures: &AggregatedCoinIndicesSignatures,
     ) -> Result<(), CredentialProxyError> {
@@ -355,7 +356,7 @@ impl CredentialProxyStorage {
         Ok(())
     }
 
-    pub(crate) async fn get_master_expiration_date_signatures(
+    pub async fn get_master_expiration_date_signatures(
         &self,
         expiration_date: Date,
         epoch_id: EpochId,
@@ -376,7 +377,7 @@ impl CredentialProxyStorage {
         Ok(Some(deserialised))
     }
 
-    pub(crate) async fn insert_master_expiration_date_signatures(
+    pub async fn insert_master_expiration_date_signatures(
         &self,
         signatures: &AggregatedExpirationDateSignatures,
     ) -> Result<(), CredentialProxyError> {
@@ -398,7 +399,7 @@ impl CredentialProxyStorage {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::http::helpers;
+    use crate::helpers::random_uuid;
     use crate::storage::models::BlindedSharesStatus;
     use nym_compact_ecash::scheme::keygen::KeyPairUser;
     use nym_crypto::asymmetric::ed25519;
@@ -480,7 +481,7 @@ mod tests {
     async fn test_add() -> anyhow::Result<()> {
         let storage = get_storage().await?;
 
-        let dummy_uuid = helpers::random_uuid();
+        let dummy_uuid = random_uuid();
         println!("🚀 insert_pending_blinded_share...");
 
         storage.insert_dummy_used_deposit(dummy_uuid).await?;

common/credential-proxy/src/storage/pruner.rs

@@ -1,5 +1,5 @@
-// Copyright 2024 Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
 
 use crate::storage::CredentialProxyStorage;
 use tokio_util::sync::CancellationToken;

common/credential-proxy/src/ticketbook_manager/mod.rs

@@ -0,0 +1,163 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use crate::deposits_buffer::DepositsBuffer;
+use crate::error::CredentialProxyError;
+use crate::quorum_checker::QuorumStateChecker;
+use crate::shared_state::ecash_state::EcashState;
+use crate::shared_state::nyxd_client::ChainClient;
+use crate::shared_state::required_deposit_cache::RequiredDepositCache;
+use crate::shared_state::CredentialProxyState;
+use crate::storage::pruner::StoragePruner;
+use crate::storage::CredentialProxyStorage;
+use crate::webhook::ZkNymWebhook;
+use nym_credentials::ecash::utils::ecash_today;
+use nym_validator_client::nym_api::EpochId;
+use std::future::Future;
+use std::time::Duration;
+use time::Date;
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+use tokio_util::task::TaskTracker;
+
+mod shares_handlers;
+pub mod ticketbook_handlers;
+pub mod wallet_shares;
+
+#[derive(Clone, Default)]
+pub struct ShutdownTracker {
+    pub shutdown_token: CancellationToken,
+    pub tracker: TaskTracker,
+}
+
+#[derive(Clone)]
+pub struct TicketbookManager {
+    pub(crate) state: CredentialProxyState,
+    pub(crate) webhook: ZkNymWebhook,
+    pub(crate) shutdown_tracker: ShutdownTracker,
+}
+
+impl TicketbookManager {
+    pub async fn new(
+        build_sha: &'static str,
+        quorum_check_interval: Duration,
+        deposits_buffer_size: usize,
+        max_concurrent_deposits: usize,
+        storage: CredentialProxyStorage,
+        mnemonic: bip39::Mnemonic,
+        webhook: ZkNymWebhook,
+    ) -> Result<Self, CredentialProxyError> {
+        let chain_client = ChainClient::new(mnemonic)?;
+        let shutdown_tracker = ShutdownTracker::default();
+
+        let quorum_state_checker = QuorumStateChecker::new(
+            chain_client.clone(),
+            quorum_check_interval,
+            shutdown_tracker.shutdown_token.clone(),
+        )
+        .await?;
+
+        let required_deposit_cache = RequiredDepositCache::default();
+
+        let deposits_buffer = DepositsBuffer::new(
+            storage.clone(),
+            chain_client.clone(),
+            required_deposit_cache.clone(),
+            build_sha,
+            deposits_buffer_size,
+            max_concurrent_deposits,
+            shutdown_tracker.shutdown_token.clone(),
+        )
+        .await?;
+
+        let storage_pruner =
+            StoragePruner::new(shutdown_tracker.shutdown_token.clone(), storage.clone());
+
+        let this = TicketbookManager {
+            state: CredentialProxyState::new(
+                storage.clone(),
+                chain_client,
+                deposits_buffer,
+                EcashState::new(
+                    required_deposit_cache,
+                    quorum_state_checker.quorum_state_ref(),
+                ),
+            ),
+            webhook,
+            shutdown_tracker,
+        };
+
+        // since this is startup,
+        // might as well do all the needed network queries to establish needed global signatures
+        // if we don't already have them
+        this.build_initial_cache().await?;
+
+        // spawn the background tasks
+        this.try_spawn_in_background(quorum_state_checker.run_forever());
+        this.try_spawn_in_background(storage_pruner.run_forever());
+
+        Ok(this)
+    }
+
+    async fn build_initial_cache(&self) -> Result<(), CredentialProxyError> {
+        let today = ecash_today().date();
+
+        let epoch_id = self.state.current_epoch_id().await?;
+        let _ = self.state.deposit_amount().await?;
+        let _ = self.state.master_verification_key(Some(epoch_id)).await?;
+        let _ = self.state.ecash_threshold(epoch_id).await?;
+        let _ = self.state.ecash_clients(epoch_id).await?;
+        let _ = self
+            .state
+            .master_coin_index_signatures(Some(epoch_id))
+            .await?;
+        let _ = self
+            .state
+            .master_expiration_date_signatures(epoch_id, today)
+            .await?;
+
+        Ok(())
+    }
+
+    pub async fn cancel_and_wait(&self) {
+        self.shutdown_tracker.shutdown_token.cancel();
+        self.state.deposits_buffer().wait_for_shutdown().await;
+        self.shutdown_tracker.tracker.wait().await
+    }
+
+    pub fn shutdown_token(&self) -> CancellationToken {
+        self.shutdown_tracker.shutdown_token.clone()
+    }
+
+    /// Ensure the required global data for the specified epoch and expiration date exists in our cache (and storage)
+    async fn ensure_global_data_cached(
+        &self,
+        epoch: EpochId,
+        expiration_date: Date,
+    ) -> Result<(), CredentialProxyError> {
+        let _ = self.state.master_verification_key(Some(epoch)).await?;
+        let _ = self.state.master_coin_index_signatures(Some(epoch)).await?;
+        let _ = self
+            .state
+            .master_expiration_date_signatures(epoch, expiration_date)
+            .await?;
+        Ok(())
+    }
+
+    pub fn try_spawn_in_background<F>(&self, task: F) -> Option<JoinHandle<F::Output>>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        // don't spawn new task if we've received cancellation token
+        if self.shutdown_tracker.shutdown_token.is_cancelled() {
+            None
+        } else {
+            self.shutdown_tracker.tracker.reopen();
+            // TODO: later use a task queue since most requests will be blocked waiting on chain permit anyway
+            let join_handle = self.shutdown_tracker.tracker.spawn(task);
+            self.shutdown_tracker.tracker.close();
+            Some(join_handle)
+        }
+    }
+}

common/credential-proxy/src/ticketbook_manager/shares_handlers.rs

@@ -0,0 +1,145 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::CredentialProxyError;
+use crate::storage::models::MinimalWalletShare;
+use crate::ticketbook_manager::TicketbookManager;
+use nym_credential_proxy_requests::api::v1::ticketbook::models::{
+    GlobalDataParams, TicketbookWalletSharesResponse,
+};
+use nym_validator_client::nym_api::EpochId;
+use tracing::{debug, span, Instrument, Level};
+use uuid::Uuid;
+
+impl TicketbookManager {
+    async fn shares_to_response(
+        &self,
+        shares: Vec<MinimalWalletShare>,
+        params: GlobalDataParams,
+    ) -> Result<TicketbookWalletSharesResponse, CredentialProxyError> {
+        // in all calls we ensured the shares are non-empty
+        #[allow(clippy::unwrap_used)]
+        let first = shares.first().unwrap();
+        let expiration_date = first.expiration_date;
+        let epoch_id = first.epoch_id as EpochId;
+
+        let threshold = self.state.ecash_threshold(epoch_id).await?;
+        if shares.len() < threshold as usize {
+            return Err(CredentialProxyError::InsufficientNumberOfCredentials {
+                available: shares.len(),
+                threshold,
+            });
+        }
+
+        // grab any requested additional data
+        let (
+            master_verification_key,
+            aggregated_expiration_date_signatures,
+            aggregated_coin_index_signatures,
+        ) = self
+            .state
+            .global_data(params, epoch_id, expiration_date)
+            .await?;
+
+        // finally produce a response
+        Ok(TicketbookWalletSharesResponse {
+            epoch_id,
+            shares: shares.into_iter().map(Into::into).collect(),
+            master_verification_key,
+            aggregated_coin_index_signatures,
+            aggregated_expiration_date_signatures,
+        })
+    }
+
+    /// Query by id for blinded shares of a bandwidth voucher
+    pub async fn query_for_shares_by_id(
+        &self,
+        uuid: Uuid,
+        params: GlobalDataParams,
+        share_id: i64,
+    ) -> Result<TicketbookWalletSharesResponse, CredentialProxyError> {
+        let span = span!(Level::INFO, "query shares by id", uuid = %uuid, share_id = %share_id);
+        async move {
+            debug!("");
+
+            // TODO: edge case: this will **NOT** work if shares got created in epoch X,
+            // but this query happened in epoch X+1
+            let shares = self
+                .state
+                .storage()
+                .load_wallet_shares_by_shares_id(share_id)
+                .await?;
+            if shares.is_empty() {
+                debug!("shares not found");
+
+                // check for explicit error
+                if let Some(error_message) = self
+                    .state
+                    .storage()
+                    .load_shares_error_by_shares_id(share_id)
+                    .await?
+                {
+                    return Err(CredentialProxyError::ShareByIdLoadError {
+                        message: error_message,
+                        id: share_id,
+                    });
+                }
+
+                return Err(CredentialProxyError::SharesByIdNotFound { id: share_id });
+            }
+
+            self.shares_to_response(shares, params).await
+        }
+        .instrument(span)
+        .await
+    }
+
+    /// Query by id for blinded  wallet shares of a ticketbook
+    pub async fn query_for_shares_by_device_id_and_credential_id(
+        &self,
+        uuid: Uuid,
+        params: GlobalDataParams,
+        device_id: String,
+        credential_id: String,
+    ) -> Result<TicketbookWalletSharesResponse, CredentialProxyError> {
+        let span = span!(Level::INFO, "query shares by device and credential ids", uuid = %uuid, device_id = %device_id, credential_id = %credential_id);
+        async move {
+            debug!("");
+
+            // TODO: edge case: this will **NOT** work if shares got created in epoch X,
+            // but this query happened in epoch X+1
+            let shares = self
+                .state
+                .storage()
+                .load_wallet_shares_by_device_and_credential_id(&device_id, &credential_id)
+                .await?;
+
+            if shares.is_empty() {
+                debug!("shares not found");
+
+                // check for explicit error
+                if let Some(error_message) = self
+                    .state
+                    .storage()
+                    .load_shares_error_by_device_and_credential_id(&device_id, &credential_id)
+                    .await?
+                {
+                    return Err(CredentialProxyError::ShareByDeviceLoadError {
+                        message: error_message,
+                        device_id,
+                        credential_id,
+                    });
+                }
+
+                return Err(CredentialProxyError::SharesByDeviceNotFound {
+                    device_id,
+                    credential_id,
+                });
+            }
+
+            self.shares_to_response(shares, params).await
+        }
+        .instrument(span)
+        .await
+    }
+}

common/credential-proxy/src/ticketbook_manager/ticketbook_handlers.rs

@@ -0,0 +1,164 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::CredentialProxyError;
+use crate::nym_api_helpers::ensure_sane_expiration_date;
+use crate::ticketbook_manager::TicketbookManager;
+use nym_compact_ecash::Base58;
+use nym_credential_proxy_requests::api::v1::ticketbook::models::{
+    CurrentEpochResponse, DepositResponse, GlobalDataParams, MasterVerificationKeyResponse,
+    PartialVerificationKey, PartialVerificationKeysResponse, TicketbookAsyncRequest,
+    TicketbookObtainParams, TicketbookRequest, TicketbookWalletSharesAsyncResponse,
+    TicketbookWalletSharesResponse,
+};
+use time::OffsetDateTime;
+use tracing::{error, info, span, warn, Instrument, Level};
+use uuid::Uuid;
+
+impl TicketbookManager {
+    pub async fn obtain_ticketbook_shares(
+        &self,
+        uuid: Uuid,
+        request: TicketbookRequest,
+        params: GlobalDataParams,
+    ) -> Result<TicketbookWalletSharesResponse, CredentialProxyError> {
+        let requested_on = OffsetDateTime::now_utc();
+        let span = span!(Level::INFO, "obtain ticketboook", uuid = %uuid);
+
+        async move {
+            info!("");
+
+            self.state.ensure_credentials_issuable().await?;
+            let epoch_id = self.state.current_epoch_id().await?;
+            ensure_sane_expiration_date(request.expiration_date)?;
+
+            // if additional data was requested, grab them first in case there are any cache/network issues
+            let (
+                master_verification_key,
+                aggregated_expiration_date_signatures,
+                aggregated_coin_index_signatures,
+            ) = self
+                .state
+                .global_data(params, epoch_id, request.expiration_date)
+                .await?;
+
+            let shares = self
+                .try_obtain_wallet_shares(uuid, requested_on, request)
+                .await
+                .inspect_err(|err| warn!("shares request failure: {err}"))?;
+
+            info!("request was successful!");
+            Ok(TicketbookWalletSharesResponse {
+                epoch_id,
+                shares,
+                master_verification_key,
+                aggregated_coin_index_signatures,
+                aggregated_expiration_date_signatures,
+            })
+        }
+        .instrument(span)
+        .await
+    }
+
+    pub async fn obtain_ticketbook_shares_async(
+        &self,
+        uuid: Uuid,
+        request: TicketbookAsyncRequest,
+        params: TicketbookObtainParams,
+    ) -> Result<TicketbookWalletSharesAsyncResponse, CredentialProxyError> {
+        let requested_on = OffsetDateTime::now_utc();
+        let span = span!(Level::INFO, "[async] obtain ticketboook", uuid = %uuid);
+        async move {
+            info!("");
+
+            // 1. perform basic validation
+            self.state.ensure_credentials_issuable().await?;
+
+            ensure_sane_expiration_date(request.inner.expiration_date)?;
+
+            // 2. store the request to retrieve the id
+            let pending = self
+                .state
+                .storage()
+                .insert_new_pending_async_shares_request(
+                    uuid,
+                    &request.device_id,
+                    &request.credential_id,
+                )
+                .await
+                .inspect_err(|err| error!("failed to insert new pending async shares: {err}"))?;
+
+            let id = pending.id;
+
+            // 3. try to spawn a new task attempting to resolve the request
+            let this = self.clone();
+            if self
+                .try_spawn_in_background(async move {
+                    this.try_obtain_blinded_ticketbook_async(
+                        uuid,
+                        requested_on,
+                        request,
+                        params,
+                        pending,
+                    )
+                    .await
+                })
+                .is_none()
+            {
+                warn!("could not start async ticketbook issuance due to shutdown in progress");
+                return Err(CredentialProxyError::ShutdownInProgress);
+            }
+
+            // 4. in the meantime, return the id to the user
+            Ok(TicketbookWalletSharesAsyncResponse { id, uuid })
+        }
+        .instrument(span)
+        .await
+    }
+
+    pub async fn current_deposit(&self) -> Result<DepositResponse, CredentialProxyError> {
+        let current_deposit = self.state.deposit_amount().await?;
+        Ok(DepositResponse {
+            current_deposit_amount: current_deposit.amount,
+            current_deposit_denom: current_deposit.denom,
+        })
+    }
+
+    pub async fn partial_verification_keys(
+        &self,
+    ) -> Result<PartialVerificationKeysResponse, CredentialProxyError> {
+        self.state.ensure_credentials_issuable().await?;
+
+        let epoch_id = self.state.current_epoch_id().await?;
+        let signers = self.state.ecash_clients(epoch_id).await?;
+        Ok(PartialVerificationKeysResponse {
+            epoch_id,
+            keys: signers
+                .iter()
+                .map(|signer| PartialVerificationKey {
+                    node_index: signer.node_id,
+                    bs58_encoded_key: signer.verification_key.to_bs58(),
+                })
+                .collect(),
+        })
+    }
+
+    pub async fn master_verification_key(
+        &self,
+    ) -> Result<MasterVerificationKeyResponse, CredentialProxyError> {
+        self.state.ensure_credentials_issuable().await?;
+
+        let epoch_id = self.state.current_epoch_id().await?;
+        let key = self.state.master_verification_key(Some(epoch_id)).await?;
+        Ok(MasterVerificationKeyResponse {
+            epoch_id,
+            bs58_encoded_key: key.to_bs58(),
+        })
+    }
+
+    pub async fn current_epoch(&self) -> Result<CurrentEpochResponse, CredentialProxyError> {
+        self.state.ensure_credentials_issuable().await?;
+        let epoch_id = self.state.current_epoch_id().await?;
+        Ok(CurrentEpochResponse { epoch_id })
+    }
+}

common/credential-proxy/src/ticketbook_manager/wallet_shares.rs

@@ -0,0 +1,344 @@
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
+
+use crate::error::CredentialProxyError;
+use crate::storage::models::BlindedShares;
+use crate::ticketbook_manager::TicketbookManager;
+use futures::{stream, StreamExt};
+use nym_compact_ecash::Base58;
+use nym_credential_proxy_requests::api::v1::ticketbook::models::{
+    TicketbookAsyncRequest, TicketbookObtainParams, TicketbookRequest,
+    TicketbookWalletSharesResponse, WalletShare, WebhookTicketbookWalletShares,
+    WebhookTicketbookWalletSharesRequest,
+};
+use nym_validator_client::client::NymApiClientExt;
+use nym_validator_client::ecash::BlindSignRequestBody;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+use time::OffsetDateTime;
+use tokio::sync::Mutex;
+use tokio::time::timeout;
+use tracing::{debug, error, info, instrument};
+use uuid::Uuid;
+
+impl TicketbookManager {
+    #[instrument(
+        skip(self, request_data, request, requested_on),
+            fields(
+                expiration_date = %request_data.expiration_date,
+                ticketbook_type = %request_data.ticketbook_type
+        )
+    )]
+    pub async fn try_obtain_wallet_shares(
+        &self,
+        request: Uuid,
+        requested_on: OffsetDateTime,
+        request_data: TicketbookRequest,
+    ) -> Result<Vec<WalletShare>, CredentialProxyError> {
+        // don't proceed if we don't have quorum available as the request will definitely fail
+        if !self.state.ecash_state().quorum_state.available() {
+            return Err(CredentialProxyError::UnavailableSigningQuorum);
+        }
+
+        let epoch = self.state.current_epoch_id().await?;
+        let threshold = self.state.ecash_threshold(epoch).await?;
+        let expiration_date = request_data.expiration_date;
+
+        // before we commit to making the deposit, ensure we have required signatures cached and stored
+        self.ensure_global_data_cached(epoch, expiration_date)
+            .await?;
+        let ecash_api_clients = self.state.ecash_clients(epoch).await?.clone();
+
+        let deposit_data = self
+            .state
+            .get_deposit(request, requested_on, request_data.ecash_pubkey)
+            .await?;
+        let deposit_id = deposit_data.deposit_id;
+        let signature = deposit_data.sign_ticketbook_plaintext(&request_data.withdrawal_request);
+
+        let credential_request = BlindSignRequestBody::new(
+            request_data.withdrawal_request.into(),
+            deposit_id,
+            signature,
+            request_data.ecash_pubkey,
+            request_data.expiration_date,
+            request_data.ticketbook_type,
+        );
+
+        let wallet_shares = Arc::new(Mutex::new(HashMap::new()));
+
+        info!("attempting to contract all nym-apis for the partial wallets...");
+        stream::iter(ecash_api_clients)
+            .for_each_concurrent(None, |client| async {
+                // move the client into the block
+                let client = client;
+
+                debug!("contacting {client} for blinded partial wallet");
+                let res = timeout(
+                    Duration::from_secs(5),
+                    client.api_client.blind_sign(&credential_request),
+                )
+                .await
+                .map_err(|_| CredentialProxyError::EcashApiRequestTimeout {
+                    client_repr: client.to_string(),
+                })
+                .and_then(|res| res.map_err(Into::into));
+
+                // 1. try to store it
+                if let Err(err) = self
+                    .state
+                    .storage()
+                    .insert_partial_wallet_share(
+                        deposit_id,
+                        epoch,
+                        expiration_date,
+                        client.node_id,
+                        &res,
+                    )
+                    .await
+                {
+                    error!("failed to persist issued partial share: {err}")
+                }
+
+                // 2. add it to the map
+                match res {
+                    Ok(share) => {
+                        wallet_shares
+                            .lock()
+                            .await
+                            .insert(client.node_id, share.blinded_signature);
+                    }
+                    Err(err) => {
+                        error!("failed to obtain partial blinded wallet share from {client}: {err}")
+                    }
+                }
+            })
+            .await;
+
+        // SAFETY: the futures have completed, so we MUST have the only arc reference
+        #[allow(clippy::unwrap_used)]
+        let wallet_shares = Arc::into_inner(wallet_shares).unwrap().into_inner();
+        let shares = wallet_shares.len();
+
+        if shares < threshold as usize {
+            let err = CredentialProxyError::InsufficientNumberOfCredentials {
+                available: shares,
+                threshold,
+            };
+            self.state
+                .insert_deposit_usage_error(deposit_id, err.to_string())
+                .await;
+            return Err(err);
+        }
+
+        Ok(wallet_shares
+            .into_iter()
+            .map(|(node_index, share)| WalletShare {
+                node_index,
+                bs58_encoded_share: share.to_bs58(),
+            })
+            .collect())
+    }
+
+    pub async fn try_obtain_wallet_shares_async(
+        &self,
+        request: Uuid,
+        requested_on: OffsetDateTime,
+        request_data: TicketbookRequest,
+        device_id: &str,
+        credential_id: &str,
+    ) -> Result<Vec<WalletShare>, CredentialProxyError> {
+        let shares = match self
+            .try_obtain_wallet_shares(request, requested_on, request_data)
+            .await
+        {
+            Ok(shares) => shares,
+            Err(err) => {
+                let obtained = match err {
+                    CredentialProxyError::InsufficientNumberOfCredentials { available, .. } => {
+                        available
+                    }
+                    _ => 0,
+                };
+
+                // currently there's no retry mechanisms, but, who knows, that might change
+                if let Err(err) = self
+                    .state
+                    .storage()
+                    .update_pending_async_blinded_shares_error(
+                        obtained,
+                        device_id,
+                        credential_id,
+                        &err.to_string(),
+                    )
+                    .await
+                {
+                    error!("failed to update database with the error information: {err}")
+                }
+                return Err(err);
+            }
+        };
+
+        Ok(shares)
+    }
+
+    async fn try_obtain_blinded_ticketbook_async_inner(
+        &self,
+        request: Uuid,
+        requested_on: OffsetDateTime,
+        request_data: TicketbookAsyncRequest,
+        params: TicketbookObtainParams,
+        pending: &BlindedShares,
+    ) -> Result<(), CredentialProxyError> {
+        let epoch_id = self.state.current_epoch_id().await?;
+
+        let device_id = &request_data.device_id;
+        let credential_id = &request_data.credential_id;
+        let secret = request_data.secret.clone();
+
+        // 1. try to obtain global data
+        let (
+            master_verification_key,
+            aggregated_expiration_date_signatures,
+            aggregated_coin_index_signatures,
+        ) = self
+            .state
+            .global_data(params.global, epoch_id, request_data.inner.expiration_date)
+            .await?;
+
+        // 2. try to obtain shares (failures are written to the DB)
+        let shares = self
+            .try_obtain_wallet_shares_async(
+                request,
+                requested_on,
+                request_data.inner,
+                device_id,
+                credential_id,
+            )
+            .await?;
+
+        // 3. update the storage, if possible
+        // (as long as we can trigger webhook, we should still be good)
+        if let Err(err) = self
+            .state
+            .storage()
+            .update_pending_async_blinded_shares_issued(shares.len(), device_id, credential_id)
+            .await
+        {
+            error!(uuid = %request, "failed to update db with issued information: {err}")
+        }
+
+        // 4. build the webhook request body
+        let data = Some(TicketbookWalletSharesResponse {
+            epoch_id,
+            shares,
+            master_verification_key,
+            aggregated_coin_index_signatures,
+            aggregated_expiration_date_signatures,
+        });
+
+        let ticketbook_wallet_shares = WebhookTicketbookWalletShares {
+            id: pending.id,
+            status: pending.status.to_string(),
+            device_id: device_id.clone(),
+            credential_id: credential_id.clone(),
+            data,
+            error_message: None,
+            created: pending.created,
+            updated: pending.updated,
+        };
+
+        let webhook_request = WebhookTicketbookWalletSharesRequest {
+            ticketbook_wallet_shares,
+            secret,
+        };
+
+        // 5. call the webhook
+        self.webhook.try_trigger(request, &webhook_request).await;
+
+        Ok(())
+    }
+
+    async fn try_trigger_webhook_request_for_error(
+        &self,
+        request: Uuid,
+        request_data: TicketbookAsyncRequest,
+        pending: &BlindedShares,
+        error_message: String,
+    ) -> Result<(), CredentialProxyError> {
+        let device_id = &request_data.device_id;
+        let credential_id = &request_data.credential_id;
+        let secret = request_data.secret.clone();
+
+        let ticketbook_wallet_shares = WebhookTicketbookWalletShares {
+            id: pending.id,
+            status: "error".to_string(),
+            device_id: device_id.clone(),
+            credential_id: credential_id.clone(),
+            data: None,
+            error_message: Some(error_message),
+            created: pending.created,
+            updated: pending.updated,
+        };
+
+        let webhook_request = WebhookTicketbookWalletSharesRequest {
+            ticketbook_wallet_shares,
+            secret,
+        };
+
+        self.webhook.try_trigger(request, &webhook_request).await;
+
+        Ok(())
+    }
+
+    #[instrument(
+        skip_all,
+        fields(
+            credential_id = %request_data.credential_id,
+            device_id = %request_data.device_id)
+        )
+    ]
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) async fn try_obtain_blinded_ticketbook_async(
+        &self,
+        request: Uuid,
+        requested_on: OffsetDateTime,
+        request_data: TicketbookAsyncRequest,
+        params: TicketbookObtainParams,
+        pending: BlindedShares,
+    ) {
+        let skip_webhook = params.skip_webhook;
+        if let Err(err) = self
+            .try_obtain_blinded_ticketbook_async_inner(
+                request,
+                requested_on,
+                request_data.clone(),
+                params,
+                &pending,
+            )
+            .await
+        {
+            if skip_webhook {
+                info!(uuid = %request,"the webhook is not going to be called for this request");
+                return;
+            }
+
+            // post to the webhook to notify of errors on this side
+            if let Err(webhook_err) = self
+                .try_trigger_webhook_request_for_error(
+                    request,
+                    request_data,
+                    &pending,
+                    format!("Failed to get ticketbook: {err}"),
+                )
+                .await
+            {
+                error!(uuid = %request, "failed to make webhook request to report error: {webhook_err}")
+            }
+            error!(uuid = %request, "failed to resolve the blinded ticketbook issuance: {err}")
+        } else {
+            info!(uuid = %request, "managed to resolve the blinded ticketbook issuance")
+        }
+    }
+}

common/credential-proxy/src/webhook.rs

@@ -1,57 +1,34 @@
-// Copyright 2024 Nym Technologies SA <contact@nymtech.net>
-// SPDX-License-Identifier: GPL-3.0-only
+// Copyright 2025 - Nym Technologies SA <contact@nymtech.net>
+// SPDX-License-Identifier: Apache-2.0
 
-use crate::error::CredentialProxyError;
-use clap::Args;
 use reqwest::header::AUTHORIZATION;
 use serde::Serialize;
 use tracing::{debug, error, instrument, span, Instrument, Level};
 use url::Url;
 use uuid::Uuid;
 
-#[derive(Args, Debug, Clone)]
-pub struct ZkNymWebHookConfig {
-    #[clap(long, env = "WEBHOOK_ZK_NYMS_URL")]
-    pub webhook_url: Url,
+#[derive(Debug, Clone)]
+pub struct ZkNymWebhook {
+    pub webhook_client_url: Url,
 
-    #[clap(long, env = "WEBHOOK_ZK_NYMS_CLIENT_ID")]
-    pub webhook_client_id: String,
-
-    #[clap(long, env = "WEBHOOK_ZK_NYMS_CLIENT_SECRET")]
     pub webhook_client_secret: String,
 }
 
-impl ZkNymWebHookConfig {
-    pub fn ensure_valid_client_url(&self) -> Result<(), CredentialProxyError> {
-        self.client_url()
-            .map_err(|_| CredentialProxyError::InvalidWebhookUrl)
-            .map(|_| ())
-    }
-
-    fn client_url(&self) -> Result<Url, url::ParseError> {
-        self.webhook_url.join(&self.webhook_client_id)
-    }
-
-    fn unchecked_client_url(&self) -> Url {
-        // we ensured we have valid url on startup
-        #[allow(clippy::unwrap_used)]
-        self.client_url().unwrap()
-    }
-
+impl ZkNymWebhook {
     fn bearer_token(&self) -> String {
         format!("Bearer {}", self.webhook_client_secret)
     }
 
     #[instrument(skip_all)]
     pub async fn try_trigger<T: Serialize + ?Sized>(&self, original_uuid: Uuid, payload: &T) {
-        let url = self.unchecked_client_url();
+        let url = self.webhook_client_url.clone();
         let span = span!(Level::DEBUG, "webhook", uuid = %original_uuid, url = %url);
 
         async move {
             debug!("🕸️ about to trigger the webhook");
 
             match reqwest::Client::new()
-                .post(url.clone())
+                .post(url)
                 .header(AUTHORIZATION, self.bearer_token())
                 .json(payload)
                 .send()

common/credential-verification/src/ecash/credential_sender.rs

@@ -14,7 +14,7 @@ use nym_api_requests::ecash::models::{BatchRedeemTicketsBody, VerifyEcashTicketB
 use nym_credentials_interface::Bandwidth;
 use nym_credentials_interface::{ClientTicket, TicketType};
 use nym_validator_client::coconut::EcashApiError;
-use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
 use nym_validator_client::nyxd::contract_traits::{
     EcashSigningClient, MultisigQueryClient, MultisigSigningClient, PagedMultisigQueryClient,
 };
@@ -126,7 +126,7 @@ pub struct CredentialHandlerConfig {
     pub maximum_time_between_redemption: Duration,
 }
 
-pub(crate) struct CredentialHandler {
+pub struct CredentialHandler {
     config: CredentialHandlerConfig,
     multisig_threshold: f32,
     ticket_receiver: UnboundedReceiver<ClientTicket>,
@@ -354,7 +354,7 @@ impl CredentialHandler {
             Err(err) => {
                 error!("failed to send ticket {ticket_id} for verification to ecash signer '{client}': {err}. if we don't reach quorum, we'll retry later");
                 Err(EcashTicketError::ApiFailure(EcashApiError::NymApi {
-                    source: err,
+                    source: nym_validator_client::ValidatorClientError::from(err),
                 }))
             }
         }
@@ -907,7 +907,7 @@ impl CredentialHandler {
         Ok(())
     }
 
-    async fn run(mut self, mut shutdown: nym_task::TaskClient) {
+    pub async fn run(mut self, shutdown: nym_task::ShutdownToken) {
         info!("Starting Ecash CredentialSender");
 
         // attempt to clear any pending operations
@@ -919,11 +919,12 @@ impl CredentialHandler {
         let start = Instant::now() + self.config.pending_poller;
         let mut resolver_interval = interval_at(start, self.config.pending_poller);
 
-        while !shutdown.is_shutdown() {
+        loop {
             tokio::select! {
                 biased;
-                _ = shutdown.recv() => {
+                _ = shutdown.cancelled() => {
                     trace!("client_handling::credentialSender : received shutdown");
+                    break
                 },
                 Some(ticket) = self.ticket_receiver.next() => {
                     let (queued_up, _) = self.ticket_receiver.size_hint();
@@ -946,8 +947,4 @@ impl CredentialHandler {
             }
         }
     }
-
-    pub(crate) fn start(self, shutdown: nym_task::TaskClient) {
-        tokio::spawn(async move { self.run(shutdown).await });
-    }
 }

common/credential-verification/src/ecash/mod.rs

@@ -82,9 +82,8 @@ impl EcashManager {
         credential_handler_cfg: CredentialHandlerConfig,
         nyxd_client: DirectSigningHttpRpcNyxdClient,
         pk_bytes: [u8; 32],
-        shutdown: nym_task::TaskClient,
         storage: GatewayStorage,
-    ) -> Result<Self, Error> {
+    ) -> Result<(Self, CredentialHandler), Error> {
         let shared_state = SharedState::new(nyxd_client, Box::new(storage)).await?;
 
         let (cred_sender, cred_receiver) = mpsc::unbounded();
@@ -92,14 +91,16 @@ impl EcashManager {
         let cs =
             CredentialHandler::new(credential_handler_cfg, cred_receiver, shared_state.clone())
                 .await?;
-        cs.start(shutdown);
 
-        Ok(EcashManager {
-            shared_state,
-            pk_bytes,
-            pay_infos: Default::default(),
-            cred_sender,
-        })
+        Ok((
+            EcashManager {
+                shared_state,
+                pk_bytes,
+                pay_infos: Default::default(),
+                cred_sender,
+            },
+            cs,
+        ))
     }
 
     pub async fn verify_pay_info(&self, pay_info: NymPayInfo) -> Result<usize, EcashTicketError> {

common/credentials/Cargo.toml

@@ -22,6 +22,7 @@ nym-ecash-time = { path = "../ecash-time", features = ["expiration"] }
 nym-credentials-interface = { path = "../credentials-interface" }
 nym-crypto = { path = "../crypto" }
 nym-api-requests = { path = "../../nym-api/nym-api-requests" }
+nym-http-api-client = { path = "../http-api-client" }
 nym-validator-client = { path = "../client-libs/validator-client", default-features = false }
 nym-ecash-contract-common = { path = "../cosmwasm-smart-contracts/ecash-contract" }
 nym-network-defaults = { path = "../network-defaults" }

common/credentials/src/ecash/bandwidth/issuance.rs

@@ -15,7 +15,7 @@ use nym_credentials_interface::{
 use nym_crypto::asymmetric::ed25519;
 use nym_ecash_contract_common::deposit::DepositId;
 use nym_ecash_time::{ecash_default_expiration_date, ecash_today, EcashTime};
-use nym_validator_client::nym_api::EpochId;
+use nym_validator_client::nym_api::{EpochId, NymApiClientExt};
 use serde::{Deserialize, Serialize};
 use time::Date;
 
@@ -116,7 +116,7 @@ impl IssuanceTicketBook {
 
     pub async fn obtain_blinded_credential(
         &self,
-        client: &nym_validator_client::client::NymApiClient,
+        client: &nym_http_api_client::Client,
         request_body: &BlindSignRequestBody,
     ) -> Result<BlindedSignature, Error> {
         let server_response = client.blind_sign(request_body).await?;
@@ -179,7 +179,7 @@ impl IssuanceTicketBook {
     // ideally this would have been generic over credential type, but we really don't need secp256k1 keys for bandwidth vouchers
     pub async fn obtain_partial_ticketbook_credential(
         &self,
-        client: &nym_validator_client::client::NymApiClient,
+        client: &nym_http_api_client::Client,
         signer_index: u64,
         validator_vk: &VerificationKeyAuth,
         signing_data: CredentialSigningData,

common/credentials/src/ecash/utils.rs

@@ -10,6 +10,7 @@ use nym_credentials_interface::{
     VerificationKeyAuth, WalletSignatures,
 };
 use nym_validator_client::client::EcashApiClient;
+use nym_validator_client::nym_api::NymApiClientExt;
 
 // so we wouldn't break all the existing imports
 pub use nym_ecash_time::{cred_exp_date, ecash_date_offset, ecash_today, EcashTime};

common/credentials/src/error.rs

@@ -4,7 +4,7 @@
 use crate::ecash::bandwidth::issued::CURRENT_SERIALIZATION_REVISION;
 use nym_credentials_interface::CompactEcashError;
 use nym_crypto::asymmetric::x25519::KeyRecoveryError;
-use nym_validator_client::ValidatorClientError;
+use nym_validator_client::{nym_api::error::NymAPIError, ValidatorClientError};
 use thiserror::Error;
 
 #[derive(Debug, Error)]
@@ -37,6 +37,9 @@ pub enum Error {
     #[error("Ran into a validator client error - {0}")]
     ValidatorClientError(#[from] ValidatorClientError),
 
+    #[error("Nym API request failed - {0}")]
+    NymAPIError(Box<NymAPIError>),
+
     #[error("Bandwidth operation overflowed. {0}")]
     BandwidthOverflow(String),
 
@@ -61,3 +64,9 @@ pub enum Error {
     #[error("failed to create a secp256k1 signature")]
     Secp256k1SignFailure,
 }
+
+impl From<NymAPIError> for Error {
+    fn from(e: NymAPIError) -> Self {
+        Error::NymAPIError(Box::new(e))
+    }
+}

common/crypto/Cargo.toml

@@ -11,6 +11,7 @@ repository = { workspace = true }
 aes-gcm-siv = { workspace = true, optional = true }
 aes = { workspace = true, optional = true }
 aead = { workspace = true, optional = true }
+base64.workspace = true
 bs58 = { workspace = true }
 blake3 = { workspace = true, features = ["traits-preview"], optional = true }
 ctr = { workspace = true, optional = true }
@@ -18,6 +19,7 @@ digest = { workspace = true, optional = true }
 generic-array = { workspace = true, optional = true }
 hkdf = { workspace = true, optional = true }
 hmac = { workspace = true, optional = true }
+jwt-simple = { workspace = true, optional = true }
 cipher = { workspace = true, optional = true }
 x25519-dalek = { workspace = true, features = ["static_secrets"], optional = true }
 ed25519-dalek = { workspace = true, features = ["rand_core"], optional = true }
@@ -39,6 +41,7 @@ rand_chacha = { workspace = true }
 [features]
 default = []
 aead = ["dep:aead", "aead/std", "aes-gcm-siv", "generic-array"]
+naive_jwt = ["asymmetric", "jwt-simple"]
 serde = ["dep:serde", "serde_bytes", "ed25519-dalek/serde", "x25519-dalek/serde"]
 asymmetric = ["x25519-dalek", "ed25519-dalek", "zeroize"]
 hashing = ["blake3", "digest", "hkdf", "hmac", "generic-array", "sha2"]

common/crypto/src/asymmetric/ed25519/mod.rs

@@ -2,8 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 pub use ed25519_dalek::SignatureError;
-use ed25519_dalek::{SecretKey, Signer, SigningKey};
 pub use ed25519_dalek::{Verifier, PUBLIC_KEY_LENGTH, SECRET_KEY_LENGTH, SIGNATURE_LENGTH};
+
+use ed25519_dalek::Signer;
 use nym_pemstore::traits::{PemStorableKey, PemStorableKeyPair};
 use std::fmt::{self, Debug, Display, Formatter};
 use std::str::FromStr;
@@ -13,6 +14,9 @@ use zeroize::{Zeroize, ZeroizeOnDrop};
 #[cfg(feature = "serde")]
 pub mod serde_helpers;
 
+#[cfg(feature = "serde")]
+pub use serde_helpers::*;
+
 #[cfg(feature = "sphinx")]
 use nym_sphinx_types::{DestinationAddressBytes, DESTINATION_ADDRESS_LENGTH};
 
@@ -81,8 +85,8 @@ impl KeyPair {
         }
     }
 
-    pub fn from_secret(secret: SecretKey, index: u32) -> Self {
-        let ed25519_signing_key = SigningKey::from(secret);
+    pub fn from_secret(secret: ed25519_dalek::SecretKey, index: u32) -> Self {
+        let ed25519_signing_key = ed25519_dalek::SigningKey::from(secret);
 
         KeyPair {
             private_key: PrivateKey(ed25519_signing_key.to_bytes()),
@@ -276,7 +280,7 @@ impl Display for PrivateKey {
 
 impl<'a> From<&'a PrivateKey> for PublicKey {
     fn from(pk: &'a PrivateKey) -> Self {
-        PublicKey(SigningKey::from_bytes(&pk.0).verifying_key())
+        PublicKey(ed25519_dalek::SigningKey::from_bytes(&pk.0).verifying_key())
     }
 }
 
@@ -320,7 +324,7 @@ impl PrivateKey {
     }
 
     pub fn sign<M: AsRef<[u8]>>(&self, message: M) -> Signature {
-        let signing_key: SigningKey = self.0.into();
+        let signing_key: ed25519_dalek::SigningKey = self.0.into();
         let sig = signing_key.sign(message.as_ref());
         Signature(sig)
     }
@@ -425,9 +429,57 @@ impl<'d> Deserialize<'d> for Signature {
     }
 }
 
+#[cfg(feature = "naive_jwt")]
+impl PublicKey {
+    pub fn to_jwt_compatible_key(&self) -> jwt_simple::algorithms::Ed25519PublicKey {
+        (*self).into()
+    }
+}
+
+#[cfg(feature = "naive_jwt")]
+impl From<PublicKey> for jwt_simple::algorithms::Ed25519PublicKey {
+    fn from(value: PublicKey) -> Self {
+        // SAFETY: we have a valid ed25519 pubkey, we're just changing to a different library wrapper
+        #[allow(clippy::unwrap_used)]
+        jwt_simple::algorithms::Ed25519PublicKey::from_bytes(&value.to_bytes()).unwrap()
+    }
+}
+
+#[cfg(feature = "naive_jwt")]
+impl PrivateKey {
+    pub fn to_jwt_compatible_keys(&self) -> jwt_simple::algorithms::Ed25519KeyPair {
+        let pub_key = self.public_key();
+        let mut bytes = zeroize::Zeroizing::new([0u8; 64]);
+
+        bytes[..SECRET_KEY_LENGTH]
+            .copy_from_slice(zeroize::Zeroizing::new(self.to_bytes()).as_ref());
+        bytes[SECRET_KEY_LENGTH..].copy_from_slice(&pub_key.to_bytes());
+
+        // SAFETY: we have a valid ed25519 keys, we're just changing to a different library wrapper
+        #[allow(clippy::unwrap_used)]
+        jwt_simple::algorithms::Ed25519KeyPair::from_bytes(bytes.as_ref()).unwrap()
+    }
+}
+
+#[cfg(feature = "naive_jwt")]
+impl KeyPair {
+    pub fn to_jwt_compatible_keys(&self) -> jwt_simple::algorithms::Ed25519KeyPair {
+        let mut bytes = zeroize::Zeroizing::new([0u8; 64]);
+
+        bytes[..SECRET_KEY_LENGTH]
+            .copy_from_slice(zeroize::Zeroizing::new(self.private_key.to_bytes()).as_ref());
+        bytes[SECRET_KEY_LENGTH..].copy_from_slice(&self.public_key.to_bytes());
+
+        // SAFETY: we have a valid ed25519 keys, we're just changing to a different library wrapper
+        #[allow(clippy::unwrap_used)]
+        jwt_simple::algorithms::Ed25519KeyPair::from_bytes(bytes.as_ref()).unwrap()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rand::thread_rng;
 
     fn assert_zeroize_on_drop<T: ZeroizeOnDrop>() {}
 
@@ -438,4 +490,29 @@ mod tests {
         assert_zeroize::<PrivateKey>();
         assert_zeroize_on_drop::<PrivateKey>();
     }
+
+    #[test]
+    #[cfg(all(feature = "naive_jwt", feature = "rand"))]
+    fn check_jwt_key_compat_conversion() {
+        let mut rng = thread_rng();
+        let keys = KeyPair::new(&mut rng);
+        let jwt_keys = keys.to_jwt_compatible_keys();
+
+        // internally they're represented by hidden `Edwards25519KeyPair` (plus key_id)
+        // which has way nicer API for assertions
+        let jwt_keys_inner =
+            jwt_simple::algorithms::Edwards25519KeyPair::from_bytes(&jwt_keys.to_bytes()).unwrap();
+
+        let compact_ed25519 = jwt_keys_inner.as_ref();
+        assert!(compact_ed25519
+            .sk
+            .validate_public_key(&compact_ed25519.pk)
+            .is_ok());
+
+        let dummy_message = "hello world";
+        let sig1 = keys.private_key.sign(dummy_message).to_bytes();
+        let sig2 = compact_ed25519.sk.sign(dummy_message, None).to_vec();
+
+        assert_eq!(sig1.to_vec(), sig2);
+    }
 }